[Transform] Create upgrade mode (#117858) (#119352)

Prevent writes to the Transform system index while it is being reindexed
during upgrade.

- Add a new REST API, `_transform/set_upgrade_mode`, similar to `_ml`,
  which will enable via `?enabled` or disable (when omitted).
- Add a new Transport action, enabling or disabling upgrade mode.  When
  enabled, all Transforms will abort and move into the `waiting` state,
  preventing Transforms from running and writing to the system index.
- Hook into the System Index upgrade action, calling the new Transform
  action to set and unset upgrade mode.

Co-authored-by: David Kyle <[email protected]>
Co-authored-by: Elastic Machine <[email protected]>

Author: 3 people

docs/changelog/117858.yaml

@@ -0,0 +1,5 @@
+pr: 117858
+summary: Create upgrade mode
+area: Transform
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -151,6 +151,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ERROR_TRACE_IN_TRANSPORT_HEADER = def(8_811_00_0);
     public static final TransportVersion FAILURE_STORE_ENABLED_BY_CLUSTER_SETTING = def(8_812_00_0);
     public static final TransportVersion SIMULATE_IGNORED_FIELDS = def(8_813_00_0);
+    public static final TransportVersion TRANSFORMS_UPGRADE_MODE = def(8_814_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.core.transform;
 
+import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
 import org.elasticsearch.xcontent.ParseField;
 
 /*
@@ -101,5 +102,14 @@ public final class TransformField {
     // internal document id
     public static String DOCUMENT_ID_FIELD = "_id";
 
+    public static final PersistentTasksCustomMetadata.Assignment AWAITING_UPGRADE = new PersistentTasksCustomMetadata.Assignment(
+        null,
+        "Transform task will not be assigned while upgrade mode is enabled."
+    );
+    public static final PersistentTasksCustomMetadata.Assignment RESET_IN_PROGRESS = new PersistentTasksCustomMetadata.Assignment(
+        null,
+        "Transform task will not be assigned as a feature reset is in progress."
+    );
+
     private TransformField() {}
 }

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

@@ -14,9 +14,9 @@
 import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
@@ -30,8 +30,9 @@
 public class TransformMetadata implements Metadata.Custom {
     public static final String TYPE = "transform";
     public static final ParseField RESET_MODE = new ParseField("reset_mode");
+    public static final ParseField UPGRADE_MODE = new ParseField("upgrade_mode");
 
-    public static final TransformMetadata EMPTY_METADATA = new TransformMetadata(false);
+    public static final TransformMetadata EMPTY_METADATA = new TransformMetadata(false, false);
     // This parser follows the pattern that metadata is parsed leniently (to allow for enhancements)
     public static final ObjectParser<TransformMetadata.Builder, Void> LENIENT_PARSER = new ObjectParser<>(
         "" + "transform_metadata",
@@ -40,19 +41,26 @@ public class TransformMetadata implements Metadata.Custom {
     );
 
     static {
-        LENIENT_PARSER.declareBoolean(TransformMetadata.Builder::isResetMode, RESET_MODE);
+        LENIENT_PARSER.declareBoolean(TransformMetadata.Builder::resetMode, RESET_MODE);
+        LENIENT_PARSER.declareBoolean(TransformMetadata.Builder::upgradeMode, UPGRADE_MODE);
     }
 
     private final boolean resetMode;
+    private final boolean upgradeMode;
 
-    private TransformMetadata(boolean resetMode) {
+    private TransformMetadata(boolean resetMode, boolean upgradeMode) {
         this.resetMode = resetMode;
+        this.upgradeMode = upgradeMode;
     }
 
-    public boolean isResetMode() {
+    public boolean resetMode() {
         return resetMode;
     }
 
+    public boolean upgradeMode() {
+        return upgradeMode;
+    }
+
     @Override
     public TransportVersion getMinimalSupportedVersion() {
         return TransportVersions.MINIMUM_COMPATIBLE;
@@ -75,28 +83,46 @@ public Diff<Metadata.Custom> diff(Metadata.Custom previousState) {
 
     public TransformMetadata(StreamInput in) throws IOException {
         this.resetMode = in.readBoolean();
+        if (in.getTransportVersion().onOrAfter(TransportVersions.TRANSFORMS_UPGRADE_MODE)) {
+            this.upgradeMode = in.readBoolean();
+        } else {
+            this.upgradeMode = false;
+        }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeBoolean(resetMode);
+        if (out.getTransportVersion().onOrAfter(TransportVersions.TRANSFORMS_UPGRADE_MODE)) {
+            out.writeBoolean(upgradeMode);
+        }
     }
 
     @Override
     public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params ignored) {
-        return ChunkedToXContentHelper.field(RESET_MODE.getPreferredName(), resetMode);
+        return Iterators.single(
+            ((builder, params) -> builder.field(UPGRADE_MODE.getPreferredName(), upgradeMode)
+                .field(RESET_MODE.getPreferredName(), resetMode))
+        );
     }
 
     public static class TransformMetadataDiff implements NamedDiff<Metadata.Custom> {
 
         final boolean resetMode;
+        final boolean upgradeMode;
 
         TransformMetadataDiff(TransformMetadata before, TransformMetadata after) {
             this.resetMode = after.resetMode;
+            this.upgradeMode = after.upgradeMode;
         }
 
         public TransformMetadataDiff(StreamInput in) throws IOException {
             resetMode = in.readBoolean();
+            if (in.getTransportVersion().onOrAfter(TransportVersions.TRANSFORMS_UPGRADE_MODE)) {
+                this.upgradeMode = in.readBoolean();
+            } else {
+                this.upgradeMode = false;
+            }
         }
 
         /**
@@ -106,12 +132,15 @@ public TransformMetadataDiff(StreamInput in) throws IOException {
          */
         @Override
         public Metadata.Custom apply(Metadata.Custom part) {
-            return new TransformMetadata(resetMode);
+            return new TransformMetadata(resetMode, upgradeMode);
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeBoolean(resetMode);
+            if (out.getTransportVersion().onOrAfter(TransportVersions.TRANSFORMS_UPGRADE_MODE)) {
+                out.writeBoolean(upgradeMode);
+            }
         }
 
         @Override
@@ -130,7 +159,7 @@ public boolean equals(Object o) {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         TransformMetadata that = (TransformMetadata) o;
-        return resetMode == that.resetMode;
+        return resetMode == that.resetMode && upgradeMode == that.upgradeMode;
     }
 
     @Override
@@ -140,12 +169,17 @@ public final String toString() {
 
     @Override
     public int hashCode() {
-        return Objects.hash(resetMode);
+        return Objects.hash(resetMode, upgradeMode);
+    }
+
+    public Builder builder() {
+        return new TransformMetadata.Builder(this);
     }
 
     public static class Builder {
 
         private boolean resetMode;
+        private boolean upgradeMode;
 
         public static TransformMetadata.Builder from(@Nullable TransformMetadata previous) {
             return new TransformMetadata.Builder(previous);
@@ -156,16 +190,22 @@ public Builder() {}
         public Builder(@Nullable TransformMetadata previous) {
             if (previous != null) {
                 resetMode = previous.resetMode;
+                upgradeMode = previous.upgradeMode;
             }
         }
 
-        public TransformMetadata.Builder isResetMode(boolean isResetMode) {
+        public TransformMetadata.Builder resetMode(boolean isResetMode) {
             this.resetMode = isResetMode;
             return this;
         }
 
+        public TransformMetadata.Builder upgradeMode(boolean upgradeMode) {
+            this.upgradeMode = upgradeMode;
+            return this;
+        }
+
         public TransformMetadata build() {
-            return new TransformMetadata(resetMode);
+            return new TransformMetadata(resetMode, upgradeMode);
         }
     }
 
@@ -176,4 +216,8 @@ public static TransformMetadata getTransformMetadata(ClusterState state) {
         }
         return TransformMetadata;
     }
+
+    public static boolean upgradeMode(ClusterState state) {
+        return getTransformMetadata(state).upgradeMode();
+    }
 }

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

@@ -0,0 +1,20 @@
+/*
+ * 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.transform.action;
+
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+
+public class SetTransformUpgradeModeAction extends ActionType<AcknowledgedResponse> {
+    public static final SetTransformUpgradeModeAction INSTANCE = new SetTransformUpgradeModeAction();
+    public static final String NAME = "cluster:admin/transform/upgrade_mode";
+
+    private SetTransformUpgradeModeAction() {
+        super(NAME);
+    }
+}

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/store/ReservedRolesStoreTests.java

@@ -197,6 +197,7 @@
 import org.elasticsearch.xpack.core.transform.action.GetTransformStatsAction;
 import org.elasticsearch.xpack.core.transform.action.PreviewTransformAction;
 import org.elasticsearch.xpack.core.transform.action.PutTransformAction;
+import org.elasticsearch.xpack.core.transform.action.SetTransformUpgradeModeAction;
 import org.elasticsearch.xpack.core.transform.action.StartTransformAction;
 import org.elasticsearch.xpack.core.transform.action.StopTransformAction;
 import org.elasticsearch.xpack.core.transform.action.UpdateTransformAction;
@@ -3442,6 +3443,7 @@ public void testTransformAdminRole() {
             assertThat(role.cluster().check(PutTransformAction.NAME, request, authentication), is(true));
             assertThat(role.cluster().check(StartTransformAction.NAME, request, authentication), is(true));
             assertThat(role.cluster().check(StopTransformAction.NAME, request, authentication), is(true));
+            assertThat(role.cluster().check(SetTransformUpgradeModeAction.NAME, request, authentication), is(true));
             assertThat(role.cluster().check(DelegatePkiAuthenticationAction.NAME, request, authentication), is(false));
 
             assertThat(role.runAs().check(randomAlphaOfLengthBetween(1, 30)), is(false));
@@ -3531,6 +3533,7 @@ public void testTransformUserRole() {
             assertThat(role.cluster().check(PutTransformAction.NAME, request, authentication), is(false));
             assertThat(role.cluster().check(StartTransformAction.NAME, request, authentication), is(false));
             assertThat(role.cluster().check(StopTransformAction.NAME, request, authentication), is(false));
+            assertThat(role.cluster().check(SetTransformUpgradeModeAction.NAME, request, authentication), is(false));
             assertThat(role.cluster().check(DelegatePkiAuthenticationAction.NAME, request, authentication), is(false));
             assertThat(role.cluster().check(ActivateProfileAction.NAME, request, authentication), is(false));
             assertThat(role.cluster().check(SuggestProfilesAction.NAME, request, authentication), is(false));

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

@@ -115,6 +115,7 @@ public class Constants {
         "cluster:admin/transform/schedule_now",
         "cluster:admin/transform/update",
         "cluster:admin/transform/upgrade",
+        "cluster:admin/transform/upgrade_mode",
         "cluster:admin/transform/validate",
         // "cluster:admin/voting_config/add_exclusions",
         // "cluster:admin/voting_config/clear_exclusions",

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformGetAndGetStatsIT.java

@@ -37,11 +37,6 @@
 import static org.hamcrest.Matchers.oneOf;
 
 public class TransformGetAndGetStatsIT extends TransformRestTestCase {
-
-    private static final String TEST_USER_NAME = "transform_user";
-    private static final String BASIC_AUTH_VALUE_TRANSFORM_USER = basicAuthHeaderValue(TEST_USER_NAME, TEST_PASSWORD_SECURE_STRING);
-    private static final String TEST_ADMIN_USER_NAME = "transform_admin";
-    private static final String BASIC_AUTH_VALUE_TRANSFORM_ADMIN = basicAuthHeaderValue(TEST_ADMIN_USER_NAME, TEST_PASSWORD_SECURE_STRING);
     private static final String DANGLING_TASK_ERROR_MESSAGE =
         "Found task for transform [pivot_continuous], but no configuration for it. To delete this transform use DELETE with force=true.";
 

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformRestTestCase.java

@@ -39,9 +39,16 @@
 import static org.hamcrest.Matchers.is;
 
 public abstract class TransformRestTestCase extends TransformCommonRestTestCase {
-
     protected static final String TEST_PASSWORD = "x-pack-test-password";
     protected static final SecureString TEST_PASSWORD_SECURE_STRING = new SecureString(TEST_PASSWORD.toCharArray());
+
+    protected static final String TEST_USER_NAME = "transform_user";
+    protected static final String BASIC_AUTH_VALUE_TRANSFORM_USER = basicAuthHeaderValue(TEST_USER_NAME, TEST_PASSWORD_SECURE_STRING);
+    protected static final String TEST_ADMIN_USER_NAME = "transform_admin";
+    protected static final String BASIC_AUTH_VALUE_TRANSFORM_ADMIN = basicAuthHeaderValue(
+        TEST_ADMIN_USER_NAME,
+        TEST_PASSWORD_SECURE_STRING
+    );
     private static final String BASIC_AUTH_VALUE_SUPER_USER = basicAuthHeaderValue("x_pack_rest_user", TEST_PASSWORD_SECURE_STRING);
 
     protected static final String REVIEWS_INDEX_NAME = "reviews";