From 3007e27df66492d7d1f7a31954e28aa13a5b8672 Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Tue, 27 Apr 2021 18:12:15 -0700 Subject: [PATCH 1/6] add auto clean up --- docs/configuration/index.md | 3 +++ docs/operations/metrics.md | 1 + .../metadata/MetadataSupervisorManager.java | 8 +++++++ .../SQLMetadataSupervisorManager.java | 24 +++++++++++++++++++ 4 files changed, 36 insertions(+) diff --git a/docs/configuration/index.md b/docs/configuration/index.md index d49bf75675fa..2bfe76269e99 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -747,6 +747,9 @@ These Coordinator static configurations can be defined in the `coordinator/runti |Property|Description|Required?|Default| |--------|-----------|---------|-------| |`druid.coordinator.period.metadataStoreManagementPeriod`|How often to run metadata management tasks in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. |No | `PT1H`| +|`druid.coordinator.kill.supervisor.on`| Boolean value for whether to enable automatic deletion of supervisors. If set to true, Coordinator will periodically remove terminated supervisors from the supervisor table in metadata storage.| No | False| +|`druid.coordinator.kill.supervisor.period`| How often to do automatic deletion of supervisor 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.supervisor.on` is set to True.| No| `P1D`| +|`druid.coordinator.kill.supervisor.durationToRetain`| Duration of terminated supervisor to be retained from created time in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Only applies if `druid.coordinator.kill.supervisor.on` is set to True.| Yes if `druid.coordinator.kill.supervisor.on` is set to True| None| |`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| diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md index 5a2d5fb76fd7..9c24c8b31dc8 100644 --- a/docs/operations/metrics.md +++ b/docs/operations/metrics.md @@ -256,6 +256,7 @@ These metrics are for the Druid Coordinator and are reset each time the Coordina |`interval/skipCompact/count`|Total number of intervals of this datasource that are skipped (not eligible for auto compaction) by the auto compaction.|datasource.|Varies.| |`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/supervisor/count`|Total number of supervisors automatically deleted from metadata store supervisor table per each Coordinator kill supervisor duty run. This metric can help adjust `druid.coordinator.kill.supervisor.durationToRetain` configuration based on if more or less 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 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.| diff --git a/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java b/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java index 3803316b3d98..71dea2cdbb21 100644 --- a/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java +++ b/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java @@ -34,4 +34,12 @@ public interface MetadataSupervisorManager Map> getAll(); Map getLatest(); + + /** + * Remove terminated supervisor created older than the given timestamp. + * + * @param timestamp timestamp in milliseconds + * @return number of supervisor removed + */ + int removeRulesOlderThan(long timestamp); } diff --git a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java index 52aee9218116..52ba5e52355d 100644 --- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java +++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java @@ -35,11 +35,13 @@ import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.lifecycle.LifecycleStart; import org.apache.druid.java.util.common.logger.Logger; +import org.joda.time.DateTime; import org.skife.jdbi.v2.FoldController; import org.skife.jdbi.v2.Folder3; import org.skife.jdbi.v2.Handle; import org.skife.jdbi.v2.IDBI; import org.skife.jdbi.v2.StatementContext; +import org.skife.jdbi.v2.Update; import org.skife.jdbi.v2.tweak.HandleCallback; import org.skife.jdbi.v2.tweak.ResultSetMapper; @@ -249,6 +251,28 @@ public Map fold( ); } + @Override + public int removeRulesOlderThan(long timestamp) + { + DateTime dateTime = DateTimes.utc(timestamp); + synchronized (lock) { + return dbi.withHandle( + handle -> { + Update sql = handle.createStatement( + 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 getSupervisorsTable() { return dbTables.get().getSupervisorTable(); From 1e4f0b5eb84bd43cba2f987bfda9211956fc169c Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Tue, 4 May 2021 23:43:48 -0700 Subject: [PATCH 2/6] add test --- docs/configuration/index.md | 4 +- docs/operations/metrics.md | 2 +- .../metadata/MetadataSupervisorManager.java | 2 +- .../SQLMetadataSupervisorManager.java | 40 +++++---- .../coordinator/DruidCoordinatorConfig.java | 8 ++ .../coordinator/duty/KillSupervisors.java | 81 +++++++++++++++++++ .../CuratorDruidCoordinatorTest.java | 2 + .../coordinator/DruidCoordinatorTest.java | 2 + .../coordinator/HttpLoadQueuePeonTest.java | 2 + .../server/coordinator/LoadQueuePeonTest.java | 6 ++ .../coordinator/LoadQueuePeonTester.java | 2 + .../TestDruidCoordinatorConfig.java | 18 +++++ .../duty/KillUnusedSegmentsTest.java | 2 + .../org/apache/druid/cli/CliCoordinator.java | 6 ++ 14 files changed, 157 insertions(+), 20 deletions(-) create mode 100644 server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java diff --git a/docs/configuration/index.md b/docs/configuration/index.md index 2bfe76269e99..131852565b4b 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -747,8 +747,8 @@ These Coordinator static configurations can be defined in the `coordinator/runti |Property|Description|Required?|Default| |--------|-----------|---------|-------| |`druid.coordinator.period.metadataStoreManagementPeriod`|How often to run metadata management tasks in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. |No | `PT1H`| -|`druid.coordinator.kill.supervisor.on`| Boolean value for whether to enable automatic deletion of supervisors. If set to true, Coordinator will periodically remove terminated supervisors from the supervisor table in metadata storage.| No | False| -|`druid.coordinator.kill.supervisor.period`| How often to do automatic deletion of supervisor 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.supervisor.on` is set to True.| No| `P1D`| +|`druid.coordinator.kill.supervisor.on`| Boolean value for whether to enable automatic deletion of terminated supervisors. If set to true, Coordinator will periodically remove terminated supervisors from the supervisor table in metadata storage.| No | False| +|`druid.coordinator.kill.supervisor.period`| How often to do automatic deletion of terminated supervisor 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.supervisor.on` is set to True.| No| `P1D`| |`druid.coordinator.kill.supervisor.durationToRetain`| Duration of terminated supervisor to be retained from created time in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Only applies if `druid.coordinator.kill.supervisor.on` is set to True.| Yes if `druid.coordinator.kill.supervisor.on` is set to True| None| |`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`| diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md index 9c24c8b31dc8..5c7860a432f7 100644 --- a/docs/operations/metrics.md +++ b/docs/operations/metrics.md @@ -256,7 +256,7 @@ These metrics are for the Druid Coordinator and are reset each time the Coordina |`interval/skipCompact/count`|Total number of intervals of this datasource that are skipped (not eligible for auto compaction) by the auto compaction.|datasource.|Varies.| |`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/supervisor/count`|Total number of supervisors automatically deleted from metadata store supervisor table per each Coordinator kill supervisor duty run. This metric can help adjust `druid.coordinator.kill.supervisor.durationToRetain` configuration based on if more or less 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/supervisor/count`|Total number of terminated supervisors automatically deleted from metadata store supervisor table per each Coordinator kill supervisor duty run. This metric can help adjust `druid.coordinator.kill.supervisor.durationToRetain` configuration based on if 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 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.| diff --git a/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java b/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java index 71dea2cdbb21..39553ea99259 100644 --- a/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java +++ b/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java @@ -41,5 +41,5 @@ public interface MetadataSupervisorManager * @param timestamp timestamp in milliseconds * @return number of supervisor removed */ - int removeRulesOlderThan(long timestamp); + int removeTerminatedSupervisorsOlderThan(long timestamp); } diff --git a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java index 52ba5e52355d..e9e5c85c6b03 100644 --- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java +++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java @@ -28,6 +28,7 @@ import com.google.inject.Inject; import org.apache.druid.guice.ManageLifecycle; import org.apache.druid.guice.annotations.Json; +import org.apache.druid.indexing.overlord.supervisor.NoopSupervisorSpec; import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec; import org.apache.druid.indexing.overlord.supervisor.VersionedSupervisorSpec; import org.apache.druid.java.util.common.DateTimes; @@ -252,27 +253,34 @@ public Map fold( } @Override - public int removeRulesOlderThan(long timestamp) + public int removeTerminatedSupervisorsOlderThan(long timestamp) { + int removedCount = 0; DateTime dateTime = DateTimes.utc(timestamp); - synchronized (lock) { - return dbi.withHandle( - handle -> { - Update sql = handle.createStatement( - 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(); - } - ); + Map supervisors = getLatest(); + for (Map.Entry supervisor : supervisors.entrySet()) { + final SupervisorSpec spec = supervisor.getValue(); + if (spec instanceof NoopSupervisorSpec) { + dbi.withHandle( + handle -> { + Update sql = handle.createStatement( + StringUtils.format( + "DELETE FROM %1$s WHERE spec_id = :spec_id AND created_date < :date_time", + getSupervisorsTable() + ) + ); + return sql.bind("spec_id", supervisor.getKey()) + .bind("date_time", dateTime.toString()) + .execute(); + } + ); + removedCount++; + } } + return removedCount; } + private String getSupervisorsTable() { return dbTables.get().getSupervisorTable(); diff --git a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java index 933b974d3ad0..f484b4f2b6b6 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinatorConfig.java @@ -55,6 +55,14 @@ public abstract class DruidCoordinatorConfig @Default("0") public abstract int getCoordinatorKillMaxSegments(); + @Config("druid.coordinator.kill.supervisor.period") + @Default("P1D") + public abstract Duration getCoordinatorSupervisorKillPeriod(); + + @Config("druid.coordinator.kill.supervisor.durationToRetain") + @Default("PT-1s") + public abstract Duration getCoordinatorSupervisorKillDurationToRetain(); + @Config("druid.coordinator.kill.audit.period") @Default("P1D") public abstract Duration getCoordinatorAuditKillPeriod(); diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java new file mode 100644 index 000000000000..b55a9cf37d78 --- /dev/null +++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java @@ -0,0 +1,81 @@ +/* + * 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.metadata.MetadataSupervisorManager; +import org.apache.druid.server.coordinator.DruidCoordinatorConfig; +import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams; + +public class KillSupervisors implements CoordinatorDuty +{ + private static final Logger log = new Logger(KillSupervisors.class); + + private final long period; + private final long retainDuration; + private long lastKillTime = 0; + + private final MetadataSupervisorManager metadataSupervisorManager; + + @Inject + public KillSupervisors( + DruidCoordinatorConfig config, + MetadataSupervisorManager metadataSupervisorManager + ) + { + this.metadataSupervisorManager = metadataSupervisorManager; + this.period = config.getCoordinatorSupervisorKillPeriod().getMillis(); + Preconditions.checkArgument( + this.period >= config.getCoordinatorMetadataStoreManagementPeriod().getMillis(), + "Coordinator supervisor kill period must be >= druid.coordinator.period.metadataStoreManagementPeriod" + ); + this.retainDuration = config.getCoordinatorSupervisorKillDurationToRetain().getMillis(); + Preconditions.checkArgument(this.retainDuration >= 0, "Coordinator supervisor kill retainDuration must be >= 0"); + log.debug( + "Supervisor 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 supervisorRemoved = metadataSupervisorManager.removeTerminatedSupervisorsOlderThan(timestamp); + ServiceEmitter emitter = params.getEmitter(); + emitter.emit( + new ServiceMetricEvent.Builder().build( + "metadata/kill/supervisor/count", + supervisorRemoved + ) + ); + log.info("Finished running KillSupervisors duty. Removed %,d supervisor", supervisorRemoved); + } + return params; + } +} diff --git a/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java b/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java index a8deb667dd05..402c92be203e 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java +++ b/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") ); diff --git a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java index 3aaef647c418..a1c3d033118d 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java +++ b/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") ); diff --git a/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java b/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java index 7b07d7ffa61c..607fb8dc3fcd 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java @@ -83,6 +83,8 @@ public class HttpLoadQueuePeonTest null, null, null, + null, + null, 10, Duration.ZERO ) diff --git a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java index 24e68063b6a7..a607423bb9a2 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java @@ -98,6 +98,8 @@ public void testMultipleLoadDropSegments() throws Exception null, null, null, + null, + null, 10, Duration.millis(0) ) @@ -296,6 +298,8 @@ public void testFailAssignForNonTimeoutFailures() throws Exception null, null, null, + null, + null, 10, new Duration("PT1s") ) @@ -351,6 +355,8 @@ public void testFailAssignForLoadDropTimeout() throws Exception null, null, null, + null, + null, 10, new Duration("PT1s") ) diff --git a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java index 5185452779c3..5540a117d4be 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java @@ -47,6 +47,8 @@ public LoadQueuePeonTester() null, null, null, + null, + null, 10, new Duration("PT1s") ) diff --git a/server/src/test/java/org/apache/druid/server/coordinator/TestDruidCoordinatorConfig.java b/server/src/test/java/org/apache/druid/server/coordinator/TestDruidCoordinatorConfig.java index 135f8d0557e5..a3d847d82f3e 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/TestDruidCoordinatorConfig.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/TestDruidCoordinatorConfig.java @@ -31,6 +31,8 @@ public class TestDruidCoordinatorConfig extends DruidCoordinatorConfig private final Duration loadTimeoutDelay; private final Duration coordinatorKillPeriod; private final Duration coordinatorKillDurationToRetain; + private final Duration coordinatorSupervisorKillPeriod; + private final Duration coordinatorSupervisorKillDurationToRetain; private final Duration coordinatorAuditKillPeriod; private final Duration coordinatorAuditKillDurationToRetain; private final Duration getLoadQueuePeonRepeatDelay; @@ -44,6 +46,8 @@ public TestDruidCoordinatorConfig( Duration loadTimeoutDelay, Duration coordinatorKillPeriod, Duration coordinatorKillDurationToRetain, + Duration coordinatorSupervisorKillPeriod, + Duration coordinatorSupervisorKillDurationToRetain, Duration coordinatorAuditKillPeriod, Duration coordinatorAuditKillDurationToRetain, int coordinatorKillMaxSegments, @@ -57,6 +61,8 @@ public TestDruidCoordinatorConfig( this.loadTimeoutDelay = loadTimeoutDelay; this.coordinatorKillPeriod = coordinatorKillPeriod; this.coordinatorKillDurationToRetain = coordinatorKillDurationToRetain; + this.coordinatorSupervisorKillPeriod = coordinatorSupervisorKillPeriod; + this.coordinatorSupervisorKillDurationToRetain = coordinatorSupervisorKillDurationToRetain; this.coordinatorAuditKillPeriod = coordinatorAuditKillPeriod; this.coordinatorAuditKillDurationToRetain = coordinatorAuditKillDurationToRetain; this.coordinatorKillMaxSegments = coordinatorKillMaxSegments; @@ -99,6 +105,18 @@ public Duration getCoordinatorKillDurationToRetain() return coordinatorKillDurationToRetain; } + @Override + public Duration getCoordinatorSupervisorKillPeriod() + { + return coordinatorSupervisorKillPeriod; + } + + @Override + public Duration getCoordinatorSupervisorKillDurationToRetain() + { + return coordinatorSupervisorKillDurationToRetain; + } + @Override public Duration getCoordinatorAuditKillPeriod() { diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java index cd766660dfbe..4a1aa0960761 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java @@ -111,6 +111,8 @@ private void testFindIntervalForKill(List segmentIntervals, Interval e Duration.parse("PT86400S"), null, null, + null, + null, 1000, Duration.ZERO ) diff --git a/services/src/main/java/org/apache/druid/cli/CliCoordinator.java b/services/src/main/java/org/apache/druid/cli/CliCoordinator.java index 9e3c65b3cf1f..cbba94a2573f 100644 --- a/services/src/main/java/org/apache/druid/cli/CliCoordinator.java +++ b/services/src/main/java/org/apache/druid/cli/CliCoordinator.java @@ -73,6 +73,7 @@ import org.apache.druid.server.coordinator.LoadQueueTaskMaster; import org.apache.druid.server.coordinator.duty.CoordinatorDuty; import org.apache.druid.server.coordinator.duty.KillAuditLog; +import org.apache.druid.server.coordinator.duty.KillSupervisors; import org.apache.druid.server.coordinator.duty.KillUnusedSegments; import org.apache.druid.server.http.ClusterResource; import org.apache.druid.server.http.CompactionResource; @@ -251,6 +252,11 @@ public void configure(Binder binder) CoordinatorDuty.class, CoordinatorMetadataStoreManagementDuty.class ); + conditionalMetadataStoreManagementDutyMultibind.addConditionBinding( + "druid.coordinator.kill.rule.on", + Predicates.equalTo("true"), + KillSupervisors.class + ); conditionalMetadataStoreManagementDutyMultibind.addConditionBinding( "druid.coordinator.kill.audit.on", Predicates.equalTo("true"), From 42e16ec88bbcedb1fb482c70f5a4facb9b0bbe8f Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Wed, 5 May 2021 00:46:53 -0700 Subject: [PATCH 3/6] add test --- .../SQLMetadataSupervisorManager.java | 3 +- .../SQLMetadataSupervisorManagerTest.java | 77 +++++++++ .../CuratorDruidCoordinatorTest.java | 2 + .../coordinator/DruidCoordinatorTest.java | 2 + .../coordinator/HttpLoadQueuePeonTest.java | 2 + .../server/coordinator/LoadQueuePeonTest.java | 6 + .../coordinator/LoadQueuePeonTester.java | 2 + .../coordinator/duty/KillAuditLogTest.java | 8 + .../coordinator/duty/KillRulesTest.java | 8 + .../coordinator/duty/KillSupervisorsTest.java | 155 ++++++++++++++++++ .../duty/KillUnusedSegmentsTest.java | 2 + 11 files changed, 265 insertions(+), 2 deletions(-) create mode 100644 server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsTest.java diff --git a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java index e9e5c85c6b03..c409f38a4209 100644 --- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java +++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java @@ -261,7 +261,7 @@ public int removeTerminatedSupervisorsOlderThan(long timestamp) for (Map.Entry supervisor : supervisors.entrySet()) { final SupervisorSpec spec = supervisor.getValue(); if (spec instanceof NoopSupervisorSpec) { - dbi.withHandle( + removedCount += dbi.withHandle( handle -> { Update sql = handle.createStatement( StringUtils.format( @@ -274,7 +274,6 @@ public int removeTerminatedSupervisorsOlderThan(long timestamp) .execute(); } ); - removedCount++; } } return removedCount; diff --git a/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java b/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java index 5c40757f61a5..789d8d01f4b8 100644 --- a/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java +++ b/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java @@ -21,12 +21,15 @@ 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.indexing.overlord.supervisor.NoopSupervisorSpec; import org.apache.druid.indexing.overlord.supervisor.Supervisor; import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec; import org.apache.druid.indexing.overlord.supervisor.VersionedSupervisorSpec; import org.apache.druid.indexing.overlord.supervisor.autoscaler.SupervisorTaskAutoScaler; import org.apache.druid.jackson.DefaultObjectMapper; +import org.apache.druid.java.util.common.DateTimes; import org.apache.druid.java.util.common.StringUtils; import org.junit.After; import org.junit.Assert; @@ -85,6 +88,80 @@ public void setUp() supervisorManager = new SQLMetadataSupervisorManager(MAPPER, connector, Suppliers.ofInstance(tablesConfig)); } + @Test + public void testRemoveTerminatedSupervisorsOlderThanSupervisorActiveShouldNotBeDeleted() + { + final String supervisor1 = "test-supervisor-1"; + final Map data1rev1 = ImmutableMap.of("key1-1", "value1-1-1", "key1-2", "value1-2-1"); + Assert.assertTrue(supervisorManager.getAll().isEmpty()); + supervisorManager.insert(supervisor1, new TestSupervisorSpec(supervisor1, data1rev1)); + // Test that supervisor was inserted + Map> supervisorSpecs = supervisorManager.getAll(); + Assert.assertEquals(1, supervisorSpecs.size()); + Map latestSpecs = supervisorManager.getLatest(); + Assert.assertEquals(1, latestSpecs.size()); + // Try delete. Supervisor should not be deleted as it is still active + supervisorManager.removeTerminatedSupervisorsOlderThan(System.currentTimeMillis()); + // Test that supervisor was not deleted + supervisorSpecs = supervisorManager.getAll(); + Assert.assertEquals(1, supervisorSpecs.size()); + latestSpecs = supervisorManager.getLatest(); + Assert.assertEquals(1, latestSpecs.size()); + } + + @Test + public void testRemoveTerminatedSupervisorsOlderThanWithSupervisorTerminatedAndOlderThanTimeShouldBeDeleted() + { + final String supervisor1 = "test-supervisor-1"; + final String datasource1 = "datasource-1"; + final Map data1rev1 = ImmutableMap.of("key1-1", "value1-1-1", "key1-2", "value1-2-1"); + Assert.assertTrue(supervisorManager.getAll().isEmpty()); + supervisorManager.insert(supervisor1, new TestSupervisorSpec(supervisor1, data1rev1)); + supervisorManager.insert(supervisor1, new NoopSupervisorSpec(supervisor1, ImmutableList.of(datasource1))); + // Test that supervisor was inserted + Map> supervisorSpecs = supervisorManager.getAll(); + Assert.assertEquals(1, supervisorSpecs.size()); + Assert.assertEquals(2, supervisorSpecs.get(supervisor1).size()); + Map latestSpecs = supervisorManager.getLatest(); + Assert.assertEquals(1, latestSpecs.size()); + Assert.assertEquals(ImmutableList.of(datasource1), ((NoopSupervisorSpec) latestSpecs.get(supervisor1)).getDataSources()); + // Do delete. Supervisor should be deleted as it is terminated + supervisorManager.removeTerminatedSupervisorsOlderThan(System.currentTimeMillis()); + // Verify that supervisor was actually deleted + supervisorSpecs = supervisorManager.getAll(); + Assert.assertEquals(0, supervisorSpecs.size()); + latestSpecs = supervisorManager.getLatest(); + Assert.assertEquals(0, latestSpecs.size()); + } + + @Test + public void testRemoveTerminatedSupervisorsOlderThanWithSupervisorTerminatedButNotOlderThanTimeShouldNotBeDeleted() + { + final String supervisor1 = "test-supervisor-1"; + final String datasource1 = "datasource-1"; + final Map data1rev1 = ImmutableMap.of("key1-1", "value1-1-1", "key1-2", "value1-2-1"); + Assert.assertTrue(supervisorManager.getAll().isEmpty()); + supervisorManager.insert(supervisor1, new TestSupervisorSpec(supervisor1, data1rev1)); + supervisorManager.insert(supervisor1, new NoopSupervisorSpec(supervisor1, ImmutableList.of(datasource1))); + // Test that supervisor was inserted + Map> supervisorSpecs = supervisorManager.getAll(); + Assert.assertEquals(1, supervisorSpecs.size()); + Assert.assertEquals(2, supervisorSpecs.get(supervisor1).size()); + Map latestSpecs = supervisorManager.getLatest(); + Assert.assertEquals(1, latestSpecs.size()); + Assert.assertEquals(ImmutableList.of(datasource1), ((NoopSupervisorSpec) latestSpecs.get(supervisor1)).getDataSources()); + // Do delete. Supervisor should not be deleted. Supervisor is terminated but it was created just now so it's + // created timestamp will be later than the timestamp 2012-01-01T00:00:00Z + supervisorManager.removeTerminatedSupervisorsOlderThan(DateTimes.of("2012-01-01T00:00:00Z").getMillis()); + // Verify that supervisor was not deleted + supervisorSpecs = supervisorManager.getAll(); + Assert.assertEquals(1, supervisorSpecs.size()); + Assert.assertEquals(2, supervisorSpecs.get(supervisor1).size()); + latestSpecs = supervisorManager.getLatest(); + Assert.assertEquals(1, latestSpecs.size()); + Assert.assertEquals(ImmutableList.of(datasource1), ((NoopSupervisorSpec) latestSpecs.get(supervisor1)).getDataSources()); + } + @Test public void testInsertAndGet() { diff --git a/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java b/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java index 402c92be203e..b8e3103681ef 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java @@ -176,6 +176,8 @@ public void setUp() throws Exception null, null, null, + null, + null, 10, new Duration("PT0s") ); diff --git a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java index a1c3d033118d..97c220d3626c 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java @@ -148,6 +148,8 @@ public void setUp() throws Exception null, null, null, + null, + null, 10, new Duration("PT0s") ); diff --git a/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java b/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java index 607fb8dc3fcd..2a8550003998 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/HttpLoadQueuePeonTest.java @@ -85,6 +85,8 @@ public class HttpLoadQueuePeonTest null, null, null, + null, + null, 10, Duration.ZERO ) diff --git a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java index a607423bb9a2..8b2f486d1ace 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTest.java @@ -100,6 +100,8 @@ public void testMultipleLoadDropSegments() throws Exception null, null, null, + null, + null, 10, Duration.millis(0) ) @@ -300,6 +302,8 @@ public void testFailAssignForNonTimeoutFailures() throws Exception null, null, null, + null, + null, 10, new Duration("PT1s") ) @@ -357,6 +361,8 @@ public void testFailAssignForLoadDropTimeout() throws Exception null, null, null, + null, + null, 10, new Duration("PT1s") ) diff --git a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java index 5540a117d4be..57ac65b88b8c 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/LoadQueuePeonTester.java @@ -49,6 +49,8 @@ public LoadQueuePeonTester() null, null, null, + null, + null, 10, new Duration("PT1s") ) diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java index 51ddc3c4a973..41dd839cabf8 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java @@ -66,6 +66,8 @@ public void testRunSkipIfLastRunLessThanPeriod() new Duration("PT1S"), null, null, + null, + null, 10, null ); @@ -90,6 +92,8 @@ public void testRunNotSkipIfLastRunMoreThanPeriod() new Duration("PT1S"), null, null, + null, + null, 10, null ); @@ -114,6 +118,8 @@ public void testConstructorFailIfInvalidPeriod() new Duration("PT1S"), null, null, + null, + null, 10, null ); @@ -137,6 +143,8 @@ public void testConstructorFailIfInvalidRetainDuration() new Duration("PT-1S"), null, null, + null, + null, 10, null ); diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillRulesTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillRulesTest.java index 98c0ad9d65e6..d3268f21150f 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillRulesTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillRulesTest.java @@ -71,6 +71,8 @@ public void testRunSkipIfLastRunLessThanPeriod() null, null, null, + null, + null, new Duration(Long.MAX_VALUE), new Duration("PT1S"), 10, @@ -95,6 +97,8 @@ public void testRunNotSkipIfLastRunMoreThanPeriod() null, null, null, + null, + null, new Duration("PT6S"), new Duration("PT1S"), 10, @@ -119,6 +123,8 @@ public void testConstructorFailIfInvalidPeriod() null, null, null, + null, + null, new Duration("PT3S"), new Duration("PT1S"), 10, @@ -142,6 +148,8 @@ public void testConstructorFailIfInvalidRetainDuration() null, null, null, + null, + null, new Duration("PT6S"), new Duration("PT-1S"), 10, diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsTest.java new file mode 100644 index 000000000000..9aeb992f9b14 --- /dev/null +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillSupervisorsTest.java @@ -0,0 +1,155 @@ +/* + * 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 org.apache.druid.java.util.emitter.service.ServiceEmitter; +import org.apache.druid.java.util.emitter.service.ServiceEventBuilder; +import org.apache.druid.metadata.MetadataSupervisorManager; +import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams; +import org.apache.druid.server.coordinator.TestDruidCoordinatorConfig; +import org.joda.time.Duration; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.junit.runner.RunWith; +import org.mockito.ArgumentMatchers; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class KillSupervisorsTest +{ + @Mock + private MetadataSupervisorManager mockMetadataSupervisorManager; + + @Mock + private DruidCoordinatorRuntimeParams mockDruidCoordinatorRuntimeParams; + + @Mock + private ServiceEmitter mockServiceEmitter; + + @Rule + public ExpectedException exception = ExpectedException.none(); + + private KillSupervisors killSupervisors; + + @Test + public void testRunSkipIfLastRunLessThanPeriod() + { + TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig( + null, + null, + null, + new Duration("PT5S"), + null, + null, + null, + new Duration(Long.MAX_VALUE), + new Duration("PT1S"), + null, + null, + null, + null, + 10, + null + ); + killSupervisors = new KillSupervisors(druidCoordinatorConfig, mockMetadataSupervisorManager); + killSupervisors.run(mockDruidCoordinatorRuntimeParams); + Mockito.verifyZeroInteractions(mockMetadataSupervisorManager); + } + + @Test + public void testRunNotSkipIfLastRunMoreThanPeriod() + { + Mockito.when(mockDruidCoordinatorRuntimeParams.getEmitter()).thenReturn(mockServiceEmitter); + TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig( + null, + null, + null, + new Duration("PT5S"), + null, + null, + null, + new Duration("PT6S"), + new Duration("PT1S"), + null, + null, + null, + null, + 10, + null + ); + killSupervisors = new KillSupervisors(druidCoordinatorConfig, mockMetadataSupervisorManager); + killSupervisors.run(mockDruidCoordinatorRuntimeParams); + Mockito.verify(mockMetadataSupervisorManager).removeTerminatedSupervisorsOlderThan(ArgumentMatchers.anyLong()); + Mockito.verify(mockServiceEmitter).emit(ArgumentMatchers.any(ServiceEventBuilder.class)); + } + + @Test + public void testConstructorFailIfInvalidPeriod() + { + TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig( + null, + null, + null, + new Duration("PT5S"), + null, + null, + null, + new Duration("PT3S"), + new Duration("PT1S"), + null, + null, + null, + null, + 10, + null + ); + exception.expect(IllegalArgumentException.class); + exception.expectMessage("Coordinator supervisor kill period must be >= druid.coordinator.period.metadataStoreManagementPeriod"); + killSupervisors = new KillSupervisors(druidCoordinatorConfig, mockMetadataSupervisorManager); + } + + @Test + public void testConstructorFailIfInvalidRetainDuration() + { + TestDruidCoordinatorConfig druidCoordinatorConfig = new TestDruidCoordinatorConfig( + null, + null, + null, + new Duration("PT5S"), + null, + null, + null, + new Duration("PT6S"), + new Duration("PT-1S"), + null, + null, + null, + null, + 10, + null + ); + exception.expect(IllegalArgumentException.class); + exception.expectMessage("Coordinator supervisor kill retainDuration must be >= 0"); + killSupervisors = new KillSupervisors(druidCoordinatorConfig, mockMetadataSupervisorManager); + } +} diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java index 4a1aa0960761..34d020f5cd72 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java @@ -113,6 +113,8 @@ private void testFindIntervalForKill(List segmentIntervals, Interval e null, null, null, + null, + null, 1000, Duration.ZERO ) From 9582b9bdfdba112a843737798ca4f8cac90bfda0 Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Wed, 5 May 2021 13:32:33 -0700 Subject: [PATCH 4/6] fix test --- .../coordinator/duty/KillAuditLogTest.java | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java index 41dd839cabf8..aeb0bd9cbc5d 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillAuditLogTest.java @@ -62,10 +62,10 @@ public void testRunSkipIfLastRunLessThanPeriod() null, null, null, - new Duration(Long.MAX_VALUE), - new Duration("PT1S"), null, null, + new Duration(Long.MAX_VALUE), + new Duration("PT1S"), null, null, 10, @@ -88,10 +88,10 @@ public void testRunNotSkipIfLastRunMoreThanPeriod() null, null, null, - new Duration("PT6S"), - new Duration("PT1S"), null, null, + new Duration("PT6S"), + new Duration("PT1S"), null, null, 10, @@ -114,10 +114,10 @@ public void testConstructorFailIfInvalidPeriod() null, null, null, - new Duration("PT3S"), - new Duration("PT1S"), null, null, + new Duration("PT3S"), + new Duration("PT1S"), null, null, 10, @@ -139,10 +139,10 @@ public void testConstructorFailIfInvalidRetainDuration() null, null, null, - new Duration("PT6S"), - new Duration("PT-1S"), null, null, + new Duration("PT6S"), + new Duration("PT-1S"), null, null, 10, From 2710017af7eec38900b2ee107f35f62beec6a632 Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Thu, 6 May 2021 17:51:17 -0700 Subject: [PATCH 5/6] Address comments --- docs/configuration/index.md | 12 ++--- docs/operations/metrics.md | 6 +-- .../metadata/MetadataSupervisorManager.java | 2 +- .../SQLMetadataSupervisorManager.java | 44 ++++++++++--------- .../coordinator/duty/KillSupervisors.java | 3 ++ .../SQLMetadataSupervisorManagerTest.java | 9 ++-- 6 files changed, 43 insertions(+), 33 deletions(-) diff --git a/docs/configuration/index.md b/docs/configuration/index.md index 7bcf17a18281..bfedf7f5f6db 100644 --- a/docs/configuration/index.md +++ b/docs/configuration/index.md @@ -748,14 +748,14 @@ These Coordinator static configurations can be defined in the `coordinator/runti |--------|-----------|---------|-------| |`druid.coordinator.period.metadataStoreManagementPeriod`|How often to run metadata management tasks in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. |No | `PT1H`| |`druid.coordinator.kill.supervisor.on`| Boolean value for whether to enable automatic deletion of terminated supervisors. If set to true, Coordinator will periodically remove terminated supervisors from the supervisor table in metadata storage.| No | False| -|`druid.coordinator.kill.supervisor.period`| How often to do automatic deletion of terminated supervisor 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.supervisor.on` is set to True.| No| `P1D`| -|`druid.coordinator.kill.supervisor.durationToRetain`| Duration of terminated supervisor to be retained from created time in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Only applies if `druid.coordinator.kill.supervisor.on` is set to True.| Yes if `druid.coordinator.kill.supervisor.on` is set to True| None| +|`druid.coordinator.kill.supervisor.period`| How often to do automatic deletion of terminated supervisor 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.supervisor.on` is set to "True".| No| `P1D`| +|`druid.coordinator.kill.supervisor.durationToRetain`| Duration of terminated supervisor to be retained from created time in [ISO 8601](https://en.wikipedia.org/wiki/ISO_8601) duration format. Only applies if `druid.coordinator.kill.supervisor.on` is set to "True".| Yes if `druid.coordinator.kill.supervisor.on` is set to "True".| None| |`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.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 equal to or 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| +|`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| ##### Segment Management |Property|Possible Values|Description|Default| diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md index 7ff90d45b427..b68dcca58307 100644 --- a/docs/operations/metrics.md +++ b/docs/operations/metrics.md @@ -256,9 +256,9 @@ These metrics are for the Druid Coordinator and are reset each time the Coordina |`interval/skipCompact/count`|Total number of intervals of this datasource that are skipped (not eligible for auto compaction) by the auto compaction.|datasource.|Varies.| |`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/supervisor/count`|Total number of terminated supervisors automatically deleted from metadata store supervisor table per each Coordinator kill supervisor duty run. This metric can help adjust `druid.coordinator.kill.supervisor.durationToRetain` configuration based on if 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 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.| +|`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.| If `emitBalancingStats` is set to `true` in the Coordinator [dynamic configuration](../configuration/index.md#dynamic-configuration), then [log entries](../configuration/logging.md) for class diff --git a/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java b/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java index 39553ea99259..591dee280a91 100644 --- a/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java +++ b/server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java @@ -36,7 +36,7 @@ public interface MetadataSupervisorManager Map getLatest(); /** - * Remove terminated supervisor created older than the given timestamp. + * Remove terminated supervisors created before the given timestamp. * * @param timestamp timestamp in milliseconds * @return number of supervisor removed diff --git a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java index c409f38a4209..d594b0102bd9 100644 --- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java +++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java @@ -36,11 +36,14 @@ import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.lifecycle.LifecycleStart; import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.timeline.DataSegment; +import org.apache.druid.timeline.partition.NoneShardSpec; import org.joda.time.DateTime; import org.skife.jdbi.v2.FoldController; import org.skife.jdbi.v2.Folder3; import org.skife.jdbi.v2.Handle; import org.skife.jdbi.v2.IDBI; +import org.skife.jdbi.v2.PreparedBatch; import org.skife.jdbi.v2.StatementContext; import org.skife.jdbi.v2.Update; import org.skife.jdbi.v2.tweak.HandleCallback; @@ -53,6 +56,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.IntStream; @ManageLifecycle public class SQLMetadataSupervisorManager implements MetadataSupervisorManager @@ -255,31 +259,31 @@ public Map fold( @Override public int removeTerminatedSupervisorsOlderThan(long timestamp) { - int removedCount = 0; DateTime dateTime = DateTimes.utc(timestamp); Map supervisors = getLatest(); - for (Map.Entry supervisor : supervisors.entrySet()) { - final SupervisorSpec spec = supervisor.getValue(); - if (spec instanceof NoopSupervisorSpec) { - removedCount += dbi.withHandle( - handle -> { - Update sql = handle.createStatement( - StringUtils.format( - "DELETE FROM %1$s WHERE spec_id = :spec_id AND created_date < :date_time", - getSupervisorsTable() - ) - ); - return sql.bind("spec_id", supervisor.getKey()) - .bind("date_time", dateTime.toString()) - .execute(); + return dbi.withHandle( + handle -> { + final PreparedBatch batch = handle.prepareBatch( + StringUtils.format( + "DELETE FROM %1$s WHERE spec_id = :spec_id AND created_date < '%2$s'", + getSupervisorsTable(), + dateTime.toString() + ) + ); + for (Map.Entry 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(); } - ); - } - } - return removedCount; + } + int[] result = batch.execute(); + return IntStream.of(result).sum(); + } + ); } - private String getSupervisorsTable() { return dbTables.get().getSupervisorTable(); diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java index b55a9cf37d78..a1c2a3f28326 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/KillSupervisors.java @@ -28,6 +28,9 @@ import org.apache.druid.server.coordinator.DruidCoordinatorConfig; import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams; +/** + * CoordinatorDuty for automatic deletion of terminated supervisors from the supervisor table in metadata storage. + */ public class KillSupervisors implements CoordinatorDuty { private static final Logger log = new Logger(KillSupervisors.class); diff --git a/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java b/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java index 789d8d01f4b8..5e4f4833b091 100644 --- a/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java +++ b/server/src/test/java/org/apache/druid/metadata/SQLMetadataSupervisorManagerTest.java @@ -101,8 +101,9 @@ public void testRemoveTerminatedSupervisorsOlderThanSupervisorActiveShouldNotBeD Map latestSpecs = supervisorManager.getLatest(); Assert.assertEquals(1, latestSpecs.size()); // Try delete. Supervisor should not be deleted as it is still active - supervisorManager.removeTerminatedSupervisorsOlderThan(System.currentTimeMillis()); + int deleteCount = supervisorManager.removeTerminatedSupervisorsOlderThan(System.currentTimeMillis()); // Test that supervisor was not deleted + Assert.assertEquals(0, deleteCount); supervisorSpecs = supervisorManager.getAll(); Assert.assertEquals(1, supervisorSpecs.size()); latestSpecs = supervisorManager.getLatest(); @@ -126,8 +127,9 @@ public void testRemoveTerminatedSupervisorsOlderThanWithSupervisorTerminatedAndO Assert.assertEquals(1, latestSpecs.size()); Assert.assertEquals(ImmutableList.of(datasource1), ((NoopSupervisorSpec) latestSpecs.get(supervisor1)).getDataSources()); // Do delete. Supervisor should be deleted as it is terminated - supervisorManager.removeTerminatedSupervisorsOlderThan(System.currentTimeMillis()); + int deleteCount = supervisorManager.removeTerminatedSupervisorsOlderThan(System.currentTimeMillis()); // Verify that supervisor was actually deleted + Assert.assertEquals(2, deleteCount); supervisorSpecs = supervisorManager.getAll(); Assert.assertEquals(0, supervisorSpecs.size()); latestSpecs = supervisorManager.getLatest(); @@ -152,8 +154,9 @@ public void testRemoveTerminatedSupervisorsOlderThanWithSupervisorTerminatedButN Assert.assertEquals(ImmutableList.of(datasource1), ((NoopSupervisorSpec) latestSpecs.get(supervisor1)).getDataSources()); // Do delete. Supervisor should not be deleted. Supervisor is terminated but it was created just now so it's // created timestamp will be later than the timestamp 2012-01-01T00:00:00Z - supervisorManager.removeTerminatedSupervisorsOlderThan(DateTimes.of("2012-01-01T00:00:00Z").getMillis()); + int deleteCount = supervisorManager.removeTerminatedSupervisorsOlderThan(DateTimes.of("2012-01-01T00:00:00Z").getMillis()); // Verify that supervisor was not deleted + Assert.assertEquals(0, deleteCount); supervisorSpecs = supervisorManager.getAll(); Assert.assertEquals(1, supervisorSpecs.size()); Assert.assertEquals(2, supervisorSpecs.get(supervisor1).size()); From e5249da8bf7827a34d30381359807aad70118586 Mon Sep 17 00:00:00 2001 From: Maytas Monsereenusorn Date: Thu, 6 May 2021 17:52:04 -0700 Subject: [PATCH 6/6] Address comments --- .../apache/druid/metadata/SQLMetadataSupervisorManager.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java index d594b0102bd9..e995a14aac16 100644 --- a/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java +++ b/server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java @@ -36,8 +36,6 @@ import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.java.util.common.lifecycle.LifecycleStart; import org.apache.druid.java.util.common.logger.Logger; -import org.apache.druid.timeline.DataSegment; -import org.apache.druid.timeline.partition.NoneShardSpec; import org.joda.time.DateTime; import org.skife.jdbi.v2.FoldController; import org.skife.jdbi.v2.Folder3; @@ -45,7 +43,6 @@ import org.skife.jdbi.v2.IDBI; import org.skife.jdbi.v2.PreparedBatch; import org.skife.jdbi.v2.StatementContext; -import org.skife.jdbi.v2.Update; import org.skife.jdbi.v2.tweak.HandleCallback; import org.skife.jdbi.v2.tweak.ResultSetMapper;