Introduce "Feature States" for managing snapshots of system indices

client/rest-high-level/src/main/java/org/elasticsearch/client/SnapshotClient.java

@@ -28,6 +28,8 @@
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotsStatusRequest;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotsStatusResponse;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.client.snapshots.GetSnapshottableFeaturesRequest;
+import org.elasticsearch.client.snapshots.GetSnapshottableFeaturesResponse;
 
 import java.io.IOException;
 
@@ -378,4 +380,47 @@ public Cancellable deleteAsync(DeleteSnapshotRequest deleteSnapshotRequest, Requ
             SnapshotRequestConverters::deleteSnapshot, options,
             AcknowledgedResponse::fromXContent, listener, emptySet());
     }
+
+    /**
+     * Get a list of features which can be included in a snapshot as feature states.
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/get-snapshottable-features-api.html"> Get Snapshottable
+     * Features API on elastic.co</a>
+     *
+     * @param getFeaturesRequest the request
+     * @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
+     * @return the response
+     * @throws IOException in case there is a problem sending the request or parsing back the response
+     */
+    public GetSnapshottableFeaturesResponse getFeatures(GetSnapshottableFeaturesRequest getFeaturesRequest, RequestOptions options)
+        throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(
+            getFeaturesRequest,
+            SnapshotRequestConverters::getSnapshottableFeatures,
+            options,
+            GetSnapshottableFeaturesResponse::parse,
+            emptySet()
+        );
+    }
+
+    /**
+     * Asynchronously get a list of features which can be included in a snapshot as feature states.
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/get-snapshottable-features-api.html"> Get Snapshottable
+     * Features API on elastic.co</a>
+     *
+     * @param getFeaturesRequest the request
+     * @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
+     * @param listener the listener to be notified upon request completion
+     * @return cancellable that may be used to cancel the request
+     */
+    public Cancellable getFeaturesAsync(GetSnapshottableFeaturesRequest getFeaturesRequest, RequestOptions options,
+                            ActionListener<GetSnapshottableFeaturesResponse> listener) {
+        return restHighLevelClient.performRequestAsyncAndParseEntity(
+            getFeaturesRequest,
+            SnapshotRequestConverters::getSnapshottableFeatures,
+            options,
+            GetSnapshottableFeaturesResponse::parse,
+            listener,
+            emptySet()
+        );
+    }
 }

client/rest-high-level/src/main/java/org/elasticsearch/client/SnapshotRequestConverters.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.action.admin.cluster.snapshots.get.GetSnapshotsRequest;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotRequest;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotsStatusRequest;
+import org.elasticsearch.client.snapshots.GetSnapshottableFeaturesRequest;
 import org.elasticsearch.common.Strings;
 
 import java.io.IOException;
@@ -190,4 +191,13 @@ static Request deleteSnapshot(DeleteSnapshotRequest deleteSnapshotRequest) {
         request.addParameters(parameters.asMap());
         return request;
     }
+
+    static Request getSnapshottableFeatures(GetSnapshottableFeaturesRequest getSnapshottableFeaturesRequest) {
+        String endpoint = "/_snapshottable_features";
+        Request request = new Request(HttpGet.METHOD_NAME, endpoint);
+        RequestConverters.Params parameters = new RequestConverters.Params();
+        parameters.withMasterTimeout(getSnapshottableFeaturesRequest.masterNodeTimeout());
+        request.addParameters(parameters.asMap());
+        return request;
+    }
 }

client/rest-high-level/src/main/java/org/elasticsearch/client/snapshots/GetSnapshottableFeaturesRequest.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.client.snapshots;
+
+import org.elasticsearch.client.TimedRequest;
+
+/**
+ * A {@link TimedRequest} to get the list of features available to be included in snapshots in the cluster.
+ */
+public class GetSnapshottableFeaturesRequest extends TimedRequest {
+}

