Add missing streaming appendator metrics.

- Adds the missing streaming appendator metrics
for ingest/merge/time, ingest/merge/cpu and ingest/persists/cpu

Author: abhishekrb19

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

@@ -186,6 +186,21 @@ public void reportMaxSegmentHandoffTime(long maxSegmentHandoffTime)
     }
   }
 
+  public void setMergeTime(long elapsedMergeTimeMillis)
+  {
+    mergeTimeMillis.set(elapsedMergeTimeMillis);
+  }
+
+  public void setMergeCpuTime(long elapsedCpuTimeNanos)
+  {
+    mergeCpuTime.set(elapsedCpuTimeNanos);
+  }
+
+  public void setPersistCpuTime(long elpasedCpuTimeNanos)
+  {
+    persistCpuTime.set(elpasedCpuTimeNanos);
+  }
+
   public void markProcessingDone()
   {
     this.processingDone.set(true);

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

@@ -24,7 +24,6 @@
 import com.google.common.base.Function;
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
-import com.google.common.base.Stopwatch;
 import com.google.common.base.Supplier;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
@@ -46,6 +45,7 @@
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.Stopwatch;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.java.util.common.concurrent.ScheduledExecutors;
@@ -81,6 +81,7 @@
 import org.apache.druid.server.coordination.DataSegmentAnnouncer;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.utils.JvmUtils;
 import org.joda.time.Interval;
 
 import javax.annotation.Nullable;
@@ -682,6 +683,7 @@ public ListenableFuture<Object> persistAll(@Nullable final Committer committer)
           @Override
           public Object call() throws IOException
           {
+            final long startPersistCpuNanos = JvmUtils.safeGetThreadCpuTime();
             try {
               setTaskThreadContext();
               for (Pair<FireHydrant, SegmentIdWithShardSpec> pair : indexesToPersist) {
@@ -748,14 +750,15 @@ public Object call() throws IOException
             }
             finally {
               metrics.incrementNumPersists();
-              metrics.incrementPersistTimeMillis(persistStopwatch.elapsed(TimeUnit.MILLISECONDS));
+              metrics.incrementPersistTimeMillis(persistStopwatch.millisElapsed());
+              metrics.setPersistCpuTime(JvmUtils.safeGetThreadCpuTime() - startPersistCpuNanos);
               persistStopwatch.stop();
             }
           }
         }
     );
 
