[Transform] finalize feature reset integration

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

@@ -203,6 +203,7 @@
 import org.elasticsearch.xpack.core.textstructure.action.FindStructureAction;
 import org.elasticsearch.xpack.core.transform.TransformFeatureSetUsage;
 import org.elasticsearch.xpack.core.transform.TransformField;
+import org.elasticsearch.xpack.core.transform.TransformMetadata;
 import org.elasticsearch.xpack.core.transform.action.DeleteTransformAction;
 import org.elasticsearch.xpack.core.transform.action.GetTransformAction;
 import org.elasticsearch.xpack.core.transform.action.GetTransformStatsAction;
@@ -500,6 +501,8 @@ public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
             new NamedWriteableRegistry.Entry(LifecycleAction.class, SearchableSnapshotAction.NAME, SearchableSnapshotAction::new),
             new NamedWriteableRegistry.Entry(LifecycleAction.class, MigrateAction.NAME, MigrateAction::new),
             // Transforms
+            new NamedWriteableRegistry.Entry(Metadata.Custom.class, TransformMetadata.TYPE, TransformMetadata::new),
+            new NamedWriteableRegistry.Entry(NamedDiff.class, TransformMetadata.TYPE, TransformMetadata.TransformMetadataDiff::new),
             new NamedWriteableRegistry.Entry(XPackFeatureSet.Usage.class, XPackField.TRANSFORM, TransformFeatureSetUsage::new),
             new NamedWriteableRegistry.Entry(PersistentTaskParams.class, TransformField.TASK_NAME, TransformTaskParams::new),
             new NamedWriteableRegistry.Entry(Task.Status.class, TransformField.TASK_NAME, TransformState::new),
@@ -580,7 +583,9 @@ public List<NamedXContentRegistry.Entry> getNamedXContent() {
                 new NamedXContentRegistry.Entry(Task.Status.class, new ParseField(TransformField.TASK_NAME),
                         TransformState::fromXContent),
                 new NamedXContentRegistry.Entry(PersistentTaskState.class, new ParseField(TransformField.TASK_NAME),
-                        TransformState::fromXContent)
+                        TransformState::fromXContent),
+                new NamedXContentRegistry.Entry(Metadata.Custom.class, new ParseField(TransformMetadata.TYPE),
+                    parser -> TransformMetadata.LENIENT_PARSER.parse(parser, null).build())
             );
     }
 }

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

@@ -82,6 +82,7 @@
 import org.elasticsearch.xpack.core.ssl.SSLConfiguration;
 import org.elasticsearch.xpack.core.ssl.SSLConfigurationReloader;
 import org.elasticsearch.xpack.core.ssl.SSLService;
+import org.elasticsearch.xpack.core.transform.TransformMetadata;
 import org.elasticsearch.xpack.core.watcher.WatcherMetadata;
 import org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsConstants;
 
@@ -240,7 +241,8 @@ private static boolean alreadyContainsXPackCustomMetadata(ClusterState clusterSt
         return metadata.custom(LicensesMetadata.TYPE) != null ||
             metadata.custom(MlMetadata.TYPE) != null ||
             metadata.custom(WatcherMetadata.TYPE) != null ||
-            clusterState.custom(TokenMetadata.TYPE) != null;
+            clusterState.custom(TokenMetadata.TYPE) != null ||
+            metadata.custom(TransformMetadata.TYPE) != null;
     }
 
     @Override

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

