You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "zzzming (via GitHub)" <gi...@apache.org> on 2023/06/05 19:28:50 UTC
[GitHub] [pulsar] zzzming opened a new pull request, #20498: [improve][io]add metrics to elastic search sink
zzzming opened a new pull request, #20498:
URL: https://github.com/apache/pulsar/pull/20498
### 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:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*If the box was checked, please highlight the changes*
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `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 -->
### Matching PR in forked repository
PR in forked repository: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1218504600
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java:
##########
@@ -277,7 +305,7 @@ public Pair<String, String> extractIdAndDocument(Record<GenericObject> record) t
}
doc = sanitizeValue(doc);
return Pair.of(id, doc);
- } else {
+ } else {
Review Comment:
Nit: please remove this white space change
```suggestion
} else {
```
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java:
##########
@@ -139,29 +161,35 @@ public void write(Record<GenericObject> record) throws Exception {
} else {
elasticsearchClient.indexDocument(record, idAndDoc);
}
+ incrementCounter(METRICS_TOTAL_SUCCESS, 1);
}
} catch (JsonProcessingException jsonProcessingException) {
switch (elasticSearchConfig.getMalformedDocAction()) {
case IGNORE:
+ incrementCounter(METRICS_TOTAL_SKIP, 1);
Review Comment:
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?
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java:
##########
@@ -67,15 +67,24 @@ public class ElasticSearchSink implements Sink<GenericObject> {
private ElasticSearchClient elasticsearchClient;
private final ObjectMapper objectMapper = new ObjectMapper();
private ObjectMapper sortedObjectMapper;
+ private SinkContext sinkContext;
private List<String> primaryFields = null;
private final Pattern nonPrintableCharactersPattern = Pattern.compile("[\\p{C}]");
private final Base64.Encoder base64Encoder = Base64.getEncoder().withoutPadding();
+ //
Review Comment:
Nit: please add a comment or remove this line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1220244979
##########
pulsar-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);
Review Comment:
Perhaps we can integrate with this interface?
https://github.com/apache/pulsar/blob/82237d3684fe506bcb6426b3b23f413422e6e4fb/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java#L106-L116
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221513263
##########
pulsar-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) {
Review Comment:
Because it will be called by ElasticSearchSink.java too
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1228450115
##########
pulsar-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);
Review Comment:
I don't get it.
You switched to 'event based' methodology in `ElasticSearchMetrics` for `recordRecordIndexedSuccess()`, why not everywhere? This should be IMO `metrics.recordRecordFailure()`
##########
pulsar-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);
Review Comment:
How do you differentiate between retry failures to complete failures?
This method is called each time an attempt has failed.
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.elasticsearch;
+
+import org.apache.pulsar.io.core.SinkContext;
+
+/*
+ * Metrics class for ElasticSearchSink
+ */
+public class ElasticSearchMetrics {
+
+ private SinkContext sinkContext;
+ // sink metrics
+ public static final String INCOMING = "_elasticsearch_incoming";
Review Comment:
why the leading underscore?
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.elasticsearch;
+
+import org.apache.pulsar.io.core.SinkContext;
+
+/*
+ * Metrics class for ElasticSearchSink
+ */
+public class ElasticSearchMetrics {
+
+ private SinkContext sinkContext;
+ // sink metrics
+ public static final String INCOMING = "_elasticsearch_incoming";
+
+ // INCOMING = SUCCESS + FAILURE + SKIP + NULLVALUE_IGNORE
+ public static final String SUCCESS = "_elasticsearch_success";
+
+ // DELETE_ATTEMPT is an attempt to delete a document by id
+ // TODO: add delete success metrics, currently it's difficult to separate delete and index from the bulk operations
+ public static final String DELETE_ATTEMPT = "elasticsearch_delete_attempt";
+
+ public static final String FAILURE = "elasticsearch_failure";
+ public static final String SKIP = "elasticsearch_skip";
+ public static final String WARN = "elasticsearch_warn";
+ public static final String MALFORMED_IGNORE = "elasticsearch_malformed_ignore";
+ public static final String NULLVALUE_IGNORE = "elasticsearch_nullvalue_ignore";
+
+ public ElasticSearchMetrics(SinkContext sinkContext) {
+ this.sinkContext = sinkContext;
+ }
+
+ public void incrementCounter(String counter, double value) {
+ this.sinkContext.recordMetric(counter, value);
Review Comment:
`this` is redundant
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.elasticsearch;
+
+import org.apache.pulsar.io.core.SinkContext;
+
+/*
+ * Metrics class for ElasticSearchSink
+ */
+public class ElasticSearchMetrics {
+
+ private SinkContext sinkContext;
Review Comment:
place below static varibles
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.elasticsearch;
+
+import org.apache.pulsar.io.core.SinkContext;
+
+/*
+ * Metrics class for ElasticSearchSink
+ */
+public class ElasticSearchMetrics {
+
+ private SinkContext sinkContext;
+ // sink metrics
+ public static final String INCOMING = "_elasticsearch_incoming";
+
+ // INCOMING = SUCCESS + FAILURE + SKIP + NULLVALUE_IGNORE
+ public static final String SUCCESS = "_elasticsearch_success";
Review Comment:
why the leading underscore?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1229592736
##########
pulsar-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);
Review Comment:
Don't you prefer to first fix the code to have everything you need and only then add the metrics?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221748203
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.elasticsearch;
+
+import org.apache.pulsar.io.core.SinkContext;
+
+/*
+ * Metrics class for ElasticSearchSink
+ */
+public class ElasticSearchMetrics {
+
+ private SinkContext sinkContext;
+ // sink metrics
+ public static final String INCOMING = "_elasticsearch_incoming_";
Review Comment:
static final goes before private variables
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.elasticsearch;
+
+import org.apache.pulsar.io.core.SinkContext;
+
+/*
+ * Metrics class for ElasticSearchSink
+ */
+public class ElasticSearchMetrics {
+
+ private SinkContext sinkContext;
+ // sink metrics
+ public static final String INCOMING = "_elasticsearch_incoming_";
+
+ // INCOMING = SUCCESS + FAILURE + SKIP + NULLVALUE_IGNORE
+ public static final String SUCCESS = "_elasticsearch_success_";
+
+ // DELETE is an attempt to delete a document by id
+ public static final String DELETE = "elasticsearch_delete_";
+ public static final String FAILURE = "elasticsearch_failure_";
+ public static final String SKIP = "elasticsearch_skip_";
+ public static final String WARN = "elasticsearch_warn_";
+ public static final String MALFORMED_IGNORE = "elasticsearch_malformed_ignore_";
+ public static final String NULLVALUE_IGNORE = "elasticsearch_nullvalue_ignore_";
+
+ public ElasticSearchMetrics(SinkContext sinkContext) {
+ this.sinkContext = sinkContext;
+ }
+
+ public void incrementCounter(String counter, double value) {
+ if (sinkContext != null) {
Review Comment:
1. Why can sinkContext be null at all?
2. Not here for sure. Put in ctor.
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java:
##########
@@ -90,8 +82,8 @@ public void afterBulk(long executionId, List<BulkProcessor.BulkOperationRequest>
record.fail();
checkForIrrecoverableError(record, result);
} else {
- incrementCounter(METRICS_TOTAL_SUCCESS, index);
record.ack();
+ metrics.incrementCounter(ElasticSearchMetrics.SUCCESS, index);
Review Comment:
If you already did the work, wouldn't you prefer to go all the way?
`metrics.incrementCounter(ElasticSearchMetrics.SUCCESS, index);` ---> `metrics.recordRecordIndexedSuccess()`?
##########
pulsar-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);
Review Comment:
`bulkDelete` or `deleteDocument` can fail and increment failure, but you increase it anyway- double counting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1223791551
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java:
##########
@@ -90,8 +82,8 @@ public void afterBulk(long executionId, List<BulkProcessor.BulkOperationRequest>
record.fail();
checkForIrrecoverableError(record, result);
} else {
- incrementCounter(METRICS_TOTAL_SUCCESS, index);
record.ack();
+ metrics.incrementCounter(ElasticSearchMetrics.SUCCESS, index);
Review Comment:
Actually it should be incremented by 1. The counter pegging along with message acknowledgement is in an iteration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221021917
##########
pulsar-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_";
Review Comment:
1. The metric name goes as a label into a Prometheus Summary collector. Why the `_` prefix?
2. 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.
##########
pulsar-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) {
Review Comment:
What's with the if? Why would both ever be null?
##########
pulsar-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);
Review Comment:
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.
##########
pulsar-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);
Review Comment:
This doesn't make much sense - Let's have this metric (actually constant) in `ElasticSearchSink` and just call sinkContext.record from there.
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java:
##########
@@ -117,14 +137,17 @@ void checkForIrrecoverableError(Record<?> record, BulkProcessor.BulkOperationRes
isMalformed = true;
switch (config.getMalformedDocAction()) {
case IGNORE:
+ incrementCounter(METRICS_TOTAL_MALFORMED_IGNORE, 1);
break;
case WARN:
+ incrementCounter(METRICS_TOTAL_SKIP, 1);
log.warn("Ignoring malformed document index={} id={}",
result.getIndex(),
result.getDocumentId(),
error);
break;
case FAIL:
+ incrementCounter(METRICS_TOTAL_FAILURE, 1);
Review Comment:
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.
##########
pulsar-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);
Review Comment:
Aren't you suppose to increment failures as well above?
##########
pulsar-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) {
Review Comment:
why public?
##########
pulsar-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);
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_DELETE, 1);
}
+ } else {
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_SKIP, 1);
}
break;
case IGNORE:
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_NULLVALUE_IGNORE, 1);
break;
case FAIL:
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_FAILURE, 1);
Review Comment:
In `elasticsearchClient.failed(` you call below, you also introduced increments, no? Won't it double count?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221518615
##########
pulsar-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) {
Review Comment:
Again, to evaluate sinkContext against null exists in KinesisSink.java. I will remove counter evaluation part.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] lhotari commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1219913003
##########
pulsar-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);
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221664570
##########
pulsar-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) {
Review Comment:
Can you please explain what is the relation of `sinkContext != null` to `KinesisSink` in this method/line?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1223786598
##########
pulsar-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);
Review Comment:
@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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1225821262
##########
pulsar-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);
Review Comment:
Ok. Ping me in the PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#issuecomment-1577358208
@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 -->
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1225821021
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java:
##########
@@ -90,8 +82,8 @@ public void afterBulk(long executionId, List<BulkProcessor.BulkOperationRequest>
record.fail();
checkForIrrecoverableError(record, result);
} else {
- incrementCounter(METRICS_TOTAL_SUCCESS, index);
record.ack();
+ metrics.incrementCounter(ElasticSearchMetrics.SUCCESS, index);
Review Comment:
My point is changing the structure of the methods, not if it's by 1 or not :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#issuecomment-1586483484
> 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1223763092
##########
pulsar-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) {
Review Comment:
I removed the if evaluation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1223762133
##########
pulsar-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) {
Review Comment:
Move the function to the metrics class as you suggested
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#issuecomment-1636624327
The pr had no activity for 30 days, mark with Stale label.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1223765979
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.elasticsearch;
+
+import org.apache.pulsar.io.core.SinkContext;
+
+/*
+ * Metrics class for ElasticSearchSink
+ */
+public class ElasticSearchMetrics {
+
+ private SinkContext sinkContext;
+ // sink metrics
+ public static final String INCOMING = "_elasticsearch_incoming_";
+
+ // INCOMING = SUCCESS + FAILURE + SKIP + NULLVALUE_IGNORE
+ public static final String SUCCESS = "_elasticsearch_success_";
+
+ // DELETE is an attempt to delete a document by id
+ public static final String DELETE = "elasticsearch_delete_";
+ public static final String FAILURE = "elasticsearch_failure_";
+ public static final String SKIP = "elasticsearch_skip_";
+ public static final String WARN = "elasticsearch_warn_";
+ public static final String MALFORMED_IGNORE = "elasticsearch_malformed_ignore_";
+ public static final String NULLVALUE_IGNORE = "elasticsearch_nullvalue_ignore_";
+
+ public ElasticSearchMetrics(SinkContext sinkContext) {
+ this.sinkContext = sinkContext;
+ }
+
+ public void incrementCounter(String counter, double value) {
+ if (sinkContext != null) {
Review Comment:
removed the if evaluation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1223765855
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchMetrics.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.elasticsearch;
+
+import org.apache.pulsar.io.core.SinkContext;
+
+/*
+ * Metrics class for ElasticSearchSink
+ */
+public class ElasticSearchMetrics {
+
+ private SinkContext sinkContext;
+ // sink metrics
+ public static final String INCOMING = "_elasticsearch_incoming_";
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1218664635
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java:
##########
@@ -139,29 +161,35 @@ public void write(Record<GenericObject> record) throws Exception {
} else {
elasticsearchClient.indexDocument(record, idAndDoc);
}
+ incrementCounter(METRICS_TOTAL_SUCCESS, 1);
}
} catch (JsonProcessingException jsonProcessingException) {
switch (elasticSearchConfig.getMalformedDocAction()) {
case IGNORE:
+ incrementCounter(METRICS_TOTAL_SKIP, 1);
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1223786598
##########
pulsar-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);
Review Comment:
@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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][io]add metrics to elastic search sink [pulsar]
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun commented on PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#issuecomment-2100119505
Closed as stale and conflict.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#issuecomment-1580140075
> @asafm @coderzc would you mind to take a look?
@mattisonchao Thanks for pinging me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1220342859
##########
pulsar-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);
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] mattisonchao commented on pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "mattisonchao (via GitHub)" <gi...@apache.org>.
mattisonchao commented on PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#issuecomment-1579707731
@asafm @coderzc would you mind to take a look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1218675384
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java:
##########
@@ -139,29 +161,35 @@ public void write(Record<GenericObject> record) throws Exception {
} else {
elasticsearchClient.indexDocument(record, idAndDoc);
}
+ incrementCounter(METRICS_TOTAL_SUCCESS, 1);
}
} catch (JsonProcessingException jsonProcessingException) {
switch (elasticSearchConfig.getMalformedDocAction()) {
case IGNORE:
+ incrementCounter(METRICS_TOTAL_SKIP, 1);
Review Comment:
Does it make sense to change this to in the second commit? I added an IGNORE counter.
```
case IGNORE:
incrementCounter(METRICS_TOTAL_IGNORE, 1);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1218659711
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java:
##########
@@ -277,7 +305,7 @@ public Pair<String, String> extractIdAndDocument(Record<GenericObject> record) t
}
doc = sanitizeValue(doc);
return Pair.of(id, doc);
- } else {
+ } else {
Review Comment:
thanks for the explanation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1218937679
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java:
##########
@@ -139,29 +161,35 @@ public void write(Record<GenericObject> record) throws Exception {
} else {
elasticsearchClient.indexDocument(record, idAndDoc);
}
+ incrementCounter(METRICS_TOTAL_SUCCESS, 1);
}
} catch (JsonProcessingException jsonProcessingException) {
switch (elasticSearchConfig.getMalformedDocAction()) {
case IGNORE:
+ incrementCounter(METRICS_TOTAL_SKIP, 1);
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1220250146
##########
pulsar-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);
Review Comment:
It almost seems like we could get a generic metrics collector for acks on all sinks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1220533310
##########
pulsar-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);
Review Comment:
@michaeljmarshall @lhotari
Added metrics to bulk request case
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#issuecomment-1579597260
> 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 to the indexDocument() function
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][io]add metrics to elastic search sink [pulsar]
Posted by "tisonkun (via GitHub)" <gi...@apache.org>.
tisonkun closed pull request #20498: [improve][io]add metrics to elastic search sink
URL: https://github.com/apache/pulsar/pull/20498
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1223762257
##########
pulsar-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_";
Review Comment:
Removed both leading and ending underscores
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1220240052
##########
pulsar-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);
Review Comment:
Fantastic point. Note also that `indexDocument` can fail without throwing an exception:
https://github.com/apache/pulsar/blob/82237d3684fe506bcb6426b3b23f413422e6e4fb/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java#L184-L190
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221660299
##########
pulsar-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) {
Review Comment:
I wrote below - IMO it's a design smell.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221762974
##########
pulsar-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_";
Review Comment:
> 1. 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:
https://github.com/apache/pulsar/blob/b1822ed2b4714e4b485ab695eb1d52634ecc8db4/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java#L212-L222
We should remove the prefix and suffix of the `_`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "michaeljmarshall (via GitHub)" <gi...@apache.org>.
michaeljmarshall commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221762974
##########
pulsar-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_";
Review Comment:
> 1. 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:
https://github.com/apache/pulsar/blob/b1822ed2b4714e4b485ab695eb1d52634ecc8db4/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java#L212-L222
We should remove the prefix and suffix of the `_`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221737192
##########
pulsar-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);
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_DELETE, 1);
}
+ } else {
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_SKIP, 1);
}
break;
case IGNORE:
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_NULLVALUE_IGNORE, 1);
break;
case FAIL:
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_FAILURE, 1);
Review Comment:
You're correct. That's super strange method :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1218675384
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java:
##########
@@ -139,29 +161,35 @@ public void write(Record<GenericObject> record) throws Exception {
} else {
elasticsearchClient.indexDocument(record, idAndDoc);
}
+ incrementCounter(METRICS_TOTAL_SUCCESS, 1);
}
} catch (JsonProcessingException jsonProcessingException) {
switch (elasticSearchConfig.getMalformedDocAction()) {
case IGNORE:
+ incrementCounter(METRICS_TOTAL_SKIP, 1);
Review Comment:
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);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1218570840
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java:
##########
@@ -277,7 +305,7 @@ public Pair<String, String> extractIdAndDocument(Record<GenericObject> record) t
}
doc = sanitizeValue(doc);
return Pair.of(id, doc);
- } else {
+ } else {
Review Comment:
This was incorrect indent originally. It was high lighted by IDE.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221516438
##########
pulsar-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_";
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221512684
##########
pulsar-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);
Review Comment:
Increment counter of failure and skip cases are in the checkIrrecoverableError() method because there are different cases to cover
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1218675384
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java:
##########
@@ -139,29 +161,35 @@ public void write(Record<GenericObject> record) throws Exception {
} else {
elasticsearchClient.indexDocument(record, idAndDoc);
}
+ incrementCounter(METRICS_TOTAL_SUCCESS, 1);
}
} catch (JsonProcessingException jsonProcessingException) {
switch (elasticSearchConfig.getMalformedDocAction()) {
case IGNORE:
+ incrementCounter(METRICS_TOTAL_SKIP, 1);
Review Comment:
Does it make sense to change this to in the second commit? I added an IGNORE counter.
``` case IGNORE:
incrementCounter(METRICS_TOTAL_IGNORE, 1);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221696464
##########
pulsar-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_";
Review Comment:
Ok. The leading underscore is removed. Thank you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221527900
##########
pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java:
##########
@@ -117,14 +137,17 @@ void checkForIrrecoverableError(Record<?> record, BulkProcessor.BulkOperationRes
isMalformed = true;
switch (config.getMalformedDocAction()) {
case IGNORE:
+ incrementCounter(METRICS_TOTAL_MALFORMED_IGNORE, 1);
break;
case WARN:
+ incrementCounter(METRICS_TOTAL_SKIP, 1);
log.warn("Ignoring malformed document index={} id={}",
result.getIndex(),
result.getDocumentId(),
error);
break;
case FAIL:
+ incrementCounter(METRICS_TOTAL_FAILURE, 1);
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#issuecomment-1586176211
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 :)"?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zzzming commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "zzzming (via GitHub)" <gi...@apache.org>.
zzzming commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221552448
##########
pulsar-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);
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_DELETE, 1);
}
+ } else {
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_SKIP, 1);
}
break;
case IGNORE:
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_NULLVALUE_IGNORE, 1);
break;
case FAIL:
+ elasticsearchClient.incrementCounter(ElasticSearchClient.METRICS_TOTAL_FAILURE, 1);
Review Comment:
An exception is thrown in this code block. Any counter incremental below is for different cases. They are invoked in different code path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] asafm commented on a diff in pull request #20498: [improve][io]add metrics to elastic search sink
Posted by "asafm (via GitHub)" <gi...@apache.org>.
asafm commented on code in PR #20498:
URL: https://github.com/apache/pulsar/pull/20498#discussion_r1221662758
##########
pulsar-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_";
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org