Add reason dimension to ingest/events/thrownAway metric

Author: jtuglu1

docs/operations/metrics.md

@@ -270,7 +270,7 @@ batch ingestion emit the following metrics. These metrics are deltas for each em
 |`ingest/events/processed`|Number of events processed per emission period.|`dataSource`, `taskId`, `taskType`, `groupId`, `tags`|Equal to the number of events per emission period.|
 |`ingest/events/processedWithError`|Number of events processed with some partial errors per emission period. Events processed with partial errors are counted towards both this metric and `ingest/events/processed`.|`dataSource`, `taskId`, `taskType`, `groupId`, `tags`|0|
 |`ingest/events/unparseable`|Number of events rejected because the events are unparseable.|`dataSource`, `taskId`, `taskType`, `groupId`, `tags`|0|
-|`ingest/events/thrownAway`|Number of events rejected because they are null, or filtered by `transformSpec`, or outside one of `lateMessageRejectionPeriod`, `earlyMessageRejectionPeriod`.|`dataSource`, `taskId`, `taskType`, `groupId`, `tags`|0|
+|`ingest/events/thrownAway`|Number of events rejected because they are null, or filtered by `transformSpec`, or outside one of `lateMessageRejectionPeriod`, `earlyMessageRejectionPeriod`. The `reason` dimension indicates why the event was thrown away: `null` (null or empty event), `beforeMinMessageTime` (late message rejection), `afterMaxMessageTime` (early message rejection), or `filtered` (filtered by transformSpec).|`dataSource`, `taskId`, `taskType`, `groupId`, `tags`, `reason`|0|
 |`ingest/events/duplicate`|Number of events rejected because the events are duplicated.|`dataSource`, `taskId`, `taskType`, `groupId`, `tags`|0|
 |`ingest/input/bytes`|Number of bytes read from input sources, after decompression but prior to parsing. This covers all data read, including data that does not end up being fully processed and ingested. For example, this includes data that ends up being rejected for being unparseable or filtered out.|`dataSource`, `taskId`, `taskType`, `groupId`, `tags`|Depends on the amount of data read.|
 |`ingest/rows/output`|Number of Druid rows persisted.|`dataSource`, `taskId`, `taskType`, `groupId`|Your number of events with rollup.|

extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json

@@ -126,7 +126,8 @@
   },
   "ingest/events/thrownAway": {
     "dimensions": [
-      "dataSource"
+      "dataSource",
+      "reason"
     ],
     "type": "counter"
   },

extensions-contrib/opentsdb-emitter/src/main/resources/defaultMetrics.json

@@ -54,7 +54,8 @@
   "query/cache/total/timeouts": [],
   "query/cache/total/errors": [],
   "ingest/events/thrownAway": [
-    "dataSource"
+    "dataSource",
+    "reason"
   ],
   "ingest/events/unparseable": [
     "dataSource"

extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json

@@ -104,7 +104,7 @@
   "ingest/events/processed" : { "dimensions" : ["dataSource"], "type" : "count", "help": "Number of events successfully processed per emission period." },
   "ingest/events/processedWithError" : { "dimensions" : ["dataSource"], "type" : "count", "help": "Number of events processed with some partial errors per emission period." },
   "ingest/events/unparseable" : { "dimensions" : ["dataSource"], "type" : "count", "help": "Number of events rejected because the events are unparseable." },
-  "ingest/events/thrownAway" : { "dimensions" : ["dataSource"], "type" : "count", "help": "Number of events rejected because they are outside the windowPeriod."},
+  "ingest/events/thrownAway" : { "dimensions" : ["dataSource", "reason"], "type" : "count", "help": "Number of events rejected because they are null, filtered by transformSpec, or outside the message rejection periods. The reason dimension indicates why: null, beforeMinMessageTime, afterMaxMessageTime, or filtered."},
   "ingest/events/duplicate" : { "dimensions" : ["dataSource"], "type" : "count", "help": "Number of events rejected because the events are duplicated."},
   "ingest/input/bytes" : { "dimensions" : ["dataSource"], "type" : "count", "help": "Number of bytes read from input sources, after decompression but prior to parsing." },
   "ingest/rows/output" : { "dimensions" : ["dataSource"], "type" : "count", "help": "Number of Druid rows persisted."},

extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json

@@ -38,7 +38,7 @@
   "query/cache/total/timeouts" : { "dimensions" : [], "type" : "gauge" },
   "query/cache/total/errors" : { "dimensions" : [], "type" : "gauge" },
 
-  "ingest/events/thrownAway" : { "dimensions" : ["dataSource"], "type" : "count" },
+  "ingest/events/thrownAway" : { "dimensions" : ["dataSource", "reason"], "type" : "count" },
   "ingest/events/unparseable" : { "dimensions" : ["dataSource"], "type" : "count" },
   "ingest/events/duplicate" : { "dimensions" : ["dataSource"], "type" : "count" },
   "ingest/events/processed" : { "dimensions" : ["dataSource"], "type" : "count" },

indexing-service/src/main/java/org/apache/druid/indexing/common/stats/DropwizardRowIngestionMeters.java

@@ -23,7 +23,9 @@
 import com.codahale.metrics.MetricRegistry;
 import org.apache.druid.segment.incremental.RowIngestionMeters;
 import org.apache.druid.segment.incremental.RowIngestionMetersTotals;
+import org.apache.druid.segment.incremental.ThrownAwayReason;
 
+import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -33,11 +35,14 @@ public class DropwizardRowIngestionMeters implements RowIngestionMeters
   public static final String FIVE_MINUTE_NAME = "5m";
   public static final String FIFTEEN_MINUTE_NAME = "15m";
 
+  private static final int NUM_THROWN_AWAY_REASONS = ThrownAwayReason.values().length;
+
   private final Meter processed;
   private final Meter processedBytes;
   private final Meter processedWithError;
   private final Meter unparseable;
   private final Meter thrownAway;
+  private final Meter[] thrownAwayByReason = new Meter[NUM_THROWN_AWAY_REASONS];
 
   public DropwizardRowIngestionMeters()
   {
@@ -47,6 +52,9 @@ public DropwizardRowIngestionMeters()
     this.processedWithError = metricRegistry.meter(PROCESSED_WITH_ERROR);
     this.unparseable = metricRegistry.meter(UNPARSEABLE);
     this.thrownAway = metricRegistry.meter(THROWN_AWAY);
+    for (ThrownAwayReason reason : ThrownAwayReason.values()) {
+      this.thrownAwayByReason[reason.ordinal()] = metricRegistry.meter(THROWN_AWAY + "_" + reason.name());
+    }
   }
 
   @Override
@@ -109,6 +117,23 @@ public void incrementThrownAway()
     thrownAway.mark();
   }
 
