Add API for resetting state of a `SystemIndexPlugin` (#69469)

When we disable access to system indices, plugins will still need
a way to erase their state. The obvious and most pressing use
case for this is in tests, which need to be able to clean up the
state of a cluster in between groups of tests.

* Use a HandledTransportAction for reset action

My initial cut used a TransportMasterNodeAction, which requires code
that carefully manipulates cluster state. At least for the first cut and
testing, it seems like it will be much easier to use a client within a
HandledTransportAction, which effectively makes the
TransportResetFeatureStateAction a class that dispatches other transport
actions to do the real work.

* Clean up code by using a GroupedActionListener

* ML feature state cleaner

* Implement Transform feature state reset

* Change _features/reset path to _features/_reset

Out of an abundance of caution, I think the "reset" part of this path
should have a leading underscore, so that if there's ever a reason to
implement "GET _features/<feature_id>" we won't have to worry about
distinguishing "reset" from a feature name.

Co-authored-by: Gordon Brown <gordon.brown@elastic.co>
Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

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

@@ -9,8 +9,10 @@
 package org.elasticsearch.client;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.client.snapshots.GetFeaturesRequest;
-import org.elasticsearch.client.snapshots.GetFeaturesResponse;
+import org.elasticsearch.client.feature.GetFeaturesRequest;
+import org.elasticsearch.client.feature.GetFeaturesResponse;
+import org.elasticsearch.client.feature.ResetFeaturesRequest;
+import org.elasticsearch.client.feature.ResetFeaturesResponse;
 
 import java.io.IOException;
 
@@ -71,4 +73,50 @@ public class FeaturesClient {
             emptySet()
         );
     }
+
+    /**
+     * Reset the state of Elasticsearch features, deleting system indices and performing other
+     * cleanup operations.
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/reset-features-api.html"> Rest
+     * Features API on elastic.co</a>
+     *
+     * @param resetFeaturesRequest 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 ResetFeaturesResponse resetFeatures(ResetFeaturesRequest resetFeaturesRequest, RequestOptions options)
+        throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(
+            resetFeaturesRequest,
+            FeaturesRequestConverters::resetFeatures,
+            options,
+            ResetFeaturesResponse::parse,
+            emptySet()
+        );
+    }
+
+    /**
+     * Asynchronously reset the state of Elasticsearch features, deleting system indices and performing other
+     * cleanup operations.
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/get-features-api.html"> Get Snapshottable
+     * Features API on elastic.co</a>
+     *
+     * @param resetFeaturesRequest 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 resetFeaturesAsync(
+        ResetFeaturesRequest resetFeaturesRequest, RequestOptions options,
+        ActionListener<ResetFeaturesResponse> listener) {
+        return restHighLevelClient.performRequestAsyncAndParseEntity(
+            resetFeaturesRequest,
+            FeaturesRequestConverters::resetFeatures,
+            options,
+            ResetFeaturesResponse::parse,
+            listener,
+            emptySet()
+        );
+    }
 }

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

@@ -9,7 +9,9 @@
 package org.elasticsearch.client;
 
 import org.apache.http.client.methods.HttpGet;
-import org.elasticsearch.client.snapshots.GetFeaturesRequest;
+import org.apache.http.client.methods.HttpPost;
+import org.elasticsearch.client.feature.GetFeaturesRequest;
+import org.elasticsearch.client.feature.ResetFeaturesRequest;
 
 public class FeaturesRequestConverters {
 
@@ -23,4 +25,9 @@ public class FeaturesRequestConverters {
         request.addParameters(parameters.asMap());
         return request;
     }
+
+    static Request resetFeatures(ResetFeaturesRequest resetFeaturesRequest) {
+        String endpoint = "/_features/_reset";
+        return new Request(HttpPost.METHOD_NAME, endpoint);
+    }
 }

client/rest-high-level/src/main/java/org/elasticsearch/client/snapshots/GetFeaturesRequest.java → client/rest-high-level/src/main/java/org/elasticsearch/client/feature/GetFeaturesRequest.java

@@ -6,7 +6,7 @@
  * Side Public License, v 1.
  */
 
-package org.elasticsearch.client.snapshots;
+package org.elasticsearch.client.feature;
 
 import org.elasticsearch.client.TimedRequest;
 

client/rest-high-level/src/main/java/org/elasticsearch/client/snapshots/GetFeaturesResponse.java → client/rest-high-level/src/main/java/org/elasticsearch/client/feature/GetFeaturesResponse.java

@@ -6,7 +6,7 @@
  * Side Public License, v 1.
  */
 
-package org.elasticsearch.client.snapshots;
+package org.elasticsearch.client.feature;
 
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.xcontent.ConstructingObjectParser;

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

