Enhance message gap metric to include min/max/avg aggregations

Author: jtuglu-netflix

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

@@ -90,13 +90,21 @@ public boolean doMonitor(ServiceEmitter emitter)
     }
     emitter.emit(builder.setMetric("ingest/events/unparseable", unparseable));
 
-    final long processedWithError = rowIngestionMetersTotals.getProcessedWithError() - previousRowIngestionMetersTotals.getProcessedWithError();
+    final long processedWithError = rowIngestionMetersTotals.getProcessedWithError()
+                                    - previousRowIngestionMetersTotals.getProcessedWithError();
     if (processedWithError > 0) {
-      log.error("[%,d] events processed with errors! Set logParseExceptions to true in the ingestion spec to log these errors.", processedWithError);
+      log.error(
+          "[%,d] events processed with errors! Set logParseExceptions to true in the ingestion spec to log these errors.",
+          processedWithError
+      );
     }
     emitter.emit(builder.setMetric("ingest/events/processedWithError", processedWithError));
 
-    emitter.emit(builder.setMetric("ingest/events/processed", rowIngestionMetersTotals.getProcessed() - previousRowIngestionMetersTotals.getProcessed()));
+    emitter.emit(builder.setMetric(
+        "ingest/events/processed",
+        rowIngestionMetersTotals.getProcessed()
+        - previousRowIngestionMetersTotals.getProcessed()
+    ));
 
     final long dedup = metrics.dedup() - previousSegmentGenerationMetrics.dedup();
     if (dedup > 0) {
@@ -110,27 +118,60 @@ public boolean doMonitor(ServiceEmitter emitter)
         )
     );
 
-    emitter.emit(builder.setMetric("ingest/rows/output", metrics.rowOutput() - previousSegmentGenerationMetrics.rowOutput()));
-    emitter.emit(builder.setMetric("ingest/persists/count", metrics.numPersists() - previousSegmentGenerationMetrics.numPersists()));
-    emitter.emit(builder.setMetric("ingest/persists/time", metrics.persistTimeMillis() - previousSegmentGenerationMetrics.persistTimeMillis()));
-    emitter.emit(builder.setMetric("ingest/persists/cpu", metrics.persistCpuTime() - previousSegmentGenerationMetrics.persistCpuTime()));
+    emitter.emit(builder.setMetric(
+        "ingest/rows/output",
+        metrics.rowOutput() - previousSegmentGenerationMetrics.rowOutput()
+    ));
+    emitter.emit(builder.setMetric(
+        "ingest/persists/count",
+        metrics.numPersists() - previousSegmentGenerationMetrics.numPersists()
+    ));
+    emitter.emit(builder.setMetric(
+        "ingest/persists/time",
+        metrics.persistTimeMillis() - previousSegmentGenerationMetrics.persistTimeMillis()
+    ));
+    emitter.emit(builder.setMetric(
+        "ingest/persists/cpu",
+        metrics.persistCpuTime() - previousSegmentGenerationMetrics.persistCpuTime()
+    ));
     emitter.emit(
         builder.setMetric(
             "ingest/persists/backPressure",
             metrics.persistBackPressureMillis() - previousSegmentGenerationMetrics.persistBackPressureMillis()
         )
     );
-    emitter.emit(builder.setMetric("ingest/persists/failed", metrics.failedPersists() - previousSegmentGenerationMetrics.failedPersists()));
-    emitter.emit(builder.setMetric("ingest/handoff/failed", metrics.failedHandoffs() - previousSegmentGenerationMetrics.failedHandoffs()));
-    emitter.emit(builder.setMetric("ingest/merge/time", metrics.mergeTimeMillis() - previousSegmentGenerationMetrics.mergeTimeMillis()));
-    emitter.emit(builder.setMetric("ingest/merge/cpu", metrics.mergeCpuTime() - previousSegmentGenerationMetrics.mergeCpuTime()));
-    emitter.emit(builder.setMetric("ingest/handoff/count", metrics.handOffCount() - previousSegmentGenerationMetrics.handOffCount()));
+    emitter.emit(builder.setMetric(
+        "ingest/persists/failed",
+        metrics.failedPersists() - previousSegmentGenerationMetrics.failedPersists()
+    ));
+    emitter.emit(builder.setMetric(
+        "ingest/handoff/failed",
+        metrics.failedHandoffs() - previousSegmentGenerationMetrics.failedHandoffs()
+    ));
+    emitter.emit(builder.setMetric(
+        "ingest/merge/time",
+        metrics.mergeTimeMillis() - previousSegmentGenerationMetrics.mergeTimeMillis()
+    ));
+    emitter.emit(builder.setMetric(
+        "ingest/merge/cpu",
+        metrics.mergeCpuTime() - previousSegmentGenerationMetrics.mergeCpuTime()
+    ));
+    emitter.emit(builder.setMetric(
+        "ingest/handoff/count",
+        metrics.handOffCount() - previousSegmentGenerationMetrics.handOffCount()
+    ));
     emitter.emit(builder.setMetric("ingest/sink/count", metrics.sinkCount()));
 