@@ -0,0 +1,131 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.action;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.ElasticsearchTimeoutException;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.action.support.master.AcknowledgedTransportMasterNodeAction;
+import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.block.ClusterBlockLevel;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+
+public abstract class AbstractTransportSetResetModeAction extends AcknowledgedTransportMasterNodeAction<SetResetModeActionRequest> {
+
+    private static final Logger logger = LogManager.getLogger(AbstractTransportSetResetModeAction.class);
+    private final ClusterService clusterService;
+
+    @Inject
+    public AbstractTransportSetResetModeAction(
+        String actionName,
+        TransportService transportService,
+        ThreadPool threadPool,
+        ClusterService clusterService,
+        ActionFilters actionFilters,
+        IndexNameExpressionResolver indexNameExpressionResolver) {
+        super(
+            actionName,
+            transportService,
+            clusterService,
+            threadPool,
+            actionFilters,
+            SetResetModeActionRequest::new,
+            indexNameExpressionResolver,
+            ThreadPool.Names.SAME
+        );
+        this.clusterService = clusterService;
+    }
+
+    protected abstract boolean isResetMode(ClusterState clusterState);
+
+    protected abstract String featureName();
+
+    protected abstract ClusterState setState(ClusterState oldState, SetResetModeActionRequest request);
+
+    @Override
+    protected void masterOperation(Task task,
+                                   SetResetModeActionRequest request,
+                                   ClusterState state,
+                                   ActionListener<AcknowledgedResponse> listener) throws Exception {
+
+        final boolean isResetModeEnabled = isResetMode(state);
+        // Noop, nothing for us to do, simply return fast to the caller
+        if (request.isEnabled() == isResetModeEnabled) {
+            logger.debug(() -> new ParameterizedMessage("Reset mode noop for [{}]", featureName()));
+            listener.onResponse(AcknowledgedResponse.TRUE);
+            return;
+        }
+
+        logger.debug(
+            () -> new ParameterizedMessage(
+                "Starting to set [reset_mode] for [{}] to [{}] from [{}]",
+                featureName(),
+                request.isEnabled(),
+                isResetModeEnabled
+            )
+        );
+
+        ActionListener<AcknowledgedResponse> wrappedListener = ActionListener.wrap(
+            r -> {
+                logger.debug(() -> new ParameterizedMessage("Completed reset mode request for [{}]", featureName()));
+                listener.onResponse(r);
+            },
+            e -> {
+                logger.debug(
+                    () -> new ParameterizedMessage("Completed reset mode for [{}] request but with failure", featureName()),
+                    e
+                );
+                listener.onFailure(e);
+            }
+        );
+
+        ActionListener<AcknowledgedResponse> clusterStateUpdateListener = ActionListener.wrap(
+            acknowledgedResponse -> {
+                if (acknowledgedResponse.isAcknowledged() == false) {
+                    wrappedListener.onFailure(new ElasticsearchTimeoutException("Unknown error occurred while updating cluster state"));
+                    return;
+                }
+                wrappedListener.onResponse(acknowledgedResponse);
+            },
+            wrappedListener::onFailure
+        );
+
+        clusterService.submitStateUpdateTask(featureName() + "-set-reset-mode",
+            new AckedClusterStateUpdateTask(request, clusterStateUpdateListener) {
+
+                @Override
+                protected AcknowledgedResponse newResponse(boolean acknowledged) {
+                    logger.trace(() -> new ParameterizedMessage("Cluster update response built for [{}]: {}", featureName(), acknowledged));
+                    return AcknowledgedResponse.of(acknowledged);
+                }
+
+                @Override
+                public ClusterState execute(ClusterState currentState) {
+                    logger.trace(() -> new ParameterizedMessage("Executing cluster state update for [{}]", featureName()));
+                    return setState(currentState, request);
+                }
+            });
+    }
+
+    @Override
+    protected ClusterBlockException checkBlock(SetResetModeActionRequest request, ClusterState state) {
+        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
+    }
+
+}

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

@@ -0,0 +1,90 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.action;
+
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.Objects;
+
+public class SetResetModeActionRequest extends AcknowledgedRequest<SetResetModeActionRequest> implements ToXContentObject {
+    public static SetResetModeActionRequest enabled() {
+        return new SetResetModeActionRequest(true);
+    }
+
+    public static SetResetModeActionRequest disabled() {
+        return new SetResetModeActionRequest(false);
+    }
+
+    private final boolean enabled;
+
+    private static final ParseField ENABLED = new ParseField("enabled");
+    public static final ConstructingObjectParser<SetResetModeActionRequest, Void> PARSER =
+        new ConstructingObjectParser<>("set_reset_mode_action_request", a -> new SetResetModeActionRequest((Boolean)a[0]));
+
+    static {
+        PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), ENABLED);
+    }
+
+    SetResetModeActionRequest(boolean enabled) {
+        this.enabled = enabled;
+    }
+
+    public SetResetModeActionRequest(StreamInput in) throws IOException {
+        super(in);
+        this.enabled = in.readBoolean();
+    }
+
+    public boolean isEnabled() {
+        return enabled;
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        return null;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeBoolean(enabled);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(enabled);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null || obj.getClass() != getClass()) {
+            return false;
+        }
+        SetResetModeActionRequest other = (SetResetModeActionRequest) obj;
+        return Objects.equals(enabled, other.enabled);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
+        builder.startObject();
+        builder.field(ENABLED.getPreferredName(), enabled);
+        builder.endObject();
+        return builder;
+    }
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/SetResetModeAction.java

@@ -6,19 +6,9 @@
  */
 package org.elasticsearch.xpack.core.ml.action;
 
-import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionType;
-import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
-import org.elasticsearch.common.ParseField;
-import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.xcontent.ConstructingObjectParser;
-import org.elasticsearch.common.xcontent.ToXContentObject;
-import org.elasticsearch.common.xcontent.XContentBuilder;
 
-import java.io.IOException;
-import java.util.Objects;
 
 public class SetResetModeAction extends ActionType<AcknowledgedResponse> {
 
@@ -29,73 +19,4 @@ private SetResetModeAction() {
         super(NAME, AcknowledgedResponse::readFrom);
     }
 
-    public static class Request extends AcknowledgedRequest<Request> implements ToXContentObject {
-
-        public static Request enabled() {
-            return new Request(true);
-        }
-
-        public static Request disabled() {
-            return new Request(false);
-        }
-
-        private final boolean enabled;
-
-        private static final ParseField ENABLED = new ParseField("enabled");
-        public static final ConstructingObjectParser<Request, Void> PARSER =
-            new ConstructingObjectParser<>(NAME, a -> new Request((Boolean)a[0]));
-
-        static {
-            PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), ENABLED);
-        }
-
-        Request(boolean enabled) {
-            this.enabled = enabled;
-        }
-
-        public Request(StreamInput in) throws IOException {
-            super(in);
-            this.enabled = in.readBoolean();
-        }
-
-        public boolean isEnabled() {
-            return enabled;
-        }
-
-        @Override
-        public ActionRequestValidationException validate() {
-            return null;
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
-            out.writeBoolean(enabled);
-        }
-
-        @Override
-        public int hashCode() {
-            return Objects.hash(enabled);
-        }
-
-        @Override
-        public boolean equals(Object obj) {
-            if (this == obj) {
-                return true;
-            }
-            if (obj == null || obj.getClass() != getClass()) {
-                return false;
-            }
-            Request other = (Request) obj;
-            return Objects.equals(enabled, other.enabled);
-        }
-
-        @Override
-        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-            builder.startObject();
-            builder.field(ENABLED.getPreferredName(), enabled);
-            builder.endObject();
-            return builder;
-        }
-    }
 }

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

@@ -29,6 +29,9 @@ public class TransformMessages {
     public static final String TRANSFORM_FAILED_TO_PERSIST_STATS = "Failed to persist transform statistics for transform [{0}]";
     public static final String UNKNOWN_TRANSFORM_STATS = "Statistics for transform [{0}] could not be found";
 
+    public static final String FAILED_TO_UNSET_RESET_MODE =
+        "Failed to set [reset_mode] to [false] after {0}. To allow transforms to run, please call the feature reset API again";
+
     public static final String REST_DEPRECATED_ENDPOINT = "[_data_frame/transforms/] is deprecated, use [_transform/] in the future.";
     public static final String REST_WARN_NO_TRANSFORM_NODES =
         "Transform requires the transform node role for at least 1 node, found no transform nodes";