[8.19] Improve security migration resilience by handling version conflicts (#137558)

docs/changelog/137558.yaml

@@ -0,0 +1,5 @@
+pr: 137558
+summary: Improve security migration resilience by handling version conflicts
+area: Security
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/index/IndexVersions.java

@@ -137,6 +137,7 @@ private static IndexVersion def(int id, Version luceneVersion) {
     public static final IndexVersion UPGRADE_TO_LUCENE_9_12_2 = def(8_534_0_00, Version.LUCENE_9_12_2);
     public static final IndexVersion SPARSE_VECTOR_PRUNING_INDEX_OPTIONS_SUPPORT_BACKPORT_8_X = def(8_535_0_00, Version.LUCENE_9_12_2);
     public static final IndexVersion MATCH_ONLY_TEXT_STORED_AS_BYTES_BACKPORT_8_X = def(8_536_0_00, Version.LUCENE_9_12_2);
+    public static final IndexVersion SECURITY_MIGRATIONS_METADATA_FLATTENED_UPDATE = def(8_537_0_00, Version.LUCENE_9_12_2);
 
     /*
      * STOP! READ THIS FIRST! No, really,

x-pack/plugin/security/build.gradle

@@ -42,6 +42,9 @@ dependencies {
   internalClusterTestImplementation(testArtifact(project(xpackModule('core'))))
   api "com.unboundid:unboundid-ldapsdk:${versions.ldapsdk}"
 
+  internalClusterTestImplementation project(path: ':modules:lang-painless')
+  internalClusterTestImplementation project(path: ':modules:lang-painless:spi')
+
   // the following are all SAML dependencies - might as well download the whole internet
   api "org.opensaml:opensaml-core:${versions.opensaml}"
   api "org.opensaml:opensaml-saml-api:${versions.opensaml}"

x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/xpack/security/support/CleanupRoleMappingDuplicatesMigrationIT.java

@@ -236,8 +236,6 @@ public void testMigrationFallbackNamePreCondition() throws Exception {
         waitForMigrationCompletion(SecurityMigrations.CLEANUP_ROLE_MAPPING_DUPLICATES_MIGRATION_VERSION);
         // First migration is on a new index, so should skip all migrations. If we reset, it should re-trigger and run all migrations
         resetMigration();
-        // Wait for the first migration to finish
-        waitForMigrationCompletion(SecurityMigrations.CLEANUP_ROLE_MAPPING_DUPLICATES_MIGRATION_VERSION - 1);
 
         // Make sure migration didn't run yet (blocked by the fallback name)
         assertMigrationLessThan(SecurityMigrations.CLEANUP_ROLE_MAPPING_DUPLICATES_MIGRATION_VERSION);
@@ -307,10 +305,7 @@ public void testNewIndexSkipMigration() {
         ensureGreen();
         deleteSecurityIndex(); // hack to force a new security index to be created
         ensureGreen();
-        CountDownLatch awaitMigrations = awaitMigrationVersionUpdates(
-            masterNode,
-            SecurityMigrations.CLEANUP_ROLE_MAPPING_DUPLICATES_MIGRATION_VERSION
-        );
+        CountDownLatch awaitMigrations = awaitMigrationVersionUpdates(masterNode, SecurityMigrations.MIGRATIONS_BY_VERSION.lastKey());
         // Create a native role mapping to create security index and trigger migration
         createNativeRoleMapping("everyone_kibana_alone");
         // Make sure no migration ran (set to current version without applying prior migrations)

x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/xpack/security/support/MetadataFlattenedMigrationIntegTests.java

@@ -0,0 +1,190 @@
+/*
+ * 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.security.support;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.painless.PainlessPlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.SecurityIntegTestCase;
+import org.elasticsearch.xcontent.ToXContent;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.core.security.action.UpdateIndexMigrationVersionAction;
+import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
+import org.elasticsearch.xpack.security.authz.store.NativeRolesStore;
+import org.junit.Before;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
+import static org.elasticsearch.xpack.core.security.action.UpdateIndexMigrationVersionAction.MIGRATION_VERSION_CUSTOM_DATA_KEY;
+import static org.elasticsearch.xpack.core.security.action.UpdateIndexMigrationVersionAction.MIGRATION_VERSION_CUSTOM_KEY;
+import static org.elasticsearch.xpack.core.security.authz.RoleDescriptor.ROLE_TYPE;
+import static org.elasticsearch.xpack.core.security.test.TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_7;
+import static org.elasticsearch.xpack.security.support.SecurityMigrations.ROLE_METADATA_FLATTENED_MIGRATION_VERSION;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0, autoManageMasterNodes = false)
+public class MetadataFlattenedMigrationIntegTests extends SecurityIntegTestCase {
+
+    private final AtomicLong versionCounter = new AtomicLong(1);
+
+    @Before
+    public void resetVersion() {
+        versionCounter.set(1);
+    }
+
+    public void testMigrationWithConcurrentUpdates() throws Exception {
+        internalCluster().setBootstrapMasterNodeIndex(0);
+        internalCluster().startNode();
+        ensureGreen();
+
+        waitForMigrationCompletion();
+        var roles = createRoles();
+        final var nativeRoleStore = internalCluster().getInstance(NativeRolesStore.class);
+
+        final ExecutorService executor = Executors.newSingleThreadExecutor();
+        final AtomicBoolean runUpdateRolesBackground = new AtomicBoolean(true);
+        try {
+            executor.submit(() -> {
+                while (runUpdateRolesBackground.get()) {
+                    // Only update half the list so the other half can be verified as migrated
+                    RoleDescriptor roleToUpdate = randomFrom(roles.subList(0, roles.size() / 2));
+
+                    RoleDescriptor updatedRole = new RoleDescriptor(
+                        roleToUpdate.getName(),
+                        new String[] { "monitor" },
+                        null,
+                        null,
+                        null,
+                        null,
+                        Map.of("test", "value", "timestamp", System.currentTimeMillis(), "random", randomAlphaOfLength(10)),
+                        null
+                    );
+                    nativeRoleStore.putRole(
+                        WriteRequest.RefreshPolicy.IMMEDIATE,
+                        updatedRole,
+                        ActionListener.wrap(resp -> {}, ESTestCase::fail)
+                    );
+                    try {
+                        Thread.sleep(10);
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+                }
+            });
+
+            resetMigration();
+            waitForMigrationCompletion();
+        } finally {
+            runUpdateRolesBackground.set(false);
+            executor.shutdown();
+        }
+        assertAllRolesHaveMetadataFlattened();
+    }
+
+    private void resetMigration() {
+        client().execute(
+            UpdateIndexMigrationVersionAction.INSTANCE,
+            new UpdateIndexMigrationVersionAction.Request(
+                TimeValue.MAX_VALUE,
+                ROLE_METADATA_FLATTENED_MIGRATION_VERSION - 1,
+                INTERNAL_SECURITY_MAIN_INDEX_7
+            )
+        ).actionGet();
+    }
+
+    private List<RoleDescriptor> createRoles() throws IOException {
+        var roles = randomList(
+            25,
+            50,
+            () -> new RoleDescriptor(
+                randomAlphaOfLength(20),
+                null,
+                null,
+                null,
+                null,
+                null,
+                Map.of("test", "value", "timestamp", System.currentTimeMillis(), "random", randomAlphaOfLength(10)),
+                Map.of()
+            )
+        );
+        for (RoleDescriptor role : roles) {
+            indexRoleDirectly(role);
+        }
+        indicesAdmin().prepareRefresh(INTERNAL_SECURITY_MAIN_INDEX_7).get();
+        return roles;
+    }
+
+    private void indexRoleDirectly(RoleDescriptor role) throws IOException {
+        XContentBuilder builder = buildRoleDocument(role);
+        prepareIndex(INTERNAL_SECURITY_MAIN_INDEX_7).setId(ROLE_TYPE + "-" + role.getName())
+            .setSource(builder)
+            .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+            .get();
+    }
+
+    private XContentBuilder buildRoleDocument(RoleDescriptor role) throws IOException {
+        XContentBuilder builder = jsonBuilder().startObject();
+        // metadata_flattened is populated by the native role store, so write directly to index to simulate pre-migration state
+        role.innerToXContent(builder, ToXContent.EMPTY_PARAMS, true);
+        builder.endObject();
+        return builder;
+    }
+
+    private int getCurrentMigrationVersion() {
+        ClusterService clusterService = internalCluster().getInstance(ClusterService.class);
+        IndexMetadata indexMetadata = clusterService.state().metadata().index(INTERNAL_SECURITY_MAIN_INDEX_7);
+        if (indexMetadata == null || indexMetadata.getCustomData(MIGRATION_VERSION_CUSTOM_KEY) == null) {
+            return 0;
+        }
+        return Integer.parseInt(indexMetadata.getCustomData(MIGRATION_VERSION_CUSTOM_KEY).get(MIGRATION_VERSION_CUSTOM_DATA_KEY));
+    }
+
+    private void waitForMigrationCompletion() throws Exception {
+        assertBusy(() -> assertThat(getCurrentMigrationVersion(), greaterThanOrEqualTo(ROLE_METADATA_FLATTENED_MIGRATION_VERSION)));
+    }
+
+    private void assertAllRolesHaveMetadataFlattened() {
+        SearchRequest searchRequest = new SearchRequest(INTERNAL_SECURITY_MAIN_INDEX_7);
+        searchRequest.source().query(QueryBuilders.termQuery("type", "role")).size(1000);
+        SearchResponse response = client().search(searchRequest).actionGet();
+        for (SearchHit hit : response.getHits().getHits()) {
+            @SuppressWarnings("unchecked")
+            Map<String, Object> metadata = (Map<String, Object>) hit.getSourceAsMap().get("metadata_flattened");
+            // Only check non-reserved roles
+            if (metadata.get("_reserved") == null) {
+                assertEquals("value", metadata.get("test"));
+            }
+        }
+        response.decRef();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Stream.concat(super.nodePlugins().stream(), Stream.of(PainlessPlugin.class)).collect(Collectors.toList());
+    }
+}

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/Security.java

@@ -482,7 +482,7 @@ public class Security extends Plugin
 
     public static final String SECURITY_CRYPTO_THREAD_POOL_NAME = XPackField.SECURITY + "-crypto";
 
-    private static final int MAX_SECURITY_MIGRATION_RETRY_COUNT = 10;
+    private static final int MAX_SECURITY_MIGRATION_RETRY_COUNT = 1000;
 
     // TODO: ip filtering does not actually track license usage yet
     public static final LicensedFeature.Momentary IP_FILTERING_FEATURE = LicensedFeature.momentaryLenient(

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/support/SecurityIndexManager.java

@@ -44,10 +44,12 @@
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.indices.IndexClosedException;
 import org.elasticsearch.indices.SystemIndexDescriptor;
+import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.threadpool.Scheduler;
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xpack.core.security.authz.RoleMappingMetadata;
+import org.elasticsearch.xpack.core.security.support.SecurityMigrationTaskParams;
 import org.elasticsearch.xpack.security.SecurityFeatures;
 import org.elasticsearch.xpack.security.action.rolemapping.ReservedRoleMappingAction;
 
@@ -341,6 +343,9 @@ public void clusterChanged(ClusterChangedEvent event) {
             event.state(),
             migrationsVersion
         );
+        var persistentTaskCustomMetadata = PersistentTasksCustomMetadata.getPersistentTasksCustomMetadata(event.state());
+        final boolean securityMigrationRunning = persistentTaskCustomMetadata != null
+            && persistentTaskCustomMetadata.getTask(SecurityMigrationTaskParams.TASK_NAME) != null;
         final boolean mappingIsUpToDate = indexMetadata == null || checkIndexMappingUpToDate(event.state());
         final SystemIndexDescriptor.MappingsVersion minClusterMappingVersion = getMinSecurityIndexMappingVersion(event.state());
         final int indexMappingVersion = loadIndexMappingVersion(systemIndexDescriptor.getAliasName(), event.state());
@@ -370,6 +375,7 @@ public void clusterChanged(ClusterChangedEvent event) {
             indexAvailableForWrite,
             mappingIsUpToDate,
             createdOnLatestVersion,
+            securityMigrationRunning,
             roleMappingsCleanupMigrationStatus,
             migrationsVersion,
             minClusterMappingVersion,
@@ -774,6 +780,7 @@ public static class State {
             false,
             false,
             false,
+            false,
             null,
             null,
             null,
@@ -790,6 +797,7 @@ public static class State {
         public final boolean indexAvailableForWrite;
         public final boolean mappingUpToDate;
         public final boolean createdOnLatestVersion;
+        public final boolean securityMigrationRunning;
         public final RoleMappingsCleanupMigrationStatus roleMappingsCleanupMigrationStatus;
         public final Integer migrationsVersion;
         // Min mapping version supported by the descriptors in the cluster
@@ -809,6 +817,7 @@ public State(
             boolean indexAvailableForWrite,
             boolean mappingUpToDate,
             boolean createdOnLatestVersion,
+            boolean securityMigrationRunning,
             RoleMappingsCleanupMigrationStatus roleMappingsCleanupMigrationStatus,
             Integer migrationsVersion,
             SystemIndexDescriptor.MappingsVersion minClusterMappingVersion,
@@ -826,6 +835,7 @@ public State(
             this.mappingUpToDate = mappingUpToDate;
             this.migrationsVersion = migrationsVersion;
             this.createdOnLatestVersion = createdOnLatestVersion;
+            this.securityMigrationRunning = securityMigrationRunning;
             this.roleMappingsCleanupMigrationStatus = roleMappingsCleanupMigrationStatus;
             this.minClusterMappingVersion = minClusterMappingVersion;
             this.indexMappingVersion = indexMappingVersion;
@@ -847,6 +857,7 @@ public boolean equals(Object o) {
                 && indexAvailableForWrite == state.indexAvailableForWrite
                 && mappingUpToDate == state.mappingUpToDate
                 && createdOnLatestVersion == state.createdOnLatestVersion
+                && securityMigrationRunning == state.securityMigrationRunning
                 && roleMappingsCleanupMigrationStatus == state.roleMappingsCleanupMigrationStatus
                 && Objects.equals(indexMappingVersion, state.indexMappingVersion)
                 && Objects.equals(migrationsVersion, state.migrationsVersion)
@@ -870,6 +881,7 @@ public int hashCode() {
                 indexAvailableForWrite,
                 mappingUpToDate,
                 createdOnLatestVersion,
+                securityMigrationRunning,
                 roleMappingsCleanupMigrationStatus,
                 migrationsVersion,
                 minClusterMappingVersion,
@@ -895,6 +907,8 @@ public String toString() {
                 + mappingUpToDate
                 + ", createdOnLatestVersion="
                 + createdOnLatestVersion
+                + ", securityMigrationRunning="
+                + securityMigrationRunning
                 + ", roleMappingsCleanupMigrationStatus="
                 + roleMappingsCleanupMigrationStatus
                 + ", migrationsVersion="

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/support/SecurityMigrationExecutor.java

@@ -47,7 +47,7 @@ public SecurityMigrationExecutor(
     @Override
     protected void nodeOperation(AllocatedPersistentTask task, SecurityMigrationTaskParams params, PersistentTaskState state) {
         ActionListener<Void> listener = ActionListener.wrap((res) -> task.markAsCompleted(), (exception) -> {
-            logger.warn("Security migration failed: " + exception);
+            logger.warn("Security migration failed", exception);
             task.markAsFailed(exception);
         });