Add feature to automatically remove rules based on retention period

docs/configuration/index.md

@@ -750,6 +750,9 @@ These Coordinator static configurations can be defined in the `coordinator/runti
 |`druid.coordinator.kill.audit.on`| Boolean value for whether to enable automatic deletion of audit logs. If set to true, Coordinator will periodically remove audit logs from the audit table entries in metadata storage.| No | False| 
 |`druid.coordinator.kill.audit.period`| How often to do automatic deletion of audit logs in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Value must be greater than `druid.coordinator.period.metadataStoreManagementPeriod`. Only applies if `druid.coordinator.kill.audit.on` is set to True.| No| `P1D`|
 |`druid.coordinator.kill.audit.durationToRetain`| Duration of audit logs to be retained from created time in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Only applies if `druid.coordinator.kill.audit.on` is set to True.| Yes if `druid.coordinator.kill.audit.on` is set to True| None|
+|`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 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|
 
 ##### Segment Management
 |Property|Possible Values|Description|Default|

docs/operations/metrics.md

@@ -257,6 +257,7 @@ These metrics are for the Druid Coordinator and are reset each time the Coordina
 |`coordinator/time`|Approximate Coordinator duty runtime in milliseconds. The duty dimension is the string alias of the Duty that is being run.|duty.|Varies.|
 |`coordinator/global/time`|Approximate runtime of a full coordination cycle in milliseconds. The `dutyGroup` dimension indicates what type of coordination this run was. i.e. Historical Management vs Indexing|`dutyGroup`|Varies.|
 |`metadata/kill/audit/count`|Total number of audit logs automatically deleted from metadata store audit table per each Coordinator kill audit duty run. This metric can help adjust `druid.coordinator.kill.audit.durationToRetain` configuration based on if 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 automatically deleted from metadata store rule table per each Coordinator kill rule duty run. This metric can help adjust `druid.coordinator.kill.rule.durationToRetain` configuration based on if 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.|
 
 
 If `emitBalancingStats` is set to `true` in the Coordinator [dynamic configuration](../configuration/index.md#dynamic-configuration), then [log entries](../configuration/logging.md) for class

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

@@ -42,4 +42,12 @@ public interface MetadataRuleManager
   List<Rule> getRulesWithDefault(String dataSource);
 
   boolean overrideRule(String dataSource, List<Rule> rulesConfig, AuditInfo auditInfo);
+
+  /**
+   * Remove rules for non-existence datasource (datasource with no segment) created older than the given timestamp.
+   *
+   * @param timestamp timestamp in milliseconds
+   * @return number of rules removed
+   */
+  int removeRulesForEmptyDatasourcesOlderThan(long timestamp);
 }

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

@@ -48,6 +48,7 @@
 import org.skife.jdbi.v2.StatementContext;
 import org.skife.jdbi.v2.TransactionCallback;
 import org.skife.jdbi.v2.TransactionStatus;
+import org.skife.jdbi.v2.Update;
 import org.skife.jdbi.v2.tweak.HandleCallback;
 import org.skife.jdbi.v2.tweak.ResultSetMapper;
 
@@ -389,6 +390,8 @@ public Void inTransaction(Handle handle, TransactionStatus transactionStatus) th
                               .build(),
                     handle
                 );
+                // Note that the method removeRulesForEmptyDatasourcesOlderThan depends on the version field
+                // to be a timestamp
                 String version = auditTime.toString();
                 handle.createStatement(
                     StringUtils.format(
@@ -421,8 +424,40 @@ public Void inTransaction(Handle handle, TransactionStatus transactionStatus) th
     return true;
   }
 
+  @Override
+  public int removeRulesForEmptyDatasourcesOlderThan(long timestamp)
+  {
+    // Note that this DELETE SQL depends on the version field to be a timestamp. Hence, this
+    // method depends on overrideRule method to set version to timestamp when the rule entry is created
+    DateTime dateTime = DateTimes.utc(timestamp);
+    synchronized (lock) {
+      return dbi.withHandle(
+          handle -> {
+            Update sql = handle.createStatement(
+                // Note that this query could be expensive when the segments table is large
+                // However, since currently this query is run very infrequent (by default once a day by the KillRules Coordinator duty)
+                // and the inner query on segment table is a READ (no locking), it is keep this way.
+                StringUtils.format(
+                    "DELETE FROM %1$s WHERE datasource NOT IN (SELECT DISTINCT datasource from %2$s) and datasource!=:default_rule and version < :date_time",
+                    getRulesTable(),
+                    getSegmentsTable()
+                )
+            );
+            return sql.bind("default_rule", config.getDefaultRule())
+                      .bind("date_time", dateTime.toString())
+                      .execute();
+          }
+      );
+    }
+  }
+
   private String getRulesTable()
   {
     return dbTables.getRulesTable();
   }
+
+  private String getSegmentsTable()
+  {
+    return dbTables.getSegmentsTable();
+  }
 }

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

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

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