-    long messageGap = metrics.messageGap();
+    final long messageGap = metrics.messageGap();
     if (messageGap >= 0) {
       emitter.emit(builder.setMetric("ingest/events/messageGap", messageGap));
     }
+    final long numMessageGapEvent = metrics.numMessageGapEvent();
+    if (numMessageGapEvent > 0) {
+      emitter.emit(builder.setMetric("ingest/events/messageGapAvg", metrics.messageGapAvg()));
+      emitter.emit(builder.setMetric("ingest/events/minMessageGap", metrics.maxMessageGap()));
+      emitter.emit(builder.setMetric("ingest/events/maxMessageGap", metrics.minMessageGap()));
+    }
 
     long maxSegmentHandoffTime = metrics.maxSegmentHandoffTime();
     if (maxSegmentHandoffTime >= 0) {

server/src/main/java/org/apache/druid/segment/realtime/SegmentGenerationMetrics.java

@@ -53,6 +53,12 @@ public class SegmentGenerationMetrics
   private final AtomicLong messageGap = new AtomicLong(0);
   private final AtomicBoolean processingDone = new AtomicBoolean(false);
 
+  // Message gap accounting
+  private final AtomicLong minMessageGap = new AtomicLong(Long.MAX_VALUE);
+  private final AtomicLong maxMessageGap = new AtomicLong(0);
+  private final AtomicLong messageGapAvg = new AtomicLong(0); // millisecond-level precision
+  private final AtomicLong numMessageGapEvents = new AtomicLong(0);
+
   private final AtomicLong maxSegmentHandoffTime = new AtomicLong(NO_EMIT_SEGMENT_HANDOFF_TIME);
 
   public void incrementRowOutputCount(long numRows)
@@ -110,6 +116,14 @@ public void reportMessageMaxTimestamp(long messageMaxTimestamp)
     this.messageMaxTimestamp.set(Math.max(messageMaxTimestamp, this.messageMaxTimestamp.get()));
   }
 
+  public void reportMessageGapAggregate(long messageGap)
+  {
+    final long newNumEvents = this.numMessageGapEvents.incrementAndGet();
+    this.minMessageGap.getAndAccumulate(messageGap, Math::min);
+    this.maxMessageGap.getAndAccumulate(messageGap, Math::max);
+    this.messageGapAvg.getAndUpdate(oldAvg -> oldAvg + ((messageGap - oldAvg) / newNumEvents));
+  }
+
   public void reportMaxSegmentHandoffTime(long maxSegmentHandoffTime)
   {
     this.maxSegmentHandoffTime.set(Math.max(maxSegmentHandoffTime, this.maxSegmentHandoffTime.get()));
@@ -170,6 +184,7 @@ public long pushedRows()
   {
     return pushedRows.get();
   }
+
   public long mergeTimeMillis()
   {
     return mergeTimeMillis.get();
@@ -195,6 +210,26 @@ public long sinkCount()
     return sinkCount.get();
   }
 
+  public long maxMessageGap()
+  {
+    return maxMessageGap.get();
+  }
+
+  public long minMessageGap()
+  {
+    return minMessageGap.get();
+  }
+
+  public long messageGapAvg()
+  {
+    return messageGapAvg.get();
+  }
+
+  public long numMessageGapEvent()
+  {
+    return numMessageGapEvents.get();
+  }
+
   public long messageGap()
   {
     return messageGap.get();
@@ -219,8 +254,8 @@ public SegmentGenerationMetrics snapshot()
     retVal.mergeCpuTime.set(mergeCpuTime.get());
     retVal.persistCpuTime.set(persistCpuTime.get());
     retVal.handOffCount.set(handOffCount.get());
-    retVal.sinkCount.set(sinkCount.get());
     retVal.messageMaxTimestamp.set(messageMaxTimestamp.get());
+    retVal.sinkCount.set(sinkCount.get());
     retVal.maxSegmentHandoffTime.set(maxSegmentHandoffTime.get());
     retVal.mergedRows.set(mergedRows.get());
     retVal.pushedRows.set(pushedRows.get());
@@ -233,6 +268,10 @@ public SegmentGenerationMetrics snapshot()
       messageGapSnapshot = System.currentTimeMillis() - maxTimestamp;
     }
     retVal.messageGap.set(messageGapSnapshot);
+    retVal.numMessageGapEvents.set(numMessageGapEvents.get());
+    retVal.minMessageGap.set(minMessageGap.get());
+    retVal.maxMessageGap.set(maxMessageGap.get());
+    retVal.messageGapAvg.set(messageGapAvg.get());
 
     reset();
 
@@ -242,5 +281,9 @@ public SegmentGenerationMetrics snapshot()
   private void reset()
   {
     maxSegmentHandoffTime.set(NO_EMIT_SEGMENT_HANDOFF_TIME);
+    numMessageGapEvents.set(0);
+    messageGapAvg.set(0);
+    minMessageGap.set(Long.MAX_VALUE);
+    maxMessageGap.set(0);
   }
 }