@@ -0,0 +1,14 @@
+/*
+ * 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.feature;
+
+import org.elasticsearch.client.TimedRequest;
+
+public class ResetFeaturesRequest extends TimedRequest {
+}

client/rest-high-level/src/main/java/org/elasticsearch/client/feature/ResetFeaturesResponse.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
+ * 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.feature;
+
+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;
+
+public class ResetFeaturesResponse {
+    private final List<ResetFeatureStateStatus> features;
+
+    private static final ParseField FEATURES = new ParseField("features");
+
+    @SuppressWarnings("unchecked")
+    private static final ConstructingObjectParser<ResetFeaturesResponse, Void> PARSER = new ConstructingObjectParser<>(
+        "snapshottable_features_response", true,
+        (a, ctx) -> new ResetFeaturesResponse((List<ResetFeatureStateStatus>) a[0])
+    );
+
+    static {
+        PARSER.declareObjectArray(
+            ConstructingObjectParser.constructorArg(),
+            ResetFeaturesResponse.ResetFeatureStateStatus::parse, FEATURES);
+    }
+
+    public ResetFeaturesResponse(List<ResetFeatureStateStatus> features) {
+        this.features = features;
+    }
+
+    public List<ResetFeatureStateStatus> getFeatures() {
+        return features;
+    }
+
+    public static ResetFeaturesResponse parse(XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+
+    public static class ResetFeatureStateStatus {
+        private final String featureName;
+        private final String status;
+
+        private static final ParseField FEATURE_NAME = new ParseField("feature_name");
+        private static final ParseField STATUS = new ParseField("status");
+
+        private static final ConstructingObjectParser<ResetFeatureStateStatus, Void> PARSER =  new ConstructingObjectParser<>(
+            "features", true, (a, ctx) -> new ResetFeatureStateStatus((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(), STATUS, ObjectParser.ValueType.STRING);
+        }
+
+        ResetFeatureStateStatus(String featureName, String status) {
+            this.featureName = featureName;
+            this.status = status;
+        }
+
+        public static ResetFeatureStateStatus parse(XContentParser parser, Void ctx) {
+            return PARSER.apply(parser, ctx);
+        }
+
+        public String getFeatureName() {
+            return featureName;
+        }
+
+        public String getStatus() {
+            return status;
+        }
+    }
+}

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

@@ -8,8 +8,10 @@
 
 package org.elasticsearch.client;
 
-import org.elasticsearch.client.snapshots.GetFeaturesRequest;
-import org.elasticsearch.client.snapshots.GetFeaturesResponse;
+import org.elasticsearch.client.feature.GetFeaturesRequest;
+import org.elasticsearch.client.feature.GetFeaturesResponse;
+import org.elasticsearch.client.feature.ResetFeaturesRequest;
+import org.elasticsearch.client.feature.ResetFeaturesResponse;
 
 import java.io.IOException;
 
@@ -28,4 +30,17 @@ public class FeaturesIT extends ESRestHighLevelClientTestCase {
         assertThat(response.getFeatures().size(), greaterThan(1));
         assertTrue(response.getFeatures().stream().anyMatch(feature -> "tasks".equals(feature.getFeatureName())));
     }
+
+    public void testResetFeatures() throws IOException {
+        ResetFeaturesRequest request = new ResetFeaturesRequest();
+
+        ResetFeaturesResponse response = execute(request,
+            highLevelClient().features()::resetFeatures, highLevelClient().features()::resetFeaturesAsync);
+
+        assertThat(response, notNullValue());
+        assertThat(response.getFeatures(), notNullValue());
+        assertThat(response.getFeatures().size(), greaterThan(1));
+        assertTrue(response.getFeatures().stream().anyMatch(
+            feature -> "tasks".equals(feature.getFeatureName()) && "SUCCESS".equals(feature.getStatus())));
+    }
 }

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

@@ -9,6 +9,7 @@
 package org.elasticsearch.client.snapshots;
 
 import org.elasticsearch.client.AbstractResponseTestCase;
+import org.elasticsearch.client.feature.GetFeaturesResponse;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
 

rest-api-spec/src/main/resources/rest-api-spec/api/features.reset_features.json

@@ -0,0 +1,23 @@
+{
+  "features.reset_features":{
+    "documentation":{
+      "url":"https://www.elastic.co/guide/en/elasticsearch/reference/master/modules-snapshots.html",
+      "description":"Resets the internal state of features, usually by deleting system indices"
+    },
+    "stability":"experimental",
+    "visibility":"public",
+    "headers":{
+      "accept": [ "application/json"]
+    },
+    "url":{
+      "paths":[
+        {
+          "path":"/_features/_reset",
+          "methods":[
+            "POST"
+          ]
+        }
+      ]
+    }
+  }
+}

rest-api-spec/src/main/resources/rest-api-spec/test/features.reset_features/10_basic.yml

@@ -0,0 +1,8 @@
+---
+"Get Features":
+  - skip:
+      features: contains
+      version: " - 7.99.99" # Adjust this after backport
+      reason: "This API was added in 7.13.0"
+  - do: { features.get_features: {}}
+  - contains: {'features': {'name': 'tasks'}}

server/src/internalClusterTest/java/org/elasticsearch/snapshots/FeatureStateResetApiIT.java

@@ -0,0 +1,148 @@
+/*
+ * 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.snapshots;
+
+import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateAction;
+import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateRequest;
+import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateResponse;
+import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.IndexNotFoundException;
+import org.elasticsearch.indices.SystemIndexDescriptor;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.plugins.SystemIndexPlugin;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import static org.hamcrest.Matchers.arrayContaining;
+import static org.hamcrest.Matchers.containsInAnyOrder;
+import static org.hamcrest.Matchers.containsString;
+
+public class FeatureStateResetApiIT extends ESIntegTestCase {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        List<Class<? extends Plugin>> plugins = new ArrayList<>(super.nodePlugins());
+        plugins.add(SystemIndexTestPlugin.class);
+        plugins.add(SecondSystemIndexTestPlugin.class);
+        return plugins;
+    }
+
+    /** Check that the reset method cleans up a feature */
+    public void testResetSystemIndices() throws Exception {
+        String systemIndex1 = ".test-system-idx-1";
+        String systemIndex2 = ".second-test-system-idx-1";
+        String associatedIndex = ".associated-idx-1";
+
+        // put a document in a system index
+        indexDoc(systemIndex1, "1", "purpose", "system index doc");
+        refresh(systemIndex1);
+
+        // put a document in a second system index
+        indexDoc(systemIndex2, "1", "purpose", "second system index doc");
+        refresh(systemIndex2);
+
+        // put a document in associated index
+        indexDoc(associatedIndex, "1", "purpose", "associated index doc");
+        refresh(associatedIndex);
+
+        // put a document in a normal index
+        indexDoc("my_index", "1", "purpose", "normal index doc");
+        refresh("my_index");
+
+        // call the reset API
+        ResetFeatureStateResponse apiResponse = client().execute(ResetFeatureStateAction.INSTANCE, new ResetFeatureStateRequest()).get();
+        assertThat(apiResponse.getItemList(), containsInAnyOrder(
+            new ResetFeatureStateResponse.ResetFeatureStateStatus("SystemIndexTestPlugin", "SUCCESS"),
+            new ResetFeatureStateResponse.ResetFeatureStateStatus("SecondSystemIndexTestPlugin", "SUCCESS"),
+            new ResetFeatureStateResponse.ResetFeatureStateStatus("tasks", "SUCCESS")
+        ));
+
+        // verify that both indices are gone
+        Exception e1 = expectThrows(IndexNotFoundException.class, () -> client().admin().indices().prepareGetIndex()
+            .addIndices(systemIndex1)
+            .get());
+
+        assertThat(e1.getMessage(), containsString("no such index"));
+
+        Exception e2 = expectThrows(IndexNotFoundException.class, () -> client().admin().indices().prepareGetIndex()
+            .addIndices(associatedIndex)
+            .get());
+
+        assertThat(e2.getMessage(), containsString("no such index"));
+
+        Exception e3 = expectThrows(IndexNotFoundException.class, () -> client().admin().indices().prepareGetIndex()
+            .addIndices(systemIndex2)
+            .get());
+
+        assertThat(e3.getMessage(), containsString("no such index"));
+
+        GetIndexResponse response = client().admin().indices().prepareGetIndex()
+            .addIndices("my_index")
+            .get();
+
+        assertThat(response.getIndices(), arrayContaining("my_index"));
+    }
+
+    /**
+     * A test plugin with patterns for system indices and associated indices.
+     */
+    public static class SystemIndexTestPlugin extends Plugin implements SystemIndexPlugin {
+
+        public static final String SYSTEM_INDEX_PATTERN = ".test-system-idx*";
+        public static final String ASSOCIATED_INDEX_PATTERN = ".associated-idx*";
+
+        @Override
+        public Collection<SystemIndexDescriptor> getSystemIndexDescriptors(Settings settings) {
+            return Collections.singletonList(new SystemIndexDescriptor(SYSTEM_INDEX_PATTERN, "System indices for tests"));
+        }
+
+        @Override
+        public Collection<String> getAssociatedIndexPatterns() {
+            return Collections.singletonList(ASSOCIATED_INDEX_PATTERN);
+        }
+
+        @Override
+        public String getFeatureName() {
+            return SystemIndexTestPlugin.class.getSimpleName();
+        }
+
+        @Override
+        public String getFeatureDescription() {
+            return "A simple test plugin";
+        }
+    }
+
+    /**
+     * A second test plugin with a patterns for system indices.
+     */
+    public static class SecondSystemIndexTestPlugin extends Plugin implements SystemIndexPlugin {
+
+        public static final String SYSTEM_INDEX_PATTERN = ".second-test-system-idx*";
+
+        @Override
+        public Collection<SystemIndexDescriptor> getSystemIndexDescriptors(Settings settings) {
+            return Collections.singletonList(new SystemIndexDescriptor(SYSTEM_INDEX_PATTERN, "System indices for tests"));
+        }
+
+        @Override
+        public String getFeatureName() {
+            return SecondSystemIndexTestPlugin.class.getSimpleName();
+        }
+
+        @Override
+        public String getFeatureDescription() {
+            return "A second test plugin";
+        }
+    }
+}

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -58,7 +58,9 @@ import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotAct
 import org.elasticsearch.action.admin.cluster.snapshots.create.TransportCreateSnapshotAction;
 import org.elasticsearch.action.admin.cluster.snapshots.delete.DeleteSnapshotAction;
 import org.elasticsearch.action.admin.cluster.snapshots.delete.TransportDeleteSnapshotAction;