@@ -0,0 +1,76 @@
+/*
+ * 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.inject.Inject;
+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.server.coordinator.DruidCoordinatorConfig;
+import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
+
+public class KillRules implements CoordinatorDuty
+{
+  private static final Logger log = new Logger(KillRules.class);
+
+  private final long period;
+  private final long retainDuration;
+  private long lastKillTime = 0;
+
+  @Inject
+  public KillRules(
+      DruidCoordinatorConfig config
+  )
+  {
+    this.period = config.getCoordinatorRuleKillPeriod().getMillis();
+    Preconditions.checkArgument(
+        this.period >= config.getCoordinatorMetadataStoreManagementPeriod().getMillis(),
+        "coordinator rule kill period must be >= druid.coordinator.period.metadataStoreManagementPeriod"
+    );
+    this.retainDuration = config.getCoordinatorRuleKillDurationToRetain().getMillis();
+    Preconditions.checkArgument(this.retainDuration >= 0, "coordinator rule kill retainDuration must be >= 0");
+    log.debug(
+        "Rule Kill Task scheduling enabled with period [%s], retainDuration [%s]",
+        this.period,
+        this.retainDuration
+    );
+  }
+
+  @Override
+  public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams params)
+  {
+    if ((lastKillTime + period) < System.currentTimeMillis()) {
+      lastKillTime = System.currentTimeMillis();
+
+      long timestamp = System.currentTimeMillis() - retainDuration;
+      int ruleRemoved = params.getDatabaseRuleManager().removeRulesForEmptyDatasourcesOlderThan(timestamp);
+      ServiceEmitter emitter = params.getEmitter();
+      emitter.emit(
+          new ServiceMetricEvent.Builder().build(
+              "metadata/kill/rule/count",
+              ruleRemoved
+          )
+      );
+      log.info("Finished running KillRules duty. Removed %,d rule", ruleRemoved);
+    }
+    return params;
+  }
+}

server/src/test/java/org/apache/druid/metadata/SQLMetadataRuleManagerTest.java

@@ -23,19 +23,24 @@
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Suppliers;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import org.apache.druid.audit.AuditEntry;
 import org.apache.druid.audit.AuditInfo;
 import org.apache.druid.audit.AuditManager;
 import org.apache.druid.client.DruidServer;
 import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.segment.TestHelper;
 import org.apache.druid.server.audit.SQLAuditManager;
 import org.apache.druid.server.audit.SQLAuditManagerConfig;
 import org.apache.druid.server.coordinator.rules.IntervalLoadRule;
 import org.apache.druid.server.coordinator.rules.Rule;
 import org.apache.druid.server.metrics.NoopServiceEmitter;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.partition.NoneShardSpec;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -55,8 +60,9 @@ public class SQLMetadataRuleManagerTest
   private MetadataStorageTablesConfig tablesConfig;
   private SQLMetadataRuleManager ruleManager;
   private AuditManager auditManager;
+  private SQLMetadataSegmentPublisher publisher;
   private final ObjectMapper mapper = new DefaultObjectMapper();
-
+  private final ObjectMapper jsonMapper = TestHelper.makeJsonMapper();
 
   @Before
   public void setUp()
@@ -80,6 +86,12 @@ public void setUp()
         connector,
         auditManager
     );
+    connector.createSegmentTable();
+    publisher = new SQLMetadataSegmentPublisher(
+        jsonMapper,
+        derbyConnectorRule.metadataTablesConfigSupplier().get(),
+        connector
+    );
   }
 
   @Test
@@ -201,6 +213,143 @@ public void testFetchAuditEntriesForAllDataSources() throws Exception
     }
   }
 
+  @Test
+  public void testRemoveRulesOlderThanWithNonExistenceDatasourceAndOlderThanTimestampShouldDelete()
+  {
+    List<Rule> rules = ImmutableList.of(
+        new IntervalLoadRule(
+            Intervals.of("2015-01-01/2015-02-01"), ImmutableMap.of(
+            DruidServer.DEFAULT_TIER,
+            DruidServer.DEFAULT_NUM_REPLICANTS
+        )
+        )
+    );
+    AuditInfo auditInfo = new AuditInfo("test_author", "test_comment", "127.0.0.1");
+    ruleManager.overrideRule(
+        "test_dataSource",
+        rules,
+        auditInfo
+    );
+    // Verify that rule was added
+    ruleManager.poll();
+    Map<String, List<Rule>> allRules = ruleManager.getAllRules();
+    Assert.assertEquals(1, allRules.size());
+    Assert.assertEquals(1, allRules.get("test_dataSource").size());
+
+    // Now delete rules
+    ruleManager.removeRulesForEmptyDatasourcesOlderThan(System.currentTimeMillis());
+
+    // Verify that rule was deleted
+    ruleManager.poll();
+    allRules = ruleManager.getAllRules();
+    Assert.assertEquals(0, allRules.size());
+  }
+
+  @Test
+  public void testRemoveRulesOlderThanWithNonExistenceDatasourceAndNewerThanTimestampShouldNotDelete()
+  {
+    List<Rule> rules = ImmutableList.of(
+        new IntervalLoadRule(
+            Intervals.of("2015-01-01/2015-02-01"), ImmutableMap.of(
+            DruidServer.DEFAULT_TIER,
+            DruidServer.DEFAULT_NUM_REPLICANTS
+        )
+        )
+    );
+    AuditInfo auditInfo = new AuditInfo("test_author", "test_comment", "127.0.0.1");
+    ruleManager.overrideRule(
+        "test_dataSource",
+        rules,
+        auditInfo
+    );
+    // Verify that rule was added
+    ruleManager.poll();
+    Map<String, List<Rule>> allRules = ruleManager.getAllRules();
+    Assert.assertEquals(1, allRules.size());
+    Assert.assertEquals(1, allRules.get("test_dataSource").size());
+
+    // This will not delete the rule as the rule was created just now so it will have the created timestamp later than
+    // the timestamp 2012-01-01T00:00:00Z
+    ruleManager.removeRulesForEmptyDatasourcesOlderThan(DateTimes.of("2012-01-01T00:00:00Z").getMillis());
+
+    // Verify that rule was not deleted
+    ruleManager.poll();
+    allRules = ruleManager.getAllRules();
+    Assert.assertEquals(1, allRules.size());
+    Assert.assertEquals(1, allRules.get("test_dataSource").size());
+  }
+
+  @Test
+  public void testRemoveRulesOlderThanWithActiveDatasourceShouldNotDelete() throws Exception
+  {
+    List<Rule> rules = ImmutableList.of(
+        new IntervalLoadRule(
+            Intervals.of("2015-01-01/2015-02-01"), ImmutableMap.of(
+            DruidServer.DEFAULT_TIER,
+            DruidServer.DEFAULT_NUM_REPLICANTS
+        )
+        )
+    );
+    AuditInfo auditInfo = new AuditInfo("test_author", "test_comment", "127.0.0.1");
+    ruleManager.overrideRule(
+        "test_dataSource",
+        rules,
+        auditInfo
+    );
+
+    // Verify that rule was added
+    ruleManager.poll();
+    Map<String, List<Rule>> allRules = ruleManager.getAllRules();
+    Assert.assertEquals(1, allRules.size());
+    Assert.assertEquals(1, allRules.get("test_dataSource").size());
+
+    // Add segment metadata to segment table so that the datasource is considered active
+    DataSegment dataSegment = new DataSegment(
+        "test_dataSource",
+        Intervals.of("2015-01-01/2015-02-01"),
+        "1",
+        ImmutableMap.of(
+            "type", "s3_zip",
+            "bucket", "test",
+            "key", "test_dataSource/xxx"
+        ),
+        ImmutableList.of("dim1", "dim2", "dim3"),
+        ImmutableList.of("count", "value"),
+        NoneShardSpec.instance(),
+        1,
+        1234L
+    );
+    publisher.publishSegment(dataSegment);
+
+    // This will not delete the rule as the datasource has segment in the segment metadata table
+    ruleManager.removeRulesForEmptyDatasourcesOlderThan(System.currentTimeMillis());
+
+    // Verify that rule was not deleted
+    ruleManager.poll();
+    allRules = ruleManager.getAllRules();
+    Assert.assertEquals(1, allRules.size());
+    Assert.assertEquals(1, allRules.get("test_dataSource").size());
+  }
+
+  @Test
+  public void testRemoveRulesOlderThanShouldNotDeleteDefault()
+  {
+    // Create the default rule
+    ruleManager.start();
+    // Verify the default rule
+    ruleManager.poll();
+    Map<String, List<Rule>> allRules = ruleManager.getAllRules();
+    Assert.assertEquals(1, allRules.size());
+    Assert.assertEquals(1, allRules.get("_default").size());
+    // Delete everything
+    ruleManager.removeRulesForEmptyDatasourcesOlderThan(System.currentTimeMillis());
+    // Verify the default rule was not deleted
+    ruleManager.poll();
+    allRules = ruleManager.getAllRules();
+    Assert.assertEquals(1, allRules.size());
+    Assert.assertEquals(1, allRules.get("_default").size());
+  }
+
   @After
   public void cleanup()
   {

server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java

@@ -174,6 +174,8 @@ public void setUp() throws Exception
         null,
         null,
         null,
+        null,
+        null,
         10,
         new Duration("PT0s")
     );

server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java

@@ -146,6 +146,8 @@ public void setUp() throws Exception
         null,
         null,
         null,
+        null,
+        null,
         10,
         new Duration("PT0s")
     );