Improve logging for exceptions (#4942)

Improve logging for exceptions Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>
opensearch-project · Sep 18, 2024 · 87c560a · 87c560a
1 parent 2919e99
commit 87c560a
Show file tree

Hide file tree

Showing 21 changed files with 200 additions and 70 deletions.
diff --git a/data-prepper-api/src/main/java/org/opensearch/dataprepper/logging/DataPrepperMarkers.java b/data-prepper-api/src/main/java/org/opensearch/dataprepper/logging/DataPrepperMarkers.java
@@ -6,6 +6,7 @@
 public final class DataPrepperMarkers {
     public static final Marker EVENT = MarkerFactory.getMarker("EVENT");
     public static final Marker SENSITIVE = MarkerFactory.getMarker("SENSITIVE");
+    public static final Marker NOISY = MarkerFactory.getMarker("NOISY");
 
     static {
         EVENT.add(SENSITIVE);

diff --git a/...ins/common/src/main/java/org/opensearch/dataprepper/plugins/processor/DelayProcessor.java b/...ins/common/src/main/java/org/opensearch/dataprepper/plugins/processor/DelayProcessor.java
@@ -8,6 +8,7 @@
 import com.fasterxml.jackson.annotation.JsonClassDescription;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
 import org.opensearch.dataprepper.model.processor.Processor;
@@ -34,7 +35,7 @@ public Collection<Record<?>> execute(final Collection<Record<?>> records) {
         try {
             Thread.sleep(delayDuration.toMillis());
         } catch (final InterruptedException ex) {
-            LOG.error("Interrupted during delay processor", ex);
+            LOG.error(NOISY, "Interrupted during delay processor", ex);
         }
         return records;
     }

diff --git a/...ns/common/src/main/java/org/opensearch/dataprepper/plugins/processor/StringProcessor.java b/...ns/common/src/main/java/org/opensearch/dataprepper/plugins/processor/StringProcessor.java
@@ -8,16 +8,18 @@
 import com.fasterxml.jackson.annotation.JsonClassDescription;
 import com.fasterxml.jackson.annotation.JsonPropertyDescription;
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.event.JacksonEvent;
 import org.opensearch.dataprepper.model.processor.Processor;
 import org.opensearch.dataprepper.model.record.Record;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -26,8 +28,6 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
-import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
-
 /**
  * A simple String implementation of {@link Processor} which generates new Records with uppercase or lowercase content. The current
  * simpler implementation does not handle errors (if any).
@@ -84,7 +84,13 @@ public Collection<Record<Event>> execute(final Collection<Record<Event>> records
                         .build();
                 modifiedRecords.add(new Record<>(newRecordEvent));
             } catch (JsonProcessingException e) {
-                LOG.error(EVENT, "Unable to process Event data: {}", eventJson, e);
+                LOG.atError()
+                        .addMarker(EVENT)
+                        .addMarker(NOISY)
+                        .setMessage("Unable to process Event data: {}")
+                        .addArgument(eventJson)
+                        .setCause(e)
+                        .log();
             }
         }
         return modifiedRecords;

diff --git a/...rocessor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessor.java b/...rocessor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessor.java
@@ -5,7 +5,14 @@
 
 package org.opensearch.dataprepper.plugins.processor.csv;
 
+import com.fasterxml.jackson.databind.MappingIterator;
+import com.fasterxml.jackson.dataformat.csv.CsvMapper;
+import com.fasterxml.jackson.dataformat.csv.CsvParser;
+import com.fasterxml.jackson.dataformat.csv.CsvSchema;
+import io.micrometer.core.instrument.Counter;
 import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
@@ -14,11 +21,6 @@
 import org.opensearch.dataprepper.model.processor.AbstractProcessor;
 import org.opensearch.dataprepper.model.processor.Processor;
 import org.opensearch.dataprepper.model.record.Record;
-import com.fasterxml.jackson.databind.MappingIterator;
-import com.fasterxml.jackson.dataformat.csv.CsvMapper;
-import com.fasterxml.jackson.dataformat.csv.CsvParser;
-import com.fasterxml.jackson.dataformat.csv.CsvSchema;
-import io.micrometer.core.instrument.Counter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -28,8 +30,6 @@
 import java.util.List;
 import java.util.Objects;
 
-import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
-
 /**
  * Processor to parse CSV data in Events.
  *
@@ -107,7 +107,13 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
                 }
             } catch (final IOException e) {
                 csvInvalidEventsCounter.increment();
-                LOG.error(EVENT, "An exception occurred while reading event [{}]", event, e);
+                LOG.atError()
+                        .addMarker(EVENT)
+                        .addMarker(NOISY)
+                        .setMessage("An exception occurred while reading event [{}]")
+                        .addArgument(event)
+                        .setCause(e)
+                        .log();
             }
         }
         return records;

diff --git a/...cessor/src/main/java/org/opensearch/dataprepper/plugins/processor/date/DateProcessor.java b/...cessor/src/main/java/org/opensearch/dataprepper/plugins/processor/date/DateProcessor.java
@@ -8,6 +8,7 @@
 import io.micrometer.core.instrument.Counter;
 import org.apache.commons.lang3.tuple.Pair;
 import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
@@ -102,7 +103,7 @@ public Collection<Record<Event>> doExecute(Collection<Record<Event>> records) {
                     record.getData().put(dateProcessorConfig.getDestination(), zonedDateTime);
                 }
             } catch (final Exception e) {
-                LOG.error("An exception occurred while attempting to process Event: ", e);
+                LOG.error(NOISY,"An exception occurred while attempting to process Event: ", e);
             }
 
         }

diff --git a/.../src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java b/.../src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java
@@ -6,6 +6,8 @@
 package org.opensearch.dataprepper.plugins.processor.dissect;
 
 import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
@@ -26,8 +28,6 @@
 import java.util.Map;
 import java.util.Objects;
 
-import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
-
 
 @DataPrepperPlugin(name = "dissect", pluginType = Processor.class, pluginConfigurationType = DissectProcessorConfig.class)
 public class DissectProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
@@ -72,7 +72,13 @@ public Collection<Record<Event>> doExecute(Collection<Record<Event>> records) {
                     }
                 }
             } catch (Exception ex){
-                LOG.error(EVENT, "Error dissecting the event [{}] ", record.getData(), ex);
+                LOG.atError()
+                        .addMarker(EVENT)
+                        .addMarker(NOISY)
+                        .setMessage("Error dissecting the event [{}]")
+                        .addArgument(record.getData())
+                        .setCause(ex)
+                        .log();
             }
         }
         return records;

diff --git a/.../src/main/java/org/opensearch/dataprepper/plugins/processor/flatten/FlattenProcessor.java b/.../src/main/java/org/opensearch/dataprepper/plugins/processor/flatten/FlattenProcessor.java
@@ -7,6 +7,7 @@
 
 import com.github.wnameless.json.flattener.JsonFlattener;
 import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
@@ -87,7 +88,7 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
 
                 updateEvent(recordEvent, flattenedJson);
             } catch (Exception e) {
-                LOG.error("Fail to perform flatten operation", e);
+                LOG.error(NOISY, "Fail to perform flatten operation", e);
                 recordEvent.getMetadata().addTags(config.getTagsOnFailure());
             } finally {
                 // Add temporarily deleted fields back

diff --git a/...ssor/src/main/java/org/opensearch/dataprepper/plugins/geoip/processor/GeoIPProcessor.java b/...ssor/src/main/java/org/opensearch/dataprepper/plugins/geoip/processor/GeoIPProcessor.java
@@ -8,6 +8,7 @@
 import io.micrometer.core.instrument.Counter;
 import org.opensearch.dataprepper.expression.ExpressionEvaluator;
 import org.opensearch.dataprepper.logging.DataPrepperMarkers;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
@@ -107,7 +108,7 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
     try (final GeoIPDatabaseReader geoIPDatabaseReader = BatchGeoIPDatabaseReader.decorate(geoIPProcessorService.getGeoIPDatabaseReader())) {
        processRecords(records, geoIPDatabaseReader);
     } catch (final Exception e) {
-      LOG.error("Encountered exception in geoip processor.", e);
+      LOG.error(NOISY, "Encountered exception in geoip processor.", e);
     }
     return records;
   }

diff --git a/...cessor/src/main/java/org/opensearch/dataprepper/plugins/processor/grok/GrokProcessor.java b/...cessor/src/main/java/org/opensearch/dataprepper/plugins/processor/grok/GrokProcessor.java
@@ -12,6 +12,8 @@
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.Timer;
 import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
@@ -21,6 +23,8 @@
 import org.opensearch.dataprepper.model.processor.AbstractProcessor;
 import org.opensearch.dataprepper.model.processor.Processor;
 import org.opensearch.dataprepper.model.record.Record;
+import static org.opensearch.dataprepper.plugins.processor.grok.GrokProcessorConfig.TOTAL_PATTERNS_ATTEMPTED_METADATA_KEY;
+import static org.opensearch.dataprepper.plugins.processor.grok.GrokProcessorConfig.TOTAL_TIME_SPENT_IN_GROK_METADATA_KEY;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -54,10 +58,6 @@
 import java.util.regex.PatternSyntaxException;
 import java.util.stream.Collectors;
 
-import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
-import static org.opensearch.dataprepper.plugins.processor.grok.GrokProcessorConfig.TOTAL_PATTERNS_ATTEMPTED_METADATA_KEY;
-import static org.opensearch.dataprepper.plugins.processor.grok.GrokProcessorConfig.TOTAL_TIME_SPENT_IN_GROK_METADATA_KEY;
-
 
 @SingleThread
 @DataPrepperPlugin(name = "grok", pluginType = Processor.class, pluginConfigurationType = GrokProcessorConfig.class)
@@ -153,11 +153,25 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
 
             } catch (final TimeoutException e) {
                 event.getMetadata().addTags(tagsOnTimeout);
-                LOG.error(EVENT, "Matching on record [{}] took longer than [{}] and timed out", record.getData(), grokProcessorConfig.getTimeoutMillis());
+                LOG.atError()
+                        .addMarker(EVENT)
+                        .addMarker(NOISY)
+                        .setMessage("Matching on record [{}] took longer than [{}] and timed out")
+                        .addArgument(record.getData())
+                        .addArgument(grokProcessorConfig.getTimeoutMillis())
+                        .log();
+
                 grokProcessingTimeoutsCounter.increment();
             } catch (final ExecutionException | InterruptedException | RuntimeException e) {
                 event.getMetadata().addTags(tagsOnMatchFailure);
-                LOG.error(EVENT, "An exception occurred when matching record [{}]", record.getData(), e);
+                LOG.atError()
+                        .addMarker(EVENT)
+                        .addMarker(NOISY)
+                        .setMessage("An exception occurred when matching record [{}]")
+                        .addArgument(record.getData())
+                        .setCause(e)
+                        .log();
+
                 grokProcessingErrorsCounter.increment();
             }
 

diff --git a/...main/java/org/opensearch/dataprepper/plugins/processor/mutateevent/AddEntryProcessor.java b/...main/java/org/opensearch/dataprepper/plugins/processor/mutateevent/AddEntryProcessor.java
@@ -6,6 +6,8 @@
 package org.opensearch.dataprepper.plugins.processor.mutateevent;
 
 import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
@@ -26,8 +28,6 @@
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
-import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
-
 @DataPrepperPlugin(name = "add_entries", pluginType = Processor.class, pluginConfigurationType = AddEntryProcessorConfig.class)
 public class AddEntryProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
     private static final Logger LOG = LoggerFactory.getLogger(AddEntryProcessor.class);
@@ -92,12 +92,28 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
                             }
                         }
                     } catch (Exception e) {
-                        LOG.error(EVENT, "Error adding entry to record [{}] with key [{}], metadataKey [{}], value_expression [{}] format [{}], value [{}]",
-                                recordEvent, entry.getKey(), entry.getMetadataKey(), entry.getValueExpression(), entry.getFormat(), entry.getValue(), e);
+                        LOG.atError()
+                                .addMarker(EVENT)
+                                .addMarker(NOISY)
+                                .setMessage("Error adding entry to record [{}] with key [{}], metadataKey [{}], value_expression [{}] format [{}], value [{}]")
+                                .addArgument(recordEvent)
+                                .addArgument(entry.getKey())
+                                .addArgument(entry.getMetadataKey())
+                                .addArgument(entry.getValueExpression())
+                                .addArgument(entry.getFormat())
+                                .addArgument(entry.getValue())
+                                .setCause(e)
+                                .log();
                     }
                 }
             } catch(final Exception e){
-                LOG.error(EVENT, "There was an exception while processing Event [{}]", recordEvent, e);
+                LOG.atError()
+                        .addMarker(EVENT)
+                        .addMarker(NOISY)
+                        .setMessage("There was an exception while processing Event [{}]")
+                        .addArgument(recordEvent)
+                        .setCause(e)
+                        .log();
             }
         }
 

diff --git a/...a/org/opensearch/dataprepper/plugins/processor/mutateevent/ConvertEntryTypeProcessor.java b/...a/org/opensearch/dataprepper/plugins/processor/mutateevent/ConvertEntryTypeProcessor.java
@@ -6,6 +6,8 @@
 package org.opensearch.dataprepper.plugins.processor.mutateevent;
 
 import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
@@ -24,8 +26,6 @@
 import java.util.List;
 import java.util.Objects;
 
-import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
-
 @DataPrepperPlugin(name = "convert_entry_type", pluginType = Processor.class, pluginConfigurationType = ConvertEntryTypeProcessorConfig.class)
 public class ConvertEntryTypeProcessor  extends AbstractProcessor<Record<Event>, Record<Event>> {
     private static final Logger LOG = LoggerFactory.getLogger(ConvertEntryTypeProcessor.class);
@@ -91,7 +91,13 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
                     }
                 }
             } catch (final Exception e) {
-                LOG.error(EVENT, "There was an exception while processing Event [{}]", recordEvent, e);
+                LOG.atError()
+                        .addMarker(EVENT)
+                        .addMarker(NOISY)
+                        .setMessage("There was an exception while processing Event [{}]")
+                        .addArgument(recordEvent)
+                        .setCause(e)
+                        .log();
                 recordEvent.getMetadata().addTags(tagsOnFailure);
             }
         }

diff --git a/...ain/java/org/opensearch/dataprepper/plugins/processor/mutateevent/CopyValueProcessor.java b/...ain/java/org/opensearch/dataprepper/plugins/processor/mutateevent/CopyValueProcessor.java
@@ -6,6 +6,8 @@
 package org.opensearch.dataprepper.plugins.processor.mutateevent;
 
 import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
@@ -24,8 +26,6 @@
 import java.util.Map;
 import java.util.Objects;
 
-import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
-
 @DataPrepperPlugin(name = "copy_values", pluginType = Processor.class, pluginConfigurationType = CopyValueProcessorConfig.class)
 public class CopyValueProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
     private static final Logger LOG = LoggerFactory.getLogger(CopyValueProcessor.class);
@@ -93,7 +93,13 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
                     }
                 }
             } catch (final Exception e) {
-                LOG.error(EVENT, "There was an exception while processing Event [{}]", recordEvent, e);
+                LOG.atError()
+                        .addMarker(EVENT)
+                        .addMarker(NOISY)
+                        .setMessage("There was an exception while processing Event [{}]")
+                        .addArgument(recordEvent)
+                        .setCause(e)
+                        .log();
             }
         }