[improve][io]add metrics to elastic search sink #20498

zzzming · 2023-06-05T19:28:48Z

Motivation

add metrics to elastic search sink

Modifications

add metrics to elastic search sink

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (10MB)
Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository:

github-actions · 2023-06-05T19:29:18Z

@zzzming Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

michaeljmarshall

Thanks for your contribution @zzzming! I left some comments.

...ar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java

michaeljmarshall · 2023-06-05T19:50:58Z

...ar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java

                }
            } catch (JsonProcessingException jsonProcessingException) {
                switch (elasticSearchConfig.getMalformedDocAction()) {
                    case IGNORE:
+                        incrementCounter(METRICS_TOTAL_SKIP, 1);


It seems like it would be valuable to differentiate this case from the other SKIP case since this one is a failure case. What do you think?

I think you might have misunderstood my question @zzzming. I think this case is different than the other skip, so we should have different metric names.

In the second commit, does it make sense to change this to a new metric? I added an IGNORE counter. I agree it should not use the SKIP metrics.

case IGNORE: incrementCounter(METRICS_TOTAL_IGNORE, 1);

There are still two calls to incrementCounter(METRICS_TOTAL_IGNORE, 1); in what seem like two different scenarios. I am suggesting they need to be differentiated.

eolivelli

+1

This will be helpful in debugging workloads

lhotari

LGTM. Good work @zzzming

lhotari · 2023-06-06T15:57:20Z

...ar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java

@@ -139,29 +163,35 @@ public void write(Record<GenericObject> record) throws Exception {
                    } else {
                        elasticsearchClient.indexDocument(record, idAndDoc);
                    }
+                    incrementCounter(METRICS_TOTAL_SUCCESS, 1);


I guess the only concern about the metrics is that when bulk indexing is used, the result is probably asynchronous. @nicoloboschi is this the case?

Fantastic point. Note also that indexDocument can fail without throwing an exception:

pulsar/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java

Lines 184 to 190 in 82237d3

final boolean createdOrUpdated = client.indexDocument(indexName, documentId, documentSource);

if (createdOrUpdated) {

record.ack();

} else {

record.fail();

}

return createdOrUpdated;

Perhaps we can integrate with this interface?

pulsar/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java

Lines 106 to 116 in 82237d3

/**

* Acknowledge that this record is fully processed.

*/

default void ack() {

}

/**

* To indicate that this record has failed to be processed.

*/

default void fail() {

}

It almost seems like we could get a generic metrics collector for acks on all sinks.

We don't need the generic solution in this PR, of course, but we do need to handle the async behavior of writing to ES.

@michaeljmarshall @lhotari
Added metrics to bulk request case

michaeljmarshall

Sorry for the late changes requested, but in order for these metrics to provide correct insight, I think we need to properly handle bulk indexing and index failures before we can merge this.

I'll also note that it'd be interesting to expose the duration of index requests, but that would belong in another PR.

michaeljmarshall · 2023-06-06T19:54:56Z

...ar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java

@@ -139,29 +163,35 @@ public void write(Record<GenericObject> record) throws Exception {
                    } else {
                        elasticsearchClient.indexDocument(record, idAndDoc);
                    }
+                    incrementCounter(METRICS_TOTAL_SUCCESS, 1);


Fantastic point. Note also that indexDocument can fail without throwing an exception:

pulsar/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java

Lines 184 to 190 in 82237d3

final boolean createdOrUpdated = client.indexDocument(indexName, documentId, documentSource);

if (createdOrUpdated) {

record.ack();

} else {

record.fail();

}

return createdOrUpdated;

zzzming · 2023-06-06T23:35:17Z

Sorry for the late changes requested, but in order for these metrics to provide correct insight, I think we need to properly handle bulk indexing and index failures before we can merge this.

I'll also note that it'd be interesting to expose the duration of index requests, but that would belong in another PR.

Moved a metrics inside the indexDocument() function

I will raise another PR to track the index request duration.

mattisonchao · 2023-06-07T01:27:44Z

@asafm @coderzc would you mind to take a look?

asafm · 2023-06-07T07:07:52Z

...-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java

@@ -78,6 +90,7 @@ public void afterBulk(long executionId, List<BulkProcessor.BulkOperationRequest>
                        record.fail();
                        checkForIrrecoverableError(record, result);
                    } else {
+                        incrementCounter(METRICS_TOTAL_SUCCESS, index);


Aren't you suppose to increment failures as well above?

Increment counter of failure and skip cases are in the checkIrrecoverableError() method because there are different cases to cover

asafm · 2023-06-07T07:08:27Z

...-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java

@@ -96,6 +110,12 @@ public void afterBulk(long executionId, List<BulkProcessor.BulkOperationRequest>
        this.client = retry(() -> RestClientFactory.createClient(config, bulkListener), -1, "client creation");
    }

+    public void incrementCounter(String counter, double value) {


why public?

Because it will be called by ElasticSearchSink.java too

I wrote below - IMO it's a design smell.

Move the function to the metrics class as you suggested

asafm · 2023-06-07T07:14:57Z

...-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java

@@ -56,8 +58,18 @@ public class ElasticSearchClient implements AutoCloseable {
    final AtomicReference<Exception> irrecoverableError = new AtomicReference<>();
    private final IndexNameFormatter indexNameFormatter;

-    public ElasticSearchClient(ElasticSearchConfig elasticSearchConfig) {
+    // sink metrics
+    public static final String METRICS_TOTAL_INCOMING = "_elasticsearch_total_incoming_";


The metric name goes as a label into a Prometheus Summary collector. Why the _ prefix?

I would remove _total_ everywhere. This is just a label in a Prometheus Summary, so this ends up exported as multiple quantiles and sum and count. So maybe let's just express what this counts.

I follow the same pattern from KinesisSink.java. I suppose the there is a prefix to be added to this metrics name. The leading underscore is a delimiter.
I can remove the _total

You can double-check me - run the connector and check the /metrics output. If I'm right, all I'm saying is previous code shouldn't be used as an example.
Removing the total is only 2nd point. Look at point 1 as well.

Ok. The leading underscore is removed. Thank you

The metric name goes as a label into a Prometheus Summary collector. Why the _ prefix?

Great point @asafm. I had assumed these were metrics names, not label names. Here is the generation code and it supports that it is in fact a label name:

pulsar/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java

Lines 212 to 222 in b1822ed

this.userMetricsSummary = collectorRegistry.registerIfNotExist(

prefix + ComponentStatsManager.USER_METRIC_PREFIX,

Summary.build()

.name(prefix + ComponentStatsManager.USER_METRIC_PREFIX)

.help("User defined metric.")

.labelNames(userMetricsLabelNames)

.quantile(0.5, 0.01)

.quantile(0.9, 0.01)

.quantile(0.99, 0.01)

.quantile(0.999, 0.01)

.create());

We should remove the prefix and suffix of the _.

Removed both leading and ending underscores

asafm · 2023-06-07T07:24:26Z

...-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java

@@ -96,6 +110,12 @@ public void afterBulk(long executionId, List<BulkProcessor.BulkOperationRequest>
        this.client = retry(() -> RestClientFactory.createClient(config, bulkListener), -1, "client creation");
    }

+    public void incrementCounter(String counter, double value) {
+        if (sinkContext != null && counter != null) {


What's with the if? Why would both ever be null?

Again, to evaluate sinkContext against null exists in KinesisSink.java. I will remove counter evaluation part.

Can you please explain what is the relation of sinkContext != null to KinesisSink in this method/line?

I removed the if evaluation.

asafm · 2023-06-07T07:31:47Z

...-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java

                        log.warn("Ignoring malformed document index={} id={}",
                                result.getIndex(),
                                result.getDocumentId(),
                                error);
                        break;
                    case FAIL:
+                        incrementCounter(METRICS_TOTAL_FAILURE, 1);


For me failed == have not been indexed, whether the reason was malformed or not. So we need to decide how to differentiate.
Maybe: es_total_failures, es_skipped_failures

Also, I wouldn't differentiate between IGNORE and WARN. Don't see a good reason for it.

In our case, we need to track different ignored cases and skipped. I'm going to make a change to count them as all failures but peg ignores and skip too.

asafm · 2023-06-07T07:49:29Z

...ar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java

@@ -109,6 +109,7 @@ void setElasticsearchClient(ElasticSearchClient elasticsearchClient) {

    @Override
    public void write(Record<GenericObject> record) throws Exception {
+        this.elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_INCOMING, 1);


This doesn't make much sense - Let's have this metric (actually constant) in ElasticSearchSink and just call sinkContext.record from there.

asafm · 2023-06-07T07:53:00Z

...ar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java

@@ -123,12 +124,18 @@ public void write(Record<GenericObject> record) throws Exception {
                                    elasticsearchClient.bulkDelete(record, idAndDoc.getLeft());
                                } else {
                                    elasticsearchClient.deleteDocument(record, idAndDoc.getLeft());
+                                    elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_SUCCESS, 1);


This needs to be cleaner. If you have two classes, use this, then let's extract it to ElasticSearchSinkMetrics, define the constants there, and record all those events.

asafm · 2023-06-07T07:54:10Z

...ar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java

                            break;
                        case FAIL:
+                            elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_FAILURE, 1);


In elasticsearchClient.failed( you call below, you also introduced increments, no? Won't it double count?

An exception is thrown in this code block. Any counter incremental below is for different cases. They are invoked in different code path.

You're correct. That's super strange method :)

asafm · 2023-06-07T07:55:48Z

@asafm @coderzc would you mind to take a look?

@mattisonchao Thanks for pinging me.

I'll review again once this is ready for review.

asafm

You have no tests to see that your metrics are correct

asafm · 2023-06-07T14:54:00Z

...-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java

                        record.ack();
+                        metrics.incrementCounter(ElasticSearchMetrics.SUCCESS, index);


If you already did the work, wouldn't you prefer to go all the way?
metrics.incrementCounter(ElasticSearchMetrics.SUCCESS, index); ---> metrics.recordRecordIndexedSuccess()?

Actually it should be incremented by 1. The counter pegging along with message acknowledgement is in an iteration.

My point is changing the structure of the methods, not if it's by 1 or not :)

asafm · 2023-06-07T14:58:23Z

...io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java

+
+    private SinkContext sinkContext;
+    // sink metrics
+    public static final String INCOMING = "_elasticsearch_incoming_";


static final goes before private variables

asafm · 2023-06-07T14:59:17Z

...io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java

+    }
+
+    public void incrementCounter(String counter, double value) {
+        if (sinkContext != null) {


Why can sinkContext be null at all?

Not here for sure. Put in ctor.

removed the if evaluation

asafm · 2023-06-07T15:02:11Z

...ar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java

@@ -120,22 +123,21 @@ public void write(Record<GenericObject> record) throws Exception {
                    switch (elasticSearchConfig.getNullValueAction()) {
                        case DELETE:
                            if (idAndDoc.getLeft() != null) {
+                                metrics.incrementCounter(ElasticSearchMetrics.DELETE, 1);


bulkDelete or deleteDocument can fail and increment failure, but you increase it anyway- double counting.

@asafm I changed to DELETE_ATTEMPT to count the number of attempts.

I'm having a problem to separate DELETE or index in the bulk operation. This requires more code change. Can I do this in another PR? We do need a delete_attempt count that helps to track if or how many delete documents requests.

Would you help me to add the DELETE counter in another PR?

Ok. Ping me in the PR.

Don't you prefer to first fix the code to have everything you need and only then add the metrics?

asafm · 2023-06-11T13:57:01Z

Can you ping me to review once you take care of my comment "My point is changing the structure of the methods, not if it's by 1 or not :)"?

zzzming · 2023-06-12T02:35:27Z

Can you ping me to review once you take care of my comment "My point is changing the structure of the methods, not if it's by 1 or not :)"?

@asafm added the function per your review comment

asafm

I haven't reviewed most of it yet, because I'm still trying to figure out how retry works.
Also if you have 0 tests for your code - You have no idea if what you wrote is working properly , counting properly. Not to mention future changes will be super afraid to touch it since they might break without the protection of an e2e test or something similar.

asafm · 2023-06-13T17:07:33Z

...-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java

@@ -89,6 +94,7 @@ public void afterBulk(long executionId, List<BulkProcessor.BulkOperationRequest>
                for (BulkProcessor.BulkOperationRequest operation: bulkOperationList) {
                    final Record record = operation.getPulsarRecord();
                    record.fail();
+                    metrics.incrementCounter(ElasticSearchMetrics.FAILURE, 1);


I don't get it.
You switched to 'event based' methodology in ElasticSearchMetrics for recordRecordIndexedSuccess(), why not everywhere? This should be IMO metrics.recordRecordFailure()

asafm · 2023-06-13T17:10:30Z

...-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java

@@ -89,6 +94,7 @@ public void afterBulk(long executionId, List<BulkProcessor.BulkOperationRequest>
                for (BulkProcessor.BulkOperationRequest operation: bulkOperationList) {
                    final Record record = operation.getPulsarRecord();
                    record.fail();
+                    metrics.incrementCounter(ElasticSearchMetrics.FAILURE, 1);


How do you differentiate between retry failures to complete failures?
This method is called each time an attempt has failed.

asafm · 2023-06-13T18:18:14Z

...io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java

+ */
+public class ElasticSearchMetrics {
+
+    private SinkContext sinkContext;


place below static varibles

asafm · 2023-06-13T18:19:25Z

...io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java

+
+    private SinkContext sinkContext;
+    // sink metrics
+    public static final String INCOMING = "_elasticsearch_incoming";


why the leading underscore?

asafm · 2023-06-13T18:19:29Z

...io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java

+    public static final String INCOMING = "_elasticsearch_incoming";
+
+    // INCOMING = SUCCESS + FAILURE + SKIP + NULLVALUE_IGNORE
+    public static final String SUCCESS = "_elasticsearch_success";


why the leading underscore?

asafm · 2023-06-13T18:20:09Z

...io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java

+    }
+
+    public void incrementCounter(String counter, double value) {
+        this.sinkContext.recordMetric(counter, value);


this is redundant

github-actions · 2023-07-15T02:08:39Z

The pr had no activity for 30 days, mark with Stale label.

tisonkun · 2024-05-08T09:10:07Z

Closed as stale and conflict.

[improve][sink]add metrics to elastic search sink

e886a89

github-actions bot added the doc-label-missing label Jun 5, 2023

michaeljmarshall reviewed Jun 5, 2023

View reviewed changes

michaeljmarshall assigned zzzming Jun 5, 2023

michaeljmarshall requested review from eolivelli, tisonkun and nicoloboschi June 5, 2023 19:51

michaeljmarshall added this to the 3.1.0 milestone Jun 5, 2023

michaeljmarshall added the area/function label Jun 5, 2023

review comments

e3309d1

github-actions bot added doc-not-needed Your PR changes do not impact docs and removed doc-label-missing labels Jun 5, 2023

eolivelli approved these changes Jun 6, 2023

View reviewed changes

nicoloboschi approved these changes Jun 6, 2023

View reviewed changes

separate two different ignore

3eeb7b5

lhotari previously approved these changes Jun 6, 2023

View reviewed changes

lhotari reviewed Jun 6, 2023

View reviewed changes

michaeljmarshall requested changes Jun 6, 2023

View reviewed changes

zzzming added 2 commits June 6, 2023 19:24

bulk ops metrics

cedaed0

move metrics to indexDocument

90e90a1

asafm requested changes Jun 7, 2023

View reviewed changes

zzzming added 2 commits June 7, 2023 10:16

review comments

2b29c53

add missing metrics file

e59871c

asafm requested changes Jun 7, 2023

View reviewed changes

review feedback update

81d8033

zzzming added 3 commits June 8, 2023 22:58

delete_attempt metrics

3b95212

add incrementSuccessCounter

11743cd

fix success increments

381293c

zzzming mentioned this pull request Jun 9, 2023

[run-tests] [improve][io]add metrics to elastic search sink zzzming/pulsar#1

Open

ci lint error

1f4329b

add recordRecordIndexedSuccess() per review comment

616ad1a

asafm requested changes Jun 13, 2023

View reviewed changes

github-actions bot added the Stale label Jul 15, 2023

Technoboy- modified the milestones: 3.1.0, 3.2.0 Jul 31, 2023

Technoboy- modified the milestones: 3.2.0, 3.3.0 Dec 22, 2023

coderzc modified the milestones: 3.3.0, 3.4.0 May 8, 2024

tisonkun closed this May 8, 2024

	final boolean createdOrUpdated = client.indexDocument(indexName, documentId, documentSource);
	if (createdOrUpdated) {
	record.ack();
	} else {
	record.fail();
	}
	return createdOrUpdated;

	/**
	* Acknowledge that this record is fully processed.
	*/
	default void ack() {
	}

	/**
	* To indicate that this record has failed to be processed.
	*/
	default void fail() {
	}

	this.userMetricsSummary = collectorRegistry.registerIfNotExist(
	prefix + ComponentStatsManager.USER_METRIC_PREFIX,
	Summary.build()
	.name(prefix + ComponentStatsManager.USER_METRIC_PREFIX)
	.help("User defined metric.")
	.labelNames(userMetricsLabelNames)
	.quantile(0.5, 0.01)
	.quantile(0.9, 0.01)
	.quantile(0.99, 0.01)
	.quantile(0.999, 0.01)
	.create());

		record.ack();
		metrics.incrementCounter(ElasticSearchMetrics.SUCCESS, index);

[improve][io]add metrics to elastic search sink #20498

[improve][io]add metrics to elastic search sink #20498

Conversation

zzzming commented Jun 5, 2023 • edited

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Matching PR in forked repository

github-actions bot commented Jun 5, 2023

michaeljmarshall left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zzzming Jun 5, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

lhotari 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

michaeljmarshall left a comment • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zzzming commented Jun 6, 2023 • edited

mattisonchao commented Jun 7, 2023

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

Choose a reason for hiding this comment

michaeljmarshall Jun 7, 2023 • edited

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

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

asafm commented Jun 7, 2023

asafm 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

zzzming Jun 9, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asafm commented Jun 11, 2023

zzzming commented Jun 12, 2023

asafm 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

zzzming commented Jun 5, 2023 •

edited

zzzming Jun 5, 2023 •

edited

michaeljmarshall left a comment •

edited

zzzming commented Jun 6, 2023 •

edited

michaeljmarshall Jun 7, 2023 •

edited

zzzming Jun 9, 2023 •

edited