CrowdStrike client and coordinator implementation by nsgupta1 · Pull Request #5678 · opensearch-project/data-prepper

nsgupta1 · 2025-05-05T06:03:26Z

Description

CrowdStrike Client implementation
TimeSlice Crawler implementation
CrowdStrike specific leader and worker implementation

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

[Y] New functionality includes testing.
[N] New functionality has a documentation issue. Please link to it in this PR.
[ Y] New functionality has javadoc added
[Y] Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: nsgupta1 <nsgupta1@users.noreply.github.com>

san81 · 2025-05-05T18:30:54Z

+        createPartitionForCrawling(leaderPartition, coordinator, latestModifiedTime);
+        long crawlTimeMillis = System.currentTimeMillis() - startTime;
+        log.debug("Crawling completed in {} ms", crawlTimeMillis);
+        crawlingTimer.record(crawlTimeMillis, TimeUnit.MILLISECONDS);


Leader Crawling in this case doesn't involve any network calls, tracking this time may not be much helpful. May be tracking the number of partitions helpful here? Something to think about.

Yes, we are tracking the partitionCounter, I will remove Crawling time.

san81 · 2025-05-05T18:32:55Z

+        for(int i = remainingDays; i > 0; i--) {
+            Instant startDate = todayUtc.minus(Duration.ofDays(i));
+            createWorkerPartition(startDate, startDate.plus(Duration.ofDays(1)), coordinator);
+            updateLeaderProgressState(leaderPartition, i-1, leaderProgressState.getLastPollTime(), coordinator);


May be updatingLeaderProgressState just once at the end of the loop is also fine I guess as there are not network calls and possibility of failures inside the loop are super minimal.

I will change this.

san81

Just added 2 suggestions. Nice work 👍

nsgupta1 · 2025-05-05T18:42:17Z

Just added 2 suggestions. Nice work 👍

Thank you for quick review

Signed-off-by: nsgupta1 <nsgupta1@users.noreply.github.com>

nishtham-amazon · 2025-05-06T14:59:11Z

+     */
    @Override
-    public void executePartition(SaasWorkerProgressState state, Buffer<Record<Event>> buffer, AcknowledgementSet acknowledgementSet) {
+    public void executePartition(CrowdStrikeWorkerProgressState state, Buffer<Record<Event>> buffer, AcknowledgementSet acknowledgementSet) {


Make the parameters final

nishtham-amazon · 2025-05-06T15:03:21Z


+        try {
+            buffer.writeAll(records, (int) Duration.ofSeconds(bufferWriteTimeoutInSeconds).toMillis());
+        } catch (Exception e) {


Its better to catch the expected exception rather than generalising it. This might mask the actual actual code error as service failure.

WriteAll method doesn't throw a specific exception, that's why I had to catch Exception. Here is the method signature. void writeAll(Collection<T> records, int timeoutInMillis) throws Exception;

nishtham-amazon · 2025-05-06T15:04:11Z

+            buffer.writeAll(records, (int) Duration.ofSeconds(bufferWriteTimeoutInSeconds).toMillis());
+        } catch (Exception e) {
+            log.error("Failed to write {} indicators to buffer", records.size(), e);
+            throw new RuntimeException("Buffer write failed for CrowdStrike indicators", e);


Convert this to a custom exception rather than throwing runtime exception

I can address it in follow-up PR, I see all the other clients also have same implementation I can check with @san81 and create a specific exception for all Saas crawlers.

nishtham-amazon · 2025-05-06T15:15:59Z

            .registerModule(new JavaTimeModule());
    private final LeaderProgressState state;

-    public LeaderPartition() {


Is this not used anywhere?

No it is not used anymore

nishtham-amazon · 2025-05-06T15:28:21Z


 @Named
-public class PaginationCrawler implements Crawler {
+public class PaginationCrawler implements Crawler<AtlassianWorkerProgressState> {


Is this crawler specific to Atlassian/Jira? Why use a Atlassian specific state for base crawler/s?

Yes PaginationCrawler is specific to Atlassian and TimeSlice is specific to CrowdStrikeWorker.

can we rename PaginationCrawler to AtlassianPaginationCrawler?

Signed-off-by: nsgupta1 <nsgupta1@users.noreply.github.com>

eirsep · 2025-05-06T16:52:14Z

+            CrowdStrikeIndicatorResult result = response.getBody();
+            List<ThreatIndicator> indicators = result.getResults();
+            if (indicators == null || indicators.isEmpty()) {
+                log.info("No threat indicators found for the time window {} to {}", startTime, endTime);


if it's empty should we continue to next page at all?

eirsep · 2025-05-06T16:57:00Z

+    @Min(0)
+    @Max(90)
+    @Valid
+    private int lookBackDays = DEFAULT_NUMBER_OF_LOOK_BACK_DAYS;


nit: can this be final?

eirsep · 2025-05-06T19:11:42Z

     */
    @JsonProperty("published_date")
-    private long publishedDate = 0L;
+    private Instant publishedDate;


can all these variables be final?

eirsep · 2025-05-06T19:12:55Z

+        @JsonSubTypes.Type(value = AtlassianLeaderProgressState.class),
+        @JsonSubTypes.Type(value = CrowdStrikeLeaderProgressState.class)
+})
+public interface LeaderProgressState {


does it not store information about failures too?

plz add java docs on what is "pollTime"

eirsep · 2025-05-06T19:23:39Z

+    }
+
+    /**
+     * Main crawling logic for CrowdStrike using time-based partitioning.


plz add code comments/java docs for each method

eirsep · 2025-05-06T19:24:24Z

+        double startCount = partitionsCreatedCounter.count();
+        createPartitionForCrawling(leaderPartition, coordinator, latestModifiedTime);
+        double partitionsInThisCrawl = partitionsCreatedCounter.count() - startCount;
+        log.info("Total partitions created in this crawl: {}", partitionsInThisCrawl);


plz have log indicate. it is crowdstrikeClient's partition count

eirsep · 2025-05-06T19:25:37Z

+    private Instant endTime;
+
+    @JsonProperty("marker")
+    private String marker;


can we add java doc on what is a marker

engechas · 2025-05-06T19:53:49Z

+    CrowdStrikeService crowdStrikeService;
+    private static final Logger log = LoggerFactory.getLogger(CrowdStrikeClient.class);
+    private final CrowdStrikeSourceConfig configuration;
+    private final int bufferWriteTimeoutInSeconds = 10;


Should this be configurable?

engechas · 2025-05-06T19:58:44Z

+    @JsonProperty("remaining_days")
+    private int remainingDays;
+
+    public CrowdStrikeLeaderProgressState(@JsonProperty("last_poll_time") final Instant lastPollTime, @JsonProperty("remaining_days") int remainingDays) {


Nit: do we need the @JsonProperty annotation on both the constructor parameters and the member variables? Jackson should be able to deserialize with just a no-args constructor + setters on the member variables

nsgupta1 · 2025-05-06T20:01:51Z

Thanks @engechas and @eirsep for reviews, since none of the comments are blockers I will address these in follow up PR.

…#5678) * CrowdStrike client and coordinator implementation Signed-off-by: nsgupta1 <nsgupta1@users.noreply.github.com>

CrowdStrike client and coordinator implementation

254ec8a

Signed-off-by: nsgupta1 <nsgupta1@users.noreply.github.com>

nsgupta1 requested review from KarstenSchnitter, chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, kkondaka, oeyh, san81, sb2k16 and srikanthjg as code owners May 5, 2025 06:03

san81 reviewed May 5, 2025

View reviewed changes

Addressing review comments

411d53c

Signed-off-by: nsgupta1 <nsgupta1@users.noreply.github.com>

nishtham-amazon reviewed May 6, 2025

View reviewed changes

Addressing Nit comments

76b34f9

Signed-off-by: nsgupta1 <nsgupta1@users.noreply.github.com>

san81 approved these changes May 6, 2025

View reviewed changes

eirsep reviewed May 6, 2025

View reviewed changes

engechas approved these changes May 6, 2025

View reviewed changes

san81 merged commit 75f7bc7 into opensearch-project:main May 6, 2025
45 of 47 checks passed

Conversation

nsgupta1 commented May 5, 2025

Description

Issues Resolved

Check List

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

san81 left a comment

Choose a reason for hiding this comment

Uh oh!

nsgupta1 commented May 5, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nsgupta1 commented May 6, 2025

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants