Introduce "Feature States" for managing snapshots of system indices (#63513)

This PR expands the meaning of `include_global_state` for snapshots to include system indices. If `include_global_state` is `true` on creation, system indices will be included in the snapshot regardless of the contents of the `indices` field. If `include_global_state` is `true` on restoration, system indices will be restored (if included in the snapshot), regardless of the contents of the `indices` field. Index renaming is not applied to system indices, as system indices rely on their names matching certain patterns. If restored system indices are already present, they are automatically deleted prior to restoration from the snapshot to avoid conflicts.

This behavior can be overridden to an extent by including a new field in the snapshot creation or restoration call, `feature_states`, which contains an array of strings indicating the "feature" for which system indices should be snapshotted or restored. For example, this call will only restore the `watcher` and `security` system indices (in addition to `index_1`):

```
POST /_snapshot/my_repository/snapshot_2/_restore
{
  "indices": "index_1",
  "include_global_state": true,
  "feature_states": ["watcher", "security"]
}
```

If `feature_states` is present, the system indices associated with those features will be snapshotted or restored regardless of the value of `include_global_state`. All system indices can be omitted by providing a special value of `none` (`"feature_states": ["none"]`), or included by omitting the field or explicitly providing an empty array (`"feature_states": []`), similar to the `indices` field.

The list of currently available features can be retrieved via a new "Get Snapshottable Features" API:
```
GET /_snapshottable_features
```

which returns a response of the form:
```
{
    "features": [
        {
            "name": "tasks",
            "description": "Manages task results"
        },
        {
            "name": "kibana",
            "description": "Manages Kibana configuration and reports"
        }
    ]
}
```

Features currently map one-to-one with `SystemIndexPlugin`s, but this should be considered an implementation detail. The Get Snapshottable Features API and snapshot creation rely upon all relevant plugins being installed on the master node.

Further, the list of feature states included in a given snapshot is exposed by the Get Snapshot API, which now includes a new field, `feature_states`, which contains a list of the feature states and their associated system indices which are included in the snapshot. All system indices in feature states are also included in the `indices` array for backwards compatibility, although explicitly requesting system indices included in a feature state is deprecated. For example, an excerpt from the Get Snapshot API showing `feature_states`:
```
"feature_states": [
    {
        "feature_name": "tasks",
        "indices": [
            ".tasks"
        ]
    }
],
"indices": [
    ".tasks",
    "test1",
    "test2"
]
```

Co-authored-by: William Brafford <[email protected]>

Author: gwbrown

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) == false) 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) == false) 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,67 @@
+/*
+ * 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",