[Pub/Sub to Elasticsearch] Fix error handling #319

Cherepushko · 2021-12-21T09:55:04Z

This pull request contains many changes in custom ElasticsearchIO to be able to handle failed inserts. Now ElasticsearchIO.Write is not a terminal step but returns JSON with error message to be sent to error-output-topic.

Now failed inserts don't slow down the pipeline and the throughput remain stable.

It's a draft PR and additional refactoring can be required.

Cherepushko · 2021-12-21T09:55:44Z

@prathapreddy123 @suyograo

...in/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessValidateJsonFields.java

suyograo · 2022-02-08T00:12:25Z

@Cherepushko can you rebase please? Also, can you add in the description and code comments we are removing invalid fields (with only dot as name) so Elasticsearch doesn't throw exception and slow the throughput considerably.

...in/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessValidateJsonFields.java

suyograo · 2022-02-08T00:14:19Z

Also left a comment for adding unit test for the method that removes dots from k8s logs

suyograo · 2022-02-14T17:23:59Z

LGTM

prathapreddy123 · 2022-02-23T01:16:25Z

v2/elasticsearch-common/pom.xml

@@ -27,6 +27,7 @@
    </parent>

    <artifactId>elasticsearch-common</artifactId>
+    <version>1.0-SNAPSHOT</version>


Is this file needs to be included as current version in master branch is already 1.0-SNAPSHOT

prathapreddy123 · 2022-02-23T01:21:15Z

...rc/main/java/com/google/cloud/teleport/v2/elasticsearch/transforms/WriteToElasticsearch.java