+  @Override
+  public void incrementThrownAway(ThrownAwayReason reason)
+  {
+    thrownAway.mark();
+    thrownAwayByReason[reason.ordinal()].mark();
+  }
+
+  @Override
+  public Map<ThrownAwayReason, Long> getThrownAwayByReason()
+  {
+    EnumMap<ThrownAwayReason, Long> result = new EnumMap<>(ThrownAwayReason.class);
+    for (ThrownAwayReason reason : ThrownAwayReason.values()) {
+      result.put(reason, thrownAwayByReason[reason.ordinal()].getCount());
+    }
+    return result;
+  }
+
   @Override
   public RowIngestionMetersTotals getTotals()
   {

indexing-service/src/main/java/org/apache/druid/indexing/common/stats/TaskRealtimeMetricsMonitor.java

@@ -25,21 +25,27 @@
 import org.apache.druid.java.util.metrics.AbstractMonitor;
 import org.apache.druid.segment.incremental.RowIngestionMeters;
 import org.apache.druid.segment.incremental.RowIngestionMetersTotals;
+import org.apache.druid.segment.incremental.ThrownAwayReason;
 import org.apache.druid.segment.realtime.SegmentGenerationMetrics;
 
+import java.util.EnumMap;
+import java.util.Map;
+
 /**
  * Emits metrics from {@link SegmentGenerationMetrics} and {@link RowIngestionMeters}.
  */
 public class TaskRealtimeMetricsMonitor extends AbstractMonitor
 {
   private static final EmittingLogger log = new EmittingLogger(TaskRealtimeMetricsMonitor.class);
+  private static final String REASON_DIMENSION = "reason";
 
   private final SegmentGenerationMetrics segmentGenerationMetrics;
   private final RowIngestionMeters rowIngestionMeters;
   private final ServiceMetricEvent.Builder builder;
 
   private SegmentGenerationMetrics previousSegmentGenerationMetrics;
   private RowIngestionMetersTotals previousRowIngestionMetersTotals;
+  private Map<ThrownAwayReason, Long> previousThrownAwayByReason;
 
   public TaskRealtimeMetricsMonitor(
       SegmentGenerationMetrics segmentGenerationMetrics,
@@ -52,6 +58,7 @@ public TaskRealtimeMetricsMonitor(
     this.builder = metricEventBuilder;
     previousSegmentGenerationMetrics = new SegmentGenerationMetrics();
     previousRowIngestionMetersTotals = new RowIngestionMetersTotals(0, 0, 0, 0, 0);
+    previousThrownAwayByReason = new EnumMap<>(ThrownAwayReason.class);
   }
 
   @Override
@@ -60,14 +67,28 @@ public boolean doMonitor(ServiceEmitter emitter)
     SegmentGenerationMetrics metrics = segmentGenerationMetrics.snapshot();
     RowIngestionMetersTotals rowIngestionMetersTotals = rowIngestionMeters.getTotals();
 
-    final long thrownAway = rowIngestionMetersTotals.getThrownAway() - previousRowIngestionMetersTotals.getThrownAway();
-    if (thrownAway > 0) {
+    // Emit per-reason metrics with the reason dimension
+    final Map<ThrownAwayReason, Long> currentThrownAwayByReason = rowIngestionMeters.getThrownAwayByReason();
+    long totalThrownAway = 0;
+    for (ThrownAwayReason reason : ThrownAwayReason.values()) {
+      final long currentCount = currentThrownAwayByReason.getOrDefault(reason, 0L);
+      final long previousCount = previousThrownAwayByReason.getOrDefault(reason, 0L);
+      final long delta = currentCount - previousCount;
+      if (delta > 0) {
+        totalThrownAway += delta;
+        emitter.emit(
+            builder.setDimension(REASON_DIMENSION, reason.getMetricValue())
+                   .setMetric("ingest/events/thrownAway", delta)
+        );
+      }
+    }
+    previousThrownAwayByReason = currentThrownAwayByReason;
+    if (totalThrownAway > 0) {
       log.warn(
           "[%,d] events thrown away. Possible causes: null events, events filtered out by transformSpec, or events outside earlyMessageRejectionPeriod / lateMessageRejectionPeriod.",
-          thrownAway
+          totalThrownAway
       );
     }
-    emitter.emit(builder.setMetric("ingest/events/thrownAway", thrownAway));
 
     final long unparseable = rowIngestionMetersTotals.getUnparseable()
                              - previousRowIngestionMetersTotals.getUnparseable();

indexing-service/src/main/java/org/apache/druid/indexing/common/task/FilteringCloseableInputRowIterator.java

@@ -24,34 +24,49 @@
 import org.apache.druid.java.util.common.parsers.ParseException;
 import org.apache.druid.segment.incremental.ParseExceptionHandler;
 import org.apache.druid.segment.incremental.RowIngestionMeters;
+import org.apache.druid.segment.incremental.ThrownAwayReason;
 
 import java.io.IOException;
 import java.util.NoSuchElementException;
 import java.util.function.Predicate;
 
 /**
  * An {@link InputRow} iterator used by ingestion {@link Task}s. It can filter out rows which do not satisfy the given
- * {@link #filter} or throw {@link ParseException} while parsing them. The relevant metric should be counted whenever
+ * {@link RowFilter} or throw {@link ParseException} while parsing them. The relevant metric should be counted whenever
  * it filters out rows based on the filter. ParseException handling is delegatged to {@link ParseExceptionHandler}.
  */
 public class FilteringCloseableInputRowIterator implements CloseableIterator<InputRow>
 {
   private final CloseableIterator<InputRow> delegate;
-  private final Predicate<InputRow> filter;
+  private final RowFilter rowFilter;
   private final RowIngestionMeters rowIngestionMeters;
   private final ParseExceptionHandler parseExceptionHandler;
 
   private InputRow next;
 
+  /**
+   * @deprecated Use the constructor with {@link RowFilter} for better observability of thrown away reasons.
+   */
+  @Deprecated
   public FilteringCloseableInputRowIterator(
       CloseableIterator<InputRow> delegate,
       Predicate<InputRow> filter,
       RowIngestionMeters rowIngestionMeters,
       ParseExceptionHandler parseExceptionHandler
   )
+  {
+    this(delegate, RowFilter.fromPredicate(filter), rowIngestionMeters, parseExceptionHandler);
+  }
+
+  public FilteringCloseableInputRowIterator(
+      CloseableIterator<InputRow> delegate,
+      RowFilter rowFilter,
+      RowIngestionMeters rowIngestionMeters,
+      ParseExceptionHandler parseExceptionHandler
+  )
   {
     this.delegate = delegate;
-    this.filter = filter;
+    this.rowFilter = rowFilter;
     this.rowIngestionMeters = rowIngestionMeters;
     this.parseExceptionHandler = parseExceptionHandler;
   }
@@ -66,11 +81,12 @@ public boolean hasNext()
         while (next == null && delegate.hasNext()) {
           // delegate.next() can throw ParseException
           final InputRow row = delegate.next();
-          // filter.test() can throw ParseException
-          if (filter.test(row)) {
+          // rowFilter.test() can throw ParseException, returns null if accepted, or reason if rejected
+          final ThrownAwayReason rejectionReason = rowFilter.test(row);
+          if (rejectionReason == null) {
             next = row;
           } else {
-            rowIngestionMeters.incrementThrownAway();
+            rowIngestionMeters.incrementThrownAway(rejectionReason);
           }
         }
         break;

indexing-service/src/main/java/org/apache/druid/indexing/common/task/RowFilter.java

@@ -0,0 +1,68 @@
+/*
+ * 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.druid.indexing.common.task;
+
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.segment.incremental.ThrownAwayReason;
+
+import javax.annotation.Nullable;
+import java.util.function.Predicate;
+
+/**
+ * A filter for input rows during ingestion that can report the reason for rejection.
+ * This is similar to {@link Predicate} but returns the rejection reason instead of just a boolean.
+ */
+@FunctionalInterface
+public interface RowFilter
+{
+  /**
+   * Tests whether the given row should be accepted.
+   *
+   * @param row the input row to test
+   * @return null if the row should be accepted, or the {@link ThrownAwayReason} if the row should be rejected
+   */
+  @Nullable
+  ThrownAwayReason test(InputRow row);
+
+  /**
+   * Creates a RowFilter from a Predicate. When the predicate returns false,
+   * the rejection reason will be {@link ThrownAwayReason#FILTERED}.
+   */
+  static RowFilter fromPredicate(Predicate<InputRow> predicate)
+  {
+    return row -> predicate.test(row) ? null : ThrownAwayReason.FILTERED;
+  }
+
+  /**
+   * Combines this filter with another filter. A row is rejected if either filter rejects it.
+   * The rejection reason from the first rejecting filter (this filter first) is returned.
+   */
+  default RowFilter and(RowFilter other)
+  {
+    return row -> {
+      ThrownAwayReason reason = this.test(row);
+      if (reason != null) {
+        return reason;
+      }
+      return other.test(row);
+    };
+  }
+}
+

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java

@@ -66,6 +66,7 @@
 import org.apache.druid.indexing.common.actions.TaskLocks;
 import org.apache.druid.indexing.common.actions.TimeChunkLockAcquireAction;
 import org.apache.druid.indexing.common.stats.TaskRealtimeMetricsMonitor;
+import org.apache.druid.indexing.common.task.RowFilter;
 import org.apache.druid.indexing.input.InputRowSchemas;
 import org.apache.druid.indexing.seekablestream.common.OrderedPartitionableRecord;
 import org.apache.druid.indexing.seekablestream.common.OrderedSequenceNumber;
@@ -81,6 +82,7 @@
 import org.apache.druid.segment.incremental.ParseExceptionHandler;
 import org.apache.druid.segment.incremental.ParseExceptionReport;
 import org.apache.druid.segment.incremental.RowIngestionMeters;
+import org.apache.druid.segment.incremental.ThrownAwayReason;
 import org.apache.druid.segment.realtime.ChatHandler;
 import org.apache.druid.segment.realtime.SegmentGenerationMetrics;
 import org.apache.druid.segment.realtime.appenderator.Appenderator;
@@ -419,7 +421,7 @@ private TaskStatus runInternal(TaskToolbox toolbox) throws Exception
         inputRowSchema,
         task.getDataSchema().getTransformSpec(),
         toolbox.getIndexingTmpDir(),
-        row -> row != null && withinMinMaxRecordTime(row),
+        this::getRowRejectionReason,
         rowIngestionMeters,
         parseExceptionHandler
     );
@@ -2144,26 +2146,36 @@ private void refreshMinMaxMessageTime()
     );
   }
 
-  public boolean withinMinMaxRecordTime(final InputRow row)
+  /**
+   * Returns the rejection reason for a row, or null if the row should be accepted.
+   * This method is used as a {@link RowFilter} for the {@link StreamChunkParser}.
+   */
+  @Nullable
+  ThrownAwayReason getRowRejectionReason(final InputRow row)
   {
-    final boolean beforeMinimumMessageTime = minMessageTime.isAfter(row.getTimestamp());
-    final boolean afterMaximumMessageTime = maxMessageTime.isBefore(row.getTimestamp());
-
-    if (log.isDebugEnabled()) {
-      if (beforeMinimumMessageTime) {
+    if (row == null) {
+      return ThrownAwayReason.NULL;
+    }
+    if (minMessageTime.isAfter(row.getTimestamp())) {
+      if (log.isDebugEnabled()) {
         log.debug(
             "CurrentTimeStamp[%s] is before MinimumMessageTime[%s]",
             row.getTimestamp(),
             minMessageTime
         );
-      } else if (afterMaximumMessageTime) {
+      }
+      return ThrownAwayReason.BEFORE_MIN_MESSAGE_TIME;
+    }
+    if (maxMessageTime.isBefore(row.getTimestamp())) {
+      if (log.isDebugEnabled()) {
         log.debug(
             "CurrentTimeStamp[%s] is after MaximumMessageTime[%s]",
             row.getTimestamp(),
             maxMessageTime
         );
       }
+      return ThrownAwayReason.AFTER_MAX_MESSAGE_TIME;
     }
-    return !beforeMinimumMessageTime && !afterMaximumMessageTime;
+    return null;
   }
 }