Introduce autoscaling policies (#54473)

This commit is the first in a series of commits that introduces
autoscaling policies, and APIs for working with them. For now, we
introduce the basic infrastructure, and a single API for putting an
autoscaling policy. We will follow in rapid succession with APIs for
getting, and deleting autoscaling policies.

docs/reference/autoscaling/apis/autoscaling-apis.asciidoc

@@ -10,6 +10,8 @@ You can use the following APIs to perform autoscaling operations.
 === Top-Level
 
 * <<autoscaling-get-autoscaling-decision,Get autoscaling decision>>
+* <<autoscaling-put-autoscaling-policy,Put autoscaling policy>>
 
 // top-level
 include::get-autoscaling-decision.asciidoc[]
+include::put-autoscaling-policy.asciidoc[]

docs/reference/autoscaling/apis/put-autoscaling-policy.asciidoc

@@ -0,0 +1,67 @@
+[role="xpack"]
+[testenv="platinum"]
+[[autoscaling-put-autoscaling-policy]]
+=== Put autoscaling policy API
+++++
+<titleabbrev>Put autoscaling policy</titleabbrev>
+++++
+
+Put autoscaling policy.
+
+[[autoscaling-put-autoscaling-policy-request]]
+==== {api-request-title}
+
+[source,console]
+--------------------------------------------------
+PUT /_autoscaling/policy/<name>
+{
+  "policy": {
+    "deciders": {
+      "always": {
+      }
+    }
+  }
+}
+--------------------------------------------------
+// TEST[s/<name>/name/]
+
+[[autoscaling-put-autoscaling-policy-prereqs]]
+==== {api-prereq-title}
+
+* If the {es} {security-features} are enabled, you must have
+`manage_autoscaling` cluster privileges. For more information, see
+<<security-privileges>>.
+
+[[autoscaling-put-autoscaling-policy-desc]]
+==== {api-description-title}
+
+This API puts an autoscaling policy with the provided name.
+
+[[autoscaling-put-autoscaling-policy-examples]]
+==== {api-examples-title}
+
+This example puts an autoscaling policy named `hot` using the always
+autoscaling decider.
+
+[source,console]
+--------------------------------------------------
+PUT /_autoscaling/policy/hot
+{
+  "policy": {
+    "deciders": {
+      "always": {
+      }
+    }
+  }
+}
+--------------------------------------------------
+// TEST
+
+The API returns the following result:
+
+[source,console-result]
+--------------------------------------------------
+{
+  "acknowledged": true
+}
+--------------------------------------------------

x-pack/plugin/autoscaling/build.gradle

@@ -14,6 +14,16 @@ archivesBaseName = 'x-pack-autoscaling'
 
 integTest.enabled = false
 
+task internalClusterTest(type: Test) {
+  description = 'Java fantasy integration tests'
+  mustRunAfter test
+
+  include '**/*IT.class'
+  systemProperty 'es.set.netty.runtime.available.processors', 'false'
+}
+
+check.dependsOn internalClusterTest
+
 dependencies {
   compileOnly project(path: xpackModule('core'), configuration: 'default')
   testCompile project(path: xpackModule('core'), configuration: 'testArtifacts')
@@ -27,3 +37,9 @@ gradle.projectsEvaluated {
     .findAll { it.path.startsWith(project.path + ":qa") }
     .each { check.dependsOn it.check }
 }
+
+testingConventions.naming {
+  IT {
+    baseClass "org.elasticsearch.xpack.autoscaling.AutoscalingIntegTestCase"
+  }
+}

x-pack/plugin/autoscaling/qa/rest/src/test/resources/rest-api-spec/test/autoscaling/put_autoscaling_policy.yml

@@ -0,0 +1,21 @@
+---
+"Test put autoscaling decision":
+  - do:
+      autoscaling.put_autoscaling_policy:
+        name: hot
+        body:
+          policy:
+            deciders:
+              always: {}
+
+  - match: { "acknowledged": true }
+
+  - do:
+      catch: bad_request
+      autoscaling.put_autoscaling_policy:
+        name: hot
+        body:
+          policy:
+            deciders:
+              does_not_exist: {}
+

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/Autoscaling.java

@@ -9,20 +9,32 @@ package org.elasticsearch.xpack.autoscaling;
 import org.elasticsearch.Build;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.IndexScopedSettings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
 import org.elasticsearch.xpack.autoscaling.action.GetAutoscalingDecisionAction;
+import org.elasticsearch.xpack.autoscaling.action.PutAutoscalingPolicyAction;
 import org.elasticsearch.xpack.autoscaling.action.TransportGetAutoscalingDecisionAction;
+import org.elasticsearch.xpack.autoscaling.action.TransportPutAutoscalingPolicyAction;
+import org.elasticsearch.xpack.autoscaling.decision.AlwaysAutoscalingDecider;
+import org.elasticsearch.xpack.autoscaling.decision.AutoscalingDecider;
 import org.elasticsearch.xpack.autoscaling.rest.RestGetAutoscalingDecisionHandler;
+import org.elasticsearch.xpack.autoscaling.rest.RestPutAutoscalingPolicyHandler;
+import org.elasticsearch.xpack.core.XPackPlugin;
 
 import java.util.List;
 import java.util.function.Supplier;
@@ -78,10 +90,17 @@ public class Autoscaling extends Plugin implements ActionPlugin {
         }
     }
 
+    boolean isSnapshot() {
+        return Build.CURRENT.isSnapshot();
+    }
+
     @Override
     public List<ActionHandler<? extends ActionRequest, ? extends ActionResponse>> getActions() {
         if (enabled) {
-            return List.of(new ActionHandler<>(GetAutoscalingDecisionAction.INSTANCE, TransportGetAutoscalingDecisionAction.class));
+            return List.of(
+                new ActionHandler<>(GetAutoscalingDecisionAction.INSTANCE, TransportGetAutoscalingDecisionAction.class),
+                new ActionHandler<>(PutAutoscalingPolicyAction.INSTANCE, TransportPutAutoscalingPolicyAction.class)
+            );
         } else {
             return List.of();
         }
@@ -98,14 +117,35 @@ public class Autoscaling extends Plugin implements ActionPlugin {
         final Supplier<DiscoveryNodes> nodesInCluster
     ) {
         if (enabled) {
-            return List.of(new RestGetAutoscalingDecisionHandler());
+            return List.of(new RestGetAutoscalingDecisionHandler(), new RestPutAutoscalingPolicyHandler());
         } else {
             return List.of();
         }
     }
 
-    boolean isSnapshot() {
-        return Build.CURRENT.isSnapshot();
+    @Override
+    public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
+        return List.of(
+            new NamedWriteableRegistry.Entry(Metadata.Custom.class, AutoscalingMetadata.NAME, AutoscalingMetadata::new),
+            new NamedWriteableRegistry.Entry(NamedDiff.class, AutoscalingMetadata.NAME, AutoscalingMetadata.AutoscalingMetadataDiff::new),
+            new NamedWriteableRegistry.Entry(AutoscalingDecider.class, AlwaysAutoscalingDecider.NAME, AlwaysAutoscalingDecider::new)
+        );
+    }
+
+    @Override
+    public List<NamedXContentRegistry.Entry> getNamedXContent() {
+        return List.of(
+            new NamedXContentRegistry.Entry(Metadata.Custom.class, new ParseField(AutoscalingMetadata.NAME), AutoscalingMetadata::parse),
+            new NamedXContentRegistry.Entry(
+                AutoscalingDecider.class,
+                new ParseField(AlwaysAutoscalingDecider.NAME),
+                AlwaysAutoscalingDecider::parse
+            )
+        );
+    }
+
+    protected XPackLicenseState getLicenseState() {
+        return XPackPlugin.getSharedLicenseState();
     }
 
 }

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/AutoscalingMetadata.java

@@ -0,0 +1,170 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.AbstractDiffable;
+import org.elasticsearch.cluster.Diff;
+import org.elasticsearch.cluster.DiffableUtils;
+import org.elasticsearch.cluster.NamedDiff;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicyMetadata;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class AutoscalingMetadata implements Metadata.Custom {
+
+    public static final String NAME = "autoscaling";
+
+    public static final AutoscalingMetadata EMPTY = new AutoscalingMetadata(Collections.emptySortedMap());
+
+    private static final ParseField POLICIES_FIELD = new ParseField("policies");
+
+    @SuppressWarnings("unchecked")
+    private static final ConstructingObjectParser<AutoscalingMetadata, Void> PARSER = new ConstructingObjectParser<>(
+        NAME,
+        c -> new AutoscalingMetadata(
+            new TreeMap<>(
+                ((List<AutoscalingPolicyMetadata>) c[0]).stream().collect(Collectors.toMap(p -> p.policy().name(), Function.identity()))
+            )
+        )
+    );
+
+    static {
+        PARSER.declareNamedObjects(
+            ConstructingObjectParser.constructorArg(),
+            (p, c, n) -> AutoscalingPolicyMetadata.parse(p, n),
+            POLICIES_FIELD
+        );
+    }
+
+    public static AutoscalingMetadata parse(final XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+
+    private final SortedMap<String, AutoscalingPolicyMetadata> policies;
+
+    public SortedMap<String, AutoscalingPolicyMetadata> policies() {
+        return policies;
+    }
+
+    public AutoscalingMetadata(final SortedMap<String, AutoscalingPolicyMetadata> policies) {
+        this.policies = policies;
+    }
+
+    public AutoscalingMetadata(final StreamInput in) throws IOException {
+        final int size = in.readVInt();
+        final SortedMap<String, AutoscalingPolicyMetadata> policies = new TreeMap<>();
+        for (int i = 0; i < size; i++) {
+            final AutoscalingPolicyMetadata policyMetadata = new AutoscalingPolicyMetadata(in);
+            policies.put(policyMetadata.policy().name(), policyMetadata);
+        }
+        this.policies = policies;
+    }
+
+    @Override
+    public void writeTo(final StreamOutput out) throws IOException {
+        out.writeVInt(policies.size());
+        for (final Map.Entry<String, AutoscalingPolicyMetadata> policy : policies.entrySet()) {
+            policy.getValue().writeTo(out);
+        }
+    }
+
+    @Override
+    public EnumSet<Metadata.XContentContext> context() {
+        return Metadata.ALL_CONTEXTS;
+    }
+
+    @Override
+    public Diff<Metadata.Custom> diff(final Metadata.Custom previousState) {
+        return new AutoscalingMetadataDiff((AutoscalingMetadata) previousState, this);
+    }
+
+    @Override
+    public String getWriteableName() {
+        return NAME;
+    }
+
+    @Override
+    public Version getMinimalSupportedVersion() {
+        // TODO: update on backport
+        return Version.V_8_0_0;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.field(POLICIES_FIELD.getPreferredName(), policies);
+        return builder;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        final AutoscalingMetadata metadata = (AutoscalingMetadata) o;
+        return policies.equals(metadata.policies);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(policies);
+    }
+
+    public static class AutoscalingMetadataDiff implements NamedDiff<Metadata.Custom> {
+
+        final Diff<Map<String, AutoscalingPolicyMetadata>> policies;
+
+        public AutoscalingMetadataDiff(final AutoscalingMetadata before, final AutoscalingMetadata after) {
+            this.policies = DiffableUtils.diff(before.policies, after.policies, DiffableUtils.getStringKeySerializer());
+        }
+
+        public AutoscalingMetadataDiff(final StreamInput in) throws IOException {
+            this.policies = DiffableUtils.readJdkMapDiff(
+                in,
+                DiffableUtils.getStringKeySerializer(),
+                AutoscalingPolicyMetadata::new,
+                AutoscalingMetadataDiff::readFrom
+            );
+        }
+
+        @Override
+        public Metadata.Custom apply(final Metadata.Custom part) {
+            return new AutoscalingMetadata(new TreeMap<>(policies.apply(((AutoscalingMetadata) part).policies)));
+        }
+
+        @Override
+        public String getWriteableName() {
+            return NAME;
+        }
+
+        @Override
+        public void writeTo(final StreamOutput out) throws IOException {
+            policies.writeTo(out);
+        }
+
+        static Diff<AutoscalingPolicyMetadata> readFrom(final StreamInput in) throws IOException {
+            return AbstractDiffable.readDiffFrom(AutoscalingPolicyMetadata::new, in);
+        }
+
+    }
+
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/action/GetAutoscalingDecisionAction.java

@@ -14,7 +14,7 @@ import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
-import org.elasticsearch.xpack.autoscaling.AutoscalingDecisions;
+import org.elasticsearch.xpack.autoscaling.decision.AutoscalingDecisions;
 
 import java.io.IOException;
 import java.util.Map;

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/action/PutAutoscalingPolicyAction.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.action;
+
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicy;
+
+import java.io.IOException;
+
+public class PutAutoscalingPolicyAction extends ActionType<AcknowledgedResponse> {
+
+    public static final PutAutoscalingPolicyAction INSTANCE = new PutAutoscalingPolicyAction();
+    public static final String NAME = "cluster:admin/autoscaling/put_autoscaling_policy";
+
+    private PutAutoscalingPolicyAction() {
+        super(NAME, AcknowledgedResponse::new);
+    }
+
+    public static class Request extends AcknowledgedRequest<Request> implements ToXContentObject {
+
+        static final ParseField POLICY_FIELD = new ParseField("policy");
+
+        @SuppressWarnings("unchecked")
+        private static final ConstructingObjectParser<Request, String> PARSER = new ConstructingObjectParser<>(
+            "put_autoscaling_policy_request",
+            a -> new Request((AutoscalingPolicy) a[0])
+        );
+
+        static {
+            PARSER.declareObject(ConstructingObjectParser.constructorArg(), AutoscalingPolicy::parse, POLICY_FIELD);
+        }
+
+        public static Request parse(final XContentParser parser, final String name) {
+            return PARSER.apply(parser, name);
+        }
+
+        private final AutoscalingPolicy policy;
+
+        public AutoscalingPolicy policy() {
+            return policy;
+        }
+
+        public Request(final AutoscalingPolicy policy) {
+            this.policy = policy;
+        }
+
+        public Request(final StreamInput in) throws IOException {
+            super(in);
+            policy = new AutoscalingPolicy(in);
+        }
+
+        @Override
+        public void writeTo(final StreamOutput out) throws IOException {
+            super.writeTo(out);
+            policy.writeTo(out);
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            // TODO: validate that the policy deciders are non-empty
+            return null;
+        }
+
+        @Override
+        public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
+            builder.startObject();
+            {
+                builder.field(POLICY_FIELD.getPreferredName(), policy);
+            }
+            builder.endObject();
+            return builder;
+        }
+
+    }
+
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/action/TransportPutAutoscalingPolicyAction.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.action;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.action.support.master.TransportMasterNodeAction;
+import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.autoscaling.AutoscalingMetadata;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicy;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicyMetadata;
+
+import java.io.IOException;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+public class TransportPutAutoscalingPolicyAction extends TransportMasterNodeAction<
+    PutAutoscalingPolicyAction.Request,
+    AcknowledgedResponse> {
+
+    private static final Logger logger = LogManager.getLogger(TransportPutAutoscalingPolicyAction.class);
+
+    @Inject
+    public TransportPutAutoscalingPolicyAction(
+        final TransportService transportService,
+        final ClusterService clusterService,
+        final ThreadPool threadPool,
+        final ActionFilters actionFilters,
+        final IndexNameExpressionResolver indexNameExpressionResolver
+    ) {
+        super(
+            PutAutoscalingPolicyAction.NAME,
+            transportService,
+            clusterService,
+            threadPool,
+            actionFilters,
+            PutAutoscalingPolicyAction.Request::new,
+            indexNameExpressionResolver
+        );
+    }
+
+    @Override
+    protected String executor() {
+        return ThreadPool.Names.SAME;
+    }
+
+    @Override
+    protected AcknowledgedResponse read(final StreamInput in) throws IOException {
+        return new AcknowledgedResponse(in);
+    }
+
+    @Override
+    protected void masterOperation(
+        final Task task,
+        final PutAutoscalingPolicyAction.Request request,
+        final ClusterState state,
+        final ActionListener<AcknowledgedResponse> listener
+    ) {
+        clusterService.submitStateUpdateTask("put-autoscaling-policy", new AckedClusterStateUpdateTask<>(request, listener) {
+
+            @Override
+            protected AcknowledgedResponse newResponse(final boolean acknowledged) {
+                return new AcknowledgedResponse(acknowledged);
+            }
+
+            @Override
+            public ClusterState execute(final ClusterState currentState) {
+                return putAutoscalingPolicy(currentState, request.policy(), logger);
+            }
+
+        });
+    }
+
+    @Override
+    protected ClusterBlockException checkBlock(final PutAutoscalingPolicyAction.Request request, final ClusterState state) {
+        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
+    }
+
+    static ClusterState putAutoscalingPolicy(final ClusterState currentState, final AutoscalingPolicy policy, final Logger logger) {
+        final ClusterState.Builder builder = ClusterState.builder(currentState);
+        final AutoscalingMetadata currentMetadata;
+        if (currentState.metadata().custom(AutoscalingMetadata.NAME) != null) {
+            currentMetadata = currentState.metadata().custom(AutoscalingMetadata.NAME);
+        } else {
+            currentMetadata = AutoscalingMetadata.EMPTY;
+        }
+        final SortedMap<String, AutoscalingPolicyMetadata> newPolicies = new TreeMap<>(currentMetadata.policies());
+        final AutoscalingPolicyMetadata newPolicyMetadata = new AutoscalingPolicyMetadata(policy);
+        final AutoscalingPolicyMetadata oldPolicyMetadata = newPolicies.put(policy.name(), newPolicyMetadata);
+        if (oldPolicyMetadata == null) {
+            logger.info("adding autoscaling policy [{}]", policy.name());
+        } else if (oldPolicyMetadata.equals(newPolicyMetadata)) {
+            logger.info("skipping updating autoscaling policy [{}] due to no change in policy", policy.name());
+            return currentState;
+        } else {
+            logger.info("updating autoscaling policy [{}]", policy.name());
+        }
+        final AutoscalingMetadata newMetadata = new AutoscalingMetadata(newPolicies);
+        builder.metadata(Metadata.builder(currentState.getMetadata()).putCustom(AutoscalingMetadata.NAME, newMetadata).build());
+        return builder.build();
+    }
+
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/decision/AlwaysAutoscalingDecider.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.decision;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ObjectParser;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+
+public class AlwaysAutoscalingDecider implements AutoscalingDecider {
+
+    public static final String NAME = "always";
+
+    private static final ObjectParser<AlwaysAutoscalingDecider, Void> PARSER = new ObjectParser<>(NAME, AlwaysAutoscalingDecider::new);
+
+    public static AlwaysAutoscalingDecider parse(final XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+
+    public AlwaysAutoscalingDecider() {}
+
+    @SuppressWarnings("unused")
+    public AlwaysAutoscalingDecider(final StreamInput in) {
+
+    }
+
+    @Override
+    public String name() {
+        return NAME;
+    }
+
+    @Override
+    public AutoscalingDecision scale() {
+        return new AutoscalingDecision(NAME, AutoscalingDecisionType.SCALE_UP, "always");
+    }
+
+    @Override
+    public String getWriteableName() {
+        return NAME;
+    }
+
+    @Override
+    public void writeTo(final StreamOutput out) {
+
+    }
+
+    @Override
+    public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
+        builder.startObject();
+        {}
+        builder.endObject();
+        return builder;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        return 0;
+    }
+
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/decision/AutoscalingDecider.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.decision;
+
+import org.elasticsearch.common.io.stream.NamedWriteable;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+
+/**
+ * Represents an autoscaling decider, a component that determines whether or not to scale.
+ */
+public interface AutoscalingDecider extends ToXContentObject, NamedWriteable {
+
+    /**
+     * The name of the autoscaling decider.
+     *
+     * @return the name
+     */
+    String name();
+
+    /**
+     * Whether or not to scale based on the current state.
+     *
+     * @return the autoscaling decision
+     */
+    AutoscalingDecision scale();
+
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/AutoscalingDecision.java → x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/decision/AutoscalingDecision.java

@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-package org.elasticsearch.xpack.autoscaling;
+package org.elasticsearch.xpack.autoscaling.decision;
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/AutoscalingDecisionType.java → x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/decision/AutoscalingDecisionType.java

@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-package org.elasticsearch.xpack.autoscaling;
+package org.elasticsearch.xpack.autoscaling.decision;
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/AutoscalingDecisions.java → x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/decision/AutoscalingDecisions.java

@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-package org.elasticsearch.xpack.autoscaling;
+package org.elasticsearch.xpack.autoscaling.decision;
 
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -65,7 +65,7 @@ public class AutoscalingDecisions implements ToXContent, Writeable {
     public boolean equals(final Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
-        final org.elasticsearch.xpack.autoscaling.AutoscalingDecisions that = (org.elasticsearch.xpack.autoscaling.AutoscalingDecisions) o;
+        final AutoscalingDecisions that = (AutoscalingDecisions) o;
         return decisions.equals(that.decisions);
     }
 

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/policy/AutoscalingPolicy.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.policy;
+
+import org.elasticsearch.cluster.AbstractDiffable;
+import org.elasticsearch.cluster.Diffable;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.xpack.autoscaling.decision.AutoscalingDecider;
+
+import java.io.IOException;
+import java.util.AbstractMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class AutoscalingPolicy extends AbstractDiffable<AutoscalingPolicy> implements Diffable<AutoscalingPolicy>, ToXContentObject {
+
+    public static final String NAME = "autoscaling_policy";
+
+    public static final ParseField DECIDERS_FIELD = new ParseField("deciders");
+
+    private static final ConstructingObjectParser<AutoscalingPolicy, String> PARSER;
+
+    static {
+        PARSER = new ConstructingObjectParser<>(NAME, false, (c, name) -> {
+            @SuppressWarnings("unchecked")
+            final List<Map.Entry<String, AutoscalingDecider>> deciders = (List<Map.Entry<String, AutoscalingDecider>>) c[0];
+            return new AutoscalingPolicy(
+                name,
+                new TreeMap<>(deciders.stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)))
+            );
+        });
+        PARSER.declareNamedObjects(
+            ConstructingObjectParser.constructorArg(),
+            (p, c, n) -> new AbstractMap.SimpleEntry<>(n, p.namedObject(AutoscalingDecider.class, n, null)),
+            DECIDERS_FIELD
+        );
+    }
+
+    public static AutoscalingPolicy parse(final XContentParser parser, final String name) {
+        return PARSER.apply(parser, name);
+    }
+
+    private final String name;
+
+    public String name() {
+        return name;
+    }
+
+    private final SortedMap<String, AutoscalingDecider> deciders;
+
+    public SortedMap<String, AutoscalingDecider> deciders() {
+        return deciders;
+    }
+
+    public AutoscalingPolicy(final String name, final SortedMap<String, AutoscalingDecider> deciders) {
+        this.name = Objects.requireNonNull(name);
+        // TODO: validate that the policy deciders are non-empty
+        this.deciders = Objects.requireNonNull(deciders);
+    }
+
+    public AutoscalingPolicy(final StreamInput in) throws IOException {
+        name = in.readString();
+        deciders = new TreeMap<>(
+            in.readNamedWriteableList(AutoscalingDecider.class)
+                .stream()
+                .collect(Collectors.toMap(AutoscalingDecider::name, Function.identity()))
+        );
+    }
+
+    @Override
+    public void writeTo(final StreamOutput out) throws IOException {
+        out.writeString(name);
+        out.writeNamedWriteableList(deciders.values().stream().collect(Collectors.toUnmodifiableList()));
+    }
+
+    @Override
+    public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
+        builder.startObject();
+        {
+            builder.startObject(DECIDERS_FIELD.getPreferredName());
+            {
+                for (final Map.Entry<String, AutoscalingDecider> entry : deciders.entrySet()) {
+                    builder.field(entry.getKey(), entry.getValue());
+                }
+            }
+            builder.endObject();
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        final AutoscalingPolicy that = (AutoscalingPolicy) o;
+        return name.equals(that.name) && deciders.equals(that.deciders);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(name, deciders);
+    }
+
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/policy/AutoscalingPolicyMetadata.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.policy;
+
+import org.elasticsearch.cluster.AbstractDiffable;
+import org.elasticsearch.cluster.Diffable;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public class AutoscalingPolicyMetadata extends AbstractDiffable<AutoscalingPolicyMetadata>
+    implements
+        Diffable<AutoscalingPolicyMetadata>,
+        ToXContentObject {
+
+    static final ParseField POLICY_FIELD = new ParseField("policy");
+
+    @SuppressWarnings("unchecked")
+    private static final ConstructingObjectParser<AutoscalingPolicyMetadata, String> PARSER;
+    static {
+        PARSER = new ConstructingObjectParser<>("autoscaling_policy_metadata", a -> {
+            final AutoscalingPolicy policy = (AutoscalingPolicy) a[0];
+            return new AutoscalingPolicyMetadata(policy);
+        });
+        PARSER.declareObject(ConstructingObjectParser.constructorArg(), AutoscalingPolicy::parse, POLICY_FIELD);
+    }
+
+    public static AutoscalingPolicyMetadata parse(final XContentParser parser, final String name) {
+        return PARSER.apply(parser, name);
+    }
+
+    private final AutoscalingPolicy policy;
+
+    public AutoscalingPolicy policy() {
+        return policy;
+    }
+
+    public AutoscalingPolicyMetadata(final AutoscalingPolicy policy) {
+        this.policy = policy;
+    }
+
+    public AutoscalingPolicyMetadata(final StreamInput in) throws IOException {
+        policy = new AutoscalingPolicy(in);
+    }
+
+    @Override
+    public void writeTo(final StreamOutput out) throws IOException {
+        policy.writeTo(out);
+    }
+
+    @Override
+    public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
+        builder.startObject();
+        {
+            builder.field(POLICY_FIELD.getPreferredName(), policy);
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        final AutoscalingPolicyMetadata that = (AutoscalingPolicyMetadata) o;
+        return policy.equals(that.policy);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(policy);
+    }
+
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/rest/RestPutAutoscalingPolicyHandler.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.rest;
+
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestToXContentListener;
+import org.elasticsearch.xpack.autoscaling.action.PutAutoscalingPolicyAction;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.elasticsearch.rest.RestRequest.Method.PUT;
+
+public class RestPutAutoscalingPolicyHandler extends BaseRestHandler {
+
+    @Override
+    public List<Route> routes() {
+        return List.of(new Route(PUT, "/_autoscaling/policy/{name}"));
+    }
+
+    @Override
+    public String getName() {
+        return "put_autoscaling_policy";
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(final RestRequest restRequest, final NodeClient client) throws IOException {
+        final String name = restRequest.param("name");
+        final PutAutoscalingPolicyAction.Request request;
+        try (XContentParser parser = restRequest.contentParser()) {
+            request = PutAutoscalingPolicyAction.Request.parse(parser, name);
+        }
+        return channel -> client.execute(PutAutoscalingPolicyAction.INSTANCE, request, new RestToXContentListener<>(channel));
+    }
+
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/AutoscalingIntegTestCase.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import java.util.Collection;
+import java.util.List;
+
+public abstract class AutoscalingIntegTestCase extends ESIntegTestCase {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return List.of(LocalStateAutoscaling.class);
+    }
+
+    @Override
+    protected Settings nodeSettings(final int nodeOrdinal) {
+        final Settings.Builder builder = Settings.builder().put(super.nodeSettings(nodeOrdinal));
+        builder.put(Autoscaling.AUTOSCALING_ENABLED_SETTING.getKey(), true);
+        return builder.build();
+    }
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/AutoscalingMetadataDiffableSerializationTests.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling;
+
+import org.elasticsearch.cluster.Diff;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractDiffableSerializationTestCase;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicy;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicyMetadata;
+
+import java.io.IOException;
+import java.util.SortedMap;
+import java.util.TreeMap;
+
+import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.mutateAutoscalingPolicy;
+import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.randomAutoscalingMetadata;
+import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.randomAutoscalingPolicy;
+
+public class AutoscalingMetadataDiffableSerializationTests extends AbstractDiffableSerializationTestCase<Metadata.Custom> {
+
+    @Override
+    protected NamedWriteableRegistry getNamedWriteableRegistry() {
+        return new NamedWriteableRegistry(new Autoscaling(Settings.EMPTY).getNamedWriteables());
+    }
+
+    @Override
+    protected NamedXContentRegistry xContentRegistry() {
+        return new NamedXContentRegistry(new Autoscaling(Settings.EMPTY).getNamedXContent());
+    }
+
+    @Override
+    protected AutoscalingMetadata doParseInstance(final XContentParser parser) throws IOException {
+        return AutoscalingMetadata.parse(parser);
+    }
+
+    @Override
+    protected Writeable.Reader<Metadata.Custom> instanceReader() {
+        return AutoscalingMetadata::new;
+    }
+
+    @Override
+    protected AutoscalingMetadata createTestInstance() {
+        return randomAutoscalingMetadata();
+    }
+
+    @Override
+    protected Metadata.Custom makeTestChanges(final Metadata.Custom testInstance) {
+        return mutateInstance(testInstance);
+    }
+
+    @Override
+    protected Metadata.Custom mutateInstance(final Metadata.Custom instance) {
+        final AutoscalingMetadata metadata = (AutoscalingMetadata) instance;
+        final SortedMap<String, AutoscalingPolicyMetadata> policies = new TreeMap<>(metadata.policies());
+        if (policies.size() == 0 || randomBoolean()) {
+            final AutoscalingPolicy policy = randomAutoscalingPolicy();
+            policies.put(policy.name(), new AutoscalingPolicyMetadata(policy));
+        } else {
+            // randomly remove a policy
+            final String name = randomFrom(policies.keySet());
+            final AutoscalingPolicyMetadata policyMetadata = policies.remove(name);
+            final AutoscalingPolicy mutatedPolicy = mutateAutoscalingPolicy(policyMetadata.policy());
+            policies.put(mutatedPolicy.name(), new AutoscalingPolicyMetadata(mutatedPolicy));
+        }
+        return new AutoscalingMetadata(policies);
+    }
+
+    @Override
+    protected Writeable.Reader<Diff<Metadata.Custom>> diffReader() {
+        return AutoscalingMetadata.AutoscalingMetadataDiff::new;
+    }
+
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/AutoscalingTestCase.java

@@ -8,21 +8,33 @@ package org.elasticsearch.xpack.autoscaling;
 
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.autoscaling.decision.AlwaysAutoscalingDecider;
+import org.elasticsearch.xpack.autoscaling.decision.AutoscalingDecider;
+import org.elasticsearch.xpack.autoscaling.decision.AutoscalingDecision;
+import org.elasticsearch.xpack.autoscaling.decision.AutoscalingDecisionType;
+import org.elasticsearch.xpack.autoscaling.decision.AutoscalingDecisions;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicy;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicyMetadata;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 public abstract class AutoscalingTestCase extends ESTestCase {
 
-    static AutoscalingDecision randomAutoscalingDecision() {
+    public static AutoscalingDecision randomAutoscalingDecision() {
         return randomAutoscalingDecisionOfType(randomFrom(AutoscalingDecisionType.values()));
     }
 
-    static AutoscalingDecision randomAutoscalingDecisionOfType(final AutoscalingDecisionType type) {
+    public static AutoscalingDecision randomAutoscalingDecisionOfType(final AutoscalingDecisionType type) {
         return new AutoscalingDecision(randomAlphaOfLength(8), type, randomAlphaOfLength(8));
     }
 
-    static AutoscalingDecisions randomAutoscalingDecisions() {
+    public static AutoscalingDecisions randomAutoscalingDecisions() {
         final int numberOfDecisions = 1 + randomIntBetween(1, 8);
         final List<AutoscalingDecision> decisions = new ArrayList<>(numberOfDecisions);
         for (int i = 0; i < numberOfDecisions; i++) {
@@ -34,7 +46,7 @@ public abstract class AutoscalingTestCase extends ESTestCase {
         return randomAutoscalingDecisions(numberOfDownDecisions, numberOfNoDecisions, numberOfUpDecisions);
     }
 
-    static AutoscalingDecisions randomAutoscalingDecisions(
+    public static AutoscalingDecisions randomAutoscalingDecisions(
         final int numberOfDownDecisions,
         final int numberOfNoDecisions,
         final int numberOfUpDecisions
@@ -53,4 +65,55 @@ public abstract class AutoscalingTestCase extends ESTestCase {
         return new AutoscalingDecisions(decisions);
     }
 
+    public static SortedMap<String, AutoscalingDecider> randomAutoscalingDeciders() {
+        return new TreeMap<>(
+            List.of(new AlwaysAutoscalingDecider()).stream().collect(Collectors.toMap(AutoscalingDecider::name, Function.identity()))
+        );
+    }
+
+    public static AutoscalingPolicy randomAutoscalingPolicy() {
+        return randomAutoscalingPolicyOfName(randomAlphaOfLength(8));
+    }
+
+    public static AutoscalingPolicy randomAutoscalingPolicyOfName(final String name) {
+        return new AutoscalingPolicy(name, randomAutoscalingDeciders());
+    }
+
+    public static AutoscalingPolicy mutateAutoscalingPolicy(final AutoscalingPolicy instance) {
+        final SortedMap<String, AutoscalingDecider> deciders;
+        if (randomBoolean()) {
+            // if the policy name did not change, or randomly, use a mutated set of deciders
+            deciders = mutateAutoscalingDeciders(instance.deciders());
+        } else {
+            deciders = instance.deciders();
+        }
+        return new AutoscalingPolicy(randomValueOtherThan(instance.name(), () -> randomAlphaOfLength(8)), deciders);
+    }
+
+    public static SortedMap<String, AutoscalingDecider> mutateAutoscalingDeciders(final SortedMap<String, AutoscalingDecider> deciders) {
+        if (deciders.size() == 0) {
+            return randomAutoscalingDeciders();
+        } else {
+            // use a proper subset of the deciders
+            return new TreeMap<>(
+                randomSubsetOf(randomIntBetween(0, deciders.size() - 1), deciders.entrySet()).stream()
+                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
+            );
+        }
+    }
+
+    public static AutoscalingMetadata randomAutoscalingMetadata() {
+        return randomAutoscalingMetadataOfPolicyCount(randomIntBetween(0, 8));
+    }
+
+    public static AutoscalingMetadata randomAutoscalingMetadataOfPolicyCount(final int numberOfPolicies) {
+        final SortedMap<String, AutoscalingPolicyMetadata> policies = new TreeMap<>();
+        for (int i = 0; i < numberOfPolicies; i++) {
+            final AutoscalingPolicy policy = randomAutoscalingPolicy();
+            final AutoscalingPolicyMetadata policyMetadata = new AutoscalingPolicyMetadata(policy);
+            policies.put(policy.name(), policyMetadata);
+        }
+        return new AutoscalingMetadata(policies);
+    }
+
 }

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/LocalStateAutoscaling.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.xpack.core.LocalStateCompositeXPackPlugin;
+
+public class LocalStateAutoscaling extends LocalStateCompositeXPackPlugin {
+
+    public LocalStateAutoscaling(final Settings settings) {
+        super(settings, null);
+        plugins.add(new Autoscaling(settings) {
+
+            @Override
+            protected XPackLicenseState getLicenseState() {
+                return LocalStateAutoscaling.this.getLicenseState();
+            }
+
+        });
+    }
+
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/action/TransportPutAutoscalingPolicyActionIT.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.action;
+
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.xpack.autoscaling.AutoscalingIntegTestCase;
+import org.elasticsearch.xpack.autoscaling.AutoscalingMetadata;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicy;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.mutateAutoscalingDeciders;
+import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.randomAutoscalingPolicy;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.Matchers.sameInstance;
+
+public class TransportPutAutoscalingPolicyActionIT extends AutoscalingIntegTestCase {
+
+    public void testAddPolicy() {
+        final AutoscalingPolicy policy = putRandomAutoscalingPolicy();
+        final ClusterState state = client().admin().cluster().prepareState().get().getState();
+        final AutoscalingMetadata metadata = state.metadata().custom(AutoscalingMetadata.NAME);
+        assertNotNull(metadata);
+        assertThat(metadata.policies(), hasKey(policy.name()));
+        assertThat(metadata.policies().get(policy.name()).policy(), equalTo(policy));
+    }
+
+    public void testUpdatePolicy() {
+        final AutoscalingPolicy policy = putRandomAutoscalingPolicy();
+        final AutoscalingPolicy updatedPolicy = new AutoscalingPolicy(policy.name(), mutateAutoscalingDeciders(policy.deciders()));
+        putAutoscalingPolicy(updatedPolicy);
+        final ClusterState state = client().admin().cluster().prepareState().get().getState();
+        final AutoscalingMetadata metadata = state.metadata().custom(AutoscalingMetadata.NAME);
+        assertNotNull(metadata);
+        assertThat(metadata.policies(), hasKey(policy.name()));
+        assertThat(metadata.policies().get(policy.name()).policy(), equalTo(updatedPolicy));
+    }
+
+    public void testNoOpPolicy() {
+        final AutoscalingPolicy policy = putRandomAutoscalingPolicy();
+        final ClusterState beforeState = internalCluster().getInstance(ClusterService.class, internalCluster().getMasterName()).state();
+        putAutoscalingPolicy(policy);
+        final ClusterState afterState = internalCluster().getInstance(ClusterService.class, internalCluster().getMasterName()).state();
+        assertThat(
+            beforeState.metadata().custom(AutoscalingMetadata.NAME),
+            sameInstance(afterState.metadata().custom(AutoscalingMetadata.NAME))
+        );
+    }
+
+    private AutoscalingPolicy putRandomAutoscalingPolicy() {
+        final AutoscalingPolicy policy = randomAutoscalingPolicy();
+        putAutoscalingPolicy(policy);
+        return policy;
+    }
+
+    private void putAutoscalingPolicy(final AutoscalingPolicy policy) {
+        final PutAutoscalingPolicyAction.Request request = new PutAutoscalingPolicyAction.Request(policy);
+        assertAcked(client().execute(PutAutoscalingPolicyAction.INSTANCE, request).actionGet());
+    }
+
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/action/TransportPutAutoscalingPolicyActionTests.java

@@ -0,0 +1,164 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.action;
+
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlocks;
+import org.elasticsearch.cluster.coordination.NoMasterBlockService;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.autoscaling.AutoscalingMetadata;
+import org.elasticsearch.xpack.autoscaling.AutoscalingTestCase;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicy;
+import org.elasticsearch.xpack.autoscaling.policy.AutoscalingPolicyMetadata;
+
+import java.util.Map;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasKey;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.sameInstance;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+public class TransportPutAutoscalingPolicyActionTests extends AutoscalingTestCase {
+
+    public void testWriteBlock() {
+        final TransportPutAutoscalingPolicyAction action = new TransportPutAutoscalingPolicyAction(
+            mock(TransportService.class),
+            mock(ClusterService.class),
+            mock(ThreadPool.class),
+            mock(ActionFilters.class),
+            mock(IndexNameExpressionResolver.class)
+        );
+        final ClusterBlocks blocks = ClusterBlocks.builder()
+            .addGlobalBlock(
+                randomFrom(
+                    Metadata.CLUSTER_READ_ONLY_BLOCK,
+                    Metadata.CLUSTER_READ_ONLY_ALLOW_DELETE_BLOCK,
+                    NoMasterBlockService.NO_MASTER_BLOCK_WRITES
+                )
+            )
+            .build();
+        final ClusterState state = ClusterState.builder(new ClusterName(randomAlphaOfLength(8))).blocks(blocks).build();
+        final ClusterBlockException e = action.checkBlock(new PutAutoscalingPolicyAction.Request(randomAutoscalingPolicy()), state);
+        assertThat(e, not(nullValue()));
+    }
+
+    public void testNoWriteBlock() {
+        final TransportPutAutoscalingPolicyAction action = new TransportPutAutoscalingPolicyAction(
+            mock(TransportService.class),
+            mock(ClusterService.class),
+            mock(ThreadPool.class),
+            mock(ActionFilters.class),
+            mock(IndexNameExpressionResolver.class)
+        );
+        final ClusterBlocks blocks = ClusterBlocks.builder().build();
+        final ClusterState state = ClusterState.builder(new ClusterName(randomAlphaOfLength(8))).blocks(blocks).build();
+        final ClusterBlockException e = action.checkBlock(new PutAutoscalingPolicyAction.Request(randomAutoscalingPolicy()), state);
+        assertThat(e, nullValue());
+    }
+
+    public void testAddPolicy() {
+        final ClusterState currentState;
+        {
+            final ClusterState.Builder builder = ClusterState.builder(new ClusterName(randomAlphaOfLength(8)));
+            if (randomBoolean()) {
+                builder.metadata(Metadata.builder().putCustom(AutoscalingMetadata.NAME, randomAutoscalingMetadata()));
+            }
+            currentState = builder.build();
+        }
+        // put an entirely new policy
+        final AutoscalingPolicy policy = randomAutoscalingPolicy();
+        final Logger mockLogger = mock(Logger.class);
+        final ClusterState state = TransportPutAutoscalingPolicyAction.putAutoscalingPolicy(currentState, policy, mockLogger);
+
+        // ensure the new policy is in the updated cluster state
+        final AutoscalingMetadata metadata = state.metadata().custom(AutoscalingMetadata.NAME);
+        assertNotNull(metadata);
+        assertThat(metadata.policies(), hasKey(policy.name()));
+        assertThat(metadata.policies().get(policy.name()).policy(), equalTo(policy));
+        verify(mockLogger).info("adding autoscaling policy [{}]", policy.name());
+        verifyNoMoreInteractions(mockLogger);
+
+        // ensure that existing policies were preserved
+        final AutoscalingMetadata currentMetadata = currentState.metadata().custom(AutoscalingMetadata.NAME);
+        if (currentMetadata != null) {
+            for (final Map.Entry<String, AutoscalingPolicyMetadata> entry : currentMetadata.policies().entrySet()) {
+                assertThat(metadata.policies(), hasKey(entry.getKey()));
+                assertThat(metadata.policies().get(entry.getKey()).policy(), equalTo(entry.getValue().policy()));
+            }
+        }
+    }
+
+    public void testUpdatePolicy() {
+        final ClusterState currentState;
+        {
+            final ClusterState.Builder builder = ClusterState.builder(new ClusterName(randomAlphaOfLength(8)));
+            builder.metadata(
+                Metadata.builder().putCustom(AutoscalingMetadata.NAME, randomAutoscalingMetadataOfPolicyCount(randomIntBetween(1, 8)))
+            );
+            currentState = builder.build();
+        }
+        final AutoscalingMetadata currentMetadata = currentState.metadata().custom(AutoscalingMetadata.NAME);
+        final String name = randomFrom(currentMetadata.policies().keySet());
+        // add to the existing deciders, to ensure the policy has changed
+        final AutoscalingPolicy policy = new AutoscalingPolicy(
+            name,
+            mutateAutoscalingDeciders(currentMetadata.policies().get(name).policy().deciders())
+        );
+        final Logger mockLogger = mock(Logger.class);
+        final ClusterState state = TransportPutAutoscalingPolicyAction.putAutoscalingPolicy(currentState, policy, mockLogger);
+
+        // ensure the updated policy is in the updated cluster state
+        final AutoscalingMetadata metadata = state.metadata().custom(AutoscalingMetadata.NAME);
+        assertNotNull(metadata);
+        assertThat(metadata.policies(), hasKey(policy.name()));
+        assertThat(metadata.policies().get(policy.name()).policy(), equalTo(policy));
+        verify(mockLogger).info("updating autoscaling policy [{}]", policy.name());
+        verifyNoMoreInteractions(mockLogger);
+
+        // ensure that existing policies were otherwise preserved
+        for (final Map.Entry<String, AutoscalingPolicyMetadata> entry : currentMetadata.policies().entrySet()) {
+            if (entry.getKey().equals(name)) {
+                continue;
+            }
+            assertThat(metadata.policies(), hasKey(entry.getKey()));
+            assertThat(metadata.policies().get(entry.getKey()).policy(), equalTo(entry.getValue().policy()));
+        }
+    }
+
+    public void testNoOpUpdatePolicy() {
+        final ClusterState currentState;
+        {
+            final ClusterState.Builder builder = ClusterState.builder(new ClusterName(randomAlphaOfLength(8)));
+            builder.metadata(
+                Metadata.builder().putCustom(AutoscalingMetadata.NAME, randomAutoscalingMetadataOfPolicyCount(randomIntBetween(1, 8)))
+            );
+            currentState = builder.build();
+        }
+        // randomly put an existing policy
+        final AutoscalingMetadata currentMetadata = currentState.metadata().custom(AutoscalingMetadata.NAME);
+        final AutoscalingPolicy policy = randomFrom(currentMetadata.policies().values()).policy();
+        final Logger mockLogger = mock(Logger.class);
+        final ClusterState state = TransportPutAutoscalingPolicyAction.putAutoscalingPolicy(currentState, policy, mockLogger);
+
+        assertThat(state, sameInstance(currentState));
+        verify(mockLogger).info("skipping updating autoscaling policy [{}] due to no change in policy", policy.name());
+        verifyNoMoreInteractions(mockLogger);
+    }
+
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/AutoscalingDecisionTests.java → x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/decision/AutoscalingDecisionTests.java

@@ -4,9 +4,10 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-package org.elasticsearch.xpack.autoscaling;
+package org.elasticsearch.xpack.autoscaling.decision;
 
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.xpack.autoscaling.AutoscalingTestCase;
 
 import java.io.IOException;
 

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/AutoscalingDecisionTypeWireSerializingTests.java → x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/decision/AutoscalingDecisionTypeWireSerializingTests.java

@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-package org.elasticsearch.xpack.autoscaling;
+package org.elasticsearch.xpack.autoscaling.decision;
 
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/AutoscalingDecisionWireSerializingTests.java → x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/decision/AutoscalingDecisionWireSerializingTests.java

@@ -4,10 +4,11 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-package org.elasticsearch.xpack.autoscaling;
+package org.elasticsearch.xpack.autoscaling.decision;
 
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
+import org.elasticsearch.xpack.autoscaling.AutoscalingTestCase;
 
 public class AutoscalingDecisionWireSerializingTests extends AbstractWireSerializingTestCase<AutoscalingDecision> {
 

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/AutoscalingDecisionsTests.java → x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/decision/AutoscalingDecisionsTests.java

@@ -4,7 +4,9 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-package org.elasticsearch.xpack.autoscaling;
+package org.elasticsearch.xpack.autoscaling.decision;
+
+import org.elasticsearch.xpack.autoscaling.AutoscalingTestCase;
 
 import java.util.List;
 

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/AutoscalingDecisionsWireSerializingTests.java → x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/decision/AutoscalingDecisionsWireSerializingTests.java

@@ -4,10 +4,11 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-package org.elasticsearch.xpack.autoscaling;
+package org.elasticsearch.xpack.autoscaling.decision;
 
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
+import org.elasticsearch.xpack.autoscaling.AutoscalingTestCase;
 
 public class AutoscalingDecisionsWireSerializingTests extends AbstractWireSerializingTestCase<AutoscalingDecisions> {
 

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/policy/AutoscalingPolicyMetadataDiffableSerializationTests.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.policy;
+
+import org.elasticsearch.cluster.AbstractDiffable;
+import org.elasticsearch.cluster.Diff;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractDiffableSerializationTestCase;
+import org.elasticsearch.xpack.autoscaling.Autoscaling;
+
+import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.mutateAutoscalingPolicy;
+import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.randomAutoscalingPolicyOfName;
+
+public class AutoscalingPolicyMetadataDiffableSerializationTests extends AbstractDiffableSerializationTestCase<AutoscalingPolicyMetadata> {
+
+    private final String name = randomAlphaOfLength(8);
+
+    @Override
+    protected NamedWriteableRegistry getNamedWriteableRegistry() {
+        return new NamedWriteableRegistry(new Autoscaling(Settings.EMPTY).getNamedWriteables());
+    }
+
+    @Override
+    protected NamedXContentRegistry xContentRegistry() {
+        return new NamedXContentRegistry(new Autoscaling(Settings.EMPTY).getNamedXContent());
+    }
+
+    @Override
+    protected AutoscalingPolicyMetadata doParseInstance(final XContentParser parser) {
+        return AutoscalingPolicyMetadata.parse(parser, name);
+    }
+
+    @Override
+    protected Writeable.Reader<AutoscalingPolicyMetadata> instanceReader() {
+        return AutoscalingPolicyMetadata::new;
+    }
+
+    @Override
+    protected AutoscalingPolicyMetadata createTestInstance() {
+        return new AutoscalingPolicyMetadata(randomAutoscalingPolicyOfName(name));
+    }
+
+    @Override
+    protected AutoscalingPolicyMetadata makeTestChanges(final AutoscalingPolicyMetadata testInstance) {
+        return mutateInstance(testInstance);
+    }
+
+    @Override
+    protected AutoscalingPolicyMetadata mutateInstance(final AutoscalingPolicyMetadata instance) {
+        return new AutoscalingPolicyMetadata(mutateAutoscalingPolicy(instance.policy()));
+    }
+
+    @Override
+    protected Writeable.Reader<Diff<AutoscalingPolicyMetadata>> diffReader() {
+        return in -> AbstractDiffable.readDiffFrom(AutoscalingPolicyMetadata::new, in);
+    }
+
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/policy/AutoscalingPolicySerializingTests.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.autoscaling.policy;
+
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractSerializingTestCase;
+import org.elasticsearch.xpack.autoscaling.Autoscaling;
+
+import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.mutateAutoscalingPolicy;
+import static org.elasticsearch.xpack.autoscaling.AutoscalingTestCase.randomAutoscalingPolicyOfName;
+
+public class AutoscalingPolicySerializingTests extends AbstractSerializingTestCase<AutoscalingPolicy> {
+
+    private final String name = randomAlphaOfLength(8);
+
+    @Override
+    protected NamedWriteableRegistry getNamedWriteableRegistry() {
+        return new NamedWriteableRegistry(new Autoscaling(Settings.EMPTY).getNamedWriteables());
+    }
+
+    @Override
+    protected NamedXContentRegistry xContentRegistry() {
+        return new NamedXContentRegistry(new Autoscaling(Settings.EMPTY).getNamedXContent());
+    }
+
+    @Override
+    protected AutoscalingPolicy doParseInstance(final XContentParser parser) {
+        return AutoscalingPolicy.parse(parser, name);
+    }
+
+    @Override
+    protected Writeable.Reader<AutoscalingPolicy> instanceReader() {
+        return AutoscalingPolicy::new;
+    }
+
+    @Override
+    protected AutoscalingPolicy createTestInstance() {
+        return randomAutoscalingPolicyOfName(name);
+    }
+
+    @Override
+    protected AutoscalingPolicy mutateInstance(final AutoscalingPolicy instance) {
+        return mutateAutoscalingPolicy(instance);
+    }
+
+}

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/LocalStateCompositeXPackPlugin.java

@@ -101,7 +101,7 @@ public class LocalStateCompositeXPackPlugin extends XPackPlugin implements Scrip
     private LicenseService licenseService;
     protected List<Plugin> plugins = new ArrayList<>();
 
-    public LocalStateCompositeXPackPlugin(final Settings settings, final Path configPath) throws Exception {
+    public LocalStateCompositeXPackPlugin(final Settings settings, final Path configPath) {
         super(settings, configPath);
     }
 

x-pack/plugin/src/test/resources/rest-api-spec/api/autoscaling.put_autoscaling_policy.json

@@ -0,0 +1,28 @@
+{
+  "autoscaling.put_autoscaling_policy":{
+    "documentation":{
+      "url":"https://www.elastic.co/guide/en/elasticsearch/reference/current/autoscaling-put-autoscaling-policy.html"
+    },
+    "stability":"experimental",
+    "url":{
+      "paths":[
+        {
+          "path":"/_autoscaling/policy/{name}",
+          "methods":[
+            "PUT"
+          ],
+          "parts":{
+            "name":{
+              "type":"string",
+              "description":"the name of the autoscaling policy"
+            }
+          }
+        }
+      ]
+    },
+    "body":{
+      "description":"the specification of the autoscaling policy",
+      "required":true
+    }
+  }
+}