Implement dimensional time slice crawler by alparish · Pull Request #6011 · opensearch-project/data-prepper

alparish · 2025-08-22T20:06:02Z

Signed-off-by: Alekhya Parisha aparisha@amazon.com

Description

Added DimensionalTimeSliceCrawler to partition data by dimensionType and time
Replaced PaginationCrawler with DimensionalTimeSliceCrawler in Office365 source
Updated tests to support dimensional partitioning

Issues Resolved

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

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
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.

bbenner7635 · 2025-08-22T20:38:01Z

+        client.executePartition(state, buffer, acknowledgementSet);
+    }
+
+    private void createPartitionsFordimensionTypes(LeaderPartition leaderPartition,


nit: capitalize Dimension

bbenner7635 · 2025-08-22T20:45:55Z

+        DimensionalTimeSliceLeaderProgressState leaderProgressState =
+                (DimensionalTimeSliceLeaderProgressState) leaderPartition.getProgressState().get();
+        int remainingHours = leaderProgressState.getRemainingHours();
+        Instant nowUtc = latestModifiedTime.truncatedTo(ChronoUnit.HOURS);


Why do we use latestModifiedTime (Instant.now) instead of leaderProgressState.getLastpollTime (ref)?

I am actually confused on the original code. Why todayUTC is equal to getLastPollTime. Can we divide into it a bit.

I agree we should be consistent with TimeSliceCrawler if possible.

Instant initialDate = leaderProgressState.getLastPollTime(); Instant todayUtc = initialDate.truncatedTo(ChronoUnit.DAYS);

Historical pull occurs only during the first run. Initially, lastPollTime is set to Instant.now() when the plugin starts, while latestModifiedTime is set to Instant.now() when the crawl begins. I previously used latestModifiedTime to accommodate Office 365's 7-day limit, but this approach isn't generic enough for all connectors. I'll refactor the crawler to consistently use lastPollTime and handle the 7-day limit specifically in the O365 connector.

We don't need to truncate to hour boundaries (nowUTC) as it causes unnecessary data pulls. For example, if lastPollTime is 14:37:22, truncating to 14:00:00 means we'd pull an extra 37 minutes of data. I'll use lastPollTime directly to ensure we only retrieve data from the exact time the plugin started.

Prefer to truncate for now to keep consistent with TimeSliceCrawler--it is better to truncate & pull in extra logs than have potential dataloss

Discussed offline, will truncate to keep it consistent with TimeSliceCrawler

san81 · 2025-08-22T23:20:53Z

+                            }
+                        } catch (Office365Exception e) {
+
+                            log.error(NOISY, "{} error processing audit log: {}",


Adding logType to the error log would help, I guess?

logType is already included in the outer catch block's error message, which covers the entire partition, so adding it to individual error logs would be redundant.

wjyao0316 · 2025-08-22T23:10:59Z

+        Instant latestModifiedTime = Instant.now();
+        double startCount = partitionsCreatedCounter.count();
+
+        createPartitionsFordimensionTypes(leaderPartition, coordinator, latestModifiedTime, dimensionTypes);


nit. Naming is createPartitionsForDimensionTypes should be capitalize D

wjyao0316 · 2025-08-22T23:17:04Z

+        DimensionalTimeSliceLeaderProgressState leaderProgressState =
+                (DimensionalTimeSliceLeaderProgressState) leaderPartition.getProgressState().get();
+        int remainingHours = leaderProgressState.getRemainingHours();
+        Instant nowUtc = latestModifiedTime.truncatedTo(ChronoUnit.HOURS);


I am actually confused on the original code. Why todayUTC is equal to getLastPollTime. Can we divide into it a bit.

I agree we should be consistent with TimeSliceCrawler if possible.

Instant initialDate = leaderProgressState.getLastPollTime(); Instant todayUtc = initialDate.truncatedTo(ChronoUnit.DAYS);

san81 · 2025-08-22T23:36:59Z

-
                    return restTemplate.exchange(
-                            url,
+                            contentUri,


We are hitting the contentUri received from external source. For security, we should validate this url before hitting.

bbenner7635 · 2025-08-23T00:35:13Z

-                    throw new RuntimeException("Interrupted while waiting to retry time window", ie);
-                }
+    public AuditLogsResponse searchAuditLogs(final String logType, final Instant startTime, final Instant endTime, final String nextPageUri) {
+        try {


Can we do a null check for startTime & endTime? Example

Added null checks

san81 · 2025-08-26T16:07:19Z

-        service.initializeSubscriptions();
-        office365Iterator.initialize(lastPollTime);
-        return office365Iterator;


Where are these steps taking place now?

I didn't add this initialization before, as subscriptions were likely already set up for this tenant, allowing log fetching without explicit initialization. I'll now add it to Office365Source's start() method, which is a more appropriate place since it's a one-time setup that should happen when the plugin starts.

Also, we no longer need the iterator pattern since we're now processing logs in time-based chunks rather than page by page.

Signed-off-by: Alekhya Parisha <aparisha@amazon.com>

graytaylor0 · 2025-08-26T22:16:28Z

+            return new Record<>(event);
+        } catch (JsonProcessingException e) {
+            // JSON parsing errors are non-retryable as they indicate malformed data
+            throw new Office365Exception("Failed to parse audit log: " + logId, e, false);


Will this cause infinite loop of processing?

This is marked as non-retryable and swallowed / sent to pipeline DLQ (when available).

alparish requested review from KarstenSchnitter, chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, kkondaka, oeyh, san81, sb2k16 and srikanthjg as code owners August 22, 2025 20:06

bbenner7635 reviewed Aug 22, 2025

View reviewed changes

san81 reviewed Aug 22, 2025

View reviewed changes

wjyao0316 reviewed Aug 22, 2025

View reviewed changes

san81 reviewed Aug 22, 2025

View reviewed changes

bbenner7635 reviewed Aug 23, 2025

View reviewed changes

alparish force-pushed the feature/dimensional-tscrawler branch from aabf54e to 243fa1a Compare August 25, 2025 20:25

san81 reviewed Aug 26, 2025

View reviewed changes

Implement dimensional time slice crawler

2f95fba

Signed-off-by: Alekhya Parisha <aparisha@amazon.com>

alparish force-pushed the feature/dimensional-tscrawler branch from 243fa1a to 2f95fba Compare August 26, 2025 17:55

graytaylor0 approved these changes Aug 26, 2025

View reviewed changes

graytaylor0 reviewed Aug 26, 2025

View reviewed changes

san81 approved these changes Aug 27, 2025

View reviewed changes

san81 merged commit a6d2a53 into opensearch-project:main Aug 27, 2025
40 of 47 checks passed

Conversation

alparish commented Aug 22, 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!

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 Aug 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

san81 Aug 22, 2025 •

edited

Loading