-    final long startDelay = runExecStopwatch.elapsed(TimeUnit.MILLISECONDS);
+    final long startDelay = runExecStopwatch.millisElapsed();
     metrics.incrementPersistBackPressureMillis(startDelay);
     if (startDelay > WARN_DELAY) {
       log.warn("Ingestion was throttled for [%,d] millis because persists were pending.", startDelay);
@@ -935,8 +938,9 @@ private DataSegmentWithMetadata mergeAndPush(
       }
 
       final File mergedFile;
-      final long mergeFinishTime;
-      final long startTime = System.nanoTime();
+      final Stopwatch stopwatch = Stopwatch.createStarted();
+      final long mergeTimeMillis;
+      final long startMergeCpuNanos = JvmUtils.safeGetThreadCpuTime();
       List<QueryableIndex> indexes = new ArrayList<>();
       Closer closer = Closer.create();
       try {
@@ -961,9 +965,11 @@ private DataSegmentWithMetadata mergeAndPush(
             tuningConfig.getMaxColumnsToMerge()
         );
 
-        mergeFinishTime = System.nanoTime();
-
-        log.debug("Segment[%s] built in %,dms.", identifier, (mergeFinishTime - startTime) / 1000000);
+        mergeTimeMillis = stopwatch.millisElapsed();
+        stopwatch.restart();
+        metrics.setMergeTime(mergeTimeMillis);
+        metrics.setMergeCpuTime(JvmUtils.safeGetThreadCpuTime() - startMergeCpuNanos);
+        log.debug("Segment[%s] built in %,dms.", identifier, mergeTimeMillis);
       }
       catch (Throwable t) {
         throw closer.rethrow(t);
@@ -979,8 +985,7 @@ private DataSegmentWithMetadata mergeAndPush(
       // dataSegmentPusher retries internally when appropriate; no need for retries here.
       final DataSegment segment = dataSegmentPusher.push(mergedFile, segmentToPush, useUniquePath);
 
-      final long pushFinishTime = System.nanoTime();
-
+      final long pushTimeMillis = stopwatch.millisElapsed();
       objectMapper.writeValue(descriptorFile, segment);
 
       log.info(
@@ -991,8 +996,8 @@ private DataSegmentWithMetadata mergeAndPush(
           identifier,
           segment.getSize(),
           indexes.size(),
-          (mergeFinishTime - startTime) / 1000000,
-          (pushFinishTime - mergeFinishTime) / 1000000,
+          mergeTimeMillis,
+          pushTimeMillis,
           objectMapper.writeValueAsString(segment.getLoadSpec())
       );
 

server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java

@@ -56,6 +56,7 @@
 import org.apache.druid.segment.incremental.RowIngestionMeters;
 import org.apache.druid.segment.incremental.SimpleRowIngestionMeters;
 import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig;
+import org.apache.druid.segment.realtime.SegmentGenerationMetrics;
 import org.apache.druid.segment.realtime.sink.Committers;
 import org.apache.druid.server.coordination.DataSegmentAnnouncer;
 import org.apache.druid.testing.InitializedNullHandlingTest;
@@ -102,8 +103,10 @@ public class StreamAppenderatorTest extends InitializedNullHandlingTest
   @Test
   public void testSimpleIngestion() throws Exception
   {
+    final SegmentGenerationMetrics segmentGenerationMetrics = new SegmentGenerationMetrics();
     try (final StreamAppenderatorTester tester =
              new StreamAppenderatorTester.Builder().maxRowsInMemory(2)
+                                                   .segmentGenerationMetrics(segmentGenerationMetrics)
                                                    .basePersistDirectory(temporaryFolder.newFolder())
                                                    .build()) {
       final Appenderator appenderator = tester.getAppenderator();
@@ -183,6 +186,14 @@ public SegmentIdWithShardSpec apply(DataSegment input)
       );
       Assert.assertEquals(sorted(tester.getPushedSegments()), sorted(segmentsAndCommitMetadata.getSegments()));
 
+      Assert.assertEquals(2, segmentGenerationMetrics.numPersists());
+      Assert.assertEquals(3, segmentGenerationMetrics.rowOutput());
+      Assert.assertTrue(segmentGenerationMetrics.persistTimeMillis() > 0);
+      Assert.assertTrue(segmentGenerationMetrics.persistCpuTime() > 0);
+
+      Assert.assertTrue(segmentGenerationMetrics.mergeTimeMillis() > 0);
+      Assert.assertTrue(segmentGenerationMetrics.mergeCpuTime() > 0);
+
       // clear
       appenderator.clear();
       Assert.assertTrue(appenderator.getSegments().isEmpty());
@@ -192,10 +203,12 @@ public SegmentIdWithShardSpec apply(DataSegment input)
   @Test
   public void testPushFailure() throws Exception
   {
+    final SegmentGenerationMetrics segmentGenerationMetrics = new SegmentGenerationMetrics();
     try (final StreamAppenderatorTester tester =
              new StreamAppenderatorTester.Builder().maxRowsInMemory(2)
                                                    .basePersistDirectory(temporaryFolder.newFolder())
                                                    .enablePushFailure(true)
+                                                   .segmentGenerationMetrics(segmentGenerationMetrics)
                                                    .build()) {
       final Appenderator appenderator = tester.getAppenderator();
       boolean thrown;
@@ -269,6 +282,8 @@ public void testPushFailure() throws Exception
               CoreMatchers.startsWith("Push failure test"))))
       );
     }
+
+    Assert.assertEquals(1, segmentGenerationMetrics.failedHandoffs());
   }
 
   @Test

server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java

@@ -115,7 +115,8 @@ public StreamAppenderatorTester(
       final ServiceEmitter serviceEmitter,
       final PolicyEnforcer policyEnforcer,
       final boolean releaseLocksOnHandoff,
-      final TaskIntervalUnlocker taskIntervalUnlocker
+      final TaskIntervalUnlocker taskIntervalUnlocker,
+      final SegmentGenerationMetrics segmentGenerationMetrics
   )
   {
     objectMapper = new DefaultObjectMapper();
@@ -165,7 +166,7 @@ public StreamAppenderatorTester(
         releaseLocksOnHandoff
     );
 
-    metrics = new SegmentGenerationMetrics();
+    metrics = segmentGenerationMetrics == null ? new SegmentGenerationMetrics() : segmentGenerationMetrics;
     queryExecutor = Execs.singleThreaded("queryExecutor(%d)");
 
     IndexIO indexIO = new IndexIO(
@@ -364,6 +365,7 @@ public static class Builder
     private PolicyEnforcer policyEnforcer = NoopPolicyEnforcer.instance();
     private boolean releaseLocksOnHandoff;
     private TaskIntervalUnlocker taskIntervalUnlocker = interval -> {};
+    private SegmentGenerationMetrics segmentGenerationMetrics;
 
     public Builder maxRowsInMemory(final int maxRowsInMemory)
     {
@@ -389,6 +391,12 @@ public Builder enablePushFailure(final boolean enablePushFailure)
       return this;
     }
 
+    public Builder segmentGenerationMetrics(final SegmentGenerationMetrics segmentGenerationMetrics)
+    {
+      this.segmentGenerationMetrics = segmentGenerationMetrics;
+      return this;
+    }
+
     public Builder rowIngestionMeters(final RowIngestionMeters rowIngestionMeters)
     {
       this.rowIngestionMeters = rowIngestionMeters;
@@ -446,7 +454,8 @@ public StreamAppenderatorTester build()
           serviceEmitter,
           policyEnforcer,
           releaseLocksOnHandoff,
-          taskIntervalUnlocker
+          taskIntervalUnlocker,
+          segmentGenerationMetrics
       );
     }
 
@@ -468,7 +477,8 @@ public StreamAppenderatorTester build(
           serviceEmitter,
           policyEnforcer,
           releaseLocksOnHandoff,
-          taskIntervalUnlocker
+          taskIntervalUnlocker,
+          segmentGenerationMetrics
       );
     }
   }