Overlord to support autoscalers per indexer/middlemanager category

docs/configuration/index.md

@@ -971,11 +971,6 @@ A sample worker config spec is shown below:
 
 Issuing a GET request at the same URL will return the current worker config spec that is currently in place. The worker config spec list above is just a sample for EC2 and it is possible to extend the code base for other deployment environments. A description of the worker config spec is shown below.
 
-|Property|Description|Default|
-|--------|-----------|-------|
-|`selectStrategy`|How to assign tasks to MiddleManagers. Choices are `fillCapacity`, `equalDistribution`, and `javascript`.|equalDistribution|
-|`autoScaler`|Only used if autoscaling is enabled. See below.|null|
-
 To view the audit history of worker config issue a GET request to the URL -
 
 ```
@@ -990,6 +985,13 @@ To view last <n> entries of the audit history of worker config issue a GET reque
 http://<OVERLORD_IP>:<port>/druid/indexer/v1/worker/history?count=<n>
 ```
 
+##### Default Worker Config
+
+|Property|Description|Default|
+|--------|-----------|-------|
+|`selectStrategy`|How to assign tasks to MiddleManagers. Choices are `fillCapacity`, `equalDistribution`, and `javascript`.|equalDistribution|
+|`autoScalers`|List of [Autoscaler](#autoscaler) to serve tasks of appropriate category. In the list, there can be one autoscaler of default category (category declaration is omitted). When [Worker Category Spec](#workercategoryspec) is not in strong assignment mode then default autoscaler will be used to serve tasks with categories which do not have corresponding autoscaler|required; At least one autoscaler should be declared and at most one autoscaler per category.|
+
 ##### Worker Select Strategy
 
 Worker select strategies control how Druid assigns tasks to MiddleManagers.
@@ -1113,6 +1115,50 @@ field. If not provided, the default is to not use it at all.
 |`defaultCategory`|Specify default category for a task type.|null|
 |`categoryAffinity`|A JSON map object mapping a datasource String name to a category String name of the MiddleManager. If category isn't specified for a datasource, then using the `defaultCategory`. If no specified category and the `defaultCategory` is also null, then tasks can run on any available MiddleManagers.|null|
 
+###### Autoscaler setups
+List of [Autoscaler](#autoscaler) instances for each category of task specified in a [CategoryConfig](#categoryconfig)
+
+Example: declare separate autoscaler setups for different categories of tasks
+
+```json
+{
+  "autoScalers": [
+    {
+      "type": "ec2",
+      "category": "category0",
+      "minNumWorkers": 0,
+      "maxNumWorkers": 12,
+      "envConfig": {
+        "availabilityZone": "us-east-1a",
+        "nodeData": {
+          "amiId": "${AMI}",
+          "instanceType": "c3.8xlarge",
+          "minInstances": 1,
+          "maxInstances": 1,
+          "securityGroupIds": ["${IDs}"],
+          "keyName": "${KEY_NAME}"
+        },
+        "userData": {
+          "impl": "string",
+          "data": "${SCRIPT_COMMAND}",
+          "versionReplacementString": ":VERSION:",
+          "version": null
+        }
+      }
+    },
+    {
+      "type": "http",
+      "category": "category1",
+      "minNumWorkers": 0,
+      "maxNumWorkers": 30,
+      "envConfig": {
+        "endpoint": "http://druid-indexer-autoscaler"
+      }
+    }
+  ]
+}
+```
+
 ##### Autoscaler
 
 Amazon's EC2 is currently the only supported autoscaler.
@@ -1121,6 +1167,7 @@ Amazon's EC2 is currently the only supported autoscaler.
 |--------|-----------|-------|
 |`minNumWorkers`|The minimum number of workers that can be in the cluster at any given time.|0|
 |`maxNumWorkers`|The maximum number of workers that can be in the cluster at any given time.|0|
+|`category`|Category of tasks which should be served by this autoscaler|`__default_worker_category`; optional|
 |`availabilityZone`|What availability zone to run in.|none|
 |`nodeData`|A JSON object that describes how to launch new nodes.|none; required|
 |`userData`|A JSON object that describes how to configure new nodes. If you have set druid.indexer.autoscale.workerVersion, this must have a versionReplacementString. Otherwise, a versionReplacementString is not necessary.|none; optional|

extensions-core/ec2-extensions/src/main/java/org/apache/druid/indexing/overlord/autoscaling/ec2/EC2AutoScaler.java

@@ -44,6 +44,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 
 /**
  */
@@ -55,6 +56,7 @@ public class EC2AutoScaler implements AutoScaler<EC2EnvironmentConfig>
 
   private final int minNumWorkers;
   private final int maxNumWorkers;
+  private final String category;
   private final EC2EnvironmentConfig envConfig;
   private final AmazonEC2 amazonEC2Client;
   private final SimpleWorkerProvisioningConfig config;
@@ -65,11 +67,13 @@ public EC2AutoScaler(
       @JsonProperty("maxNumWorkers") int maxNumWorkers,
       @JsonProperty("envConfig") EC2EnvironmentConfig envConfig,
       @JacksonInject AmazonEC2 amazonEC2Client,
-      @JacksonInject SimpleWorkerProvisioningConfig config
+      @JacksonInject SimpleWorkerProvisioningConfig config,
+      @JsonProperty("category") String category
   )
   {
     this.minNumWorkers = minNumWorkers;
     this.maxNumWorkers = maxNumWorkers;
+    this.category = category;
     this.envConfig = envConfig;
     this.amazonEC2Client = amazonEC2Client;
     this.config = config;
@@ -89,6 +93,13 @@ public int getMaxNumWorkers()
     return maxNumWorkers;
   }
 
+  @Override
+  @JsonProperty
+  public String getCategory()
+  {
+    return category;
+  }
+
   @Override
   @JsonProperty
   public EC2EnvironmentConfig getEnvConfig()
@@ -331,6 +342,7 @@ public String toString()
            "envConfig=" + envConfig +
            ", maxNumWorkers=" + maxNumWorkers +
            ", minNumWorkers=" + minNumWorkers +
+           ", category=" + category +
            '}';
   }
 
@@ -343,28 +355,16 @@ public boolean equals(Object o)
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
-
     EC2AutoScaler that = (EC2AutoScaler) o;
-
-    if (maxNumWorkers != that.maxNumWorkers) {
-      return false;
-    }
-    if (minNumWorkers != that.minNumWorkers) {
-      return false;
-    }
-    if (envConfig != null ? !envConfig.equals(that.envConfig) : that.envConfig != null) {
-      return false;
-    }
-
-    return true;
+    return minNumWorkers == that.minNumWorkers &&
+           maxNumWorkers == that.maxNumWorkers &&
+           Objects.equals(category, that.category) &&
+           Objects.equals(envConfig, that.envConfig);
   }
 
   @Override
   public int hashCode()
   {
-    int result = minNumWorkers;
-    result = 31 * result + maxNumWorkers;
-    result = 31 * result + (envConfig != null ? envConfig.hashCode() : 0);
-    return result;
+    return Objects.hash(minNumWorkers, maxNumWorkers, category, envConfig);
   }
 }

