Add metrics for number of segments generated per task in MSQ

apache · Sep 13, 2023 · d286380 · d286380
1 parent 0f38a37
commit d286380
Show file tree

Hide file tree

Showing 4 changed files with 23 additions and 0 deletions.
diff --git a/...ons-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerContext.java b/...ons-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerContext.java
@@ -25,6 +25,7 @@
 import org.apache.druid.indexing.common.TaskReport;
 import org.apache.druid.indexing.common.actions.TaskActionClient;
 import org.apache.druid.java.util.common.io.Closer;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.server.DruidNode;
 
 import java.util.Map;
@@ -36,6 +37,8 @@
  */
 public interface ControllerContext
 {
+  ServiceEmitter emitter();
+
   ObjectMapper jsonMapper();
 
   /**

diff --git a/...nsions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java b/...nsions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java
@@ -1330,6 +1330,7 @@ private void publishAllSegments(final Set<DataSegment> segments) throws IOExcept
     final DataSourceMSQDestination destination =
         (DataSourceMSQDestination) task.getQuerySpec().getDestination();
     final Set<DataSegment> segmentsWithTombstones = new HashSet<>(segments);
+    int tombstoneSize = 0;
 
     if (destination.isReplaceTimeChunks()) {
       final List<Interval> intervalsToDrop = findIntervalsToDrop(Preconditions.checkNotNull(segments, "segments"));
@@ -1344,6 +1345,7 @@ private void publishAllSegments(final Set<DataSegment> segments) throws IOExcept
               destination.getSegmentGranularity()
           );
           segmentsWithTombstones.addAll(tombstones);
+          tombstoneSize = tombstones.size();
         }
         catch (IllegalStateException e) {
           throw new MSQException(e, InsertLockPreemptedFault.instance());
@@ -1389,6 +1391,10 @@ private void publishAllSegments(final Set<DataSegment> segments) throws IOExcept
           SegmentTransactionalInsertAction.appendAction(segments, null, null)
       );
     }
+
+    task.emitMetric(context.emitter(), "ingest/tombstones/count", tombstoneSize);
+    // segments count metric is documented to include tombstones
+    task.emitMetric(context.emitter(), "ingest/segments/count", segmentsWithTombstones.size());
   }
 
   /**

diff --git a/...lti-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerControllerContext.java b/...lti-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerControllerContext.java
@@ -28,6 +28,7 @@
 import org.apache.druid.indexing.common.TaskToolbox;
 import org.apache.druid.indexing.common.actions.TaskActionClient;
 import org.apache.druid.java.util.common.io.Closer;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.msq.exec.Controller;
 import org.apache.druid.msq.exec.ControllerContext;
 import org.apache.druid.msq.exec.WorkerClient;
@@ -67,6 +68,12 @@ public IndexerControllerContext(
     this.workerManager = new IndexerWorkerManagerClient(overlordClient);
   }
 
+  @Override
+  public ServiceEmitter emitter()
+  {
+    return toolbox.getEmitter();
+  }
+
   @Override
   public ObjectMapper jsonMapper()
   {

diff --git a/...e/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestControllerContext.java b/...e/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestControllerContext.java
@@ -36,6 +36,7 @@
 import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.java.util.common.io.Closer;
 import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
 import org.apache.druid.msq.exec.Controller;
 import org.apache.druid.msq.exec.ControllerContext;
 import org.apache.druid.msq.exec.Worker;
@@ -215,6 +216,12 @@ public void close()
     }
   };
 
+  @Override
+  public ServiceEmitter emitter()
+  {
+    return null;
+  }
+
   @Override
   public ObjectMapper jsonMapper()
   {