Addressing review comments

Signed-off-by: ngsupta1 <guptaneha.e@gmail.com>

Author: nsgupta1

data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/CrowdStrikeService.java

@@ -13,6 +13,9 @@
 import java.net.URI;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Optional;
+
 import static org.opensearch.dataprepper.plugins.source.crowdstrike.utils.Constants.BATCH_SIZE;
 import static org.opensearch.dataprepper.plugins.source.crowdstrike.utils.Constants.FILTER_KEY;
 import static org.opensearch.dataprepper.plugins.source.crowdstrike.utils.Constants.LAST_UPDATED;
@@ -44,35 +47,30 @@ public CrowdStrikeService(CrowdStrikeRestClient crowdStrikeRestClient, PluginMet
      *   @param paginationLink  An optional pagination URL suffix (used when fetching next pages).
      *   @return                A {@link CrowdStrikeApiResponse} containing response body and headers.
      */
-    public CrowdStrikeApiResponse getAllContent(Long startTime, Long endTime, String paginationLink) {
+    public CrowdStrikeApiResponse getThreatIndicators(Instant startTime, Instant endTime, Optional<String> paginationLink) {
+        if (startTime == null || endTime == null) {
+            throw new IllegalArgumentException("startTime and endTime must not be null");
+        }
         URI uri = buildCrowdStrikeUri(startTime, endTime, paginationLink);
-
         return searchCallLatencyTimer.record(() -> {
-            try {
-                log.debug("Calling CrowdStrike API with URI: {}", uri);
-                ResponseEntity<CrowdStrikeIndicatorResult> responseEntity = crowdStrikeRestClient.invokeGetApi(uri, CrowdStrikeIndicatorResult.class);
 
-                CrowdStrikeApiResponse response = new CrowdStrikeApiResponse();
-                response.setBody(responseEntity.getBody());
-                response.setHeaders(responseEntity.getHeaders());
-                return response;
-            } catch (Exception e) {
-                log.error("Error fetching CrowdStrike content from URI: {}", uri, e);
-                throw new RuntimeException("CrowdStrike API call failed", e);
-            }
+            log.debug("Calling CrowdStrike API with URI: {}", uri);
+            ResponseEntity<CrowdStrikeIndicatorResult> responseEntity = crowdStrikeRestClient.invokeGetApi(uri, CrowdStrikeIndicatorResult.class);
+
+            return new CrowdStrikeApiResponse(responseEntity.getBody(), responseEntity.getHeaders());
         });
     }
 
-    protected URI buildCrowdStrikeUri(Long startTime, Long endTime, String paginationLink) {
+    protected URI buildCrowdStrikeUri(Instant startTime, Instant endTime,  Optional<String> paginationLink) {
         try {
-            if (paginationLink != null) {
-                String urlString = BASE_URL + paginationLink;
+            if (paginationLink.isPresent()) {
+                String urlString = BASE_URL + paginationLink.get();
                 urlString = CrowdStrikeNextLinkValidator.validateAndSanitizeURL(urlString);
                 return new URI(urlString);
             } else {
                 // Manually construct and encode the query string
-                String filter1 = URLEncoder.encode(LAST_UPDATED + ":>=" + startTime, StandardCharsets.UTF_8);
-                String filter2 = URLEncoder.encode(LAST_UPDATED + ":<" + endTime, StandardCharsets.UTF_8);
+                String filter1 = URLEncoder.encode(LAST_UPDATED + ":>=" + startTime.getEpochSecond(), StandardCharsets.UTF_8);
+                String filter2 = URLEncoder.encode(LAST_UPDATED + ":<" + endTime.getEpochSecond(), StandardCharsets.UTF_8);
                 String encodedFilter = filter1 + "%2B" + filter2;  // Use literal '+' // ensure literal '+'
 
                 UriComponentsBuilder builder = UriComponentsBuilder
@@ -86,4 +84,4 @@ protected URI buildCrowdStrikeUri(Long startTime, Long endTime, String paginatio
             throw new RuntimeException("Failed to construct CrowdStrike request URI", e);
         }
     }
-}
+}

data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/models/CrowdStrikeApiResponse.java

@@ -16,6 +16,11 @@ public class CrowdStrikeApiResponse {
     private CrowdStrikeIndicatorResult body;
     private Map<String, List<String>> headers;
 
+    public CrowdStrikeApiResponse(CrowdStrikeIndicatorResult body, Map<String, List<String>> headers) {
+        this.body = body;
+        this.headers = headers;
+    }
+
     // Convenience method to get a specific header
     public List<String> getHeader(String headerName) {
         return headers.getOrDefault(headerName, Collections.emptyList());

data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/models/CrowdStrikeIndicatorResult.java

@@ -3,16 +3,20 @@
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import lombok.Getter;
+import lombok.Setter;
+
 import java.util.List;
 
 /**
  * The result of Falcon query search.
  */
 @Getter
+@Setter
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class CrowdStrikeIndicatorResult {
 
     @JsonProperty("resources")
     private List<ThreatIndicator> results = null;
 
+
 }

data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/models/ThreatIndicator.java

@@ -9,15 +9,6 @@
  * Represents a threat intelligence indicator from CrowdStrike's API.
  * This class encapsulates information about potential security threats,
  * including indicators of compromise (IoCs) and associated metadata.
- *
- * <p>A threat indicator may include various attributes such as:
- * <ul>
- *     <li>Type of the indicator (IP, Domain, Hash, etc.)</li>
- *     <li>Value of the indicator</li>
- *     <li>Confidence score</li>
- *     <li>Associated timestamps</li>
- * </ul>
- * </p>
  */
 @Setter
 @Getter

data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/rest/CrowdStrikeAuthClient.java

@@ -67,37 +67,43 @@ public void initCredentials() {
      * @throws UnauthorizedException Runtime exception if the token cannot be retrieved.
      */
     protected void getAuthToken() {
-        log.info(NOISY, "You are trying to access token");
-        HttpHeaders headers = new HttpHeaders();
-        headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
-        headers.setBasicAuth(this.clientId, this.clientSecret);
-        HttpEntity<String> request = new HttpEntity<>(headers);
-        int retryCount = 0;
-        while(retryCount < MAX_RETRIES) {
-            try {
-                ResponseEntity<Map> response = restTemplate.postForEntity(OAUTH_TOKEN_URL, request, Map.class);
-                Map tokenData = response.getBody();
-                this.bearerToken = (String) tokenData.get(ACCESS_TOKEN);
-                this.expireTime = Instant.now().plusSeconds((Integer) tokenData.get(EXPIRE_IN));
-                log.info("Access token acquired successfully");
+        synchronized (tokenRenewLock) {
+            if (isTokenValid()) {
+                //Someone else must have already renewed it
                 return;
-            } catch (HttpClientErrorException ex) {
-                this.expireTime = Instant.ofEpochMilli(0);
-                HttpStatus statusCode = ex.getStatusCode();
-                String statusMessage = ex.getMessage();
-                log.error("Failed to acquire access token. Status code: {}, Error Message: {}",
-                        statusCode, statusMessage);
+            }
+            log.info(NOISY, "You are trying to access token");
+            HttpHeaders headers = new HttpHeaders();
+            headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
+            headers.setBasicAuth(this.clientId, this.clientSecret);
+            HttpEntity<String> request = new HttpEntity<>(headers);
+            int retryCount = 0;
+            while (retryCount < MAX_RETRIES) {
                 try {
-                    Thread.sleep((long) RETRY_ATTEMPT_SLEEP_TIME.get(retryCount) * SLEEP_TIME_MULTIPLIER);
-                } catch (InterruptedException e) {
-                    throw new RuntimeException("Sleep in the retry attempt got interrupted", e);
+                    ResponseEntity<Map> response = restTemplate.postForEntity(OAUTH_TOKEN_URL, request, Map.class);
+                    Map tokenData = response.getBody();
+                    this.bearerToken = (String) tokenData.get(ACCESS_TOKEN);
+                    this.expireTime = Instant.now().plusSeconds((Integer) tokenData.get(EXPIRE_IN));
+                    log.info("Access token acquired successfully");
+                    return;
+                } catch (HttpClientErrorException ex) {
+                    this.expireTime = Instant.ofEpochMilli(0);
+                    HttpStatus statusCode = ex.getStatusCode();
+                    String statusMessage = ex.getMessage();
+                    log.error("Failed to acquire access token. Status code: {}, Error Message: {}",
+                            statusCode, statusMessage);
+                    try {
+                        Thread.sleep((long) RETRY_ATTEMPT_SLEEP_TIME.get(retryCount) * SLEEP_TIME_MULTIPLIER);
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException("Sleep in the retry attempt got interrupted", e);
+                    }
                 }
+                retryCount++;
             }
-            retryCount++;
+            String errorMessage = String.format("Failed to acquire access token even after %s retry attempts", MAX_RETRIES);
+            log.error(errorMessage);
+            throw new UnauthorizedException(errorMessage);
         }
-        String errorMessage = String.format("Failed to acquire access token even after %s retry attempts", MAX_RETRIES);
-        log.error(errorMessage);
-        throw new UnauthorizedException(errorMessage);
     }
 
     protected boolean isTokenValid() {
@@ -113,13 +119,7 @@ public void refreshToken() {
             //There is still time to renew, or someone else must have already renewed it
             return;
         }
-        synchronized (tokenRenewLock) {
-            if (isTokenValid()) {
-                //Someone else must have already renewed it
-                return;
-            }
-            log.info("Renewing authentication token for CrowdStrike Connector.");
-            getAuthToken();
-        }
+        log.info("Renewing authentication token for CrowdStrike Connector.");
+        getAuthToken();
     }
 }

data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/test/java/org/opensearch/dataprepper/plugins/source/crowdstrike/CrowdStrikeNextLinkValidatorTest.java

@@ -11,7 +11,7 @@ public class CrowdStrikeNextLinkValidatorTest {
 
     @Test
     void testValidEncodedCrowdStrikeUrlPreserved() throws MalformedURLException {
-        String url = "https://api.crowdstrike.com//intel/combined/indicators/v1" +
+        String url = "https://api.crowdstrike.com/intel/combined/indicators/v1" +
                 "?filter=last_updated%3A%3E%3D1745519529%2Blast_updated%3A%3C1745523129%2B_marker%3A%3C%2717455225567d09efadf14547a1aee2bc25cabc525e%27" +
                 "&limit=10000";
 

data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/test/java/org/opensearch/dataprepper/plugins/source/crowdstrike/CrowdStrikeServiceTest.java

@@ -11,14 +11,16 @@
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.ResponseEntity;
 import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
 import java.util.function.Supplier;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -32,24 +34,29 @@ class CrowdStrikeServiceTest {
     private CrowdStrikeRestClient restClient;
     private PluginMetrics pluginMetrics;
     private CrowdStrikeService service;
+    Instant startTime = Instant.now().minus(Duration.ofHours(1));
+    Instant endTime = Instant.now();
+    Timer mockTimer;
 
     @BeforeEach
     void setup() {
         restClient = mock(CrowdStrikeRestClient.class);
         pluginMetrics = mock(PluginMetrics.class);
-        Timer mockTimer = mock(Timer.class);
+        mockTimer = mock(Timer.class);
+
         when(pluginMetrics.timer(any())).thenReturn(mockTimer);
-        // Timer records calls synchronously for test
-        lenient().when(mockTimer.record(Mockito.<Supplier<Object>>any()))
+
+        service = new CrowdStrikeService(restClient, pluginMetrics);
+
+        when(mockTimer.record(Mockito.<Supplier<Object>>any()))
                 .thenAnswer(invocation -> {
                     Supplier<Object> supplier = invocation.getArgument(0);
                     return supplier.get();
                 });
-        service = new CrowdStrikeService(restClient, pluginMetrics);
     }
 
     @Test
-    void testGetAllContentWithValidTimeRange() {
+    void testGetThreatIndicatorsWithValidTimeRange() {
         CrowdStrikeIndicatorResult result = new CrowdStrikeIndicatorResult();
         ResponseEntity<CrowdStrikeIndicatorResult> responseEntity = ResponseEntity.ok()
                 .headers(new HttpHeaders())
@@ -58,15 +65,15 @@ void testGetAllContentWithValidTimeRange() {
         when(restClient.invokeGetApi(any(), eq(CrowdStrikeIndicatorResult.class)))
                 .thenReturn(responseEntity);
 
-        CrowdStrikeApiResponse response = service.getAllContent(1745000000L, 1745003600L, null);
+        CrowdStrikeApiResponse response = service.getThreatIndicators(startTime, endTime, Optional.empty());
 
         assertNotNull(response.getBody());
         assertNotNull(response.getHeaders());
         verify(restClient, times(1)).invokeGetApi(any(), eq(CrowdStrikeIndicatorResult.class));
     }
 
     @Test
-    void testGetAllContentWithPaginationLink() throws Exception {
+    void testGetThreatIndicatorsWithPaginationLink() throws Exception {
         String paginationLink = "intel/combined/indicators/v1?filter=last_updated%3A%3E%3D1745%2B_marker%3A%3C%27123%27";
         URI sanitizedUri = new URI("https://api.crowdstrike.com/" + paginationLink);
         CrowdStrikeIndicatorResult result = new CrowdStrikeIndicatorResult();
@@ -78,7 +85,7 @@ void testGetAllContentWithPaginationLink() throws Exception {
         when(restClient.invokeGetApi(eq(sanitizedUri), eq(CrowdStrikeIndicatorResult.class)))
                 .thenReturn(responseEntity);
 
-        CrowdStrikeApiResponse response = service.getAllContent(null, null, paginationLink);
+        CrowdStrikeApiResponse response = service.getThreatIndicators(startTime, endTime, Optional.of(paginationLink));
         assertNotNull(response.getBody());
         verify(restClient).invokeGetApi(eq(sanitizedUri), eq(CrowdStrikeIndicatorResult.class));
     }
@@ -89,24 +96,39 @@ void testRestClientThrowsException() {
                 .thenThrow(new RuntimeException("API failure"));
 
         RuntimeException ex = assertThrows(RuntimeException.class, () ->
-                service.getAllContent(1745000000L, 1745003600L, null));
+                service.getThreatIndicators(startTime, endTime, Optional.empty()));
 
-        assertTrue(ex.getMessage().contains("CrowdStrike API call failed"));
+        assertTrue(ex.getMessage().contains("API failure"));
     }
 
     @Test
     void testBuildUriFailsGracefully() {
         CrowdStrikeService faultyService = new CrowdStrikeService(restClient, pluginMetrics) {
             @Override
-            protected URI buildCrowdStrikeUri(Long startTime, Long endTime, String paginationLink) {
+            protected URI buildCrowdStrikeUri(Instant startTime, Instant endTime, Optional<String> paginationLink) {
                 throw new RuntimeException("URI construction failed");
             }
         };
 
         RuntimeException ex = assertThrows(RuntimeException.class, () ->
-                faultyService.getAllContent(1745000000L, 1745003600L, null));
+                faultyService.getThreatIndicators(startTime, endTime, Optional.empty()));
 
         assertEquals("URI construction failed", ex.getMessage());
     }
 
-}
+    @Test
+    void testSearchCallLatencyTimerIsRecorded() {
+        CrowdStrikeIndicatorResult result = new CrowdStrikeIndicatorResult();
+        ResponseEntity<CrowdStrikeIndicatorResult> responseEntity = ResponseEntity.ok()
+                .headers(new HttpHeaders())
+                .body(result);
+
+        when(restClient.invokeGetApi(any(), eq(CrowdStrikeIndicatorResult.class)))
+                .thenReturn(responseEntity);
+
+        service.getThreatIndicators(startTime, endTime, Optional.empty());
+
+        verify(mockTimer, times(1)).record(any(Supplier.class));
+    }
+
+}