Limit max number of built-in roles sync attempts on unexpected failures (#120149)

slobodanadamovic · web-flow · commit ab15edd8a58b · 2025-01-17T07:03:51.000+11:00
Currently, we retry indefinitely on unexpected errors.  This PR adds a
limit to avoid flooding logs with potential unknown errors.

The counter of unexpected sync failures will be reset to 0  when a
successful sync occurs or when a master node is restarted.
diff --git a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/support/QueryableBuiltInRolesSynchronizer.java b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/support/QueryableBuiltInRolesSynchronizer.java
@@ -52,6 +52,7 @@
 import java.util.Set;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static java.util.stream.Collectors.toMap;
 import static java.util.stream.Collectors.toSet;
@@ -130,6 +131,16 @@ public void taskSucceeded(MarkRolesAsSyncedTask task, Map<String, String> value)
 
     private volatile boolean securityIndexDeleted = false;
 
+    /**
+     * The max consecutive failed sync attempts before skipping further sync attempts.
+     */
+    static final int MAX_FAILED_SYNC_ATTEMPTS = 10;
+
+    /**
+     * The counter of unexpected sync failures. Reset to 0 when a successful sync occurs or when a master node is restarted.
+     */
+    private final AtomicInteger failedSyncAttempts = new AtomicInteger(0);
+
     /**
      * Constructs a new built-in roles synchronizer.
      *
@@ -212,10 +223,12 @@ private void syncBuiltInRoles(final QueryableBuiltInRoles roles) {
                 final Map<String, String> indexedRolesDigests = readIndexedBuiltInRolesDigests(clusterService.state());
                 if (roles.rolesDigest().equals(indexedRolesDigests)) {
                     logger.debug("Security index already contains the latest built-in roles indexed, skipping roles synchronization");
+                    resetFailedSyncAttempts();
                     synchronizationInProgress.set(false);
                 } else {
                     executor.execute(() -> doSyncBuiltinRoles(indexedRolesDigests, roles, ActionListener.wrap(v -> {
-                        logger.info("Successfully synced [" + roles.roleDescriptors().size() + "] built-in roles to .security index");
+                        logger.info("Successfully synced [{}] built-in roles to .security index", roles.roleDescriptors().size());
+                        resetFailedSyncAttempts();
                         synchronizationInProgress.set(false);
                     }, e -> {
                         handleException(e);
@@ -224,12 +237,14 @@ private void syncBuiltInRoles(final QueryableBuiltInRoles roles) {
                 }
             } catch (Exception e) {
                 logger.error("Failed to sync built-in roles", e);
+                failedSyncAttempts.incrementAndGet();
                 synchronizationInProgress.set(false);
             }
         }
     }
 
-    private static void handleException(Exception e) {
+    private void handleException(Exception e) {
+        boolean isUnexpectedFailure = false;
         if (e instanceof BulkRolesResponseException bulkException) {
             final boolean isBulkDeleteFailure = bulkException instanceof BulkDeleteRolesResponseException;
             for (final Map.Entry<String, Exception> bulkFailure : bulkException.getFailures().entrySet()) {
@@ -241,14 +256,35 @@ private static void handleException(Exception e) {
                 if (isExpectedFailure(bulkFailure.getValue())) {
                     logger.info(logMessage, bulkFailure.getValue());
                 } else {
+                    isUnexpectedFailure = true;
                     logger.warn(logMessage, bulkFailure.getValue());
                 }
             }
         } else if (isExpectedFailure(e)) {
             logger.info("Failed to sync built-in roles to .security index", e);
         } else {
+            isUnexpectedFailure = true;
             logger.warn("Failed to sync built-in roles to .security index due to unexpected exception", e);
         }
+        if (isUnexpectedFailure) {
+            failedSyncAttempts.incrementAndGet();
+        }
+    }
+
+    private void resetFailedSyncAttempts() {
+        if (failedSyncAttempts.get() > 0) {
+            logger.trace("resetting failed sync attempts to 0");
+            failedSyncAttempts.set(0);
+        }
+    }
+
+    /**
+     * Package protected for testing purposes.
+     *
+     * @return the number of failed sync attempts
+     */
+    int getFailedSyncAttempts() {
+        return failedSyncAttempts.get();
     }
 
     /**
@@ -279,6 +315,13 @@ private boolean shouldSyncBuiltInRoles(final ClusterState state) {
             logger.trace("Local node is not the master, skipping built-in roles synchronization");
             return false;
         }
+        if (failedSyncAttempts.get() >= MAX_FAILED_SYNC_ATTEMPTS) {
+            logger.debug(
+                "Failed to sync built-in roles to .security index [{}] times. Skipping built-in roles synchronization.",
+                failedSyncAttempts.get()
+            );
+            return false;
+        }
         if (false == state.clusterRecovered()) {
             logger.trace("Cluster state has not recovered yet, skipping built-in roles synchronization");
             return false;
diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/support/QueryableBuiltInRolesSynchronizerTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/support/QueryableBuiltInRolesSynchronizerTests.java
@@ -9,6 +9,7 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.UnavailableShardsException;
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterName;
@@ -45,8 +46,11 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.gateway.GatewayService.STATE_NOT_RECOVERED_BLOCK;
 import static org.elasticsearch.xpack.security.support.QueryableBuiltInRolesSynchronizer.QUERYABLE_BUILT_IN_ROLES_FEATURE;
@@ -164,6 +168,7 @@ public void testSuccessfulSync() {
         verify(clusterService, times(3)).state();
         verifyNoMoreInteractions(nativeRolesStore, featureService, taskQueue, reservedRolesProvider, threadPool, clusterService);
         assertThat(synchronizer.isSynchronizationInProgress(), equalTo(false));
+        assertThat(synchronizer.getFailedSyncAttempts(), equalTo(0));
     }
 
     public void testNotMaster() {
@@ -384,6 +389,151 @@ public void testSecurityIndexClosed() {
         verifyNoMoreInteractions(nativeRolesStore, featureService, taskQueue, reservedRolesProvider, threadPool, clusterService);
     }
 
+    public void testUnexpectedSyncFailures() {
+        assertInitialState();
+
+        ClusterState clusterState = markShardsAvailable(createClusterStateWithOpenSecurityIndex()).nodes(localNodeMaster())
+            .blocks(emptyClusterBlocks())
+            .build();
+
+        when(clusterService.state()).thenReturn(clusterState);
+        when(featureService.clusterHasFeature(any(), eq(QUERYABLE_BUILT_IN_ROLES_FEATURE))).thenReturn(true);
+
+        final Set<String> roles = randomReservedRoles(randomIntBetween(1, 10));
+        final QueryableBuiltInRoles builtInRoles = buildQueryableBuiltInRoles(
+            roles.stream().map(ReservedRolesStore::roleDescriptor).collect(Collectors.toSet())
+        );
+        when(reservedRolesProvider.getRoles()).thenReturn(builtInRoles);
+        mockNativeRolesStoreWithFailure(builtInRoles.roleDescriptors(), Set.of(), new IllegalStateException("unexpected failure"));
+        assertThat(synchronizer.isSynchronizationInProgress(), equalTo(false));
+
+        for (int i = 1; i <= QueryableBuiltInRolesSynchronizer.MAX_FAILED_SYNC_ATTEMPTS + 5; i++) {
+            synchronizer.clusterChanged(event(clusterState));
+            if (i < QueryableBuiltInRolesSynchronizer.MAX_FAILED_SYNC_ATTEMPTS) {
+                assertThat(synchronizer.getFailedSyncAttempts(), equalTo(i));
+            } else {
+                assertThat(synchronizer.getFailedSyncAttempts(), equalTo(QueryableBuiltInRolesSynchronizer.MAX_FAILED_SYNC_ATTEMPTS));
+            }
+        }
+
+        verify(nativeRolesStore, times(QueryableBuiltInRolesSynchronizer.MAX_FAILED_SYNC_ATTEMPTS)).isEnabled();
+        verify(featureService, times(QueryableBuiltInRolesSynchronizer.MAX_FAILED_SYNC_ATTEMPTS)).clusterHasFeature(
+            any(),
+            eq(QUERYABLE_BUILT_IN_ROLES_FEATURE)
+        );
+        verify(reservedRolesProvider, times(QueryableBuiltInRolesSynchronizer.MAX_FAILED_SYNC_ATTEMPTS)).getRoles();
+        verify(nativeRolesStore, times(QueryableBuiltInRolesSynchronizer.MAX_FAILED_SYNC_ATTEMPTS)).putRoles(
+            eq(WriteRequest.RefreshPolicy.IMMEDIATE),
+            eq(builtInRoles.roleDescriptors()),
+            eq(false),
+            any()
+        );
+        verify(clusterService, times(QueryableBuiltInRolesSynchronizer.MAX_FAILED_SYNC_ATTEMPTS)).state();
+        verifyNoMoreInteractions(nativeRolesStore, featureService, taskQueue, reservedRolesProvider, threadPool, clusterService);
+        assertThat(synchronizer.isSynchronizationInProgress(), equalTo(false));
+    }
+
+    public void testFailedSyncAttemptsGetsResetAfterSuccessfulSync() {
+        assertInitialState();
+
+        ClusterState clusterState = markShardsAvailable(createClusterStateWithOpenSecurityIndex()).nodes(localNodeMaster())
+            .blocks(emptyClusterBlocks())
+            .build();
+
+        when(clusterService.state()).thenReturn(clusterState);
+        when(featureService.clusterHasFeature(any(), eq(QUERYABLE_BUILT_IN_ROLES_FEATURE))).thenReturn(true);
+
+        final Set<String> roles = randomReservedRoles(randomIntBetween(1, 10));
+        final QueryableBuiltInRoles builtInRoles = buildQueryableBuiltInRoles(
+            roles.stream().map(ReservedRolesStore::roleDescriptor).collect(Collectors.toSet())
+        );
+        when(reservedRolesProvider.getRoles()).thenReturn(builtInRoles);
+        mockNativeRolesStoreWithFailure(builtInRoles.roleDescriptors(), Set.of(), new IllegalStateException("unexpected failure"));
+        assertThat(synchronizer.isSynchronizationInProgress(), equalTo(false));
+
+        // assert failed sync attempts are counted
+        int numOfSimulatedFailures = randomIntBetween(1, QueryableBuiltInRolesSynchronizer.MAX_FAILED_SYNC_ATTEMPTS - 1);
+        for (int i = 0; i < numOfSimulatedFailures; i++) {
+            synchronizer.clusterChanged(event(clusterState));
+            assertThat(synchronizer.getFailedSyncAttempts(), equalTo(i + 1));
+        }
+        assertThat(synchronizer.getFailedSyncAttempts(), equalTo(numOfSimulatedFailures));
+
+        // assert successful sync resets the failed sync attempts
+        mockEnabledNativeStore(builtInRoles.roleDescriptors(), Set.of());
+        synchronizer.clusterChanged(event(clusterState));
+        assertThat(synchronizer.getFailedSyncAttempts(), equalTo(0));
+
+        verify(nativeRolesStore, times(numOfSimulatedFailures + 1)).isEnabled();
+        verify(featureService, times(numOfSimulatedFailures + 1)).clusterHasFeature(any(), eq(QUERYABLE_BUILT_IN_ROLES_FEATURE));
+        verify(reservedRolesProvider, times(numOfSimulatedFailures + 1)).getRoles();
+        verify(nativeRolesStore, times(numOfSimulatedFailures + 1)).putRoles(
+            eq(WriteRequest.RefreshPolicy.IMMEDIATE),
+            eq(builtInRoles.roleDescriptors()),
+            eq(false),
+            any()
+        );
+        verify(taskQueue, times(1)).submitTask(any(), argThat(task -> task.getNewRoleDigests().equals(builtInRoles.rolesDigest())), any());
+        verify(clusterService, times(numOfSimulatedFailures + 3)).state();
+        verifyNoMoreInteractions(nativeRolesStore, featureService, taskQueue, reservedRolesProvider, threadPool, clusterService);
+        assertThat(synchronizer.isSynchronizationInProgress(), equalTo(false));
+    }
+
+    public void testExpectedSyncFailuresAreNotCounted() {
+        assertInitialState();
+
+        ClusterState clusterState = markShardsAvailable(createClusterStateWithOpenSecurityIndex()).nodes(localNodeMaster())
+            .blocks(emptyClusterBlocks())
+            .build();
+
+        when(clusterService.state()).thenReturn(clusterState);
+        when(featureService.clusterHasFeature(any(), eq(QUERYABLE_BUILT_IN_ROLES_FEATURE))).thenReturn(true);
+
+        final Set<String> roles = randomReservedRoles(randomIntBetween(1, 10));
+        final QueryableBuiltInRoles builtInRoles = buildQueryableBuiltInRoles(
+            roles.stream().map(ReservedRolesStore::roleDescriptor).collect(Collectors.toSet())
+        );
+        when(reservedRolesProvider.getRoles()).thenReturn(builtInRoles);
+        mockNativeRolesStoreWithFailure(builtInRoles.roleDescriptors(), Set.of(), new UnavailableShardsException(null, "expected failure"));
+        assertThat(synchronizer.isSynchronizationInProgress(), equalTo(false));
+
+        synchronizer.clusterChanged(event(clusterState));
+
+        assertThat(synchronizer.getFailedSyncAttempts(), equalTo(0));
+
+        verify(nativeRolesStore, times(1)).isEnabled();
+        verify(featureService, times(1)).clusterHasFeature(any(), eq(QUERYABLE_BUILT_IN_ROLES_FEATURE));
+        verify(reservedRolesProvider, times(1)).getRoles();
+        verify(nativeRolesStore, times(1)).putRoles(
+            eq(WriteRequest.RefreshPolicy.IMMEDIATE),
+            eq(builtInRoles.roleDescriptors()),
+            eq(false),
+            any()
+        );
+        verify(clusterService, times(1)).state();
+        verifyNoMoreInteractions(nativeRolesStore, featureService, taskQueue, reservedRolesProvider, threadPool, clusterService);
+        assertThat(synchronizer.isSynchronizationInProgress(), equalTo(false));
+    }
+
+    private Set<String> randomReservedRoles(int count) {
+        assert count >= 0;
+        if (count == 0) {
+            return Set.of();
+        }
+        if (count == 1) {
+            return Set.of(ReservedRolesStore.SUPERUSER_ROLE_DESCRIPTOR.getName());
+        }
+
+        final Set<String> reservedRoles = new HashSet<>();
+        reservedRoles.add(ReservedRolesStore.SUPERUSER_ROLE_DESCRIPTOR.getName());
+        final Set<String> allReservedRolesExceptSuperuser = ReservedRolesStore.names()
+            .stream()
+            .filter(role -> false == role.equals(ReservedRolesStore.SUPERUSER_ROLE_DESCRIPTOR.getName()))
+            .collect(Collectors.toSet());
+        reservedRoles.addAll(randomUnique(() -> randomFrom(allReservedRolesExceptSuperuser), count - 1));
+        return Collections.unmodifiableSet(reservedRoles);
+    }
+
     private static ClusterState.Builder markShardsAvailable(ClusterState.Builder clusterStateBuilder) {
         final ClusterState cs = clusterStateBuilder.build();
         return ClusterState.builder(cs)
@@ -508,6 +658,31 @@ private void mockEnabledNativeStore(final Collection<RoleDescriptor> rolesToUpse
             .deleteRoles(eq(rolesToDelete), eq(WriteRequest.RefreshPolicy.IMMEDIATE), eq(false), any(ActionListener.class));
     }
 
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    private void mockNativeRolesStoreWithFailure(
+        final Collection<RoleDescriptor> rolesToUpsert,
+        final Collection<String> rolesToDelete,
+        Exception failure
+    ) {
+        when(nativeRolesStore.isEnabled()).thenReturn(true);
+        doAnswer(i -> {
+            assertThat(synchronizer.isSynchronizationInProgress(), equalTo(true));
+            ((ActionListener) i.getArgument(3)).onResponse(
+                new BulkRolesResponse(rolesToUpsert.stream().map(role -> BulkRolesResponse.Item.failure(role.getName(), failure)).toList())
+            );
+            return null;
+        }).when(nativeRolesStore)
+            .putRoles(eq(WriteRequest.RefreshPolicy.IMMEDIATE), eq(rolesToUpsert), eq(false), any(ActionListener.class));
+
+        doAnswer(i -> {
+            ((ActionListener) i.getArgument(3)).onResponse(
+                new BulkRolesResponse(rolesToDelete.stream().map(role -> BulkRolesResponse.Item.failure(role, failure)).toList())
+            );
+            return null;
+        }).when(nativeRolesStore)
+            .deleteRoles(eq(rolesToDelete), eq(WriteRequest.RefreshPolicy.IMMEDIATE), eq(false), any(ActionListener.class));
+    }
+
     private static ClusterState.Builder createClusterStateWithOpenSecurityIndex() {
         return SecurityIndexManagerTests.createClusterState(
             TestRestrictedIndices.INTERNAL_SECURITY_MAIN_INDEX_7,