extensions-core/ec2-extensions/src/test/java/org/apache/druid/indexing/overlord/autoscaling/ec2/EC2AutoScalerTest.java

@@ -35,6 +35,7 @@
 import com.google.common.collect.Range;
 import org.apache.druid.indexing.overlord.autoscaling.AutoScalingData;
 import org.apache.druid.indexing.overlord.autoscaling.SimpleWorkerProvisioningConfig;
+import org.apache.druid.indexing.overlord.setup.DefaultWorkerBehaviorConfig;
 import org.apache.druid.jackson.DefaultObjectMapper;
 import org.easymock.EasyMock;
 import org.junit.After;
@@ -101,7 +102,8 @@ public void testScale()
         1,
         ENV_CONFIG,
         amazonEC2Client,
-        managementConfig
+        managementConfig,
+        DefaultWorkerBehaviorConfig.DEFAULT_AUTOSCALER_CATEGORY
     );
 
     EasyMock.expect(amazonEC2Client.runInstances(EasyMock.anyObject(RunInstancesRequest.class))).andReturn(
@@ -145,7 +147,8 @@ public void testIptoIdLookup()
         1,
         ENV_CONFIG,
         amazonEC2Client,
-        managementConfig
+        managementConfig,
+        DefaultWorkerBehaviorConfig.DEFAULT_AUTOSCALER_CATEGORY
     );
 
     final int n = 150;
