Add embedded kill tasks that run on the Overlord

extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentKiller.java

@@ -149,7 +149,7 @@ private boolean deleteKeysForBucket(
       try {
         deleteObjectsRequest.setKeys(chunkOfKeys);
         log.info(
-            "Removing from bucket: [%s] the following index files: [%s] from s3!",
+            "Deleting the following segment files from S3 bucket[%s]: [%s]",
             s3Bucket,
             keysToDeleteStrings
         );

indexing-service/pom.xml

@@ -267,6 +267,11 @@
             <artifactId>maven-resolver-api</artifactId>
             <version>1.3.1</version>
         </dependency>
+        <dependency>
+            <groupId>org.jdbi</groupId>
+            <artifactId>jdbi</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <profiles>

indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentNukeAction.java

@@ -27,16 +27,23 @@
 import org.apache.druid.indexing.common.task.Task;
 import org.apache.druid.indexing.overlord.CriticalAction;
 import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
 import org.apache.druid.query.DruidMetrics;
 import org.apache.druid.segment.SegmentUtils;
 import org.apache.druid.timeline.DataSegment;
+import org.joda.time.Interval;
 
 import java.util.Set;
 import java.util.stream.Collectors;
 
+/**
+ * Permanently deletes unused segments from the metadata store.
+ */
 public class SegmentNukeAction implements TaskAction<Void>
 {
+  private static final Logger log = new Logger(SegmentNukeAction.class);
+
   private final Set<DataSegment> segments;
 
   @JsonCreator
@@ -65,22 +72,25 @@ public Void perform(Task task, TaskActionToolbox toolbox)
     TaskLocks.checkLockCoversSegments(task, toolbox.getTaskLockbox(), segments);
 
     try {
-      toolbox.getTaskLockbox().doInCriticalSection(
+      final Set<Interval> intervals = segments.stream().map(DataSegment::getInterval).collect(Collectors.toSet());
+      int numDeletedSegments = toolbox.getTaskLockbox().doInCriticalSection(
           task,
-          segments.stream().map(DataSegment::getInterval).collect(Collectors.toSet()),
-          CriticalAction.builder()
-                        .onValidLocks(
-                            () -> {
-                              toolbox.getIndexerMetadataStorageCoordinator().deleteSegments(segments);
-                              return null;
-                            }
-                        )
-                        .onInvalidLocks(
-                            () -> {
-                              throw new ISE("Some locks for task[%s] are already revoked", task.getId());
-                            }
-                        )
-                        .build()
+          intervals,
+          CriticalAction.<Integer>builder().onValidLocks(
+              () -> toolbox.getIndexerMetadataStorageCoordinator().deleteSegments(segments)
+          ).onInvalidLocks(
+              () -> {
+                throw new ISE("Some locks for task[%s] are already revoked", task.getId());
+              }
+          ).build()
+      );
+
+      log.info(
+          "Deleted [%d] segments from metadata store out of requested[%d],"
+          + " across [%d] intervals[%s], for task[%s] of datasource[%s].",
+          numDeletedSegments, segments.size(),
+          intervals.size(), intervals,
+          task.getId(), task.getDataSource()
       );
     }
     catch (Exception e) {

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

@@ -212,7 +212,7 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
     int nextBatchSize = computeNextBatchSize(numSegmentsKilled);
     @Nullable Integer numTotalBatches = getNumTotalBatches();
     List<DataSegment> unusedSegments;
-    LOG.info(
+    logInfo(
         "Starting kill for datasource[%s] in interval[%s] and versions[%s] with batchSize[%d], up to limit[%d]"
         + " segments before maxUsedStatusLastUpdatedTime[%s] will be deleted%s",
         getDataSource(), getInterval(), getVersions(), batchSize, limit, maxUsedStatusLastUpdatedTime,
@@ -236,9 +236,7 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
         break;
       }
 
-      unusedSegments = toolbox.getTaskActionClient().submit(
-          new RetrieveUnusedSegmentsAction(getDataSource(), getInterval(), getVersions(), nextBatchSize, maxUsedStatusLastUpdatedTime)
-      );
+      unusedSegments = fetchNextBatchOfUnusedSegments(toolbox, nextBatchSize);
 
       // Fetch locks each time as a revokal could have occurred in between batches
       final NavigableMap<DateTime, List<TaskLock>> taskLockMap
@@ -283,29 +281,35 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
 
       // Nuke Segments
       taskActionClient.submit(new SegmentNukeAction(new HashSet<>(unusedSegments)));
+      emitMetric(toolbox.getEmitter(), TaskMetrics.SEGMENTS_DELETED_FROM_METADATA_STORE, unusedSegments.size());
 
       // Determine segments to be killed
       final List<DataSegment> segmentsToBeKilled
           = getKillableSegments(unusedSegments, upgradedFromSegmentIds, usedSegmentLoadSpecs, taskActionClient);
 
       final Set<DataSegment> segmentsNotKilled = new HashSet<>(unusedSegments);
       segmentsToBeKilled.forEach(segmentsNotKilled::remove);
-      LOG.infoSegments(
-          segmentsNotKilled,
-          "Skipping segment kill from deep storage as their load specs are referenced by other segments."
-      );
+
+      if (!segmentsNotKilled.isEmpty()) {
+        LOG.warn(
+            "Skipping kill of [%d] segments from deep storage as their load specs are used by other segments.",
+            segmentsNotKilled.size()
+        );
+      }
 
       toolbox.getDataSegmentKiller().kill(segmentsToBeKilled);
+      emitMetric(toolbox.getEmitter(), TaskMetrics.SEGMENTS_DELETED_FROM_DEEPSTORE, segmentsToBeKilled.size());
+
       numBatchesProcessed++;
       numSegmentsKilled += segmentsToBeKilled.size();
 
-      LOG.info("Processed [%d] batches for kill task[%s].", numBatchesProcessed, getId());
+      logInfo("Processed [%d] batches for kill task[%s].", numBatchesProcessed, getId());
 
       nextBatchSize = computeNextBatchSize(numSegmentsKilled);
     } while (!unusedSegments.isEmpty() && (null == numTotalBatches || numBatchesProcessed < numTotalBatches));
 
     final String taskId = getId();
-    LOG.info(
+    logInfo(
         "Finished kill task[%s] for dataSource[%s] and interval[%s]."
         + " Deleted total [%d] unused segments in [%d] batches.",
         taskId, getDataSource(), getInterval(), numSegmentsKilled, numBatchesProcessed
@@ -322,9 +326,8 @@ taskId, getDataSource(), getInterval(), numSegmentsKilled, numBatchesProcessed
   }
 
   @JsonIgnore
-  @VisibleForTesting
   @Nullable
-  Integer getNumTotalBatches()
+  protected Integer getNumTotalBatches()
   {
     return null != limit ? (int) Math.ceil((double) limit / batchSize) : null;
   }
@@ -336,6 +339,31 @@ int computeNextBatchSize(int numSegmentsKilled)
     return null != limit ? Math.min(limit - numSegmentsKilled, batchSize) : batchSize;
   }
 
+  /**
+   * Fetches the next batch of unused segments that are eligible for kill.
+   */
+  protected List<DataSegment> fetchNextBatchOfUnusedSegments(TaskToolbox toolbox, int nextBatchSize) throws IOException
+  {
+    return toolbox.getTaskActionClient().submit(
+        new RetrieveUnusedSegmentsAction(
+            getDataSource(),
+            getInterval(),
+            getVersions(),
+            nextBatchSize,
+            maxUsedStatusLastUpdatedTime
+        )
+    );
+  }
+
+  /**
+   * Logs the given info message. Exposed here to allow embedded kill tasks to
+   * suppress info logs.
+   */
+  protected void logInfo(String message, Object... args)
+  {
+    LOG.info(message, args);
+  }
+
   private NavigableMap<DateTime, List<TaskLock>> getNonRevokedTaskLockMap(TaskActionClient client) throws IOException
   {
     final NavigableMap<DateTime, List<TaskLock>> taskLockMap = new TreeMap<>();
@@ -385,6 +413,10 @@ private List<DataSegment> getKillableSegments(
         response.getUpgradedToSegmentIds().forEach((parent, children) -> {
           if (!CollectionUtils.isNullOrEmpty(children)) {
             // Do not kill segment if its parent or any of its siblings still exist in metadata store
+            LOG.info(
+                "Skipping kill of segments[%s] as its load spec is also used by segment IDs[%s].",
+                parentIdToUnusedSegments.get(parent), children
+            );
             parentIdToUnusedSegments.remove(parent);
           }
         });
@@ -402,10 +434,25 @@ private List<DataSegment> getKillableSegments(
     return parentIdToUnusedSegments.values()
                                    .stream()
                                    .flatMap(Set::stream)
-                                   .filter(segment -> !usedSegmentLoadSpecs.contains(segment.getLoadSpec()))
+                                   .filter(segment -> !isSegmentLoadSpecPresentIn(segment, usedSegmentLoadSpecs))
                                    .collect(Collectors.toList());
   }
 
+  /**
+   * @return true if the load spec of the segment is present in the given set of
+   * used load specs.
+   */
+  private boolean isSegmentLoadSpecPresentIn(
+      DataSegment segment,
+      Set<Map<String, Object>> usedSegmentLoadSpecs
+  )
+  {
+    boolean isPresent = usedSegmentLoadSpecs.contains(segment.getLoadSpec());
+    if (isPresent) {
+      LOG.info("Skipping kill of segment[%s] as its load spec is also used by other segments.", segment);
+    }
+    return isPresent;
+  }
 
   @Override
   public LookupLoadingSpec getLookupLoadingSpec()

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

@@ -0,0 +1,37 @@
+/*
+ * 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;
+
+/**
+ * Task-related metrics emitted by the Druid cluster.
+ */
+public class TaskMetrics
+{
+  private TaskMetrics()
+  {
+    // no instantiation
+  }
+
+  public static final String RUN_DURATION = "task/run/time";
+
+  public static final String SEGMENTS_DELETED_FROM_METADATA_STORE = "segment/killed/metadataStore/count";
+  public static final String SEGMENTS_DELETED_FROM_DEEPSTORE = "segment/killed/deepStorage/count";
+  public static final String FILES_DELETED_FROM_DEEPSTORE = "segment/killed/deepStorageFile/count";
+}

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

@@ -32,6 +32,12 @@ public class Tasks
   public static final int DEFAULT_BATCH_INDEX_TASK_PRIORITY = 50;
   public static final int DEFAULT_MERGE_TASK_PRIORITY = 25;
 
+  /**
+   * Priority of embedded kill tasks. Kept lower than batch and realtime tasks
+   * to allow them to preempt embbedded kill tasks.
+   */
+  public static final int DEFAULT_EMBEDDED_KILL_TASK_PRIORITY = 25;
+
   static {
     Verify.verify(DEFAULT_MERGE_TASK_PRIORITY == DataSourceCompactionConfig.DEFAULT_COMPACTION_TASK_PRIORITY);
   }

indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskLockbox.java

@@ -1232,16 +1232,17 @@ public void remove(final Task task)
   {
     giant.lock();
     try {
-      try {
-        log.info("Removing task[%s] from activeTasks", task.getId());
-        cleanupUpgradeAndPendingSegments(task);
-        unlockAll(task);
-      }
-      finally {
-        activeTasks.remove(task.getId());
+      if (!activeTasks.contains(task.getId())) {
+        return;
       }
+      log.info("Removing task[%s] from activeTasks", task.getId());
+      cleanupUpgradeAndPendingSegments(task);
+      unlockAll(task);
     }
     finally {
+      if (task != null) {
+        activeTasks.remove(task.getId());
+      }
       giant.unlock();
     }
   }