server/src/main/java/org/apache/druid/segment/realtime/appenderator/AppenderatorConfig.java

@@ -76,4 +76,22 @@ default int getMaxColumnsToMerge()
   {
     return -1;
   }
+
+  /**
+   * Number of rows to skip when taking message gap aggregate samples.
+   * Has no effect unless `isAggregateMessageGapStatsEnabled` is true.
+   */
+  default int getMessageGapRowSampleSize()
+  {
+    return 1_000;
+  }
+
+  /**
+   * Whether to capture aggregate message gap metrics.
+   * Defaults to false.
+   */
+  default boolean isAggregateMessageGapStatsEnabled()
+  {
+    return false;
+  }
 }

server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java

@@ -121,6 +121,8 @@ public class BatchAppenderator implements Appenderator
   private final long maxBytesTuningConfig;
   private final boolean skipBytesInMemoryOverheadCheck;
   private final boolean useMaxMemoryEstimates;
+  private final int messageGapRowSampleSize;
+  private final boolean aggregateMessageGapStats;
 
   private volatile ListeningExecutorService persistExecutor = null;
   private volatile ListeningExecutorService pushExecutor = null;
@@ -190,6 +192,8 @@ public class BatchAppenderator implements Appenderator
     skipBytesInMemoryOverheadCheck = tuningConfig.isSkipBytesInMemoryOverheadCheck();
     maxPendingPersists = tuningConfig.getMaxPendingPersists();
     this.useMaxMemoryEstimates = useMaxMemoryEstimates;
+    this.aggregateMessageGapStats = tuningConfig.isAggregateMessageGapStatsEnabled();
+    this.messageGapRowSampleSize = tuningConfig.getMessageGapRowSampleSize();
     this.centralizedDatasourceSchemaConfig = centralizedDatasourceSchemaConfig;
     this.fingerprintGenerator = new FingerprintGenerator(objectMapper);
   }
@@ -287,6 +291,9 @@ public AppenderatorAddResult add(
     }
 
     final Sink sink = getOrCreateSink(identifier);
+    if (aggregateMessageGapStats && totalRows % messageGapRowSampleSize == 0) {
+      metrics.reportMessageGapAggregate(System.currentTimeMillis() - row.getTimestampFromEpoch());
+    }
     metrics.reportMessageMaxTimestamp(row.getTimestampFromEpoch());
     final int sinkRowsInMemoryBeforeAdd = sink.getNumRowsInMemory();
     final int sinkRowsInMemoryAfterAdd;
