Base Livecapture Config and Manager Implementations by K4ran30 · Pull Request #5971 · opensearch-project/data-prepper

K4ran30 · 2025-08-06T05:13:04Z

Description

This PR introduces the core live capture functionality to Data Prepper, enabling real-time event tracing capabilities. This first commit includes the foundational classes including configuration, event capture coordination, and rate limiting.

LiveCaptureAppConfig.java - Spring configuration for live capture initialization
LiveCaptureConfiguration.java - Configuration class mapping to YAML settings
LiveCaptureManager.java - Central singleton managing live capture state

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
Support live tracking of event transformation #4711
- New functionality has javadoc added
Commits are signed with a real name per the DCO

Notes

This is the first commit in a multi-part implementation of live capture
Documentation will be added in a separate issue/PR once the complete feature is implemented
Additional commits will follow with interceptors and pipeline integration

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: Karan Bhakta <kabhakta@amazon.com>

kkondaka · 2025-08-08T05:43:44Z

@K4ran30 every java file in src/main should have a test file in src/test. Please write tests for all the code.

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

kkondaka · 2025-08-08T05:39:21Z

+
+        try {
+            Map<String, Object> eventData = event.toMap();
+            return activeFilters.entrySet().stream().allMatch(filter -> {


You should use DataPrepper expressions instead.

Would I do this by using the DataPrepper's ExpressionEvaluator?

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

kkondaka · 2025-08-11T04:47:40Z

+    public static final String PROCESSOR = "Processor";
+    public static final String ROUTE = "Route";
+    public static final String SINK = "Sink";
+    private static volatile LiveCaptureManager INSTANCE;


I am not sure if we need this. For now, we can keep it but we should not need this.

We should remove this. This LiveCaptureManager can be created as a Spring Bean so that we don't need this at all.

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

dlvenable

Thanks for this initial PR!

dlvenable · 2025-08-12T19:35:37Z

+    public static final String PROCESSOR = "Processor";
+    public static final String ROUTE = "Route";
+    public static final String SINK = "Sink";
+    private static volatile LiveCaptureManager INSTANCE;


We should remove this. This LiveCaptureManager can be created as a Spring Bean so that we don't need this at all.

dlvenable · 2025-08-12T19:36:41Z

+        return INSTANCE;
+    }
+
+    public static boolean shouldLiveCapture(final Event event) {


We should avoid all of these static methods. Use dependency injection where need and have a single instance provided by Spring.

dlvenable · 2025-08-12T21:52:31Z

+    }
+
+    // turn every entry into a dataprepper event so we can output to sink
+    public void processEventLiveCapture(final List<Map<String, Object>> liveCaptureEntries) {


We should have a more concrete model than Map<String, Object>.

Should I introduce model classes for the entries?

dlvenable · 2025-08-12T21:53:12Z

+                }
+                break;
+            case PROCESSOR:
+                if (name != null) captureMetaData.put("processorName", name);


I'm not sure that using Name in the suffix here is quite correct.

For one, we may add an Id concept in the future. For another, these are processor/sink types.

Changed from name to type

dlvenable · 2025-08-12T21:54:37Z

+        Map<String, Object> configMap = (Map<String, Object>) sinkConfig;
+        // Find the first non-metadata entry to create the sink
+        for (Map.Entry<String, Object> entry : configMap.entrySet()) {
+            if (!"entry_threshold".equals(entry.getKey()) && !"batch_size".equals(entry.getKey())) {


What are these things? entry_threshold and batch_size?

entry_threshold and batch_size are live capture configuration metadata that users can control through the config file.

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

…into live-capture # Conflicts: # data-prepper-core/src/main/java/org/opensearch/dataprepper/core/livecapture/LiveCaptureAppConfig.java # data-prepper-core/src/main/java/org/opensearch/dataprepper/core/livecapture/LiveCaptureManager.java # data-prepper-core/src/test/java/org/opensearch/dataprepper/core/livecapture/LiveCaptureAppConfigTest.java

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

Signed-off-by: Karan Bhakta <123035205+K4ran30@users.noreply.github.com>

Base Livecapture Config and Manager Implementations

f6f5111

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

K4ran30 marked this pull request as ready for review August 6, 2025 21:32

K4ran30 requested review from KarstenSchnitter, chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, kkondaka, oeyh, san81, sb2k16 and srikanthjg as code owners August 6, 2025 21:32

K4ran30 force-pushed the live-capture branch from ce1d02d to f6f5111 Compare August 8, 2025 18:28

Added tests

24693b8

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

kkondaka reviewed Aug 8, 2025

View reviewed changes

Karan Bhakta added 2 commits August 9, 2025 20:17

Fixed how livecaptures are stored and the ratelimiter

2bc5cf9

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

Removed old rate limiter files

0747e38

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

kkondaka reviewed Aug 11, 2025

View reviewed changes

Comment thread ...core/src/main/java/org/opensearch/dataprepper/core/livecapture/LiveCaptureOutputManager.java Outdated

Karan Bhakta added 2 commits August 11, 2025 10:40

Removed livecapuredoutputmanager and simplifed livecapturemanager

19ca74c

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

Final Manager Code

dac6ba9

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

dlvenable requested changes Aug 12, 2025

View reviewed changes

Karan Bhakta and others added 4 commits August 13, 2025 14:44

Inital refactoring of LiveCaptureManager Spring

11b9c61

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

Initial refactoring of LiveCaptureManager for Spring integration

94c1bd6

Signed-off-by: Karan Bhakta <kabhakta@amazon.com>

Merge branch 'main' into live-capture

7886877

Signed-off-by: Karan Bhakta <123035205+K4ran30@users.noreply.github.com>

Conversation

K4ran30 commented Aug 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Check List

Notes

Uh oh!

kkondaka commented Aug 8, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

dlvenable left a comment

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

K4ran30 Aug 12, 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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

K4ran30 commented Aug 6, 2025 •

edited

Loading

K4ran30 Aug 12, 2025 •

edited

Loading