Merge pull request #2 from Galactus22625/saas-connector-sdk

log edits and sign off test
san81 · Oct 21, 2024 · e932289 · e932289
2 parents 07be050 + 82b5605
commit e932289
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 17 deletions.
diff --git a/...source/src/main/java/org/opensearch/dataprepper/plugins/source/saas/jira/JiraService.java b/...source/src/main/java/org/opensearch/dataprepper/plugins/source/saas/jira/JiraService.java
@@ -19,7 +19,9 @@
 import org.slf4j.Logger;
 import org.springframework.security.oauth2.client.ClientAuthorizationRequiredException;
 import org.springframework.util.CollectionUtils;
+import org.springframework.web.client.HttpClientErrorException;
 import org.springframework.web.client.RestTemplate;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 
 import javax.inject.Named;
 import java.io.ByteArrayInputStream;
@@ -208,7 +210,6 @@ private void addItemsToQueue(List<IssueBean> issueList, Queue<ItemInfo> itemInfo
       issueMetadata.put(CONTENT_TYPE, JiraContentType.ISSUE.getType());
       String id = _ISSUE + issueMetadata.get(PROJECT_KEY) + "-" + issue.getKey();
 
-      log.info("Creating issue information {}", id);
       itemInfoQueue.add(createItemInfo(id, issueMetadata));
 
       if (Objects.nonNull(issueMetadata.get(PROJECT_KEY)) && !jiraProjectCache
@@ -229,7 +230,6 @@ private void addItemsToQueue(List<IssueBean> issueList, Queue<ItemInfo> itemInfo
    */
   public SearchResults getAllIssues(StringBuilder jql, int startAt,
                                     JiraSourceConfig configuration) {
-    log.info("Started to fetch all issues information");
     SearchResults results = null;
     HttpResponse<JsonNode> response;
     com.mashape.unirest.request.HttpRequest request;
@@ -245,8 +245,6 @@ public SearchResults getAllIssues(StringBuilder jql, int startAt,
                 .queryString(START_AT, startAt)
                 .queryString(JQL_FIELD, jql)
                 .queryString(EXPAND_FIELD, EXPAND_VALUE);
-        log.info("Search result api call request is : {}",
-                new Gson().toJson(request, com.mashape.unirest.request.HttpRequest.class));
 
         response = request.asJson();
         /*appLog.info("Search result api call response is: {}",
@@ -274,11 +272,7 @@ public SearchResults getAllIssues(StringBuilder jql, int startAt,
                   .queryString(START_AT, startAt)
                   .queryString(JQL_FIELD, jql)
                   .queryString(EXPAND_FIELD, EXPAND_VALUE);
-          log.info("Search result api call request is :",
-                  new Gson().toJson(request, com.mashape.unirest.request.HttpRequest.class));
           response = request.asJson();
-          log.info("Search result api call response is:",
-                  new Gson().toJson(response, com.mashape.unirest.http.HttpResponse.class));
           if (response.getStatus() == TOKEN_EXPIRED) {
             JiraOauthConfig.changeAccessAndRefreshToken(configuration);
             retryCount++;
@@ -361,17 +355,17 @@ public final InputStream buildTextInputStream(String text) {
    * @return the issue
    */
   public String getIssue(String issueKey, JiraSourceConfig configuration) {
-    log.info("Started to fetch issue information");
+    log.debug("Started to fetch issue information");
     Queue<Integer> waitTimeQueue = new ConcurrentLinkedQueue<>(waitTimeList);
 
     while(true) {
       String url = configuration.getAccountUrl() + REST_API_FETCH_ISSUE + "/" + issueKey;
-      log.info("Issue Fetching api call request is : {}", url);
       try {
         return restTemplate.getForEntity(url, String.class).getBody();
       } catch (ClientAuthorizationRequiredException ex) {
 
-        log.error("Failed to execute the rest call ",ex);
+        log.error(NOISY, "Failed to execute the rest call ",ex);
+
 
       }
     }

diff --git a/...n/java/org/opensearch/dataprepper/plugins/source/saas/jira/rest/BasicAuthInterceptor.java b/...n/java/org/opensearch/dataprepper/plugins/source/saas/jira/rest/BasicAuthInterceptor.java
@@ -17,9 +17,8 @@ public class BasicAuthInterceptor implements ClientHttpRequestInterceptor {
     private final String password;
 
     public BasicAuthInterceptor(JiraSourceConfig config) {
-        //TODO: fix it once the ingestion of config works
-        this.username = ""; //config.getJiraId();
-        this.password = ""; //config.getJiraCredential();
+        this.username = config.getJiraId();
+        this.password = config.getJiraCredential();
     }
 
     @Override

diff --git a/...ce/src/main/java/org/opensearch/dataprepper/plugins/source/saas/jira/utils/Constants.java b/...ce/src/main/java/org/opensearch/dataprepper/plugins/source/saas/jira/utils/Constants.java
@@ -77,7 +77,7 @@ public class Constants {
   public static final String UPDATE_AUTHOR = "updateAuthor";
   public static final String TIME_SPENT = "timeSpent";
   public static final String LOGGED = " logged ";
-  public static final String BASIC = "Basic";
+  public static final String BASIC = "Basic ";
   public static final String CID = "c_id";
   public static final String WID = "w_id";
   public static final String BROWSE = "browse/";

diff --git a/...er/src/main/java/org/opensearch/dataprepper/plugins/source/saas/crawler/base/Crawler.java b/...er/src/main/java/org/opensearch/dataprepper/plugins/source/saas/crawler/base/Crawler.java
@@ -38,6 +38,7 @@ public long crawl(SaasSourceConfig sourceConfig,
         Iterator<ItemInfo> itemInfoIterator = client.listItems();
         log.info("Starting to crawl the source");
         long updatedPollTime = 0;
+        log.info("Creating Partitions");
         do {
             final List<ItemInfo> itemInfoList = new ArrayList<>();
             for (int i = 0; i < maxItemsPerPage && itemInfoIterator.hasNext(); i++) {
@@ -74,7 +75,6 @@ private void createPartition(List<ItemInfo> itemInfoList, EnhancedSourceCoordina
         state.setItemIds(itemIds);
         state.setExportStartTime(System.currentTimeMillis());
         state.setLoadedItems(itemInfoList.size());
-        log.info("Creating a new partition");
         SaasSourcePartition sourcePartition = new SaasSourcePartition(state, partitionKey);
         coordinator.createPartition(sourcePartition);
     }

diff --git a/...earch/dataprepper/plugins/source/saas/crawler/coordination/scheduler/WorkerScheduler.java b/...earch/dataprepper/plugins/source/saas/crawler/coordination/scheduler/WorkerScheduler.java
@@ -42,6 +42,7 @@ public WorkerScheduler(Buffer<Record<Event>> buffer,
     @Override
     public void run() {
         log.info("Worker thread started");
+        log.info("Processing Partitions");
         while (!Thread.currentThread().isInterrupted()) {
             try {
                 // Get the next available partition from the coordinator
@@ -72,7 +73,6 @@ public void run() {
     }
 
     private void processPartition(EnhancedSourcePartition partition, Buffer<Record<Event>> buffer, SaasSourceConfig sourceConfig) {
-        log.info("Processing partition: {}", partition.getPartitionKey());
         // Implement your source extraction logic here
         // Update the partition state or commit the partition as needed
         // Commit the partition to mark it as processed