emit handoff count metrics

docs/content/operations/metrics.md

@@ -86,6 +86,8 @@ Memcached client metrics are reported as per the following. These metrics come d
 
 ## Ingestion Metrics
 
+These metrics are only available if the RealtimeMetricsMonitor is included in the monitors list for the Realtime node. These metrics are deltas for each emission period.
+
 |Metric|Description|Dimensions|Normal Value|
 |------|-----------|----------|------------|
 |`ingest/events/thrownAway`|Number of events rejected because they are outside the windowPeriod.|dataSource.|0|
@@ -100,6 +102,7 @@ Memcached client metrics are reported as per the following. These metrics come d
 |`ingest/handoff/failed`|Number of handoffs that failed.|dataSource.|0|
 |`ingest/merge/time`|Milliseconds spent merging intermediate segments|dataSource.|Depends on configuration. Generally a few minutes at most.|
 |`ingest/merge/cpu`|Cpu time in Nanoseconds spent on merging intermediate segments.|dataSource.|Depends on configuration. Generally a few minutes at most.|
+|`ingest/handoff/count`|Number of handoffs that happened.|dataSource.|Varies. Generally greater than 0 once every segment granular period if cluster operating normally|
 
 Note: If the JVM does not support CPU time measurement for the current thread, ingest/merge/cpu and ingest/persists/cpu will be 0. 
 

server/src/main/java/io/druid/segment/realtime/FireDepartmentMetrics.java

@@ -39,6 +39,7 @@ public class FireDepartmentMetrics
   private final AtomicLong mergeTimeMillis = new AtomicLong(0);
   private final AtomicLong mergeCpuTime = new AtomicLong(0);
   private final AtomicLong persistCpuTime = new AtomicLong(0);
+  private final AtomicLong handOffCount = new AtomicLong(0);
 
   public void incrementProcessed()
   {
@@ -98,6 +99,10 @@ public class FireDepartmentMetrics
     persistCpuTime.addAndGet(persistTime);
   }
 
+  public void incrementHandOffCount(){
+    handOffCount.incrementAndGet();
+  }
+
   public long processed()
   {
     return processedCount.get();
@@ -158,6 +163,11 @@ public class FireDepartmentMetrics
     return persistCpuTime.get();
   }
 
+  public long handOffCount()
+  {
+    return handOffCount.get();
+  }
+
 
   public FireDepartmentMetrics snapshot()
   {
@@ -174,6 +184,7 @@ public class FireDepartmentMetrics
     retVal.mergeTimeMillis.set(mergeTimeMillis.get());
     retVal.mergeCpuTime.set(mergeCpuTime.get());
     retVal.persistCpuTime.set(persistCpuTime.get());
+    retVal.handOffCount.set(handOffCount.get());
     return retVal;
   }
 
@@ -198,6 +209,7 @@ public class FireDepartmentMetrics
     mergeTimeMillis.addAndGet(otherSnapshot.mergeTimeMillis());
     mergeCpuTime.addAndGet(otherSnapshot.mergeCpuTime());
     persistCpuTime.addAndGet(otherSnapshot.persistCpuTime());
+    handOffCount.addAndGet(otherSnapshot.handOffCount());
     return this;
   }
 

server/src/main/java/io/druid/segment/realtime/RealtimeMetricsMonitor.java

@@ -95,6 +95,7 @@ public class RealtimeMetricsMonitor extends AbstractMonitor
       emitter.emit(builder.build("ingest/handoff/failed", metrics.failedHandoffs() - previous.failedHandoffs()));
       emitter.emit(builder.build("ingest/merge/time", metrics.mergeTimeMillis() - previous.mergeTimeMillis()));
       emitter.emit(builder.build("ingest/merge/cpu", metrics.mergeCpuTime() - previous.mergeCpuTime()));
+      emitter.emit(builder.build("ingest/handoff/count", metrics.handOffCount() - previous.handOffCount()));
       previousValues.put(fireDepartment, metrics);
     }
 

server/src/main/java/io/druid/segment/realtime/plumber/RealtimePlumber.java

@@ -592,6 +592,7 @@ public class RealtimePlumber implements Plumber
           public void run()
           {
             abandonSegment(sink.getInterval().getStartMillis(), sink);
+            metrics.incrementHandOffCount();
           }
         }
     );