Adds a new Rollup V2 Action

This commit adds a new endpoint for Rollup V2, a new way to rollup indices.

Instead of relying on a cron-job, this action will rollup a whole index on the spot.

When an index is rolled up using a Rollup Config, it does the following

1. check that original index is read-only
2. runs an aggregation and indexes results into a temporary hidden rollup index
3. "resizes" the temporary index into the final rollup-index (in-place segment pointer juggling)
4. adds RollupMetadata about the rollup-group (keyed by original index name) and adds custom index-metadata with data about what the rollup index's original index is so that its group information in RollupMetadata can be looked up

example usage:

```
POST /_rollup_vtwo/index
{
    "rollup_index": "index_rolled",
    "groups": {
        "date_histogram": {
            "field": "date",
            "calendar_interval": "1M"
        },
        "terms": {
            "fields": ["unit"]
        }
    },
    "metrics": [
        {
            "field": "temperature",
            "metrics": ["sum"]
        }
    ]
}
```

Author: talevy

server/src/main/java/org/elasticsearch/cluster/ClusterModule.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.cluster;
 
+import org.elasticsearch.Build;
 import org.elasticsearch.cluster.action.index.MappingUpdatedAction;
 import org.elasticsearch.cluster.action.index.NodeMappingRefreshAction;
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
@@ -143,7 +144,8 @@ public static List<Entry> getNamedWriteables() {
             ComposableIndexTemplateMetadata::readDiffFrom);
         registerMetadataCustom(entries, DataStreamMetadata.TYPE, DataStreamMetadata::new, DataStreamMetadata::readDiffFrom);
 
-        if (RollupV2.ROLLUPV2_FEATURE_FLAG_REGISTERED != null && RollupV2.ROLLUPV2_FEATURE_FLAG_REGISTERED) {
+        if (Build.CURRENT.isSnapshot() ||
+                (RollupV2.ROLLUPV2_FEATURE_FLAG_REGISTERED != null && RollupV2.ROLLUPV2_FEATURE_FLAG_REGISTERED)) {
             registerMetadataCustom(entries, RollupMetadata.TYPE, RollupMetadata::new, RollupMetadata::readDiffFrom);
         }
         // Task Status (not Diffable)
@@ -170,7 +172,8 @@ public static List<NamedXContentRegistry.Entry> getNamedXWriteables() {
             ComposableIndexTemplateMetadata::fromXContent));
         entries.add(new NamedXContentRegistry.Entry(Metadata.Custom.class, new ParseField(DataStreamMetadata.TYPE),
             DataStreamMetadata::fromXContent));
-        if (RollupV2.ROLLUPV2_FEATURE_FLAG_REGISTERED != null && RollupV2.ROLLUPV2_FEATURE_FLAG_REGISTERED) {
+        if (Build.CURRENT.isSnapshot() ||
+                (RollupV2.ROLLUPV2_FEATURE_FLAG_REGISTERED != null && RollupV2.ROLLUPV2_FEATURE_FLAG_REGISTERED)) {
             entries.add(new NamedXContentRegistry.Entry(Metadata.Custom.class, new ParseField(RollupMetadata.TYPE),
                 RollupMetadata::fromXContent));
         }

server/src/main/java/org/elasticsearch/rollup/RollupV2.java

@@ -29,6 +29,7 @@ public class RollupV2 {
         if (Build.CURRENT.isSnapshot() && property != null) {
             throw new IllegalArgumentException("es.rollupv2_feature_flag_registered is only supported in non-snapshot builds");
         }
+
         if ("true".equals(property)) {
             ROLLUPV2_FEATURE_FLAG_REGISTERED = true;
         } else if ("false".equals(property)) {

server/src/main/java/org/elasticsearch/tasks/Task.java

@@ -185,6 +185,10 @@ public String getHeader(String header) {
         return headers.get(header);
     }
 
+    public Map<String, String> headers() {
+        return headers;
+    }
+
     public TaskResult result(DiscoveryNode node, Exception error) throws IOException {
         return new TaskResult(taskInfo(node.getId(), true), error);
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackClientPlugin.java

@@ -50,8 +50,8 @@ public class MetricConfig implements Writeable, ToXContentObject {
     public static final ParseField SUM = new ParseField("sum");
     public static final ParseField AVG = new ParseField("avg");
     public static final ParseField VALUE_COUNT = new ParseField("value_count");
+    public static final String NAME = "metrics";
 
-    static final String NAME = "metrics";
     private static final String FIELD = "field";
     private static final String METRICS = "metrics";
     private static final ConstructingObjectParser<MetricConfig, Void> PARSER;
@@ -84,7 +84,7 @@ public MetricConfig(final String field, final List<String> metrics) {
         this.metrics = metrics;
     }
 
-    MetricConfig(final StreamInput in) throws IOException {
+    public MetricConfig(final StreamInput in) throws IOException {
         field = in.readString();
         metrics = in.readStringList();
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/rollup/job/MetricConfig.java

@@ -0,0 +1,146 @@
+/*
+ * 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.core.rollup.v2;
+
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestBuilder;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.client.ElasticsearchClient;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+
+public class RollupV2Action extends ActionType<RollupV2Action.Response> {
+
+    public static final RollupV2Action INSTANCE = new RollupV2Action();
+    public static final String NAME = "cluster:admin/xpack/rollupV2";
+
+    private RollupV2Action() {
+        super(NAME, RollupV2Action.Response::new);
+    }
+
+    public static class Request extends ActionRequest implements ToXContentObject {
+        private RollupV2Config rollupConfig;
+
+        public Request(RollupV2Config rollupConfig) {
+            this.rollupConfig = rollupConfig;
+        }
+
+        public Request() {}
+
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            rollupConfig = new RollupV2Config(in);
+        }
+
+        @Override
+        public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+            return new RollupV2Task(id, type, action, parentTaskId, rollupConfig, headers);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            rollupConfig.writeTo(out);
+        }
+
+        public RollupV2Config getRollupConfig() {
+            return rollupConfig;
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            return null;
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            rollupConfig.toXContent(builder, params);
+            builder.endObject();
+            return builder;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(rollupConfig);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (obj == null) {
+                return false;
+            }
+            if (getClass() != obj.getClass()) {
+                return false;
+            }
+            Request other = (Request) obj;
+            return Objects.equals(rollupConfig, other.rollupConfig);
+        }
+    }
+
+    public static class RequestBuilder extends ActionRequestBuilder<Request, Response> {
+
+        protected RequestBuilder(ElasticsearchClient client, RollupV2Action action) {
+            super(client, action, new Request());
+        }
+    }
+
+    public static class Response extends ActionResponse implements Writeable, ToXContentObject {
+
+        private final boolean created;
+
+        public Response(boolean created) {
+            this.created = created;
+        }
+
+        public Response(StreamInput in) throws IOException {
+            super(in);
+            created = in.readBoolean();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeBoolean(created);
+        }
+
+        public boolean isCreated() {
+            return created;
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            builder.field("created", created);
+            builder.endObject();
+            return builder;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Response response = (Response) o;
+            return created == response.created;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(created);
+        }
+    }
+}