@@ -198,7 +201,8 @@ public void testIdToIpLookup()
         1,
         ENV_CONFIG,
         amazonEC2Client,
-        managementConfig
+        managementConfig,
+        DefaultWorkerBehaviorConfig.DEFAULT_AUTOSCALER_CATEGORY
     );
 
     final int n = 150;

extensions-core/ec2-extensions/src/test/java/org/apache/druid/indexing/overlord/setup/WorkerBehaviorConfigTest.java

@@ -70,8 +70,10 @@ public void testSerde() throws Exception
                 )
             ),
             null,
-            null
-        )
+            null,
+            DefaultWorkerBehaviorConfig.DEFAULT_AUTOSCALER_CATEGORY
+        ),
+        null
     );
 
     final ObjectMapper mapper = new DefaultObjectMapper();

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

@@ -21,6 +21,7 @@
 
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import org.apache.druid.guice.annotations.ExtensionPoint;
+import org.apache.druid.indexing.overlord.setup.DefaultWorkerBehaviorConfig;
 
 import javax.annotation.Nullable;
 import java.util.List;
@@ -36,6 +37,11 @@ public interface AutoScaler<T>
 
   int getMaxNumWorkers();
 
+  default String getCategory()
+  {
+    return DefaultWorkerBehaviorConfig.DEFAULT_AUTOSCALER_CATEGORY;
+  }
+
   /**
    * This method is unused, but AutoScaler is an {@link ExtensionPoint}, so we cannot remove it.
    */

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

@@ -19,6 +19,8 @@
 
 package org.apache.druid.indexing.overlord.autoscaling;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.druid.java.util.common.UOE;
 import org.apache.druid.java.util.emitter.EmittingLogger;
 
@@ -30,6 +32,15 @@
 public class NoopAutoScaler<Void> implements AutoScaler<Void>
 {
   private static final EmittingLogger log = new EmittingLogger(NoopAutoScaler.class);
+  private final String category;
+
+  @JsonCreator
+  public NoopAutoScaler(
+      @JsonProperty("category") String category
+  )
+  {
+    this.category = category;
+  }
 
   @Override
   public int getMinNumWorkers()
@@ -43,6 +54,13 @@ public int getMaxNumWorkers()
     return 0;
   }
 
+  @Override
+  @JsonProperty
+  public String getCategory()
+  {
+    return category;
+  }
+
   @Override
   public Void getEnvConfig()
   {
@@ -52,35 +70,43 @@ public Void getEnvConfig()
   @Override
   public AutoScalingData provision()
   {
-    log.info("If I were a real strategy I'd create something now");
+    log.info("If I were a real strategy I'd create something now [category: %s]", category);
     return null;
   }
 
   @Override
   public AutoScalingData terminate(List<String> ips)
   {
-    log.info("If I were a real strategy I'd terminate %s now", ips);
+    log.info("If I were a real strategy I'd terminate %s now [category: %s]", ips, category);
     return null;
   }
 
   @Override
   public AutoScalingData terminateWithIds(List<String> ids)
   {
-    log.info("If I were a real strategy I'd terminate %s now", ids);
+    log.info("If I were a real strategy I'd terminate %s now [category: %s]", ids, category);
     return null;
   }
 
   @Override
   public List<String> ipToIdLookup(List<String> ips)
   {
-    log.info("I'm not a real strategy so I'm returning what I got %s", ips);
+    log.info("I'm not a real strategy so I'm returning what I got %s [category: %s]", ips, category);
     return ips;
   }
 
   @Override
   public List<String> idToIpLookup(List<String> nodeIds)
   {
-    log.info("I'm not a real strategy so I'm returning what I got %s", nodeIds);
+    log.info("I'm not a real strategy so I'm returning what I got %s [category: %s]", nodeIds, category);
     return nodeIds;
   }
+
+  @Override
+  public String toString()
+  {
+    return "NoopAutoScaler{" +
+           "category='" + category + '\'' +
+           '}';
+  }
 }