[Transform] Consolidate permissions checks (#106413)

* [Transform] Consolidate permissions checks

When we defer permissions checks, unattended Transforms will start and
fail immediately with errors related to the internal transform index.
The transform will progress beyond the 0th checkpoint, but the search
repeatedly fails for missing permissions.

Rather than searching and failing, we will reuse the initial permissions
check error, which includes the correct permission to set to get the
Transform working. The check will happen before the initial search, so
it will not progress beyond the 0th checkpoint.

Fix #105794

---------

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

docs/changelog/106413.yaml

@@ -0,0 +1,6 @@
+pr: 106413
+summary: Consolidate permissions checks
+area: Transform
+type: bug
+issues:
+ - 105794

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

@@ -51,6 +51,9 @@ public class TransformMessages {
         "Failed to parse transform statistics for transform [{0}]";
     public static final String FAILED_TO_LOAD_TRANSFORM_CHECKPOINT = "Failed to load transform checkpoint for transform [{0}]";
     public static final String FAILED_TO_LOAD_TRANSFORM_STATE = "Failed to load transform state for transform [{0}]";
+
+    public static final String TRANSFORM_CANNOT_START_WITHOUT_PERMISSIONS = "Cannot start transform [{0}] because user lacks required "
+        + "permissions, see privileges_check_failed issue for more details";
     public static final String TRANSFORM_CONFIGURATION_BAD_FUNCTION_COUNT = "Transform configuration must specify exactly 1 function";
     public static final String TRANSFORM_CONFIGURATION_PIVOT_NO_GROUP_BY = "Pivot transform configuration must specify at least 1 group_by";
     public static final String TRANSFORM_CONFIGURATION_PIVOT_NO_AGGREGATION =

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

@@ -420,12 +420,15 @@ public class TransformInsufficientPermissionsIT extends TransformRestTestCase {
 
         startTransform(transformId, RequestOptions.DEFAULT);
 
-        String destIndexIssue = Strings.format("Could not create destination index [%s] for transform [%s]", destIndexName, transformId);
+        var permissionIssues = Strings.format(
+            "org.elasticsearch.ElasticsearchSecurityException: Cannot start transform [%s] because user lacks required permissions, "
+                + "see privileges_check_failed issue for more details",
+            transformId
+        );
         // transform's auth state status is still RED due to:
         // - lacking permissions
-        // - and the inability to create destination index in the indexer (which is also a consequence of lacking permissions)
-        // wait for 10 seconds to give the transform indexer enough time to try creating destination index
-        assertBusy(() -> { assertRed(transformId, authIssue, destIndexIssue); });
+        // - and the inability to start the indexer (which is also a consequence of lacking permissions)
+        assertBusy(() -> { assertRed(transformId, authIssue, permissionIssues); });
 
         // update transform's credentials so that the transform has permission to access source/dest indices
         updateConfig(transformId, "{}", RequestOptions.DEFAULT.toBuilder().addHeader(AUTH_KEY, Users.SENIOR.header).build());
@@ -440,7 +443,6 @@ public class TransformInsufficientPermissionsIT extends TransformRestTestCase {
      * unattended              = true
      * pre-existing dest index = true
      */
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/105794")
     public void testTransformPermissionsDeferUnattendedDest() throws Exception {
         String transformId = "transform-permissions-defer-unattended-dest-exists";
         String sourceIndexName = transformId + "-index";
@@ -467,8 +469,15 @@ public class TransformInsufficientPermissionsIT extends TransformRestTestCase {
 
         startTransform(config.getId(), RequestOptions.DEFAULT);
 
-        // transform's auth state status is still RED, but the health status is GREEN (because dest index exists)
-        assertRed(transformId, authIssue);
+        var permissionIssues = Strings.format(
+            "org.elasticsearch.ElasticsearchSecurityException: Cannot start transform [%s] because user lacks required permissions, "
+                + "see privileges_check_failed issue for more details",
+            transformId
+        );
+        // transform's auth state status is still RED due to:
+        // - lacking permissions
+        // - and the inability to start the indexer (which is also a consequence of lacking permissions)
+        assertBusy(() -> { assertRed(transformId, authIssue, permissionIssues); });
 
         // update transform's credentials so that the transform has permission to access source/dest indices
         updateConfig(transformId, "{}", RequestOptions.DEFAULT.toBuilder().addHeader(AUTH_KEY, Users.SENIOR.header).build());

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformIndexer.java

@@ -11,6 +11,7 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ElasticsearchSecurityException;
 import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.index.IndexRequest;
@@ -21,6 +22,7 @@ import org.elasticsearch.common.logging.LoggerMessageFormat;
 import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
+import org.elasticsearch.health.HealthStatus;
 import org.elasticsearch.index.query.BoolQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
@@ -271,22 +273,25 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
             return;
         }
 
-        SetOnce<Map<String, String>> deducedDestIndexMappings = new SetOnce<>();
-
-        ActionListener<Void> finalListener = ActionListener.wrap(r -> {
-            try {
-                // if we haven't set the page size yet, if it is set we might have reduced it after running into an out of memory
-                if (context.getPageSize() == 0) {
-                    configurePageSize(getConfig().getSettings().getMaxPageSearchSize());
-                }
+        if (context.getAuthState() != null && HealthStatus.RED.equals(context.getAuthState().getStatus())) {
+            // AuthorizationState status is RED which means there was permission check error during PUT or _update.
+            listener.onFailure(
+                new ElasticsearchSecurityException(
+                    TransformMessages.getMessage(TransformMessages.TRANSFORM_CANNOT_START_WITHOUT_PERMISSIONS, getConfig().getId())
+                )
+            );
+            return;
+        }
 
-                runState = determineRunStateAtStart();
-                listener.onResponse(true);
-            } catch (Exception e) {
-                listener.onFailure(e);
-                return;
+        ActionListener<Void> finalListener = listener.delegateFailureAndWrap((l, r) -> {
+            // if we haven't set the page size yet, if it is set we might have reduced it after running into an out of memory
+            if (context.getPageSize() == 0) {
+                configurePageSize(getConfig().getSettings().getMaxPageSearchSize());
             }
-        }, listener::onFailure);
+
+            runState = determineRunStateAtStart();
+            l.onResponse(true);
+        });
 
         // On each run, we need to get the total number of docs and reset the count of processed docs
         // Since multiple checkpoints can be executed in the task while it is running on the same node, we need to gather
@@ -334,6 +339,7 @@ public abstract class TransformIndexer extends AsyncTwoPhaseIndexer<TransformInd
             }
         }, listener::onFailure);
 
+        var deducedDestIndexMappings = new SetOnce<Map<String, String>>();
         var shouldMaybeCreateDestIndexForUnattended = context.getCheckpoint() == 0
             && TransformEffectiveSettings.isUnattended(transformConfig.getSettings());