Make system index migration project-aware

Author: nielsbauman

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

@@ -1008,18 +1008,13 @@ public static void cleanUpFeature(
         }
 
         // No-op pre-migration function to be used as the default in case none are provided.
-        private static void noopPreMigrationFunction(
-            ClusterService clusterService,
-            Client client,
-            ActionListener<Map<String, Object>> listener
-        ) {
+        private static void noopPreMigrationFunction(ProjectMetadata project, Client client, ActionListener<Map<String, Object>> listener) {
             listener.onResponse(Collections.emptyMap());
         }
 
         // No-op pre-migration function to be used as the default in case none are provided.
         private static void noopPostMigrationFunction(
             Map<String, Object> preUpgradeMetadata,
-            ClusterService clusterService,
             Client client,
             ActionListener<Boolean> listener
         ) {
@@ -1028,25 +1023,20 @@ private static void noopPostMigrationFunction(
 
         /**
          * Type for the handler that's invoked prior to migrating a Feature's system indices.
-         * See {@link SystemIndexPlugin#prepareForIndicesMigration(ClusterService, Client, ActionListener)}.
+         * See {@link SystemIndexPlugin#prepareForIndicesMigration(ProjectMetadata, Client, ActionListener)}.
          */
         @FunctionalInterface
         public interface MigrationPreparationHandler {
-            void prepareForIndicesMigration(ClusterService clusterService, Client client, ActionListener<Map<String, Object>> listener);
+            void prepareForIndicesMigration(ProjectMetadata project, Client client, ActionListener<Map<String, Object>> listener);
         }
 
         /**
          * Type for the handler that's invoked when all of a feature's system indices have been migrated.
-         * See {@link SystemIndexPlugin#indicesMigrationComplete(Map, ClusterService, Client, ActionListener)}.
+         * See {@link SystemIndexPlugin#indicesMigrationComplete(Map, Client, ActionListener)}.
          */
         @FunctionalInterface
         public interface MigrationCompletionHandler {
-            void indicesMigrationComplete(
-                Map<String, Object> preUpgradeMetadata,
-                ClusterService clusterService,
-                Client client,
-                ActionListener<Boolean> listener
-            );
+            void indicesMigrationComplete(Map<String, Object> preUpgradeMetadata, Client client, ActionListener<Boolean> listener);
         }
 
         public interface CleanupFunction {

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

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.cluster.snapshots.features.ResetFeatureStateResponse;
 import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Settings;
@@ -45,8 +46,8 @@
  * in order to provide a “factory reset” of the plugin state. This can be useful for testing. The default method will simply retrieve a list
  * of system and associated indices and delete them.
  *
- * <p>An implementation may also override {@link #prepareForIndicesMigration(ClusterService, Client, ActionListener)}  and
- * {@link #indicesMigrationComplete(Map, ClusterService, Client, ActionListener)}  in order to take special action before and after a
+ * <p>An implementation may also override {@link #prepareForIndicesMigration(ProjectMetadata, Client, ActionListener)}  and
+ * {@link #indicesMigrationComplete(Map, Client, ActionListener)}  in order to take special action before and after a
  * feature migration, which will temporarily block access to system indices. For example, a plugin might want to enter a safe mode and
  * reject certain requests while the migration is in progress. See org.elasticsearch.upgrades.SystemIndexMigrationExecutor for
  * more details.
@@ -126,19 +127,19 @@ default void cleanUpFeature(
      * very rare.
      *
      * This method can also store metadata to be passed to
-     * {@link SystemIndexPlugin#indicesMigrationComplete(Map, ClusterService, Client, ActionListener)} when it is called; see the
+     * {@link SystemIndexPlugin#indicesMigrationComplete(Map, Client, ActionListener)} when it is called; see the
      * {@code listener} parameter for details.
      *
-     * @param clusterService The cluster service.
+     * @param project The project metadata.
      * @param client A {@link org.elasticsearch.client.internal.ParentTaskAssigningClient} with the parent task set to the upgrade task.
      *               Does not set the origin header, so implementors of this method will likely want to wrap it in an
      *               {@link org.elasticsearch.client.internal.OriginSettingClient}.
      * @param listener A listener that should have {@link ActionListener#onResponse(Object)} called once all necessary preparations for the
      *                 upgrade of indices owned by this plugin have been completed. The {@link Map} passed to the listener will be stored
-     *                 and passed to {@link #indicesMigrationComplete(Map, ClusterService, Client, ActionListener)}. Note the contents of
+     *                 and passed to {@link #indicesMigrationComplete(Map, Client, ActionListener)}. Note the contents of
      *                 the map *must* be writeable using {@link org.elasticsearch.common.io.stream.StreamOutput#writeGenericValue(Object)}.
      */
-    default void prepareForIndicesMigration(ClusterService clusterService, Client client, ActionListener<Map<String, Object>> listener) {
+    default void prepareForIndicesMigration(ProjectMetadata project, Client client, ActionListener<Map<String, Object>> listener) {
         listener.onResponse(Collections.emptyMap());
     }
 
@@ -155,20 +156,14 @@ default void prepareForIndicesMigration(ClusterService clusterService, Client cl
      * with no data format changes allowed).
      *
      * @param preUpgradeMetadata The metadata that was given to the listener by
-     *                           {@link #prepareForIndicesMigration(ClusterService, Client, ActionListener)}.
-     * @param clusterService The cluster service.
+     *                           {@link #prepareForIndicesMigration(ProjectMetadata, Client, ActionListener)}.
      * @param client A {@link org.elasticsearch.client.internal.ParentTaskAssigningClient} with the parent task set to the upgrade task.
      *               Does not set the origin header, so implementors of this method will likely want to wrap it in an
      *               {@link org.elasticsearch.client.internal.OriginSettingClient}.
      * @param listener A listener that should have {@code ActionListener.onResponse(true)} called once all actions following the upgrade
      *                 of this plugin's system indices have been completed.
      */
-    default void indicesMigrationComplete(
-        Map<String, Object> preUpgradeMetadata,
-        ClusterService clusterService,
-        Client client,
-        ActionListener<Boolean> listener
-    ) {
+    default void indicesMigrationComplete(Map<String, Object> preUpgradeMetadata, Client client, ActionListener<Boolean> listener) {
         listener.onResponse(true);
     }
 }

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

@@ -14,6 +14,7 @@
 import org.elasticsearch.cluster.NamedDiff;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -252,4 +253,12 @@ public static MlMetadata getMlMetadata(ClusterState state) {
         }
         return mlMetadata;
     }
+
+    public static MlMetadata getMlMetadata(ProjectMetadata project) {
+        MlMetadata mlMetadata = project == null ? null : project.custom(TYPE);
+        if (mlMetadata == null) {
+            return EMPTY_METADATA;
+        }
+        return mlMetadata;
+    }
 }

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

@@ -15,10 +15,12 @@
 import org.elasticsearch.cluster.ProjectState;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
 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.core.FixForMultiProject;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
@@ -212,8 +214,9 @@ public TransformMetadata build() {
     }
 
     /**
-     * @deprecated use {@link #transformMetadata(ClusterState, ProjectId)}
+     * @deprecated use {@link #transformMetadata(ProjectMetadata)}
      */
+    @FixForMultiProject
     @Deprecated(forRemoval = true)
     public static TransformMetadata getTransformMetadata(ClusterState state) {
         TransformMetadata TransformMetadata = (state == null) ? null : state.metadata().getSingleProjectCustom(TYPE);
@@ -223,25 +226,44 @@ public static TransformMetadata getTransformMetadata(ClusterState state) {
         return TransformMetadata;
     }
 
+    /**
+     * @deprecated use {@link #transformMetadata(ProjectMetadata)}
+     */
+    @FixForMultiProject
+    @Deprecated(forRemoval = true)
     public static TransformMetadata transformMetadata(@Nullable ClusterState state, @Nullable ProjectId projectId) {
         if (state == null || projectId == null) {
             return EMPTY_METADATA;
         }
         return transformMetadata(state.projectState(projectId));
     }
 
+    /**
+     * @deprecated use {@link #transformMetadata(ProjectMetadata)}
+     */
+    @FixForMultiProject
+    @Deprecated(forRemoval = true)
     public static TransformMetadata transformMetadata(@Nullable ProjectState projectState) {
         if (projectState == null) {
             return EMPTY_METADATA;
         }
-        TransformMetadata transformMetadata = projectState.metadata().custom(TYPE);
+        return transformMetadata(projectState.metadata());
+    }
+
+    public static TransformMetadata transformMetadata(ProjectMetadata project) {
+        TransformMetadata transformMetadata = project == null ? null : project.custom(TYPE);
         if (transformMetadata == null) {
             return EMPTY_METADATA;
         }
         return transformMetadata;
     }
 
+    @Deprecated(forRemoval = true)
     public static boolean upgradeMode(ClusterState state) {
         return getTransformMetadata(state).upgradeMode();
     }
+
+    public static boolean upgradeMode(ProjectMetadata project) {
+        return transformMetadata(project).upgradeMode();
+    }
 }

x-pack/plugin/migrate/src/internalClusterTest/java/org/elasticsearch/system_indices/action/AbstractFeatureMigrationIntegTest.java

@@ -20,10 +20,9 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.client.internal.Client;
-import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
-import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.IndexVersion;
@@ -57,7 +56,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
@@ -151,7 +150,11 @@ public abstract class AbstractFeatureMigrationIntegTest extends ESIntegTestCase
     protected String masterName;
 
     protected static ProjectMetadata assertMetadataAfterMigration(String featureName) {
-        ProjectMetadata finalMetadata = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT).get().getState().metadata().getProject();
+        ProjectMetadata finalMetadata = clusterAdmin().prepareState(TEST_REQUEST_TIMEOUT)
+            .get()
+            .getState()
+            .metadata()
+            .getProject(ProjectId.DEFAULT);
         // Check that the results metadata is what we expect.
         FeatureMigrationResults currentResults = finalMetadata.custom(FeatureMigrationResults.TYPE);
         assertThat(currentResults, notNullValue());
@@ -169,8 +172,8 @@ public void setup() {
         masterAndDataNode = internalCluster().startNode();
 
         TestPlugin testPlugin = getPlugin(TestPlugin.class);
-        testPlugin.preMigrationHook.set((state) -> Collections.emptyMap());
-        testPlugin.postMigrationHook.set((state, metadata) -> {});
+        testPlugin.preMigrationHook.set((project) -> Collections.emptyMap());
+        testPlugin.postMigrationHook.set((metadata) -> {});
     }
 
     protected <T extends Plugin> T getPlugin(Class<T> type) {
@@ -340,8 +343,8 @@ protected static TestPlugin.BlockingActionFilter blockAction(String actionTypeNa
     }
 
     public static class TestPlugin extends Plugin implements SystemIndexPlugin, ActionPlugin {
-        public final AtomicReference<Function<ClusterState, Map<String, Object>>> preMigrationHook = new AtomicReference<>();
-        public final AtomicReference<BiConsumer<ClusterState, Map<String, Object>>> postMigrationHook = new AtomicReference<>();
+        public final AtomicReference<Function<ProjectMetadata, Map<String, Object>>> preMigrationHook = new AtomicReference<>();
+        public final AtomicReference<Consumer<Map<String, Object>>> postMigrationHook = new AtomicReference<>();
         private final BlockingActionFilter blockingActionFilter;
 
         public TestPlugin() {
@@ -375,18 +378,13 @@ public Collection<AssociatedIndexDescriptor> getAssociatedIndexDescriptors() {
         }
 
         @Override
-        public void prepareForIndicesMigration(ClusterService clusterService, Client client, ActionListener<Map<String, Object>> listener) {
-            listener.onResponse(preMigrationHook.get().apply(clusterService.state()));
+        public void prepareForIndicesMigration(ProjectMetadata project, Client client, ActionListener<Map<String, Object>> listener) {
+            listener.onResponse(preMigrationHook.get().apply(project));
         }
 
         @Override
-        public void indicesMigrationComplete(
-            Map<String, Object> preUpgradeMetadata,
-            ClusterService clusterService,
-            Client client,
-            ActionListener<Boolean> listener
-        ) {
-            postMigrationHook.get().accept(clusterService.state(), preUpgradeMetadata);
+        public void indicesMigrationComplete(Map<String, Object> preUpgradeMetadata, Client client, ActionListener<Boolean> listener) {
+            postMigrationHook.get().accept(preUpgradeMetadata);
             listener.onResponse(true);
         }
 

x-pack/plugin/migrate/src/internalClusterTest/java/org/elasticsearch/system_indices/action/FeatureMigrationIT.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.cluster.metadata.Template;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -153,7 +154,7 @@ public void testMigrateSystemIndex() throws Exception {
 
         SetOnce<Boolean> preUpgradeHookCalled = new SetOnce<>();
         SetOnce<Boolean> postUpgradeHookCalled = new SetOnce<>();
-        getPlugin(TestPlugin.class).preMigrationHook.set(clusterState -> {
+        getPlugin(TestPlugin.class).preMigrationHook.set(project -> {
             // Check that the ordering of these calls is correct.
             assertThat(postUpgradeHookCalled.get(), nullValue());
             Map<String, Object> metadata = new HashMap<>();
@@ -170,7 +171,7 @@ public void testMigrateSystemIndex() throws Exception {
             return metadata;
         });
 
-        getPlugin(TestPlugin.class).postMigrationHook.set((clusterState, metadata) -> {
+        getPlugin(TestPlugin.class).postMigrationHook.set((metadata) -> {
             assertThat(preUpgradeHookCalled.get(), is(true));
 
             assertThat(metadata, hasEntry("stringKey", "stringValue"));
@@ -182,7 +183,8 @@ public void testMigrateSystemIndex() throws Exception {
             assertThat(innerMap, hasEntry("innerKey", "innerValue"));
 
             // We shouldn't have any results in the cluster state as no features have fully finished yet.
-            FeatureMigrationResults currentResults = clusterState.metadata().getProject().custom(FeatureMigrationResults.TYPE);
+            final var project = clusterService().state().metadata().getProject(ProjectId.DEFAULT);
+            FeatureMigrationResults currentResults = project.custom(FeatureMigrationResults.TYPE);
             assertThat(currentResults, nullValue());
             postUpgradeHookCalled.set(true);
         });