@@ -513,8 +520,10 @@ public void clear()
   private void clear(Map<SegmentIdWithShardSpec, Sink> sinksToClear, boolean removeOnDiskData)
   {
     // Drop commit metadata, then abandon all segments.
-    log.info("Clearing all[%d] sinks & their hydrants, removing data on disk: [%s]",
-             sinksToClear.size(), removeOnDiskData);
+    log.info(
+        "Clearing all[%d] sinks & their hydrants, removing data on disk: [%s]",
+        sinksToClear.size(), removeOnDiskData
+    );
     // Drop everything.
     Iterator<Map.Entry<SegmentIdWithShardSpec, Sink>> sinksIterator = sinksToClear.entrySet().iterator();
     sinksIterator.forEachRemaining(entry -> {
@@ -582,8 +591,9 @@ public ListenableFuture<Object> persistAll(@Nullable final Committer committer)
             // the invariant of exactly one, always swappable, sink with exactly one unpersisted hydrant must hold
             int totalHydrantsForSink = hydrants.size();
             if (totalHydrantsForSink != 1) {
-              throw new ISE("There should be only one hydrant for identifier[%s] but there are[%s]",
-                            identifier, totalHydrantsForSink
+              throw new ISE(
+                  "There should be only one hydrant for identifier[%s] but there are[%s]",
+                  identifier, totalHydrantsForSink
               );
             }
             totalHydrantsCount++;
@@ -635,8 +645,9 @@ public ListenableFuture<Object> persistAll(@Nullable final Committer committer)
             metrics.incrementPersistTimeMillis(persistMillis);
             persistStopwatch.stop();
             // make sure no push can start while persisting:
-            log.info("Persisted rows[%,d] and bytes[%,d] and removed all sinks & hydrants from memory in[%d] millis",
-                     numPersistedRows, bytesPersisted, persistMillis
+            log.info(
+                "Persisted rows[%,d] and bytes[%,d] and removed all sinks & hydrants from memory in[%d] millis",
+                numPersistedRows, bytesPersisted, persistMillis
             );
             log.info("Persist is done.");
           }
@@ -656,6 +667,7 @@ public ListenableFuture<Object> persistAll(@Nullable final Committer committer)
   /**
    * All sinks will be persisted so do a shallow copy of the Sinks map, reset
    * the map and metadata (i.e. memory consumption counters) so that ingestion can go on
+   *
    * @return The map of sinks to persist, this map will be garbage collected after
    * persist is complete since we will not be keeping a reference to it...
    */
@@ -740,8 +752,9 @@ public ListenableFuture<SegmentsAndCommitMetadata> push(
               log.warn("mergeAndPush[%s] returned null, skipping.", identifier);
             }
           }
-          log.info("Push done: total sinks merged[%d], total hydrants merged[%d]",
-                   identifiers.size(), totalHydrantsMerged
+          log.info(
+              "Push done: total sinks merged[%d], total hydrants merged[%d]",
+              identifiers.size(), totalHydrantsMerged
           );
           return new SegmentsAndCommitMetadata(dataSegments, commitMetadata, segmentSchemaMapping);
         },
@@ -753,8 +766,8 @@ public ListenableFuture<SegmentsAndCommitMetadata> push(
   /**
    * Merge segment, push to deep storage. Should only be used on segments that have been fully persisted.
    *
-   * @param identifier    sink identifier
-   * @param sink          sink to push
+   * @param identifier sink identifier
+   * @param sink       sink to push
    * @return segment descriptor along with schema, or null if the sink is no longer valid
    */
   private DataSegmentWithMetadata mergeAndPush(
@@ -784,8 +797,10 @@ private DataSegmentWithMetadata mergeAndPush(
     if (sm == null) {
       log.warn("Sink metadata not found just before merge for identifier [%s]", identifier);
     } else if (numHydrants != sm.getNumHydrants()) {
-      throw new ISE("Number of restored hydrants[%d] for identifier[%s] does not match expected value[%d]",
-                    numHydrants, identifier, sm.getNumHydrants());
+      throw new ISE(
+          "Number of restored hydrants[%d] for identifier[%s] does not match expected value[%d]",
+          numHydrants, identifier, sm.getNumHydrants()
+      );
     }
 
     try {
@@ -1260,12 +1275,14 @@ private int calculateSinkMemoryInUsed()
    */
   private static class SinkMetadata
   {
-    /** This is used to maintain the rows in the sink accross persists of the sink
+    /**
+     * This is used to maintain the rows in the sink accross persists of the sink
      * used for functionality (i.e. to detect whether an incremental push
      * is needed {@link AppenderatorDriverAddResult#isPushRequired(Integer, Long)}
      **/
     private int numRowsInSegment;
-    /** For sanity check as well as functionality: to make sure that all hydrants for a sink are restored from disk at
+    /**
+     * For sanity check as well as functionality: to make sure that all hydrants for a sink are restored from disk at
      * push time and also to remember the fire hydrant "count" when persisting it.
      */
     private int numHydrants;