+import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateAction;
 import org.elasticsearch.action.admin.cluster.snapshots.features.SnapshottableFeaturesAction;
+import org.elasticsearch.action.admin.cluster.snapshots.features.TransportResetFeatureStateAction;
 import org.elasticsearch.action.admin.cluster.snapshots.features.TransportSnapshottableFeaturesAction;
 import org.elasticsearch.action.admin.cluster.snapshots.get.GetSnapshotsAction;
 import org.elasticsearch.action.admin.cluster.snapshots.get.TransportGetSnapshotsAction;
@@ -284,6 +286,7 @@ import org.elasticsearch.rest.action.admin.cluster.RestPutRepositoryAction;
 import org.elasticsearch.rest.action.admin.cluster.RestPutStoredScriptAction;
 import org.elasticsearch.rest.action.admin.cluster.RestReloadSecureSettingsAction;
 import org.elasticsearch.rest.action.admin.cluster.RestRemoteClusterInfoAction;
+import org.elasticsearch.rest.action.admin.cluster.RestResetFeatureStateAction;
 import org.elasticsearch.rest.action.admin.cluster.RestRestoreSnapshotAction;
 import org.elasticsearch.rest.action.admin.cluster.RestSnapshotsStatusAction;
 import org.elasticsearch.rest.action.admin.cluster.RestSnapshottableFeaturesAction;
