Add feature to automatically remove datasource metadata based on retention period

docs/configuration/index.md

@@ -756,6 +756,9 @@ These Coordinator static configurations can be defined in the `coordinator/runti
 |`druid.coordinator.kill.rule.on`| Boolean value for whether to enable automatic deletion of rules. If set to true, Coordinator will periodically remove rules of inactive datasource (datasource with no used and unused segments) from the rule table in metadata storage.| No | False| 
 |`druid.coordinator.kill.rule.period`| How often to do automatic deletion of rules in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Value must be equal to or greater than  `druid.coordinator.period.metadataStoreManagementPeriod`. Only applies if `druid.coordinator.kill.rule.on` is set to "True".| No| `P1D`|
 |`druid.coordinator.kill.rule.durationToRetain`| Duration of rules to be retained from created time in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Only applies if `druid.coordinator.kill.rule.on` is set to "True".| Yes if `druid.coordinator.kill.rule.on` is set to "True".| None|
+|`druid.coordinator.kill.datasource.on`| Boolean value for whether to enable automatic deletion of datasource metadata (Note: datasource metadata only exists for datasource created from supervisor). If set to true, Coordinator will periodically remove datasource metadata of terminated supervisor from the datasource table in metadata storage.  | No | False| 
+|`druid.coordinator.kill.datasource.period`| How often to do automatic deletion of datasource metadata in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Value must be equal to or greater than  `druid.coordinator.period.metadataStoreManagementPeriod`. Only applies if `druid.coordinator.kill.datasource.on` is set to "True".| No| `P1D`|
+|`druid.coordinator.kill.datasource.durationToRetain`| Duration of datasource metadata to be retained from created time in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Only applies if `druid.coordinator.kill.datasource.on` is set to "True".| Yes if `druid.coordinator.kill.datasource.on` is set to "True".| None|
 
 ##### Segment Management
 |Property|Possible Values|Description|Default|

docs/operations/metrics.md

@@ -259,6 +259,7 @@ These metrics are for the Druid Coordinator and are reset each time the Coordina
 |`metadata/kill/supervisor/count`|Total number of terminated supervisors that were automatically deleted from metadata store per each Coordinator kill supervisor duty run. This metric can help adjust `druid.coordinator.kill.supervisor.durationToRetain` configuration based on whether more or less terminated supervisors need to be deleted per cycle. Note that this metric is only emitted when `druid.coordinator.kill.supervisor.on` is set to true.| |Varies.|
 |`metadata/kill/audit/count`|Total number of audit logs that were automatically deleted from metadata store per each Coordinator kill audit duty run. This metric can help adjust `druid.coordinator.kill.audit.durationToRetain` configuration based on whether more or less audit logs need to be deleted per cycle. Note that this metric is only emitted when `druid.coordinator.kill.audit.on` is set to true.| |Varies.|
 |`metadata/kill/rule/count`|Total number of rules that were automatically deleted from metadata store per each Coordinator kill rule duty run. This metric can help adjust `druid.coordinator.kill.rule.durationToRetain` configuration based on whether more or less rules need to be deleted per cycle. Note that this metric is only emitted when `druid.coordinator.kill.rule.on` is set to true.| |Varies.|
+|`metadata/kill/datasource/count`|Total number of datasource metadata that were automatically deleted from metadata store per each Coordinator kill datasource duty run (Note: datasource metadata only exists for datasource created from supervisor). This metric can help adjust `druid.coordinator.kill.datasource.durationToRetain` configuration based on whether more or less datasource metadata need to be deleted per cycle. Note that this metric is only emitted when `druid.coordinator.kill.datasource.on` is set to true.| |Varies.|
 
 
 If `emitBalancingStats` is set to `true` in the Coordinator [dynamic configuration](../configuration/index.md#dynamic-configuration), then [log entries](../configuration/logging.md) for class

indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java

@@ -34,6 +34,7 @@
 import org.apache.druid.timeline.partition.PartialShardSpec;
 import org.joda.time.Interval;
 
+import javax.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
@@ -139,6 +140,13 @@ public SegmentPublishResult commitMetadataOnly(
     throw new UnsupportedOperationException("Not implemented, no test uses this currently.");
   }
 
+  @Override
+  public int removeDataSourceMetadataOlderThan(long timestamp, @Nullable Set<String> excludeDatasources)
+  {
+    throw new UnsupportedOperationException("Not implemented, no test uses this currently.");
+  }
+
+
   @Override
   public SegmentIdWithShardSpec allocatePendingSegment(
       String dataSource,

server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java

@@ -26,6 +26,7 @@
 import org.joda.time.Interval;
 
 import javax.annotation.Nullable;
+import javax.validation.constraints.NotNull;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
@@ -267,6 +268,15 @@ SegmentPublishResult announceHistoricalSegments(
    */
   boolean insertDataSourceMetadata(String dataSource, DataSourceMetadata dataSourceMetadata);
 
+  /**
+   * Remove datasource metadata created before the given timestamp and not in given excludeDatasources set.
+   *
+   * @param timestamp timestamp in milliseconds
+   * @param excludeDatasources set of datasource names to exclude from removal
+   * @return number of datasource metadata removed
+   */
+  int removeDataSourceMetadataOlderThan(long timestamp, @NotNull Set<String> excludeDatasources);
+
   /**
    * Similar to {@link #announceHistoricalSegments(Set)}, but meant for streaming ingestion tasks for handling
    * the case where the task ingested no records and created no segments, but still needs to update the metadata

server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java

@@ -57,6 +57,7 @@
 import org.apache.druid.timeline.partition.PartitionChunk;
 import org.apache.druid.timeline.partition.PartitionIds;
 import org.apache.druid.timeline.partition.SingleDimensionShardSpec;
+import org.joda.time.DateTime;
 import org.joda.time.Interval;
 import org.joda.time.chrono.ISOChronology;
 import org.skife.jdbi.v2.Batch;
@@ -73,6 +74,7 @@
 import org.skife.jdbi.v2.util.ByteArrayMapper;
 
 import javax.annotation.Nullable;
+import javax.validation.constraints.NotNull;
 import java.io.IOException;
 import java.sql.ResultSet;
 import java.util.ArrayList;
@@ -1423,4 +1425,39 @@ public boolean insertDataSourceMetadata(String dataSource, DataSourceMetadata me
             .execute()
     );
   }
+
+  @Override
+  public int removeDataSourceMetadataOlderThan(long timestamp, @NotNull Set<String> excludeDatasources)
+  {
+    DateTime dateTime = DateTimes.utc(timestamp);
+    List<String> datasourcesToDelete = connector.getDBI().withHandle(
+        handle -> handle
+            .createQuery(
+                StringUtils.format(
+                    "SELECT dataSource FROM %1$s WHERE created_date < '%2$s'",
+                    dbTables.getDataSourceTable(),
+                    dateTime.toString()
+                )
+            )
+            .mapTo(String.class)
+            .list()
+    );
+    datasourcesToDelete.removeAll(excludeDatasources);
+    return connector.getDBI().withHandle(
+        handle -> {
+          final PreparedBatch batch = handle.prepareBatch(
+              StringUtils.format(
+                  "DELETE FROM %1$s WHERE dataSource = :dataSource AND created_date < '%2$s'",
+                  dbTables.getDataSourceTable(),
+                  dateTime.toString()
+              )
+          );
+          for (String datasource : datasourcesToDelete) {
+            batch.bind("dataSource", datasource).add();
+          }
+          int[] result = batch.execute();
+          return IntStream.of(result).sum();
+        }
+    );
+  }
 }

server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java

@@ -33,8 +33,27 @@ public interface MetadataSupervisorManager
 
   Map<String, List<VersionedSupervisorSpec>> getAll();
 
+  /**
+   * Return latest supervisors (both active and terminated)
+   *
+   * @return latest terminated supervisors
+   */
   Map<String, SupervisorSpec> getLatest();
 
+  /**
+   * Only return the latest active supervisors
+   *
+   * @return latest active supervisors
+   */
+  Map<String, SupervisorSpec> getLatestActiveOnly();
+
+  /**
+   * Only return the latest terminated supervisors
+   *
+   * @return latest terminated supervisors
+   */
+  Map<String, SupervisorSpec> getLatestTerminatedOnly();
+
   /**
    * Remove terminated supervisors created before the given timestamp.
    *

server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java

@@ -253,11 +253,41 @@ public Map<String, SupervisorSpec> fold(
     );
   }
 
+  @Override
+  public Map<String, SupervisorSpec> getLatestActiveOnly()
+  {
+    Map<String, SupervisorSpec> supervisors = getLatest();
+    Map<String, SupervisorSpec> activeSupervisors = new HashMap<>();
+    for (Map.Entry<String, SupervisorSpec> entry : supervisors.entrySet()) {
+      // Terminated supervisor will have it's latest supervisorSpec as NoopSupervisorSpec
+      // (NoopSupervisorSpec is used as a tombstone marker)
+      if (!(entry.getValue() instanceof NoopSupervisorSpec)) {
+        activeSupervisors.put(entry.getKey(), entry.getValue());
+      }
+    }
+    return ImmutableMap.copyOf(activeSupervisors);
+  }
+
+  @Override
+  public Map<String, SupervisorSpec> getLatestTerminatedOnly()
+  {
+    Map<String, SupervisorSpec> supervisors = getLatest();
+    Map<String, SupervisorSpec> activeSupervisors = new HashMap<>();
+    for (Map.Entry<String, SupervisorSpec> entry : supervisors.entrySet()) {
+      // Terminated supervisor will have it's latest supervisorSpec as NoopSupervisorSpec
+      // (NoopSupervisorSpec is used as a tombstone marker)
+      if (entry.getValue() instanceof NoopSupervisorSpec) {
+        activeSupervisors.put(entry.getKey(), entry.getValue());
+      }
+    }
+    return ImmutableMap.copyOf(activeSupervisors);
+  }
+
   @Override
   public int removeTerminatedSupervisorsOlderThan(long timestamp)
   {
     DateTime dateTime = DateTimes.utc(timestamp);
-    Map<String, SupervisorSpec> supervisors = getLatest();
+    Map<String, SupervisorSpec> terminatedSupervisors = getLatestTerminatedOnly();
     return dbi.withHandle(
         handle -> {
           final PreparedBatch batch = handle.prepareBatch(
@@ -267,13 +297,8 @@ public int removeTerminatedSupervisorsOlderThan(long timestamp)
                   dateTime.toString()
               )
           );
-          for (Map.Entry<String, SupervisorSpec> supervisor : supervisors.entrySet()) {
-            final SupervisorSpec spec = supervisor.getValue();
-            // Terminated supervisor will have it's latest supervisorSpec as NoopSupervisorSpec
-            // (NoopSupervisorSpec is used as a tombstone marker)
-            if (spec instanceof NoopSupervisorSpec) {
-              batch.bind("spec_id", supervisor.getKey()).add();
-            }
+          for (Map.Entry<String, SupervisorSpec> supervisor : terminatedSupervisors.entrySet()) {
+            batch.bind("spec_id", supervisor.getKey()).add();
           }
           int[] result = batch.execute();
           return IntStream.of(result).sum();

server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java

@@ -79,6 +79,14 @@ public abstract class DruidCoordinatorConfig
   @Default("PT-1s")
   public abstract Duration getCoordinatorRuleKillDurationToRetain();
 
+  @Config("druid.coordinator.kill.datasource.period")
+  @Default("P1D")
+  public abstract Duration getCoordinatorDatasourceKillPeriod();
+
+  @Config("druid.coordinator.kill.datasource.durationToRetain")
+  @Default("PT-1s")
+  public abstract Duration getCoordinatorDatasourceKillDurationToRetain();
+
   @Config("druid.coordinator.load.timeout")
   public Duration getLoadTimeoutDelay()
   {

server/src/main/java/org/apache/druid/server/coordinator/duty/KillAuditLog.java

@@ -51,6 +51,7 @@ public KillAuditLog(
     );
     this.retainDuration = config.getCoordinatorAuditKillDurationToRetain().getMillis();
     Preconditions.checkArgument(this.retainDuration >= 0, "coordinator audit kill retainDuration must be >= 0");
+    Preconditions.checkArgument(this.retainDuration < System.currentTimeMillis(), "Coordinator audit kill retainDuration cannot be greater than current time in ms");
     log.debug(
         "Audit Kill Task scheduling enabled with period [%s], retainDuration [%s]",
         this.period,
@@ -62,19 +63,24 @@ public KillAuditLog(
   @Override
   public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
   {
-    if ((lastKillTime + period) < System.currentTimeMillis()) {
-      lastKillTime = System.currentTimeMillis();
-
-      long timestamp = System.currentTimeMillis() - retainDuration;
-      int auditRemoved = auditManager.removeAuditLogsOlderThan(timestamp);
-      ServiceEmitter emitter = params.getEmitter();
-      emitter.emit(
-          new ServiceMetricEvent.Builder().build(
-              "metadata/kill/audit/count",
-              auditRemoved
-          )
-      );
-      log.info("Finished running KillAuditLog duty. Removed %,d audit logs", auditRemoved);
+    long currentTimeMillis = System.currentTimeMillis();
+    if ((lastKillTime + period) < currentTimeMillis) {
+      lastKillTime = currentTimeMillis;
+      long timestamp = currentTimeMillis - retainDuration;
+      try {
+        int auditRemoved = auditManager.removeAuditLogsOlderThan(timestamp);
+        ServiceEmitter emitter = params.getEmitter();
+        emitter.emit(
+            new ServiceMetricEvent.Builder().build(
+                "metadata/kill/audit/count",
+                auditRemoved
+            )
+        );
+        log.info("Finished running KillAuditLog duty. Removed %,d audit logs", auditRemoved);
+      }
+      catch (Exception e) {
+        log.error(e, "Failed to kill audit log");
+      }
     }
     return params;
   }

server/src/main/java/org/apache/druid/server/coordinator/duty/KillDatasourceMetadata.java

@@ -0,0 +1,121 @@
+/*
+ * 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.server.coordinator.duty;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.inject.Inject;
+import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataSupervisorManager;
+import org.apache.druid.server.coordinator.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * CoordinatorDuty for automatic deletion of datasource metadata from the datasource table in metadata storage.
+ * (Note: datasource metadata only exists for datasource created from supervisor).
+ * Note that this class relies on the supervisorSpec.getDataSources names to match with the
+ * 'datasource' column of the datasource metadata table.
+ */
+public class KillDatasourceMetadata implements CoordinatorDuty
+{
+  private static final Logger log = new Logger(KillDatasourceMetadata.class);
+
+  private final long period;
+  private final long retainDuration;
+  private long lastKillTime = 0;
+
+  private final IndexerMetadataStorageCoordinator indexerMetadataStorageCoordinator;
+  private final MetadataSupervisorManager metadataSupervisorManager;
+
+  @Inject
+  public KillDatasourceMetadata(
+      DruidCoordinatorConfig config,
+      IndexerMetadataStorageCoordinator indexerMetadataStorageCoordinator,
+      MetadataSupervisorManager metadataSupervisorManager
+  )
+  {
+    this.indexerMetadataStorageCoordinator = indexerMetadataStorageCoordinator;
+    this.metadataSupervisorManager = metadataSupervisorManager;
+    this.period = config.getCoordinatorDatasourceKillPeriod().getMillis();
+    Preconditions.checkArgument(
+        this.period >= config.getCoordinatorMetadataStoreManagementPeriod().getMillis(),
+        "Coordinator datasource metadata kill period must be >= druid.coordinator.period.metadataStoreManagementPeriod"
+    );
+    this.retainDuration = config.getCoordinatorDatasourceKillDurationToRetain().getMillis();
+    Preconditions.checkArgument(this.retainDuration >= 0, "Coordinator datasource metadata kill retainDuration must be >= 0");
+    Preconditions.checkArgument(this.retainDuration < System.currentTimeMillis(), "Coordinator datasource metadata kill retainDuration cannot be greater than current time in ms");
+    log.debug(
+        "Datasource Metadata Kill Task scheduling enabled with period [%s], retainDuration [%s]",
+        this.period,
+        this.retainDuration
+    );
+  }
+
+  @Override
+  public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
+  {
+    long currentTimeMillis = System.currentTimeMillis();
+    if ((lastKillTime + period) < currentTimeMillis) {
+      lastKillTime = currentTimeMillis;
+      long timestamp = currentTimeMillis - retainDuration;
+      try {
+        // Datasource metadata only exists for datasource with supervisor
+        // To determine if datasource metadata is still active, we check if the supervisor for that particular datasource
+        // is still active or not
+        Map<String, SupervisorSpec> allActiveSupervisor = metadataSupervisorManager.getLatestActiveOnly();
+        Set<String> allDatasourceWithActiveSupervisor = allActiveSupervisor.values()
+                                                                           .stream()
+                                                                           .map(supervisorSpec -> supervisorSpec.getDataSources())
+                                                                           .flatMap(Collection::stream)
+                                                                           .filter(datasource -> !Strings.isNullOrEmpty(datasource))
+                                                                           .collect(Collectors.toSet());
+        // We exclude removing datasource metadata with active supervisor
+        int datasourceMetadataRemovedCount = indexerMetadataStorageCoordinator.removeDataSourceMetadataOlderThan(
+            timestamp,
+            allDatasourceWithActiveSupervisor
+        );
+        ServiceEmitter emitter = params.getEmitter();
+        emitter.emit(
+            new ServiceMetricEvent.Builder().build(
+                "metadata/kill/datasource/count",
+                datasourceMetadataRemovedCount
+            )
+        );
+        log.info(
+            "Finished running KillDatasourceMetadata duty. Removed %,d datasource metadata",
+            datasourceMetadataRemovedCount
+        );
+      }
+      catch (Exception e) {
+        log.error(e, "Failed to kill datasource metadata");
+      }
+    }
+    return params;
+  }
+}