Run compaction as a supervisor on Overlord

benchmarks/src/test/java/org/apache/druid/server/coordinator/NewestSegmentFirstPolicyBenchmark.java

@@ -24,9 +24,10 @@
 import org.apache.druid.client.DataSourcesSnapshot;
 import org.apache.druid.jackson.DefaultObjectMapper;
 import org.apache.druid.java.util.common.DateTimes;
-import org.apache.druid.server.coordinator.compact.CompactionSegmentIterator;
-import org.apache.druid.server.coordinator.compact.CompactionSegmentSearchPolicy;
-import org.apache.druid.server.coordinator.compact.NewestSegmentFirstPolicy;
+import org.apache.druid.server.compaction.CompactionSegmentIterator;
+import org.apache.druid.server.compaction.CompactionSegmentSearchPolicy;
+import org.apache.druid.server.compaction.CompactionStatusTracker;
+import org.apache.druid.server.compaction.NewestSegmentFirstPolicy;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.SegmentTimeline;
 import org.apache.druid.timeline.partition.NumberedShardSpec;
@@ -61,7 +62,7 @@ public class NewestSegmentFirstPolicyBenchmark
 {
   private static final String DATA_SOURCE_PREFIX = "dataSource_";
 
-  private final CompactionSegmentSearchPolicy policy = new NewestSegmentFirstPolicy(new DefaultObjectMapper());
+  private final CompactionSegmentSearchPolicy policy = new NewestSegmentFirstPolicy(null);
 
   @Param("100")
   private int numDataSources;
@@ -132,7 +133,12 @@ public void setup()
   @Benchmark
   public void measureNewestSegmentFirstPolicy(Blackhole blackhole)
   {
-    final CompactionSegmentIterator iterator = policy.createIterator(compactionConfigs, dataSources, Collections.emptyMap());
+    final CompactionSegmentIterator iterator = policy.createIterator(
+        compactionConfigs,
+        dataSources,
+        Collections.emptyMap(),
+        new CompactionStatusTracker(new DefaultObjectMapper())
+    );
     for (int i = 0; i < numCompactionTaskSlots && iterator.hasNext(); i++) {
       blackhole.consume(iterator.next());
     }

indexing-service/src/main/java/org/apache/druid/guice/SupervisorModule.java

@@ -0,0 +1,52 @@
+/*
+ * 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.guice;
+
+import com.fasterxml.jackson.databind.Module;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Binder;
+import org.apache.druid.indexing.compact.CompactionSupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
+import org.apache.druid.initialization.DruidModule;
+
+import java.util.List;
+
+public class SupervisorModule implements DruidModule
+{
+  @Override
+  public void configure(Binder binder)
+  {
+    JsonConfigProvider.bind(binder, "druid.supervisor", SupervisorStateManagerConfig.class);
+  }
+
+  @Override
+  public List<? extends Module> getJacksonModules()
+  {
+    return ImmutableList.of(
+        new SimpleModule(getClass().getSimpleName())
+            .registerSubtypes(
+                new NamedType(CompactionSupervisorSpec.class, CompactionSupervisorSpec.TYPE)
+            )
+    );
+  }
+}

server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorModule.java → indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionStatusReport.java

@@ -17,18 +17,12 @@
  * under the License.
  */
 
+package org.apache.druid.indexing.compact;
 
-package org.apache.druid.indexing.overlord.supervisor;
-
-import com.google.inject.Binder;
-import com.google.inject.Module;
-import org.apache.druid.guice.JsonConfigProvider;
-
-public class SupervisorModule implements Module
+/**
+ * This can contain stats and progress and stuff.
+ * For simulation, we will still have the separate API.
+ */
+public class CompactionStatusReport
 {
-  @Override
-  public void configure(Binder binder)
-  {
-    JsonConfigProvider.bind(binder, "druid.supervisor", SupervisorStateManagerConfig.class);
-  }
 }

indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionSupervisor.java

@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.compact;
+
+import org.apache.druid.indexing.overlord.DataSourceMetadata;
+import org.apache.druid.indexing.overlord.supervisor.Supervisor;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorReport;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStateManager;
+import org.apache.druid.indexing.overlord.supervisor.autoscaler.LagStats;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.server.compaction.CompactionScheduler;
+import org.apache.druid.server.coordinator.AutoCompactionSnapshot;
+
+/**
+ * Supervisor for compaction of a single datasource.
+ */
+public class CompactionSupervisor implements Supervisor
+{
+  private static final Logger log = new Logger(CompactionSupervisor.class);
+
+  private final CompactionScheduler scheduler;
+  private final CompactionSupervisorSpec supervisorSpec;
+
+  public CompactionSupervisor(
+      CompactionSupervisorSpec supervisorSpec,
+      CompactionScheduler scheduler
+  )
+  {
+    this.supervisorSpec = supervisorSpec;
+    this.scheduler = scheduler;
+  }
+
+  @Override
+  public void start()
+  {
+    final String dataSource = getDataSource();
+    if (supervisorSpec.isSuspended()) {
+      log.info("Suspending compaction for dataSource[%s].", dataSource);
+      scheduler.stopCompaction(dataSource);
+    } else {
+      log.info("Starting compaction for dataSource[%s].", dataSource);
+      scheduler.startCompaction(dataSource, supervisorSpec.getSpec());
+    }
+  }
+
+  @Override
+  public void stop(boolean stopGracefully)
+  {
+    final String dataSource = getDataSource();
+    log.info("Stopping compaction for dataSource[%s].", dataSource);
+    scheduler.stopCompaction(dataSource);
+  }
+
+  @Override
+  public SupervisorReport<AutoCompactionSnapshot> getStatus()
+  {
+    final AutoCompactionSnapshot snapshot;
+    if (supervisorSpec.isSuspended()) {
+      snapshot = AutoCompactionSnapshot.builder(getDataSource())
+                                       .withStatus(AutoCompactionSnapshot.AutoCompactionScheduleStatus.NOT_ENABLED)
+                                       .build();
+    } else {
+      snapshot = scheduler.getCompactionSnapshot(getDataSource());
+    }
+
+    return new SupervisorReport<>(supervisorSpec.getId(), DateTimes.nowUtc(), snapshot);
+  }
+
+  @Override
+  public SupervisorStateManager.State getState()
+  {
+    if (!scheduler.isRunning()) {
+      return State.SCHEDULER_DISABLED;
+    } else if (supervisorSpec.isSuspended()) {
+      return State.SUSPENDED;
+    } else {
+      return State.RUNNING;
+    }
+  }
+
+  private String getDataSource()
+  {
+    return supervisorSpec.getSpec().getDataSource();
+  }
+
+  // Un-implemented methods used only by streaming supervisors
+
+  @Override
+  public void reset(DataSourceMetadata dataSourceMetadata)
+  {
+    // Do nothing
+  }
+
+  @Override
+  public void resetOffsets(DataSourceMetadata resetDataSourceMetadata)
+  {
+    // Do nothing
+  }
+
+  @Override
+  public void checkpoint(int taskGroupId, DataSourceMetadata checkpointMetadata)
+  {
+    // Do nothing
+  }
+
+  @Override
+  public LagStats computeLagStats()
+  {
+    return new LagStats(0L, 0L, 0L);
+  }
+
+  @Override
+  public int getActiveTaskGroupsCount()
+  {
+    return 0;
+  }
+
+  public enum State implements SupervisorStateManager.State
+  {
+    SCHEDULER_DISABLED(true),
+    RUNNING(true),
+    SUSPENDED(true),
+    UNHEALTHY(false);
+
+    private final boolean healthy;
+
+    State(boolean healthy)
+    {
+      this.healthy = healthy;
+    }
+
+    @Override
+    public boolean isFirstRunOnly()
+    {
+      return false;
+    }
+
+    @Override
+    public boolean isHealthy()
+    {
+      return healthy;
+    }
+  }
+}

indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionSupervisorSpec.java

@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.compact;
+
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.indexing.overlord.supervisor.Supervisor;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
+import org.apache.druid.server.compaction.CompactionScheduler;
+import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+
+import javax.annotation.Nullable;
+import java.util.Collections;
+import java.util.List;
+
+public class CompactionSupervisorSpec implements SupervisorSpec
+{
+  public static final String TYPE = "autocompact";
+  public static final String ID_PREFIX = "autocompact__";
+
+  private final boolean suspended;
+  private final DataSourceCompactionConfig spec;
+  private final CompactionScheduler scheduler;
+
+  @JsonCreator
+  public CompactionSupervisorSpec(
+      @JsonProperty("spec") DataSourceCompactionConfig spec,
+      @JsonProperty("suspended") @Nullable Boolean suspended,
+      @JacksonInject CompactionScheduler scheduler
+  )
+  {
+    if (spec == null) {
+      throw InvalidInput.exception("'spec' must be specified for a compaction supervisor.");
+    }
+
+    this.spec = spec;
+    this.suspended = Configs.valueOrDefault(suspended, false);
+    this.scheduler = scheduler;
+  }
+
+  @JsonProperty
+  public DataSourceCompactionConfig getSpec()
+  {
+    return spec;
+  }
+
+  @Override
+  @JsonProperty
+  public boolean isSuspended()
+  {
+    return suspended;
+  }
+
+  @Override
+  public String getId()
+  {
+    return ID_PREFIX + spec.getDataSource();
+  }
+
+  @Override
+  public Supervisor createSupervisor()
+  {
+    return new CompactionSupervisor(this, scheduler);
+  }
+
+  @Override
+  public List<String> getDataSources()
+  {
+    return Collections.singletonList(spec.getDataSource());
+  }
+
+  @Override
+  public SupervisorSpec createSuspendedSpec()
+  {
+    return new CompactionSupervisorSpec(spec, true, scheduler);
+  }
+
+  @Override
+  public SupervisorSpec createRunningSpec()
+  {
+    return new CompactionSupervisorSpec(spec, false, scheduler);
+  }
+
+  @Override
+  public String getType()
+  {
+    return TYPE;
+  }
+
+  @Override
+  public String getSource()
+  {
+    return "";
+  }
+}