@@ -511,6 +514,7 @@ public class ActionModule extends AbstractModule {
         actions.register(RestoreSnapshotAction.INSTANCE, TransportRestoreSnapshotAction.class);
         actions.register(SnapshotsStatusAction.INSTANCE, TransportSnapshotsStatusAction.class);
         actions.register(SnapshottableFeaturesAction.INSTANCE, TransportSnapshottableFeaturesAction.class);
+        actions.register(ResetFeatureStateAction.INSTANCE, TransportResetFeatureStateAction.class);
 
         actions.register(IndicesStatsAction.INSTANCE, TransportIndicesStatsAction.class);
         actions.register(IndicesSegmentsAction.INSTANCE, TransportIndicesSegmentsAction.class);
@@ -661,6 +665,7 @@ public class ActionModule extends AbstractModule {
         registerHandler.accept(new RestDeleteSnapshotAction());
         registerHandler.accept(new RestSnapshotsStatusAction());
         registerHandler.accept(new RestSnapshottableFeaturesAction());
+        registerHandler.accept(new RestResetFeatureStateAction());
         registerHandler.accept(new RestGetIndicesAction());
         registerHandler.accept(new RestIndicesStatsAction());
         registerHandler.accept(new RestIndicesSegmentsAction(threadPool));

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/features/ResetFeatureStateAction.java

@@ -0,0 +1,22 @@
+/*
+ * 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.action.admin.cluster.snapshots.features;
+
+import org.elasticsearch.action.ActionType;
+
+/** Action for resetting feature states, mostly meaning system indices */
+public class ResetFeatureStateAction extends ActionType<ResetFeatureStateResponse> {
+
+    public static final ResetFeatureStateAction INSTANCE = new ResetFeatureStateAction();
+    public static final String NAME = "cluster:admin/features/reset";
+
+    private ResetFeatureStateAction() {
+        super(NAME, ResetFeatureStateResponse::new);
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/features/ResetFeatureStateRequest.java

@@ -0,0 +1,37 @@
+/*
+ * 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.action.admin.cluster.snapshots.features;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+
+import java.io.IOException;
+
+/** Request for resetting feature state */
+public class ResetFeatureStateRequest extends ActionRequest {
+
+    public ResetFeatureStateRequest() {
+    }
+
+    public ResetFeatureStateRequest(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        return null;
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/features/ResetFeatureStateResponse.java

@@ -0,0 +1,151 @@
+/*
+ * 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.action.admin.cluster.snapshots.features;
+
+import org.elasticsearch.action.ActionResponse;
+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 java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+
+/** Response to a feature state reset request. */
+public class ResetFeatureStateResponse extends ActionResponse implements ToXContentObject {
+
+    List<ResetFeatureStateStatus> resetFeatureStateStatusList;
+
+    /**
+     * Create a response showing which features have had state reset and success
+     * or failure status.
+     *
+     * @param statusList A list of status responses
+     */
+    public ResetFeatureStateResponse(List<ResetFeatureStateStatus> statusList) {
+        resetFeatureStateStatusList = new ArrayList<>();
+        resetFeatureStateStatusList.addAll(statusList);
+        resetFeatureStateStatusList.sort(Comparator.comparing(ResetFeatureStateStatus::getFeatureName));
+    }
+
+    public ResetFeatureStateResponse(StreamInput in) throws IOException {
+        super(in);
+        this.resetFeatureStateStatusList = in.readList(ResetFeatureStateStatus::new);
+    }
+
+    public List<ResetFeatureStateStatus> getItemList() {
+        return this.resetFeatureStateStatusList;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        {
+            builder.startArray("features");
+            for (ResetFeatureStateStatus resetFeatureStateStatus : this.resetFeatureStateStatusList) {
+                builder.value(resetFeatureStateStatus);
+            }
+            builder.endArray();
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeList(this.resetFeatureStateStatusList);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ResetFeatureStateResponse that = (ResetFeatureStateResponse) o;
+        return Objects.equals(resetFeatureStateStatusList, that.resetFeatureStateStatusList);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(resetFeatureStateStatusList);
+    }
+
+    @Override
+    public String toString() {
+        return "ResetFeatureStateResponse{" +
+            "resetFeatureStateStatusList=" + resetFeatureStateStatusList +
+            '}';
+    }
+
+    /**
+     * An object with the name of a feature and a message indicating success or
+     * failure.
+     */
+    public static class ResetFeatureStateStatus implements Writeable, ToXContentObject {
+        private final String featureName;
+        private final String status;
+
+        public ResetFeatureStateStatus(String featureName, String status) {
+            this.featureName = featureName;
+            this.status = status;
+        }
+
+        ResetFeatureStateStatus(StreamInput in) throws IOException {
+            this.featureName = in.readString();
+            this.status = in.readString();
+        }
+
+        public String getFeatureName() {
+            return this.featureName;
+        }
+
+        public String getStatus() {
+            return this.status;
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            builder.field("feature_name", this.featureName);
+            builder.field("status", this.status);
+            builder.endObject();
+            return builder;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeString(this.featureName);
+            out.writeString(this.status);
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            ResetFeatureStateStatus that = (ResetFeatureStateStatus) o;
+            return Objects.equals(featureName, that.featureName) && Objects.equals(status, that.status);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(featureName, status);
+        }
+
+        @Override
+        public String toString() {
+            return "ResetFeatureStateStatus{" +
+                "featureName='" + featureName + '\'' +
+                ", status='" + status + '\'' +
+                '}';
+        }
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/features/TransportResetFeatureStateAction.java

@@ -0,0 +1,72 @@
+/*
+ * 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.action.admin.cluster.snapshots.features;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.GroupedActionListener;
+import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.indices.SystemIndices;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.transport.TransportService;
+
+import java.util.ArrayList;
+import java.util.Collections;
+
+/**
+ * Transport action for cleaning up feature index state.
+ */
+public class TransportResetFeatureStateAction extends HandledTransportAction<ResetFeatureStateRequest, ResetFeatureStateResponse> {
+
+    private final SystemIndices systemIndices;
+    private final NodeClient client;
+    private final ClusterService clusterService;
+
+    @Inject
+    public TransportResetFeatureStateAction(
+        TransportService transportService,
+        ActionFilters actionFilters,
+        SystemIndices systemIndices,
+        NodeClient client,
+        ClusterService clusterService
+    ) {
+        super(ResetFeatureStateAction.NAME, transportService, actionFilters,
+            ResetFeatureStateRequest::new);
+        this.systemIndices = systemIndices;
+        this.client = client;
+        this.clusterService = clusterService;
+    }
+
+    @Override
+    protected void doExecute(
+        Task task,
+        ResetFeatureStateRequest request,
+        ActionListener<ResetFeatureStateResponse> listener) {
+
+        if (systemIndices.getFeatures().size() == 0) {
+            listener.onResponse(new ResetFeatureStateResponse(Collections.emptyList()));
+        }
+
+        final int features = systemIndices.getFeatures().size();
+        GroupedActionListener<ResetFeatureStateResponse.ResetFeatureStateStatus> groupedActionListener = new GroupedActionListener<>(
+            listener.map(responses -> {
+                assert features == responses.size();
+                return new ResetFeatureStateResponse(new ArrayList<>(responses));
+            }),
+            systemIndices.getFeatures().size()
+        );
+
+        for (SystemIndices.Feature feature : systemIndices.getFeatures().values()) {
+            feature.getCleanUpFunction().apply(clusterService, client, groupedActionListener);
+        }
+    }
+}

server/src/main/java/org/elasticsearch/indices/SystemIndices.java

@@ -13,8 +13,17 @@ import org.apache.lucene.util.automaton.Automaton;
 import org.apache.lucene.util.automaton.CharacterRunAutomaton;
 import org.apache.lucene.util.automaton.MinimizationOperations;
 import org.apache.lucene.util.automaton.Operations;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateResponse.ResetFeatureStateStatus;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.TriConsumer;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.snapshots.SnapshotsService;
@@ -29,6 +38,7 @@ import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static java.util.stream.Collectors.toUnmodifiableList;
 import static org.elasticsearch.tasks.TaskResultsService.TASKS_DESCRIPTOR;
@@ -41,13 +51,18 @@ import static org.elasticsearch.tasks.TaskResultsService.TASKS_FEATURE_NAME;
  */
 public class SystemIndices {
     private static final Map<String, Feature> SERVER_SYSTEM_INDEX_DESCRIPTORS = Map.of(
-        TASKS_FEATURE_NAME, new Feature("Manages task results", List.of(TASKS_DESCRIPTOR))
+        TASKS_FEATURE_NAME, new Feature(TASKS_FEATURE_NAME, "Manages task results", List.of(TASKS_DESCRIPTOR))
     );
 
     private final CharacterRunAutomaton runAutomaton;
     private final Map<String, Feature> featureDescriptors;
     private final Map<String, CharacterRunAutomaton> productToSystemIndicesMatcher;
 
+    /**
+     * Initialize the SystemIndices object
+     * @param pluginAndModulesDescriptors A map of this node's feature names to
+     *                                    feature objects.
+     */
     public SystemIndices(Map<String, Feature> pluginAndModulesDescriptors) {
         featureDescriptors = buildSystemIndexDescriptorMap(pluginAndModulesDescriptors);
         checkForOverlappingPatterns(featureDescriptors);
@@ -236,6 +251,11 @@ public class SystemIndices {
             .collect(Collectors.toList());
     }
 
+    /**
+     * Check that a feature name is not reserved
+     * @param name Name of feature
+     * @param plugin Name of plugin providing the feature
+     */
     public static void validateFeatureName(String name, String plugin) {
         if (SnapshotsService.NO_FEATURE_STATES_VALUE.equalsIgnoreCase(name)) {
             throw new IllegalArgumentException("feature name cannot be reserved name [\"" + SnapshotsService.NO_FEATURE_STATES_VALUE +
@@ -243,19 +263,44 @@ public class SystemIndices {
         }
     }
 
+    /**
+     * Class holding a description of a stateful feature.
+     */
     public static class Feature {
         private final String description;
         private final Collection<SystemIndexDescriptor> indexDescriptors;
         private final Collection<String> associatedIndexPatterns;
-
-        public Feature(String description, Collection<SystemIndexDescriptor> indexDescriptors, Collection<String> associatedIndexPatterns) {
+        private final TriConsumer<ClusterService, Client, ActionListener<ResetFeatureStateStatus>> cleanUpFunction;
+
+        /**
+         * Construct a Feature with a custom cleanup function
+         * @param description Description of the feature
+         * @param indexDescriptors Patterns describing system indices for this feature
+         * @param associatedIndexPatterns Patterns describing associated indices
+         * @param cleanUpFunction A function that will clean up the feature's state
+         */
+        public Feature(
+            String description,
+            Collection<SystemIndexDescriptor> indexDescriptors,
+            Collection<String> associatedIndexPatterns,
+            TriConsumer<ClusterService, Client, ActionListener<ResetFeatureStateStatus>> cleanUpFunction) {
             this.description = description;
             this.indexDescriptors = indexDescriptors;
             this.associatedIndexPatterns = associatedIndexPatterns;
+            this.cleanUpFunction = cleanUpFunction;
         }
 
-        public Feature(String description, Collection<SystemIndexDescriptor> indexDescriptors) {
-            this(description, indexDescriptors, Collections.emptyList());
+        /**
+         * Construct a Feature using the default clean-up function
+         * @param name Name of the feature, used in logging
+         * @param description Description of the feature
+         * @param indexDescriptors Patterns describing system indices for this feature
+         */
+        public Feature(String name, String description, Collection<SystemIndexDescriptor> indexDescriptors) {
+            this(description, indexDescriptors, Collections.emptyList(),
+                (clusterService, client, listener) ->
+                    cleanUpFeature(indexDescriptors, Collections.emptyList(), name, clusterService, client, listener)
+            );
         }
 
         public String getDescription() {
@@ -269,5 +314,53 @@ public class SystemIndices {
         public Collection<String> getAssociatedIndexPatterns() {
             return associatedIndexPatterns;
         }
+
+        public TriConsumer<ClusterService, Client, ActionListener<ResetFeatureStateStatus>> getCleanUpFunction() {
+            return cleanUpFunction;
+        }
+
+        /**
+         * Clean up the state of a feature
+         * @param indexDescriptors List of descriptors of a feature's system indices
+         * @param associatedIndexPatterns List of patterns of a feature's associated indices
+         * @param name Name of the feature, used in logging
+         * @param clusterService A clusterService, for retrieving cluster metadata
+         * @param client A client, for issuing delete requests
+         * @param listener A listener to return success or failure of cleanup
+         */
+        public static void cleanUpFeature(
+            Collection<SystemIndexDescriptor> indexDescriptors,
+            Collection<String> associatedIndexPatterns,
+            String name,
+            ClusterService clusterService,
+            Client client,
+            ActionListener<ResetFeatureStateStatus> listener) {
+            Stream<String> systemIndices = indexDescriptors.stream()
+                .map(sid -> sid.getMatchingIndices(clusterService.state().getMetadata()))
+                .flatMap(List::stream);
+
+            List<String> allIndices = Stream.concat(systemIndices, associatedIndexPatterns.stream())
+                .collect(Collectors.toList());
+
+            if (allIndices.isEmpty()) {
+                // if no actual indices match the pattern, we can stop here
+                listener.onResponse(new ResetFeatureStateStatus(name, "SUCCESS"));
+                return;
+            }
+
+            DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest();
+            deleteIndexRequest.indices(allIndices.toArray(Strings.EMPTY_ARRAY));
+            client.execute(DeleteIndexAction.INSTANCE, deleteIndexRequest, new ActionListener<>() {
+                @Override
+                public void onResponse(AcknowledgedResponse acknowledgedResponse) {
+                    listener.onResponse(new ResetFeatureStateStatus(name, "SUCCESS"));
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    listener.onResponse(new ResetFeatureStateStatus(name, "FAILURE: " + e.getMessage()));
+                }
+            });
+        }
     }
 }

server/src/main/java/org/elasticsearch/node/Node.java

@@ -412,7 +412,8 @@ public class Node implements Closeable {
                     plugin -> new SystemIndices.Feature(
                         plugin.getFeatureDescription(),
                         plugin.getSystemIndexDescriptors(settings),
-                        plugin.getAssociatedIndexPatterns()
+                        plugin.getAssociatedIndexPatterns(),
+                        plugin::cleanUpFeature
                     ))
                 );
             final SystemIndices systemIndices = new SystemIndices(featuresMap);

server/src/main/java/org/elasticsearch/plugins/SystemIndexPlugin.java

@@ -8,8 +8,13 @@
 
 package org.elasticsearch.plugins;
 
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.indices.SystemIndexDescriptor;
+import org.elasticsearch.indices.SystemIndices;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -49,4 +54,25 @@ public interface SystemIndexPlugin extends ActionPlugin {
     default Collection<String> getAssociatedIndexPatterns() {
         return Collections.emptyList();
     }
+
+    /**
+     * Cleans up the state of the feature by deleting system indices and associated indices.
+     * Override to do more for cleanup (e.g. cancelling tasks).
+     * @param clusterService Cluster service to provide cluster state
+     * @param client A client, for executing actions
+     * @param listener Listener for post-cleanup result
+     */
+    default void cleanUpFeature(
+        ClusterService clusterService, Client client,
+        ActionListener<ResetFeatureStateResponse.ResetFeatureStateStatus> listener) {
+
+        SystemIndices.Feature.cleanUpFeature(
+            getSystemIndexDescriptors(clusterService.getSettings()),
+            getAssociatedIndexPatterns(),
+            getFeatureName(),
+            clusterService,
+            client,
+            listener
+        );
+    }
 }

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestResetFeatureStateAction.java

@@ -0,0 +1,44 @@
+/*
+ * 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.rest.action.admin.cluster;
+
+import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateAction;
+import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateRequest;
+import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestToXContentListener;
+
+import java.io.IOException;
+import java.util.List;
+
+/** Rest handler for feature state reset requests */
+public class RestResetFeatureStateAction extends BaseRestHandler {
+
+    @Override public boolean allowSystemIndexAccessByDefault() {
+        return true;
+    }
+
+    @Override
+    public List<Route> routes() {
+        return List.of(new Route(RestRequest.Method.POST, "/_features/_reset"));
+    }
+
+    @Override
+    public String getName() {
+        return "reset_feature_state";
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient client) throws IOException {
+        final ResetFeatureStateRequest req = new ResetFeatureStateRequest();
+
+        return restChannel -> client.execute(ResetFeatureStateAction.INSTANCE, req, new RestToXContentListener<>(restChannel));
+    }
+}

server/src/test/java/org/elasticsearch/action/admin/cluster/snapshots/features/ResetFeatureStateResponseTests.java

@@ -0,0 +1,53 @@
+/*
+ * 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.action.admin.cluster.snapshots.features;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class ResetFeatureStateResponseTests extends AbstractWireSerializingTestCase<ResetFeatureStateResponse> {
+
+    @Override
+    protected Writeable.Reader<ResetFeatureStateResponse> instanceReader() {
+        return ResetFeatureStateResponse::new;
+    }
+
+    @Override
+    protected ResetFeatureStateResponse createTestInstance() {
+        List<ResetFeatureStateResponse.ResetFeatureStateStatus> resetStatuses = new ArrayList<>();
+        String feature1 = randomAlphaOfLengthBetween(4, 10);
+        String feature2 = randomValueOtherThan(feature1, () -> randomAlphaOfLengthBetween(4, 10));
+        resetStatuses.add(new ResetFeatureStateResponse.ResetFeatureStateStatus(
+            feature1, randomFrom("SUCCESS", "FAILURE")));
+        resetStatuses.add(new ResetFeatureStateResponse.ResetFeatureStateStatus(
+            feature2, randomFrom("SUCCESS", "FAILURE")));
+        return new ResetFeatureStateResponse(resetStatuses);
+    }
+
+    @Override
+    protected ResetFeatureStateResponse mutateInstance(ResetFeatureStateResponse instance) throws IOException {
+        int minSize = 0;
+        if (instance.getItemList().size() == 0) {
+            minSize = 1;
+        }
+        Set<String> existingFeatureNames = instance.getItemList().stream()
+            .map(ResetFeatureStateResponse.ResetFeatureStateStatus::getFeatureName)
+            .collect(Collectors.toSet());
+        return new ResetFeatureStateResponse(randomList(minSize, 10,
+            () -> new ResetFeatureStateResponse.ResetFeatureStateStatus(
+                randomValueOtherThanMany(existingFeatureNames::contains, () -> randomAlphaOfLengthBetween(4, 10)),
+                randomAlphaOfLengthBetween(5, 10))));
+    }
+}

server/src/test/java/org/elasticsearch/action/admin/indices/alias/get/TransportGetAliasesActionTests.java

@@ -159,7 +159,7 @@ public class TransportGetAliasesActionTests extends ESTestCase {
         ClusterState state = systemIndexTestClusterState();
         SystemIndices systemIndices = new SystemIndices(Collections.singletonMap(
             this.getTestName(),
-            new SystemIndices.Feature("test feature",
+            new SystemIndices.Feature(this.getTestName(), "test feature",
                 Collections.singletonList(new SystemIndexDescriptor(".y", "an index that doesn't exist")))));
 
         GetAliasesRequest request = new GetAliasesRequest(".y");

server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionTests.java

@@ -245,7 +245,7 @@ public class TransportBulkActionTests extends ESTestCase {
         indicesLookup.put(".bar",
             new Index(IndexMetadata.builder(".bar").settings(settings).system(true).numberOfShards(1).numberOfReplicas(0).build()));
         SystemIndices systemIndices = new SystemIndices(
-            Map.of("plugin", new SystemIndices.Feature("test feature", List.of(new SystemIndexDescriptor(".test", "")))));
+            Map.of("plugin", new SystemIndices.Feature("plugin", "test feature", List.of(new SystemIndexDescriptor(".test", "")))));
         List<String> onlySystem = List.of(".foo", ".bar");
         assertTrue(bulkAction.isOnlySystem(buildBulkRequest(onlySystem), indicesLookup, systemIndices));
 

server/src/test/java/org/elasticsearch/action/support/AutoCreateIndexTests.java

@@ -302,7 +302,7 @@ public class AutoCreateIndexTests extends ESTestCase {
 
     private AutoCreateIndex newAutoCreateIndex(Settings settings) {
         SystemIndices systemIndices = new SystemIndices(Map.of(
-            "plugin", new SystemIndices.Feature("test feature", List.of(new SystemIndexDescriptor(TEST_SYSTEM_INDEX_NAME, "")))));
+            "plugin", new SystemIndices.Feature("plugin", "test feature", List.of(new SystemIndexDescriptor(TEST_SYSTEM_INDEX_NAME, "")))));
         return new AutoCreateIndex(settings, new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
             TestIndexNameExpressionResolver.newInstance(systemIndices), systemIndices);
     }

server/src/test/java/org/elasticsearch/cluster/metadata/IndexNameExpressionResolverTests.java

@@ -1922,13 +1922,15 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
             Map.of(
                 "ml",
                 new Feature(
+                    "ml",
                     "ml indices",
                     List.of(new SystemIndexDescriptor(".ml-meta", "ml meta"), new SystemIndexDescriptor(".ml-stuff", "other ml"))
                 ),
                 "watcher",
-                new Feature("watcher indices", List.of(new SystemIndexDescriptor(".watches", "watches index"))),
+                new Feature("watcher", "watcher indices", List.of(new SystemIndexDescriptor(".watches", "watches index"))),
                 "stack-component",
-                new Feature("stack component",
+                new Feature("stack-component",
+                    "stack component",
                     List.of(
                         new SystemIndexDescriptor(
                             ".external-sys-idx",
@@ -2316,11 +2318,11 @@ public class IndexNameExpressionResolverTests extends ESTestCase {
             .put(indexBuilder("some-other-index").state(State.OPEN));
         SystemIndices systemIndices = new SystemIndices(
             Map.of("ml",
-                new Feature("ml indices",
+                new Feature("ml", "ml indices",
                     List.of(new SystemIndexDescriptor(".ml-meta", "ml meta"), new SystemIndexDescriptor(".ml-stuff", "other ml"))
                 ),
                 "watcher",
-                new Feature("watcher indices", List.of(new SystemIndexDescriptor(".watches", "watches index")))
+                new Feature("watcher", "watcher indices", List.of(new SystemIndexDescriptor(".watches", "watches index")))
             )
         );
         indexNameExpressionResolver = new IndexNameExpressionResolver(threadContext, systemIndices);

server/src/test/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexServiceTests.java

@@ -522,7 +522,8 @@ public class MetadataCreateIndexServiceTests extends ESTestCase {
                 null,
                 threadPool,
                 null,
-                new SystemIndices(Collections.singletonMap("foo", new SystemIndices.Feature("test feature", systemIndexDescriptors))),
+                new SystemIndices(Collections.singletonMap("foo", new SystemIndices.Feature("foo", "test feature",
+                    systemIndexDescriptors))),
                 false
             );
             // Check deprecations

server/src/test/java/org/elasticsearch/indices/SystemIndexManagerTests.java

@@ -72,7 +72,7 @@ public class SystemIndexManagerTests extends ESTestCase {
         .setOrigin("FAKE_ORIGIN")
         .build();
 
-    private static final SystemIndices.Feature FEATURE = new SystemIndices.Feature("a test feature", List.of(DESCRIPTOR));
+    private static final SystemIndices.Feature FEATURE = new SystemIndices.Feature("foo", "a test feature", List.of(DESCRIPTOR));
 
     private Client client;
 
@@ -101,8 +101,8 @@ public class SystemIndexManagerTests extends ESTestCase {
             .build();
 
         SystemIndices systemIndices = new SystemIndices(Map.of(
-            "index 1", new SystemIndices.Feature("index 1 feature", List.of(d1)),
-            "index 2", new SystemIndices.Feature("index 2 feature", List.of(d2))));
+            "index 1", new SystemIndices.Feature("index 1", "index 1 feature", List.of(d1)),
+            "index 2", new SystemIndices.Feature("index 2", "index 2 feature", List.of(d2))));
         SystemIndexManager manager = new SystemIndexManager(systemIndices, client);
 
         final List<SystemIndexDescriptor> eligibleDescriptors = manager.getEligibleDescriptors(
@@ -139,8 +139,8 @@ public class SystemIndexManagerTests extends ESTestCase {
             .build();
 
         SystemIndices systemIndices = new SystemIndices(Map.of(
-            "index 1", new SystemIndices.Feature("index 1 feature", List.of(d1)),
-            "index 2", new SystemIndices.Feature("index 2 feature", List.of(d2))));;
+            "index 1", new SystemIndices.Feature("index 1", "index 1 feature", List.of(d1)),
+            "index 2", new SystemIndices.Feature("index 2", "index 2 feature", List.of(d2))));;
         SystemIndexManager manager = new SystemIndexManager(systemIndices, client);
 
         final List<SystemIndexDescriptor> eligibleDescriptors = manager.getEligibleDescriptors(

server/src/test/java/org/elasticsearch/indices/SystemIndicesTests.java

@@ -34,9 +34,9 @@ public class SystemIndicesTests extends ESTestCase {
         String broadPatternSource = "AAA" + randomAlphaOfLength(5);
         String otherSource = "ZZZ" + randomAlphaOfLength(6);
         Map<String, SystemIndices.Feature> descriptors = new HashMap<>();
-        descriptors.put(broadPatternSource, new SystemIndices.Feature("test feature", List.of(broadPattern)));
+        descriptors.put(broadPatternSource, new SystemIndices.Feature(broadPatternSource, "test feature", List.of(broadPattern)));
         descriptors.put(otherSource,
-            new SystemIndices.Feature("test 2", List.of(notOverlapping, overlapping1, overlapping2, overlapping3)));
+            new SystemIndices.Feature(otherSource, "test 2", List.of(notOverlapping, overlapping1, overlapping2, overlapping3)));
 
         IllegalStateException exception = expectThrows(IllegalStateException.class,
             () -> SystemIndices.checkForOverlappingPatterns(descriptors));
@@ -62,8 +62,8 @@ public class SystemIndicesTests extends ESTestCase {
         String source1 = "AAA" + randomAlphaOfLength(5);
         String source2 = "ZZZ" + randomAlphaOfLength(6);
         Map<String, SystemIndices.Feature> descriptors = new HashMap<>();
-        descriptors.put(source1, new SystemIndices.Feature("test", List.of(pattern1)));
-        descriptors.put(source2, new SystemIndices.Feature("test", List.of(pattern2)));
+        descriptors.put(source1, new SystemIndices.Feature(source1, "test", List.of(pattern1)));
+        descriptors.put(source2, new SystemIndices.Feature(source2, "test", List.of(pattern2)));
 
         IllegalStateException exception = expectThrows(IllegalStateException.class,
             () -> SystemIndices.checkForOverlappingPatterns(descriptors));
@@ -84,7 +84,8 @@ public class SystemIndicesTests extends ESTestCase {
 
     public void testPluginCannotOverrideBuiltInSystemIndex() {
         Map<String, SystemIndices.Feature> pluginMap = Map.of(
-            TASKS_FEATURE_NAME, new SystemIndices.Feature("test", List.of(new SystemIndexDescriptor(TASK_INDEX, "Task Result Index")))
+            TASKS_FEATURE_NAME, new SystemIndices.Feature(TASKS_FEATURE_NAME, "test", List.of(new SystemIndexDescriptor(TASK_INDEX, "Task" +
+                " Result Index")))
         );
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> new SystemIndices(pluginMap));
         assertThat(e.getMessage(), containsString("plugin or module attempted to define the same source"));
@@ -93,7 +94,7 @@ public class SystemIndicesTests extends ESTestCase {
     public void testPatternWithSimpleRange() {
 
         final SystemIndices systemIndices = new SystemIndices(Map.of(
-            "test", new SystemIndices.Feature("test feature", List.of(new SystemIndexDescriptor(".test-[abc]", "")))
+            "test", new SystemIndices.Feature("test", "test feature", List.of(new SystemIndexDescriptor(".test-[abc]", "")))
         ));
 
         assertThat(systemIndices.isSystemIndex(".test-a"), equalTo(true));
@@ -108,7 +109,7 @@ public class SystemIndicesTests extends ESTestCase {
 
     public void testPatternWithSimpleRangeAndRepeatOperator() {
         final SystemIndices systemIndices = new SystemIndices(Map.of(
-            "test", new SystemIndices.Feature("test feature", List.of(new SystemIndexDescriptor(".test-[a]+", "")))
+            "test", new SystemIndices.Feature("test", "test feature", List.of(new SystemIndexDescriptor(".test-[a]+", "")))
         ));
 
         assertThat(systemIndices.isSystemIndex(".test-a"), equalTo(true));
@@ -120,7 +121,7 @@ public class SystemIndicesTests extends ESTestCase {
 
     public void testPatternWithComplexRange() {
         final SystemIndices systemIndices = new SystemIndices(Map.of(
-            "test", new SystemIndices.Feature("test feature", List.of(new SystemIndexDescriptor(".test-[a-c]", "")))
+            "test", new SystemIndices.Feature("test", "test feature", List.of(new SystemIndexDescriptor(".test-[a-c]", "")))
         ));
 
         assertThat(systemIndices.isSystemIndex(".test-a"), equalTo(true));
@@ -141,8 +142,8 @@ public class SystemIndicesTests extends ESTestCase {
         SystemIndexDescriptor pattern2 = new SystemIndexDescriptor(".test-a*", "");
 
         Map<String, SystemIndices.Feature> descriptors = new HashMap<>();
-        descriptors.put(source1, new SystemIndices.Feature("source 1", List.of(pattern1)));
-        descriptors.put(source2, new SystemIndices.Feature("source 2", List.of(pattern2)));
+        descriptors.put(source1, new SystemIndices.Feature(source1, "source 1", List.of(pattern1)));
+        descriptors.put(source2, new SystemIndices.Feature(source2, "source 2", List.of(pattern2)));
 
         IllegalStateException exception = expectThrows(IllegalStateException.class,
             () -> SystemIndices.checkForOverlappingPatterns(descriptors));

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/MachineLearning.java

@@ -10,8 +10,10 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateResponse;
 import org.elasticsearch.action.support.ActionFilter;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.client.OriginSettingClient;
@@ -364,8 +366,10 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
+import java.util.stream.Collectors;
 
 import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
@@ -1245,6 +1249,62 @@ public class MachineLearning extends Plugin implements SystemIndexPlugin,
         return "Provides anomaly detection and forecasting functionality";
     }
 
+    @Override public void cleanUpFeature(
+        ClusterService clusterService,
+        Client client,
+        ActionListener<ResetFeatureStateResponse.ResetFeatureStateStatus> listener) {
+
+        Map<String, Boolean> results = new ConcurrentHashMap<>();
+
+        ActionListener<StopDataFrameAnalyticsAction.Response> afterDataframesStopped = ActionListener.wrap(dataFrameStopResponse -> {
+            // Handle the response
+            results.put("data_frame/analytics", dataFrameStopResponse.isStopped());
+
+            if (results.values().stream().allMatch(b -> b)) {
+                // Call into the original listener to clean up the indices
+                SystemIndexPlugin.super.cleanUpFeature(clusterService, client, listener);
+            } else {
+                final List<String> failedComponents = results.entrySet().stream()
+                    .filter(result -> result.getValue() == false)
+                    .map(Map.Entry::getKey)
+                    .collect(Collectors.toList());
+                listener.onFailure(new RuntimeException("Some components failed to reset: " + failedComponents));
+            }
+        }, listener::onFailure);
+
+
+        ActionListener<CloseJobAction.Response> afterAnomalyDetectionClosed = ActionListener.wrap(closeJobResponse -> {
+            // Handle the response
+            results.put("anomaly_detectors", closeJobResponse.isClosed());
+
+            // Stop data frame analytics
+            StopDataFrameAnalyticsAction.Request  stopDataFramesReq = new StopDataFrameAnalyticsAction.Request("_all");
+            stopDataFramesReq.setForce(true);
+            stopDataFramesReq.setAllowNoMatch(true);
+            client.execute(StopDataFrameAnalyticsAction.INSTANCE, stopDataFramesReq, afterDataframesStopped);
+        }, listener::onFailure);
+
+        // Close anomaly detection jobs
+        ActionListener<StopDatafeedAction.Response> afterDataFeedsStopped = ActionListener.wrap(datafeedResponse -> {
+            // Handle the response
+            results.put("datafeeds", datafeedResponse.isStopped());
+
+            // Close anomaly detection jobs
+            CloseJobAction.Request closeJobsRequest = new CloseJobAction.Request();
+            closeJobsRequest.setForce(true);
+            closeJobsRequest.setAllowNoMatch(true);
+            closeJobsRequest.setJobId("_all");
+            client.execute(CloseJobAction.INSTANCE, closeJobsRequest, afterAnomalyDetectionClosed);
+        }, listener::onFailure);
+
+        // Stop data feeds
+        StopDatafeedAction.Request stopDatafeedsReq = new StopDatafeedAction.Request("_all");
+        stopDatafeedsReq.setAllowNoMatch(true);
+        stopDatafeedsReq.setForce(true);
+        client.execute(StopDatafeedAction.INSTANCE, stopDatafeedsReq,
+            afterDataFeedsStopped);
+    }
+
     @Override
     public BreakerSettings getCircuitBreaker(Settings settings) {
         return BreakerSettings.updateFromSettings(

x-pack/plugin/security/qa/operator-privileges-tests/src/javaRestTest/java/org/elasticsearch/xpack/security/operator/Constants.java

@@ -82,6 +82,7 @@ public class Constants {
         "cluster:admin/snapshot/status",
         "cluster:admin/snapshot/status[nodes]",
         "cluster:admin/features/get",
+        "cluster:admin/features/reset",
         "cluster:admin/tasks/cancel",
         "cluster:admin/transform/delete",
         "cluster:admin/transform/preview",

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/Transform.java

@@ -10,11 +10,14 @@ package org.elasticsearch.xpack.transform;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.IndexTemplateMetadata;
+import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
@@ -354,6 +357,35 @@ public class Transform extends Plugin implements SystemIndexPlugin, PersistentTa
         return List.of(AUDIT_INDEX_PATTERN);
     }
 
+    @Override
+    public void cleanUpFeature(
+        ClusterService clusterService,
+        Client client,
+        ActionListener<ResetFeatureStateResponse.ResetFeatureStateStatus> listener
+    ) {
+        ActionListener<StopTransformAction.Response> afterStoppingTransforms = ActionListener.wrap(stopTransformsResponse -> {
+            if (stopTransformsResponse.isAcknowledged()
+                && stopTransformsResponse.getTaskFailures().isEmpty()
+                && stopTransformsResponse.getNodeFailures().isEmpty()) {
+
+                SystemIndexPlugin.super.cleanUpFeature(clusterService, client, listener);
+            } else {
+                String errMsg = "Failed to reset Transform: "
+                    + (stopTransformsResponse.isAcknowledged() ? "" : "not acknowledged ")
+                    + (stopTransformsResponse.getNodeFailures().isEmpty()
+                        ? ""
+                        : "node failures: " + stopTransformsResponse.getNodeFailures() + " ")
+                    + (stopTransformsResponse.getTaskFailures().isEmpty()
+                        ? ""
+                        : "task failures: " + stopTransformsResponse.getTaskFailures());
+                listener.onResponse(new ResetFeatureStateResponse.ResetFeatureStateStatus(this.getFeatureName(), errMsg));
+            }
+        }, listener::onFailure);
+
+        StopTransformAction.Request stopTransformsRequest = new StopTransformAction.Request(Metadata.ALL, true, true, null, true, false);
+        client.execute(StopTransformAction.INSTANCE, stopTransformsRequest, afterStoppingTransforms);
+    }
+
     @Override
     public String getFeatureName() {
         return "transform";