client/rest-high-level/src/main/java/org/elasticsearch/client/snapshots/GetSnapshottableFeaturesResponse.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.client.snapshots;
+
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ObjectParser;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.util.List;
+import java.util.Objects;
+
+public class GetSnapshottableFeaturesResponse {
+
+    private final List<SnapshottableFeature> features;
+
+    private static final ParseField FEATURES = new ParseField("features");
+
+    @SuppressWarnings("unchecked")
+    private static final ConstructingObjectParser<GetSnapshottableFeaturesResponse, Void> PARSER = new ConstructingObjectParser<>(
+        "snapshottable_features_response", true, (a, ctx) -> new GetSnapshottableFeaturesResponse((List<SnapshottableFeature>) a[0])
+    );
+
+    static {
+        PARSER.declareObjectArray(ConstructingObjectParser.constructorArg(), SnapshottableFeature::parse, FEATURES);
+    }
+
+    public GetSnapshottableFeaturesResponse(List<SnapshottableFeature> features) {
+        this.features = features;
+    }
+
+    public List<SnapshottableFeature> getFeatures() {
+        return features;
+    }
+
+    public static GetSnapshottableFeaturesResponse parse(XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof GetSnapshottableFeaturesResponse)) return false;
+        GetSnapshottableFeaturesResponse that = (GetSnapshottableFeaturesResponse) o;
+        return getFeatures().equals(that.getFeatures());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(getFeatures());
+    }
+
+    public static class SnapshottableFeature {
+
+        private final String featureName;
+        private final String description;
+
+        private static final ParseField FEATURE_NAME = new ParseField("name");
+        private static final ParseField DESCRIPTION = new ParseField("description");
+
+        private static final ConstructingObjectParser<SnapshottableFeature, Void> PARSER =  new ConstructingObjectParser<>(
+            "feature", true, (a, ctx) -> new SnapshottableFeature((String) a[0], (String) a[1])
+        );
+
+        static {
+            PARSER.declareField(ConstructingObjectParser.constructorArg(),
+                (p, c) -> p.text(), FEATURE_NAME, ObjectParser.ValueType.STRING);
+            PARSER.declareField(ConstructingObjectParser.constructorArg(),
+                (p, c) -> p.text(), DESCRIPTION, ObjectParser.ValueType.STRING);
+        }
+
+        public SnapshottableFeature(String featureName, String description) {
+            this.featureName = featureName;
+            this.description = description;
+        }
+
+        public static SnapshottableFeature parse(XContentParser parser, Void ctx) {
+            return PARSER.apply(parser, ctx);
+        }
+
+        public String getFeatureName() {
+            return featureName;
+        }
+
+        public String getDescription() {
+            return description;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (!(o instanceof SnapshottableFeature)) return false;
+            SnapshottableFeature feature = (SnapshottableFeature) o;
+            return Objects.equals(getFeatureName(), feature.getFeatureName());
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(getFeatureName());
+        }
+    }
+}

client/rest-high-level/src/test/java/org/elasticsearch/client/SnapshotIT.java

@@ -28,6 +28,8 @@
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotsStatusRequest;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotsStatusResponse;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.client.snapshots.GetSnapshottableFeaturesRequest;
+import org.elasticsearch.client.snapshots.GetSnapshottableFeaturesResponse;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentType;
@@ -39,12 +41,16 @@
 import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
