Make migrate/10_reindex/* pass

Author: joegallo

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamTransportAction.java

@@ -15,7 +15,7 @@
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.metadata.DataStream;
-import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.injection.guice.Inject;
@@ -41,14 +41,16 @@ public class ReindexDataStreamTransportAction extends HandledTransportAction<Rei
     private final TransportService transportService;
     private final ClusterService clusterService;
     private final Client client;
+    private final ProjectResolver projectResolver;
 
     @Inject
     public ReindexDataStreamTransportAction(
         TransportService transportService,
         ActionFilters actionFilters,
         PersistentTasksService persistentTasksService,
         ClusterService clusterService,
-        Client client
+        Client client,
+        ProjectResolver projectResolver
     ) {
         super(
             ReindexDataStreamAction.NAME,
@@ -62,21 +64,22 @@ public ReindexDataStreamTransportAction(
         this.persistentTasksService = persistentTasksService;
         this.clusterService = clusterService;
         this.client = client;
+        this.projectResolver = projectResolver;
     }
 
     @Override
     protected void doExecute(Task task, ReindexDataStreamRequest request, ActionListener<AcknowledgedResponse> listener) {
         String sourceDataStreamName = request.getSourceDataStream();
-        Metadata metadata = clusterService.state().metadata();
-        DataStream dataStream = metadata.getProject().dataStreams().get(sourceDataStreamName);
+        final var projectMetadata = projectResolver.getProjectMetadata(clusterService.state());
+        DataStream dataStream = projectMetadata.dataStreams().get(sourceDataStreamName);
         if (dataStream == null) {
             listener.onFailure(new ResourceNotFoundException("Data stream named [{}] does not exist", sourceDataStreamName));
             return;
         }
         int totalIndices = dataStream.getIndices().size();
         int totalIndicesToBeUpgraded = (int) dataStream.getIndices()
             .stream()
-            .filter(getReindexRequiredPredicate(metadata.getProject(), false, dataStream.isSystem()))
+            .filter(getReindexRequiredPredicate(projectMetadata, false, dataStream.isSystem()))
             .count();
         ReindexDataStreamTaskParams params = new ReindexDataStreamTaskParams(
             sourceDataStreamName,
@@ -86,7 +89,7 @@ protected void doExecute(Task task, ReindexDataStreamRequest request, ActionList
             ClientHelper.getPersistableSafeSecurityHeaders(transportService.getThreadPool().getThreadContext(), clusterService.state())
         );
         String persistentTaskId = getPersistentTaskId(sourceDataStreamName);
-        final var persistentTask = PersistentTasksCustomMetadata.getTaskWithId(clusterService.state(), persistentTaskId);
+        final var persistentTask = PersistentTasksCustomMetadata.getTaskWithId(projectMetadata, persistentTaskId);
 
         if (persistentTask == null) {
             startTask(listener, persistentTaskId, params);

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/task/ReindexDataStreamPersistentTaskExecutor.java

@@ -28,6 +28,8 @@
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.DataStreamAction;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.project.ProjectResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
@@ -70,12 +72,14 @@ public class ReindexDataStreamPersistentTaskExecutor extends PersistentTasksExec
     private final Client client;
     private final ClusterService clusterService;
     private final ThreadPool threadPool;
+    private final ProjectResolver projectResolver;
 
     public ReindexDataStreamPersistentTaskExecutor(Client client, ClusterService clusterService, String taskName, ThreadPool threadPool) {
         super(taskName, threadPool.generic());
         this.client = client;
         this.clusterService = clusterService;
         this.threadPool = threadPool;
+        this.projectResolver = client.projectResolver();
     }
 
     @Override
@@ -87,8 +91,10 @@ protected ReindexDataStreamTask createTask(
         PersistentTasksCustomMetadata.PersistentTask<ReindexDataStreamTaskParams> taskInProgress,
         Map<String, String> headers
     ) {
+        ProjectId projectId = projectResolver.getProjectId();
         ReindexDataStreamTaskParams params = taskInProgress.getParams();
         return new ReindexDataStreamTask(
+            projectId,
             clusterService,
             params.startTime(),
             params.totalIndices(),
@@ -125,7 +131,7 @@ protected void nodeOperation(
             if (dataStreamInfos.size() == 1) {
                 DataStream dataStream = dataStreamInfos.getFirst().getDataStream();
                 boolean includeSystem = dataStream.isSystem();
-                if (getReindexRequiredPredicate(clusterService.state().metadata().getProject(), false, includeSystem).test(
+                if (getReindexRequiredPredicate(projectResolver.getProjectMetadata(clusterService.state()), false, includeSystem).test(
                     dataStream.getWriteIndex()
                 )) {
                     RolloverRequest rolloverRequest = new RolloverRequest(sourceDataStream, null);
@@ -174,7 +180,7 @@ private void reindexIndices(
     ) {
         List<Index> indices = dataStream.getIndices();
         List<Index> indicesToBeReindexed = indices.stream()
-            .filter(getReindexRequiredPredicate(clusterService.state().metadata().getProject(), false, dataStream.isSystem()))
+            .filter(getReindexRequiredPredicate(projectResolver.getProjectMetadata(clusterService.state()), false, dataStream.isSystem()))
             .toList();
         final ReindexDataStreamPersistentTaskState updatedState;
         if (params.totalIndices() != totalIndicesInDataStream
@@ -337,7 +343,7 @@ private TimeValue updateCompletionTimeAndGetTimeToLive(
         @Nullable ReindexDataStreamPersistentTaskState state
     ) {
         PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = PersistentTasksCustomMetadata.getTaskWithId(
-            clusterService.state(),
+            projectResolver.getProjectMetadata(clusterService.state()),
             reindexDataStreamTask.getPersistentTaskId()
         );
         if (persistentTask == null) {

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/task/ReindexDataStreamTask.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.migrate.task;
 
+import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.util.concurrent.RunOnce;
 import org.elasticsearch.core.TimeValue;
@@ -26,6 +27,7 @@
 
 public class ReindexDataStreamTask extends AllocatedPersistentTask {
     public static final String TASK_NAME = "reindex-data-stream";
+    private final ProjectId projectId;
     private final ClusterService clusterService;
     private final long persistentTaskStartTime;
     private final int initialTotalIndices;
@@ -39,6 +41,7 @@ public class ReindexDataStreamTask extends AllocatedPersistentTask {
 
     @SuppressWarnings("this-escape")
     public ReindexDataStreamTask(
+        ProjectId projectId,
         ClusterService clusterService,
         long persistentTaskStartTime,
         int initialTotalIndices,
@@ -51,6 +54,7 @@ public ReindexDataStreamTask(
         Map<String, String> headers
     ) {
         super(id, type, action, description, parentTask, headers);
+        this.projectId = projectId;
         this.clusterService = clusterService;
         this.persistentTaskStartTime = persistentTaskStartTime;
         this.initialTotalIndices = initialTotalIndices;
@@ -70,7 +74,7 @@ public ReindexDataStreamStatus getStatus() {
         int totalIndices = initialTotalIndices;
         int totalIndicesToBeUpgraded = initialTotalIndicesToBeUpgraded;
         PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = PersistentTasksCustomMetadata.getTaskWithId(
-            clusterService.state(),
+            clusterService.state().projectState(projectId).metadata(),
             getPersistentTaskId()
         );
         boolean isComplete;
@@ -130,7 +134,7 @@ public void incrementInProgressIndicesCount(String index) {
 
     private boolean isCompleteInClusterState() {
         PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = PersistentTasksCustomMetadata.getTaskWithId(
-            clusterService.state(),
+            clusterService.state().projectState(projectId).metadata(),
             getPersistentTaskId()
         );
         if (persistentTask != null) {

x-pack/qa/multi-project/xpack-rest-tests-with-multiple-projects/build.gradle

@@ -49,7 +49,6 @@ tasks.named("yamlRestTest").configure {
     '^ilm/60_operation_mode/*',
     '^ilm/80_health/*',
     '^logsdb/10_usage/*',
-    '^migrate/10_reindex/*',
     '^migrate/20_reindex_status/*',
     '^migrate/30_create_from/*',
     '^migration/10_get_feature_upgrade_status/*',