Add Internal and external latency to OpenSearch and S3 sinks. #3583

kkondaka · 2023-11-02T22:45:46Z

Description

Add Internal and external latency to OpenSearch and S3 sinks
Date processor modified to support new option to put external origination time in the event metadata
OpenSearch and S3 sinks calculate the internal and external latencies when the event handle is released.

Issues Resolved

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

Check List

[ X] New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
[ X] 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: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

graytaylor0 · 2023-11-03T05:15:51Z

...nsearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

@@ -141,6 +144,7 @@ public OpenSearchSink(final PluginSetting pluginSetting,
    this.awsCredentialsSupplier = awsCredentialsSupplier;
    this.sinkContext = sinkContext != null ? sinkContext : new SinkContext(null, Collections.emptyList(), Collections.emptyList(), Collections.emptyList());
    this.expressionEvaluator = expressionEvaluator;
+    this.latencyMetrics = new LatencyMetrics(pluginMetrics);


Is there really no way to give the EventHandle class a copy of PluginMetrics, and then to just have the releaseEventHandle call automatically populate the metric so we don't have to pass them to every sink, and we keep to just one thing that everyone has to call (releaseEventHandle). You could even add to the function, and make it releaseEventHandle(final EventHandle eventHandle, final String pluginId). The pluginId could just be the plugin name for now, but it can be used in the metric.

How do we do that? Metrics are sink specific, right?

Yes but we could "fake" that the metrics are owned by the sink by adding the name of the sink here (since you can name the metrics anything and that it can still follow our metric naming pattern

I do not think that's the clean way because an event handle can only have one name for metrics. How can it have multiple names? I think this is way cleaner.

Every sink will need to provide this latency and we want to make it as easy as possible for sinks to use this feature.

I think that data-prepper-core could hold a LatencyMetrics per Sink. Just before calling Sink::output, data-prepper-core registers an onRelease method similar to what I suggested in another comment. Then it receives the callback when release is called.

One thing we would need to be sure to do here is ensure that sinks always call release. This is possible now that #3546 is merged.

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

graytaylor0 · 2023-11-04T01:13:58Z

...per-plugins/common/src/main/java/org/opensearch/dataprepper/plugins/sink/LatencyMetrics.java

+        final long internalLatency = Duration.between(eventHandle.getInternalOriginationTime(), Instant.now()).toMillis();
+        pluginMetrics.gauge(INTERNAL_LATENCY_MIN_MS, internalLatency, latency -> Math.min(internalLatencyMin, latency));
+        pluginMetrics.gauge(INTERNAL_LATENCY_MAX_MS, internalLatency, latency -> Math.max(internalLatencyMax, latency));
+        pluginMetrics.gauge(INTERNAL_LATENCY_AVG_MS, internalLatency, latency -> {


Why are we using guage here? Shouldn't this be DistributionSummary?

Good point. I wasn't aware of that. If everyone prefers that, I can modify to use DistributionSummary.

graytaylor0

This metric can just be a DistributionSummary

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

dlvenable · 2023-11-06T12:54:45Z

...processor/src/main/java/org/opensearch/dataprepper/plugins/processor/date/DateProcessor.java

+                    Instant timeStamp = result.getRight();
+                    if (dateProcessorConfig.getToOriginationMetadata()) {
+                        Event event = (Event)record.getData();
+                        event.getMetadata().setExternalOriginationTime(timeStamp);


Can we have EventMetadata call setExternalOriginationTime with the value when it is set? This way we can avoid making two calls here.

I did not do that because I am not sure if we want to tie them together. What if we don't want external latency metric but set metadata only? This way, we could introduce another flag in date processor in future if such requirement arises.

dlvenable · 2023-11-06T13:02:11Z

...h/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkOperationWrapper.java

@@ -76,6 +73,7 @@ public EventHandle getEventHandle() {

    public void releaseEventHandle(boolean result) {
        if (eventHandle != null) {
+            sink.updateLatencyMetrics(eventHandle);


Ultimately we want this called because release is called. Instead of doubling up calls, how about we configure the EventHandle to provide this type of callback?

interface EventHandle { ... void onRelease(BiConsumer<EventHandle, Boolean> reseaseConsumer); ... }

Then, in OpenSearchSink, be sure to make this call.

event.getHandle().onRelease((handle, result) -> updateLatencyMetrics(handle));

The current design results in too many classes having to be aware of these interactions.

dlvenable · 2023-11-06T13:08:39Z

...nsearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

@@ -141,6 +144,7 @@ public OpenSearchSink(final PluginSetting pluginSetting,
    this.awsCredentialsSupplier = awsCredentialsSupplier;
    this.sinkContext = sinkContext != null ? sinkContext : new SinkContext(null, Collections.emptyList(), Collections.emptyList(), Collections.emptyList());
    this.expressionEvaluator = expressionEvaluator;
+    this.latencyMetrics = new LatencyMetrics(pluginMetrics);


Every sink will need to provide this latency and we want to make it as easy as possible for sinks to use this feature.

I think that data-prepper-core could hold a LatencyMetrics per Sink. Just before calling Sink::output, data-prepper-core registers an onRelease method similar to what I suggested in another comment. Then it receives the callback when release is called.

One thing we would need to be sure to do here is ensure that sinks always call release. This is possible now that #3546 is merged.

dlvenable · 2023-11-06T13:09:13Z

...src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkRetryStrategyTests.java

@@ -68,6 +69,10 @@ public class BulkRetryStrategyTests {

    @BeforeEach
    public void setUp() {
+        sink = mock(OpenSearchSink.class);


Using some of the other suggestions, all of these changes should be unnecessary.

dlvenable · 2023-11-06T13:09:43Z

...ataprepper/plugins/sink/opensearch/bulk/JavaClientAccumulatingCompressedBulkRequestTest.java

@@ -36,10 +37,12 @@
 class JavaClientAccumulatingCompressedBulkRequestTest {

    private BulkRequest.Builder bulkRequestBuilder;
+    private OpenSearchSink sink;


Again, we can make all these changes unnecessary with the callback approaches recommended above.

dlvenable · 2023-11-06T13:10:40Z

...per-plugins/common/src/main/java/org/opensearch/dataprepper/plugins/sink/LatencyMetrics.java

+import java.time.Duration;
+import java.time.Instant;
+
+public class LatencyMetrics {


If we are able to move the tracking of this into data-prepper-core, please move this class into data-prepper-core. There should be no need to expose it in that situation.

dlvenable · 2023-11-06T13:11:36Z

...per-plugins/common/src/main/java/org/opensearch/dataprepper/plugins/sink/LatencyMetrics.java

+        internalLatencySummary = pluginMetrics.summary(INTERNAL_LATENCY);
+        externalLatencySummary = pluginMetrics.summary(EXTERNAL_LATENCY);
+    }
+    public void update(final EventHandle eventHandle) {


I tend to think that this should take the result passed into the EventHandle and make the decision as to whether or not to record here.

public void update(final EventHandle eventHandle, final boolean result) { if(result == false) return; ... }

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

dlvenable

Thanks @kkondaka , this is looking good. There are a few small corrections to make.

We can follow-up after merging this PR with some alternate names for the metric.

dlvenable · 2023-11-07T13:41:16Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/sink/SinkLatencyMetrics.java

+import java.time.Instant;
+
+public class SinkLatencyMetrics {
+    public static final String INTERNAL_LATENCY = "internalLatency";


I wonder if we should have different names here. What would a user interpret "internal" and "external" to mean?

Maybe we could follow up in a different PR before releasing 2.6.0.

dlvenable · 2023-11-07T13:43:00Z

...nsearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

@@ -91,6 +91,7 @@ public class OpenSearchSink extends AbstractSink<Record<Event>> {
  private static final Logger LOG = LoggerFactory.getLogger(OpenSearchSink.class);
  private static final int INITIALIZE_RETRY_WAIT_TIME_MS = 5000;
  private final AwsCredentialsSupplier awsCredentialsSupplier;
+  //private final LatencyMetrics latencyMetrics;


Please remove this commented out code.

dlvenable · 2023-11-07T13:43:21Z

...nsearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/OpenSearchSink.java

@@ -256,6 +257,20 @@ private void doInitializeInternal() throws IOException {
    return invalidActionErrorsCounter.count();
  }

+  /*


Please remove this commented out code.

dlvenable · 2023-11-07T13:44:15Z

...h/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/BulkOperationWrapper.java

-        this.bulkOperation = bulkOperation;
-        this.eventHandle = eventHandle;
-        this.jsonNode = null;
+        this(bulkOperation, eventHandle, null);


Thanks for the code improvements!

dlvenable · 2023-11-07T13:45:22Z

...ssor/src/test/java/org/opensearch/dataprepper/plugins/processor/date/DateProcessorTests.java

+        Assertions.assertTrue(event.getMetadata().getExternalOriginationTime() != null);
+        Assertions.assertTrue(event.getEventHandle().getExternalOriginationTime() != null);
+        ZonedDateTime expectedZonedDatetime = expectedDateTime.atZone(mockDateProcessorConfig.getSourceZoneId()).truncatedTo(ChronoUnit.SECONDS);
+        Assertions.assertTrue(expectedZonedDatetime.equals(event.getMetadata().getExternalOriginationTime().atZone(mockDateProcessorConfig.getSourceZoneId())));


The indentation on this line appears off.

Yeah. It appears off. But it's not. If you reduce your font size, you can see that it is fine.

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

dlvenable · 2023-11-07T17:42:32Z

Whitesource is failing, but this PR does not add the library being reported. We can ignore this failure.

juergen-walter · 2024-01-15T12:30:16Z

@kkondaka @dlvenable could you also update documentation for the new metrics
https://github.com/opensearch-project/data-prepper/blob/main/docs/monitoring.md

Please also state if a metric is given as millisecond or nanosecond.

JannikBrand · 2024-01-15T12:47:48Z

We could only get the name of the metric (e.g. <pipeline_name>_opensearch_PipelineLatency_sum) by looking at all exported metrics of the Data Prepper server (port: 4900; path: /metrics/prometheus).
I also wonder how it works with sub-pipelines, e.g. if we take a look at the trace analytics use case, which consists of three pipelines (one entry pipeline which redirects to the raw-pipeline as well as the service-map pipeline), I saw that only the raw and service-map pipeline include the PipelineLatency metric, which makes sense. Does it mean the latency of the first (entry-) pipeline is added on top of that (e.g. total_latency_of_raw_pipeline> = latency_of_entry_pipeline> + <latency_of_raw_pipeline>) ?

More general: Is it planned to document other (missing) metrics? I know about the doc that @juergen-walter linked and about the READMEs of the individual plugins which contain a metric section, but they are not including every metric. For example, the README of the opensearch sink lists some metrics and points to the AbstractSink, but this is only the java file. Another example would be circuit breaker metrics.

Add Internal and external latency to OpenSearch and S3 sinks

16cfa3f

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

kkondaka requested review from chenqi0805, engechas, graytaylor0, dinujoh, asifsmohammed, dlvenable and oeyh as code owners November 2, 2023 22:45

Added tests for Date Processor

72eda21

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

graytaylor0 reviewed Nov 3, 2023

View reviewed changes

Krishna Kondaka added 7 commits November 3, 2023 15:26

Added LatencyMetrics main and test files

a31d9bb

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

Fixed failing tests

fa8e91e

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

Fixed checkstyle failure

4c2af3b

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

Added test code to pass code coverage

123109a

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

Fixed tests

c2bbbfb

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

Fixed tests

07a0896

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

Fixed checkstyle failures

b22a682

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

graytaylor0 reviewed Nov 4, 2023

View reviewed changes

graytaylor0 requested changes Nov 4, 2023

View reviewed changes

Krishna Kondaka added 3 commits November 4, 2023 02:35

Replaced guage with distribution summary

e014114

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

Fixed failing s3 sink tests

2d32e77

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

Fixed failing s3 sink tests

7709f0e

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

graytaylor0 previously approved these changes Nov 4, 2023

View reviewed changes

dlvenable requested changes Nov 6, 2023

View reviewed changes

Addressed review comments

39c6160

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

kkondaka dismissed graytaylor0’s stale review via 39c6160 November 6, 2023 22:45

Krishna Kondaka added 2 commits November 7, 2023 00:04

Modified tests

e794577

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

Addressed review comments

ed07c9d

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

dlvenable requested changes Nov 7, 2023

View reviewed changes

Addressed review comments

b7a574c

Signed-off-by: Krishna Kondaka <krishkdk@dev-dsk-krishkdk-2c-bd29c437.us-west-2.amazon.com>

dlvenable approved these changes Nov 7, 2023

View reviewed changes

graytaylor0 approved these changes Nov 7, 2023

View reviewed changes

dlvenable merged commit a40f4b8 into opensearch-project:main Nov 7, 2023
62 of 63 checks passed

dlvenable mentioned this pull request Nov 13, 2023

Add S3 SQS Data Event Notification message as metadata to records #3641

Open

kkondaka deleted the os-s3-e2e-latency-metric branch May 13, 2024 05:51

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add Internal and external latency to OpenSearch and S3 sinks. #3583

Add Internal and external latency to OpenSearch and S3 sinks. #3583

kkondaka commented Nov 2, 2023

graytaylor0 Nov 3, 2023

kkondaka Nov 3, 2023

graytaylor0 Nov 3, 2023

kkondaka Nov 4, 2023

dlvenable Nov 6, 2023

graytaylor0 Nov 4, 2023

kkondaka Nov 4, 2023

graytaylor0 left a comment

dlvenable Nov 6, 2023

kkondaka Nov 6, 2023

dlvenable Nov 6, 2023

dlvenable Nov 6, 2023

dlvenable Nov 6, 2023

dlvenable Nov 6, 2023

dlvenable Nov 6, 2023

dlvenable Nov 6, 2023

dlvenable left a comment

dlvenable Nov 7, 2023

dlvenable Nov 7, 2023

dlvenable Nov 7, 2023

dlvenable Nov 7, 2023

dlvenable Nov 7, 2023

kkondaka Nov 7, 2023

dlvenable commented Nov 7, 2023

juergen-walter commented Jan 15, 2024

JannikBrand commented Jan 15, 2024

Add Internal and external latency to OpenSearch and S3 sinks. #3583

Add Internal and external latency to OpenSearch and S3 sinks. #3583

Conversation

kkondaka commented Nov 2, 2023

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

graytaylor0 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable commented Nov 7, 2023

juergen-walter commented Jan 15, 2024

JannikBrand commented Jan 15, 2024