+import static org.elasticsearch.snapshots.SnapshotsService.NO_FEATURE_STATES_VALUE;
+import static org.elasticsearch.tasks.TaskResultsService.TASKS_FEATURE_NAME;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
 
 public class SnapshotIT extends ESRestHighLevelClientTestCase {
 
@@ -150,6 +156,14 @@ public void testCreateSnapshot() throws Exception {
         }
         request.partial(randomBoolean());
         request.includeGlobalState(randomBoolean());
+        final List<String> featureStates = randomFrom(
+            List.of(
+                Collections.emptyList(),
+                Collections.singletonList(TASKS_FEATURE_NAME),
+                Collections.singletonList(NO_FEATURE_STATES_VALUE)
+            )
+        );
+        request.featureStates(featureStates);
 
         CreateSnapshotResponse response = createTestSnapshot(request);
         assertEquals(waitForCompletion ? RestStatus.OK : RestStatus.ACCEPTED, response.status());
@@ -262,9 +276,14 @@ public void testRestoreSnapshot() throws IOException {
         assertFalse("index [" + testIndex + "] should have been deleted", indexExists(testIndex));
 
         RestoreSnapshotRequest request = new RestoreSnapshotRequest(testRepository, testSnapshot);
+        request.indices(testIndex);
         request.waitForCompletion(true);
         request.renamePattern(testIndex);
         request.renameReplacement(restoredIndex);
+        if (randomBoolean()) {
+            request.includeGlobalState(true);
+            request.featureStates(Collections.singletonList(NO_FEATURE_STATES_VALUE));
+        }
 
         RestoreSnapshotResponse response = execute(request, highLevelClient().snapshot()::restore,
                 highLevelClient().snapshot()::restoreAsync);
@@ -364,6 +383,18 @@ public void testCloneSnapshot() throws IOException {
         assertTrue(response.isAcknowledged());
     }
 
+    public void testGetFeatures() throws IOException {
+        GetSnapshottableFeaturesRequest request = new GetSnapshottableFeaturesRequest();
+
+        GetSnapshottableFeaturesResponse response = execute(request,
+            highLevelClient().snapshot()::getFeatures, highLevelClient().snapshot()::getFeaturesAsync);
+
+        assertThat(response, notNullValue());
+        assertThat(response.getFeatures(), notNullValue());
+        assertThat(response.getFeatures().size(), greaterThan(1));
+        assertTrue(response.getFeatures().stream().anyMatch(feature -> "tasks".equals(feature.getFeatureName())));
+    }
+
     private static Map<String, Object> randomUserMetadata() {
         if (randomBoolean()) {
             return null;

client/rest-high-level/src/test/java/org/elasticsearch/client/snapshots/GetSnapshottableFeaturesResponseTests.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.client.snapshots;
+
+import org.elasticsearch.client.AbstractResponseTestCase;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.hamcrest.Matchers.everyItem;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.in;
+import static org.hamcrest.Matchers.is;
+
+public class GetSnapshottableFeaturesResponseTests extends AbstractResponseTestCase<
+    org.elasticsearch.action.admin.cluster.snapshots.features.GetSnapshottableFeaturesResponse,
+    GetSnapshottableFeaturesResponse> {
+
+    @Override
+    protected org.elasticsearch.action.admin.cluster.snapshots.features.GetSnapshottableFeaturesResponse createServerTestInstance(
+        XContentType xContentType
+    ) {
+        return new org.elasticsearch.action.admin.cluster.snapshots.features.GetSnapshottableFeaturesResponse(
+            randomList(
+                10,
+                () -> new org.elasticsearch.action.admin.cluster.snapshots.features.GetSnapshottableFeaturesResponse.SnapshottableFeature(
+                    randomAlphaOfLengthBetween(4, 10),
+                    randomAlphaOfLengthBetween(5, 10)
+                )
+            )
+        );
+
+    }
+
+    @Override
+    protected GetSnapshottableFeaturesResponse doParseToClientInstance(XContentParser parser) throws IOException {
+        return GetSnapshottableFeaturesResponse.parse(parser);
+    }
+
+    @Override
+    protected void assertInstances(
+        org.elasticsearch.action.admin.cluster.snapshots.features.GetSnapshottableFeaturesResponse serverTestInstance,
+        GetSnapshottableFeaturesResponse clientInstance
+    ) {
+        assertNotNull(serverTestInstance.getSnapshottableFeatures());
+        assertNotNull(serverTestInstance.getSnapshottableFeatures());
+
+        assertThat(clientInstance.getFeatures(), hasSize(serverTestInstance.getSnapshottableFeatures().size()));
+
+        Map<String, String> clientFeatures = clientInstance.getFeatures()
+            .stream()
+            .collect(Collectors.toMap(f -> f.getFeatureName(), f -> f.getDescription()));
+        Map<String, String> serverFeatures = serverTestInstance.getSnapshottableFeatures()
+            .stream()
+            .collect(Collectors.toMap(f -> f.getFeatureName(), f -> f.getDescription()));
+
+        assertThat(clientFeatures.entrySet(), everyItem(is(in(serverFeatures.entrySet()))));
+    }
+}

docs/reference/snapshot-restore/apis/create-snapshot-api.asciidoc

@@ -99,20 +99,35 @@ argument is provided, the snapshot only includes the specified data streams and
 +
 --
 (Optional, Boolean)
-If `true`, the current cluster state is included in the snapshot.
+If `true`, the current global state is included in the snapshot.
 Defaults to `true`.
 
-The cluster state includes:
+The global state includes:
 
 * Persistent cluster settings
 * Index templates
 * Legacy index templates
 * Ingest pipelines
 * {ilm-init} lifecycle policies
+* Data stored in system indices, such as Watches and task records (configurable via `feature_states`)
 --
 +
 IMPORTANT: By default, the entire snapshot will fail if one or more indices included in the snapshot do not have all primary shards available. You can change this behavior by setting <<create-snapshot-api-partial,`partial`>> to `true`.
 
+[[create-snapshot-api-feature-states]]
+`feature_states`::
+(Optional, array of strings)
+A list of feature states to be included in this snapshot. A list of features
+available for inclusion in the snapshot and their descriptions be can be
+retrieved using the <<get-snapshottable-features-api,get snapshottable features API>>.
+Each feature state includes one or more system indices containing data necessary
+for the function of that feature. Providing an empty array will include no feature
+states in the snapshot, regardless of the value of `include_global_state`.
++
+By default, all available feature states will be included in the snapshot if
+`include_global_state` is `true`, or no feature states if `include_global_state`
+is `false`.
+
 include::{es-repo-dir}/rest-api/common-parms.asciidoc[tag=master-timeout]
 
 `metadata`::
@@ -163,6 +178,7 @@ The API returns the following response:
     "version": <version>,
     "indices": [],
     "data_streams": [],
+    "feature_states": [],
     "include_global_state": false,
     "metadata": {
       "taken_by": "user123",