review feedback

Author: parkertimmins

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/user/InternalUsers.java

@@ -25,6 +25,7 @@
 import org.elasticsearch.action.datastreams.ModifyDataStreamsAction;
 import org.elasticsearch.action.downsample.DownsampleAction;
 import org.elasticsearch.action.index.TransportIndexAction;
+import org.elasticsearch.action.ingest.PutPipelineTransportAction;
 import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.search.TransportSearchScrollAction;
 import org.elasticsearch.index.reindex.ReindexAction;
@@ -196,7 +197,9 @@ public class InternalUsers {
         UsernamesField.REINDEX_DATA_STREAM_NAME,
         new RoleDescriptor(
             UsernamesField.REINDEX_DATA_STREAM_ROLE,
-            new String[] {},
+            new String[] {
+                PutPipelineTransportAction.TYPE.name()
+            },
             new RoleDescriptor.IndicesPrivileges[] {
                 RoleDescriptor.IndicesPrivileges.builder()
                     .indices("*")
@@ -225,6 +228,7 @@ public class InternalUsers {
                     .build() },
             null,
             null,
+
             new String[] {},
             MetadataUtils.DEFAULT_RESERVED_METADATA,
             Map.of()

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/user/InternalUsersTests.java

@@ -31,6 +31,7 @@
 import org.elasticsearch.action.downsample.DownsampleAction;
 import org.elasticsearch.action.get.TransportGetAction;
 import org.elasticsearch.action.index.TransportIndexAction;
+import org.elasticsearch.action.ingest.PutPipelineTransportAction;
 import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.search.TransportSearchScrollAction;
 import org.elasticsearch.cluster.metadata.DataStream;
@@ -302,7 +303,8 @@ public void testReindexDataStreamUser() {
 
         final SimpleRole role = getLocalClusterRole(InternalUsers.REINDEX_DATA_STREAM_USER);
 
-        assertThat(role.cluster(), is(ClusterPermission.NONE));
+        assertThat(role.cluster().privileges(), hasSize(1));
+        checkClusterAccess(InternalUsers.REINDEX_DATA_STREAM_USER, role, PutPipelineTransportAction.TYPE.name(), true);
         assertThat(role.runAs(), is(RunAsPermission.NONE));
         assertThat(role.application(), is(ApplicationPermission.NONE));
         assertThat(role.remoteIndices(), is(RemoteIndicesPermission.NONE));

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

@@ -17,6 +17,7 @@
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.ingest.IngestMetadata;
 import org.elasticsearch.ingest.PipelineConfiguration;
+import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentType;
 
@@ -61,20 +62,19 @@ public static boolean exists(ClusterState clusterState) {
      * Creates a pipeline with the current version's pipeline definition
      * @param client Client used to execute put pipeline
      * @param listener Callback used after pipeline has been created
+     * @param parentTaskId parent task id so that request can be cancelled
      */
-    public static void create(Client client, ActionListener<AcknowledgedResponse> listener) {
+    public static void create(Client client, ActionListener<AcknowledgedResponse> listener, TaskId parentTaskId) {
         final BytesReference pipeline = BytesReference.bytes(currentPipelineDefinition());
-        client.execute(
-            PutPipelineTransportAction.TYPE,
-            new PutPipelineRequest(
-                MasterNodeRequest.INFINITE_MASTER_NODE_TIMEOUT,
-                MasterNodeRequest.INFINITE_MASTER_NODE_TIMEOUT,
-                PIPELINE_NAME,
-                pipeline,
-                XContentType.JSON
-            ),
-            listener
+        var putPipelineRequest = new PutPipelineRequest(
+            MasterNodeRequest.INFINITE_MASTER_NODE_TIMEOUT,
+            MasterNodeRequest.INFINITE_MASTER_NODE_TIMEOUT,
+            PIPELINE_NAME,
+            pipeline,
+            XContentType.JSON
         );
+        putPipelineRequest.setParentTask(parentTaskId);
+        client.execute(PutPipelineTransportAction.TYPE, putPipelineRequest, listener);
     }
 
     private static XContentBuilder currentPipelineDefinition() {

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

@@ -154,7 +154,7 @@ protected void doExecute(
             return;
         }
         final boolean wasClosed = isClosed(sourceIndex);
-        SubscribableListener.newForked(this::prepareReindexOperation)
+        SubscribableListener.<AcknowledgedResponse>newForked(l -> prepareReindexOperation(l, taskId))
             .<AcknowledgedResponse>andThen(l -> setBlockWrites(sourceIndexName, l, taskId))
             .<OpenIndexResponse>andThen(l -> openIndexIfClosed(sourceIndexName, wasClosed, l, taskId))
             .<BroadcastResponse>andThen(l -> refresh(sourceIndexName, l, taskId))
@@ -199,11 +199,11 @@ private static boolean isClosed(IndexMetadata indexMetadata) {
         return indexMetadata.getState().equals(IndexMetadata.State.CLOSE);
     }
 
-    private void prepareReindexOperation(ActionListener<AcknowledgedResponse> listener) {
+    private void prepareReindexOperation(ActionListener<AcknowledgedResponse> listener, TaskId parentTaskId) {
         if (ReindexDataStreamPipeline.exists(clusterService.state())) {
             listener.onResponse(null);
         } else {
-            ReindexDataStreamPipeline.create(client, listener);
+            ReindexDataStreamPipeline.create(client, listener, parentTaskId);
         }
     }