Merge branch 'main' into rtv0

jdconrad · web-flow · commit 04bb6c3bc0d5 · 2025-10-10T09:06:19.000-07:00
diff --git a/docs/changelog/136060.yaml b/docs/changelog/136060.yaml
@@ -0,0 +1,5 @@
+pr: 136060
+summary: Limit number of allocation explanations in `shards_availability` health indicator
+area: Health
+type: bug
+issues: []
diff --git a/server/src/main/java/org/elasticsearch/TransportVersions.java b/server/src/main/java/org/elasticsearch/TransportVersions.java
@@ -60,7 +60,6 @@ static TransportVersion def(int id) {
     public static final TransportVersion V_7_8_1 = def(7_08_01_99);
     public static final TransportVersion V_7_9_0 = def(7_09_00_99);
     public static final TransportVersion V_7_10_0 = def(7_10_00_99);
-    public static final TransportVersion V_8_0_0 = def(8_00_00_99);
     public static final TransportVersion V_8_8_0 = def(8_08_00_99);
     public static final TransportVersion V_8_8_1 = def(8_08_01_99);
     /*
diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java b/server/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java
@@ -9,7 +9,6 @@
 
 package org.elasticsearch.action.admin.indices.flush;
 
-import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.ActionFilters;
@@ -18,17 +17,11 @@
 import org.elasticsearch.cluster.action.shard.ShardStateAction;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.index.shard.IndexShard;
-import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.injection.guice.Inject;
-import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.transport.AbstractTransportRequest;
-import org.elasticsearch.transport.TransportChannel;
-import org.elasticsearch.transport.TransportRequestHandler;
 import org.elasticsearch.transport.TransportService;
 
 import java.io.IOException;
@@ -64,12 +57,6 @@ public TransportShardFlushAction(
             PrimaryActionExecution.RejectOnOverload,
             ReplicaActionExecution.SubjectToCircuitBreaker
         );
-        transportService.registerRequestHandler(
-            PRE_SYNCED_FLUSH_ACTION_NAME,
-            threadPool.executor(ThreadPool.Names.FLUSH),
-            PreShardSyncedFlushRequest::new,
-            new PreSyncedFlushTransportHandler(indicesService)
-        );
     }
 
     @Override
@@ -96,43 +83,4 @@ protected void shardOperationOnReplica(ShardFlushRequest request, IndexShard rep
             return new ReplicaResult();
         }));
     }
-
-    // TODO: Remove this transition in 9.0
-    private static final String PRE_SYNCED_FLUSH_ACTION_NAME = "internal:indices/flush/synced/pre";
-
-    private static class PreShardSyncedFlushRequest extends AbstractTransportRequest {
-        private final ShardId shardId;
-
-        private PreShardSyncedFlushRequest(StreamInput in) throws IOException {
-            super(in);
-            assert in.getTransportVersion().before(TransportVersions.V_8_0_0) : "received pre_sync request from a new node";
-            this.shardId = new ShardId(in);
-        }
-
-        @Override
-        public String toString() {
-            return "PreShardSyncedFlushRequest{" + "shardId=" + shardId + '}';
-        }
-
-        @Override
-        public void writeTo(StreamOutput out) throws IOException {
-            assert false : "must not send pre_sync request from a new node";
-            throw new UnsupportedOperationException("");
-        }
-    }
-
-    private static final class PreSyncedFlushTransportHandler implements TransportRequestHandler<PreShardSyncedFlushRequest> {
-        private final IndicesService indicesService;
-
-        PreSyncedFlushTransportHandler(IndicesService indicesService) {
-            this.indicesService = indicesService;
-        }
-
-        @Override
-        public void messageReceived(PreShardSyncedFlushRequest request, TransportChannel channel, Task task) {
-            IndexShard indexShard = indicesService.indexServiceSafe(request.shardId.getIndex()).getShard(request.shardId.id());
-            indexShard.flush(new FlushRequest().force(false).waitIfOngoing(true));
-            throw new UnsupportedOperationException("Synced flush was removed and a normal flush was performed instead.");
-        }
-    }
 }
diff --git a/server/src/main/java/org/elasticsearch/cluster/routing/allocation/shards/ShardsAvailabilityHealthIndicatorService.java b/server/src/main/java/org/elasticsearch/cluster/routing/allocation/shards/ShardsAvailabilityHealthIndicatorService.java
@@ -166,17 +166,18 @@ public String name() {
      * primary and replica availability, providing the color, diagnosis, and
      * messages about the available or unavailable shards in the cluster.
      * @param metadata Metadata for the cluster
+     * @param maxAffectedResourcesCount Max number of affect resources to return
      * @return A new ShardAllocationStatus that has not yet been filled.
      */
-    public ShardAllocationStatus createNewStatus(Metadata metadata) {
-        return new ShardAllocationStatus(metadata);
+    public ShardAllocationStatus createNewStatus(Metadata metadata, int maxAffectedResourcesCount) {
+        return new ShardAllocationStatus(metadata, maxAffectedResourcesCount);
     }
 
     @Override
     public HealthIndicatorResult calculate(boolean verbose, int maxAffectedResourcesCount, HealthInfo healthInfo) {
         var state = clusterService.state();
         var shutdown = state.getMetadata().custom(NodesShutdownMetadata.TYPE, NodesShutdownMetadata.EMPTY);
-        var status = createNewStatus(state.getMetadata());
+        var status = createNewStatus(state.getMetadata(), maxAffectedResourcesCount);
         updateShardAllocationStatus(status, state, shutdown, verbose, replicaUnassignedBufferTime);
         return createIndicator(
             status.getStatus(),
@@ -464,18 +465,33 @@ static void updateShardAllocationStatus(
         );
 
     public class ShardAllocationCounts {
-        int unassigned = 0;
-        int unassigned_new = 0;
-        int unassigned_restarting = 0;
-        int initializing = 0;
-        int started = 0;
-        int relocating = 0;
-        public final Set<ProjectIndexName> indicesWithUnavailableShards = new HashSet<>();
-        public final Set<ProjectIndexName> indicesWithAllShardsUnavailable = new HashSet<>();
+        final int maxAffectedResourcesCount;
+        int unassigned;
+        int unassigned_new;
+        int unassigned_restarting;
+        int initializing;
+        int started;
+        int relocating;
+        public final Set<ProjectIndexName> indicesWithUnavailableShards;
+        public final Set<ProjectIndexName> indicesWithAllShardsUnavailable;
         // We keep the searchable snapshots separately as long as the original index is still available
         // This is checked during the post-processing
-        public SearchableSnapshotsState searchableSnapshotsState = new SearchableSnapshotsState();
-        final Map<Diagnosis.Definition, Set<ProjectIndexName>> diagnosisDefinitions = new HashMap<>();
+        public SearchableSnapshotsState searchableSnapshotsState;
+        final Map<Diagnosis.Definition, Set<ProjectIndexName>> diagnosisDefinitions;
+
+        public ShardAllocationCounts(int maxAffectedResourcesCount) {
+            this.maxAffectedResourcesCount = maxAffectedResourcesCount;
+            unassigned = 0;
+            unassigned_new = 0;
+            unassigned_restarting = 0;
+            initializing = 0;
+            started = 0;
+            relocating = 0;
+            indicesWithUnavailableShards = new HashSet<>();
+            indicesWithAllShardsUnavailable = new HashSet<>();
+            searchableSnapshotsState = new SearchableSnapshotsState();
+            diagnosisDefinitions = new HashMap<>();
+        }
 
         public void increment(
             ProjectId projectId,
@@ -512,7 +528,15 @@ public void increment(
                         unassigned_restarting++;
                     } else {
                         unassigned++;
-                        if (verbose) {
+                        // Computing the diagnosis can be very expensive in large clusters, so we limit the number of
+                        // computations to the maxAffectedResourcesCount. The main negative side effect of this is that
+                        // we might miss some diagnoses. We are willing to take this risk, and users can always
+                        // use the allocation explain API for more details or increase the maxAffectedResourcesCount.
+                        // Since we have two ShardAllocationCounts instances (primaries and replicas), we technically
+                        // do 2 * maxAffectedResourcesCount computations, but the added complexity of accurately
+                        // limiting the number of calls doesn't outweigh the benefits, as the main goal is to limit
+                        // the number of computations to a constant rather than a number that grows with the cluster size.
+                        if (verbose && unassigned <= maxAffectedResourcesCount) {
                             diagnoseUnassignedShardRouting(routing, state).forEach(definition -> addDefinition(definition, projectIndex));
                         }
                     }
@@ -957,12 +981,16 @@ public Diagnosis.Definition getIncreaseNodeWithRoleCapacityAction(String role) {
     }
 
     public class ShardAllocationStatus {
-        protected final ShardAllocationCounts primaries = new ShardAllocationCounts();
-        protected final ShardAllocationCounts replicas = new ShardAllocationCounts();
+        protected final ShardAllocationCounts primaries;
+        protected final ShardAllocationCounts replicas;
         protected final Metadata clusterMetadata;
+        protected final int maxAffectedResourcesCount;
 
-        public ShardAllocationStatus(Metadata clusterMetadata) {
+        public ShardAllocationStatus(Metadata clusterMetadata, int maxAffectedResourcesCount) {
             this.clusterMetadata = clusterMetadata;
+            this.maxAffectedResourcesCount = maxAffectedResourcesCount;
+            primaries = new ShardAllocationCounts(maxAffectedResourcesCount);
+            replicas = new ShardAllocationCounts(maxAffectedResourcesCount);
         }
 
         void addPrimary(ProjectId projectId, ShardRouting routing, ClusterState state, NodesShutdownMetadata shutdowns, boolean verbose) {
diff --git a/server/src/test/java/org/elasticsearch/cluster/routing/allocation/shards/ShardsAvailabilityHealthIndicatorServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/routing/allocation/shards/ShardsAvailabilityHealthIndicatorServiceTests.java
@@ -361,7 +361,7 @@ public void testAllReplicasUnassigned() {
                 List.of()
             );
             var service = createShardsAvailabilityIndicatorService(projectId, clusterState);
-            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata());
+            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata(), randomNonNegativeInt());
             ShardsAvailabilityHealthIndicatorService.updateShardAllocationStatus(
                 status,
                 clusterState,
@@ -386,7 +386,7 @@ public void testAllReplicasUnassigned() {
                 List.of()
             );
             var service = createShardsAvailabilityIndicatorService(projectId, clusterState);
-            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata());
+            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata(), randomNonNegativeInt());
             ShardsAvailabilityHealthIndicatorService.updateShardAllocationStatus(
                 status,
                 clusterState,
@@ -411,7 +411,7 @@ public void testAllReplicasUnassigned() {
                 List.of()
             );
             var service = createShardsAvailabilityIndicatorService(projectId, clusterState);
-            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata());
+            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata(), randomNonNegativeInt());
             ShardsAvailabilityHealthIndicatorService.updateShardAllocationStatus(
                 status,
                 clusterState,
@@ -438,7 +438,7 @@ public void testAllReplicasUnassigned() {
             );
 
             var service = createShardsAvailabilityIndicatorService(projectId, clusterState);
-            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata());
+            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata(), randomNonNegativeInt());
             ShardsAvailabilityHealthIndicatorService.updateShardAllocationStatus(
                 status,
                 clusterState,
@@ -477,7 +477,7 @@ public void testAllReplicasUnassigned() {
                 List.of()
             );
             var service = createShardsAvailabilityIndicatorService(projectId, clusterState);
-            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata());
+            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata(), randomNonNegativeInt());
             ShardsAvailabilityHealthIndicatorService.updateShardAllocationStatus(
                 status,
                 clusterState,
@@ -508,7 +508,7 @@ public void testAllReplicasUnassigned() {
             ProjectId projectId = randomProjectIdOrDefault();
             var clusterState = createClusterStateWith(projectId, List.of(routingTable), List.of());
             var service = createShardsAvailabilityIndicatorService(projectId, clusterState);
-            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata());
+            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata(), randomNonNegativeInt());
             ShardsAvailabilityHealthIndicatorService.updateShardAllocationStatus(
                 status,
                 clusterState,
@@ -534,7 +534,7 @@ public void testAllReplicasUnassigned() {
                 List.of()
             );
             var service = createShardsAvailabilityIndicatorService(projectId, clusterState);
-            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata());
+            ShardAllocationStatus status = service.createNewStatus(clusterState.metadata(), randomNonNegativeInt());
             ShardsAvailabilityHealthIndicatorService.updateShardAllocationStatus(
                 status,
                 clusterState,
@@ -1791,32 +1791,31 @@ public void testLimitNumberOfAffectedResources() {
 
         {
             // assert the full result to check that details, impacts, and symptoms use the correct count of affected indices (5)
-            assertThat(
-                service.calculate(true, 2, HealthInfo.EMPTY_HEALTH_INFO),
-                equalTo(
-                    createExpectedResult(
-                        RED,
-                        "This cluster has 5 unavailable primary shards.",
-                        Map.of("unassigned_primaries", 5),
-                        List.of(
-                            new HealthIndicatorImpact(
-                                NAME,
-                                ShardsAvailabilityHealthIndicatorService.PRIMARY_UNASSIGNED_IMPACT_ID,
-                                1,
-                                "Cannot add data to 5 indices [red-index1, red-index2, red-index3, red-index4, red-index5]. Searches might "
-                                    + "return incomplete results.",
-                                List.of(ImpactArea.INGEST, ImpactArea.SEARCH)
-                            )
-                        ),
-                        List.of(
-                            new Diagnosis(
-                                ACTION_CHECK_ALLOCATION_EXPLAIN_API,
-                                List.of(new Diagnosis.Resource(INDEX, List.of("red-index1", "red-index2")))
-                            )
-                        )
+            // since we limit the number of allocation explanations while looping over the shards, we can't guarantee
+            // which indices end up in the affected resources list, but we can at least check that the size is correct
+            var calculatedResult = service.calculate(true, 2, HealthInfo.EMPTY_HEALTH_INFO);
+            assertEquals(RED, calculatedResult.status());
+            assertEquals("This cluster has 5 unavailable primary shards.", calculatedResult.symptom());
+            assertEquals(new SimpleHealthIndicatorDetails(addDefaults(Map.of("unassigned_primaries", 5))), calculatedResult.details());
+            assertEquals(
+                List.of(
+                    new HealthIndicatorImpact(
+                        NAME,
+                        ShardsAvailabilityHealthIndicatorService.PRIMARY_UNASSIGNED_IMPACT_ID,
+                        1,
+                        "Cannot add data to 5 indices [red-index1, red-index2, red-index3, red-index4, red-index5]. Searches might "
+                            + "return incomplete results.",
+                        List.of(ImpactArea.INGEST, ImpactArea.SEARCH)
                     )
-                )
+                ),
+                calculatedResult.impacts()
             );
+            assertEquals("Expected 1 diagnosis but got " + calculatedResult.diagnosisList(), 1, calculatedResult.diagnosisList().size());
+            var diagnosis = calculatedResult.diagnosisList().get(0);
+            assertEquals(ACTION_CHECK_ALLOCATION_EXPLAIN_API, diagnosis.definition());
+            assertEquals("Expected 1 affected resource but got " + diagnosis.affectedResources(), 1, diagnosis.affectedResources().size());
+            var affectedResource = diagnosis.affectedResources().get(0);
+            assertEquals("Expected 2 indices but got " + affectedResource.getValues(), 2, affectedResource.getValues().size());
         }
 
         {
@@ -1838,11 +1837,8 @@ public void testLimitNumberOfAffectedResources() {
         }
 
         {
-            // 0 affected resources
-            assertThat(
-                service.calculate(true, 0, HealthInfo.EMPTY_HEALTH_INFO).diagnosisList(),
-                equalTo(List.of(new Diagnosis(ACTION_CHECK_ALLOCATION_EXPLAIN_API, List.of(new Diagnosis.Resource(INDEX, List.of())))))
-            );
+            // 0 affected resources means we don't do any shard allocation explanation and thus do not report any diagnosis
+            assertThat(service.calculate(true, 0, HealthInfo.EMPTY_HEALTH_INFO).diagnosisList(), equalTo(List.of()));
         }
     }
 