@@ -69,7 +69,7 @@ public static Builder newBuilder() {
  private static final String DOCUMENT_TYPE = "_doc";

  @Override
-  public PDone expand(PCollection<String> jsonStrings) {
+  public PCollection<String> expand(PCollection<String> jsonStrings) {


Since this function uses ElasticsearchIO.write() which still returns PDone can you explain how can we get PCollection

We get PCollection in modified ElasticsearchIO

prathapreddy123 · 2022-02-23T01:24:23Z

v2/googlecloud-to-elasticsearch/docs/PubSubToElasticsearch/README.md

@@ -19,6 +19,10 @@ To enable the out of the box integration:
 1. Install the Google Cloud [Logs integration](https://www.elastic.co/integrations?solution=observability&category=google-cloud) from Kibana UI
 2. [Export logs](https://cloud.google.com/logging/docs/export) from above data source to a separate Pub/Sub subscription. You can then use the Dataflow template's `dataset` parameter to specify the data source name and their corresponding `inputSubscription`.

+## Data validation
+
+Invalid input fields in messages will be fixed in runtime to not slowdown the pipeline and to prevent throwing exceptions.


Can you add some examples that were being taken care since code might not be handling all invalid fields

prathapreddy123 · 2022-02-23T01:25:37Z

...h-common/src/main/java/com/google/cloud/teleport/v2/elasticsearch/utils/ElasticsearchIO.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 1260 Google LLC


Since templates already uses 2.36 version should we directly consider using Beam ElasticSearch IO and drop the teleport version. However this needs proper validation of all 3 existing templates

prathapreddy123 · 2022-02-23T01:28:27Z

v2/googlecloud-to-elasticsearch/pom.xml

+        <dependency>
+            <groupId>com.jayway.jsonpath</groupId>
+            <artifactId>json-path</artifactId>
+            <version>2.4.0</version>


Can you upgrade to most recent version (2.6 or 2.7)

prathapreddy123 · 2022-02-23T01:39:26Z

...le/cloud/teleport/v2/elasticsearch/transforms/FailedElasticsearchMessageToPubsubTopicFn.java

+    try {
+      outputMessage = OBJECT_MAPPER.readTree(input);
+    } catch (JsonProcessingException e) {
+      e.printStackTrace();


Avoid using printStackTrace instead log the warning message using Log.warn

prathapreddy123 · 2022-02-23T01:41:06Z

...in/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessValidateJsonFields.java

+      String input = context.element();
+      ObjectNode node = null;
+      try {
+        JsonPath.using(configuration)


What is the use of this code as neither output was capture nor an exception was raised in case of any error

prathapreddy123 · 2022-02-23T01:44:03Z

...in/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessValidateJsonFields.java

+      } catch (Exception ignored) {
+      }
+
+      JsonNode requestApp = null;


Since there is a lot of repeated code, add functions shown below and call the functions for different jpaths (i.e requestAPP, responseApp etc)

parseJson(String json, String jpath) to handle read functionality

updateJson(String json, ... other parameters) to handle put functionality

deleteJsonString json, String jpath) to handle delete functionality

prathapreddy123 · 2022-02-23T01:44:37Z

...in/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessValidateJsonFields.java

+      context.output(node.toString());
+    }
+
+    public static ObjectNode removeEmptyFields(final ObjectNode jsonNode) {


Add Java doc comments to all the methods

prathapreddy123 · 2022-02-23T01:50:40Z

...in/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessValidateJsonFields.java

+import org.slf4j.LoggerFactory;
+
+/** ProcessValidateJsonFields is used to fix malformed fields, (ex. de-dot). */
+public class ProcessValidateJsonFields


Better name would be ResolveInvalidJsonFields

stale · 2022-04-18T20:34:35Z

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

bvolpato · 2023-04-10T13:21:44Z

Thanks for contributing!

There have been code reviews here that were not addressed (and the same thing about the structure of the project changing) -- do we still want to work on this PR?

cuigle · 2023-08-29T09:07:02Z

Hi team, I'm a Google Cloud TAM that has a customer that needs this issue to be addressed. They have 30+ projects that sends logs to ElasticSearch and this error constantly appears. They will need the Pub/Sub to Elasticsearch template to catch such errors and make necessary formatting changes so that all logs from Google Cloud are accepted on ElasticSearch side. Could you please reopen this issue?

vidok · 2024-03-26T11:28:35Z

Hi team, I'm a Google Cloud TAM that has a customer that needs this issue to be addressed. They have 30+ projects that sends logs to ElasticSearch and this error constantly appears. They will need the Pub/Sub to Elasticsearch template to catch such errors and make necessary formatting changes so that all logs from Google Cloud are accepted on ElasticSearch side. Could you please reopen this issue?

Hey @cuigle! It's been a while but could you let me know if your customer still needs this improvement? Our team at Elastic consider maintaining this template and we can pick up this issue. It would be helpful if you share more detailed information about the issues your customer 's been facing.

Thanks.

pull-request-size bot added the size/XXL label Dec 21, 2021

suyograo reviewed Feb 8, 2022

View reviewed changes

...in/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessValidateJsonFields.java Show resolved Hide resolved

suyograo reviewed Feb 8, 2022

View reviewed changes

...in/java/com/google/cloud/teleport/v2/elasticsearch/transforms/ProcessValidateJsonFields.java Show resolved Hide resolved

Cherepushko added 6 commits February 11, 2022 13:30

suppressed errors, routed errors to corresponding topic

c4dfab9

improved error topic messages

7df6d87

fixed error handling

707b5a1

removed unused code

880ae4d

added unit test for ProcessValidateJsonFields

ca134ee

updated README

96a352c

Cherepushko force-pushed the error-handling-fix branch from 5412975 to 96a352c Compare February 11, 2022 11:33

spotless apply

ed36d71

prathapreddy123 suggested changes Feb 23, 2022

View reviewed changes

stale bot added the stale label Apr 18, 2022

github-actions bot removed the stale label Jul 26, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Pub/Sub to Elasticsearch] Fix error handling #319

[Pub/Sub to Elasticsearch] Fix error handling #319

Cherepushko commented Dec 21, 2021

Cherepushko commented Dec 21, 2021

suyograo commented Feb 8, 2022

suyograo commented Feb 8, 2022

suyograo commented Feb 14, 2022

prathapreddy123 Feb 23, 2022

prathapreddy123 Feb 23, 2022

Cherepushko Mar 4, 2022

prathapreddy123 Feb 23, 2022

prathapreddy123 Feb 23, 2022

prathapreddy123 Feb 23, 2022

prathapreddy123 Feb 23, 2022

prathapreddy123 Feb 23, 2022

prathapreddy123 Feb 23, 2022

prathapreddy123 Feb 23, 2022

prathapreddy123 Feb 23, 2022

stale bot commented Apr 18, 2022

bvolpato commented Apr 10, 2023

cuigle commented Aug 29, 2023

vidok commented Mar 26, 2024

[Pub/Sub to Elasticsearch] Fix error handling #319

Are you sure you want to change the base?

[Pub/Sub to Elasticsearch] Fix error handling #319

Conversation

Cherepushko commented Dec 21, 2021

Cherepushko commented Dec 21, 2021

suyograo commented Feb 8, 2022

suyograo commented Feb 8, 2022

suyograo commented Feb 14, 2022

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

stale bot commented Apr 18, 2022

bvolpato commented Apr 10, 2023

cuigle commented Aug 29, 2023

vidok commented Mar 26, 2024