Revert #135052 #133630 (#135369)

* Revert "Move individual stats fields to AsyncRefresh (#135052)"

This reverts commit 2b0153b.

* Revert "Track shardStarted events for simulation in DesiredBalanceComputer (#133630)"

This reverts commit f248596.

* [CI] Update transport version definitions

* Revert "[CI] Update transport version definitions"

This reverts commit 90f38b0.

* Don't reset upper bounds (#135226)

Transport version upper bounds are currently set to their values from
upstream main whenever no new definition is detected. However, this
is like a partial merge of upstream, and produces broken state files.
This commit temporarily comments out resetting until a more robust
solution is built.

* Revert "Don't reset upper bounds (#135226)"

This reverts commit ddbac68.

---------

Co-authored-by: elasticsearchmachine <[email protected]>
Co-authored-by: Ryan Ernst <[email protected]>

Author: 3 people

docs/changelog/133630.yaml

@@ -20,15 +20,12 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.xcontent.ChunkedToXContent;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 
 import java.io.IOException;
-import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Map;
@@ -68,7 +65,6 @@ public class ClusterInfo implements ChunkedToXContent, Writeable {
     final Map<ShardId, Double> shardWriteLoads;
     // max heap size per node ID
     final Map<String, ByteSizeValue> maxHeapSizePerNode;
-    private final Map<ShardId, Set<String>> shardToNodeIds;
 
     protected ClusterInfo() {
         this(Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of());
@@ -110,7 +106,6 @@ public ClusterInfo(
         this.nodeUsageStatsForThreadPools = Map.copyOf(nodeUsageStatsForThreadPools);
         this.shardWriteLoads = Map.copyOf(shardWriteLoads);
         this.maxHeapSizePerNode = Map.copyOf(maxHeapSizePerNode);
-        this.shardToNodeIds = computeShardToNodeIds(dataPath);
     }
 
     public ClusterInfo(StreamInput in) throws IOException {
@@ -140,23 +135,6 @@ public ClusterInfo(StreamInput in) throws IOException {
         } else {
             this.maxHeapSizePerNode = Map.of();
         }
-        this.shardToNodeIds = computeShardToNodeIds(dataPath);
-    }
-
-    private static Map<ShardId, Set<String>> computeShardToNodeIds(Map<NodeAndShard, String> dataPath) {
-        if (dataPath.isEmpty()) {
-            return Map.of();
-        }
-        final var shardToNodeIds = new HashMap<ShardId, Set<String>>();
-        for (NodeAndShard nodeAndShard : dataPath.keySet()) {
-            shardToNodeIds.computeIfAbsent(nodeAndShard.shardId, ignore -> new HashSet<>()).add(nodeAndShard.nodeId);
-        }
-        Maps.transformValues(shardToNodeIds, nodeIds -> Collections.unmodifiableSet(nodeIds));
-        return shardToNodeIds;
-    }
-
-    public Set<String> getNodeIdsForShard(ShardId shardId) {
-        return shardToNodeIds.getOrDefault(shardId, Set.of());
     }
 
     @Override
@@ -365,16 +343,6 @@ public Map<String, ByteSizeValue> getMaxHeapSizePerNode() {
         return this.maxHeapSizePerNode;
     }
 
-    /**
-     * Return true if the shard has moved since the time ClusterInfo was created.
-     */
-    public boolean hasShardMoved(ShardRouting shardRouting) {
-        // We use dataPath to find out whether a shard is allocated on a node.
-        // TODO: DataPath is sent with disk usages but thread pool usage is sent separately so that local shard allocation
-        // may change between the two calls.
-        return getDataPath(shardRouting) == null;
-    }
-
     /**
      * Method that incorporates the ShardId for the shard into a string that
      * includes a 'p' or 'r' depending on whether the shard is a primary.

server/src/main/java/org/elasticsearch/cluster/ClusterInfo.java

@@ -9,16 +9,12 @@
 
 package org.elasticsearch.cluster;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.elasticsearch.cluster.ClusterInfo.NodeAndShard;
 import org.elasticsearch.cluster.routing.ShardMovementWriteLoadSimulator;
 import org.elasticsearch.cluster.routing.ShardRouting;
-import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.CopyOnFirstWriteMap;
-import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.shard.ShardId;
 
 import java.util.HashMap;
@@ -29,12 +25,9 @@
 import static org.elasticsearch.cluster.routing.ExpectedShardSizeEstimator.getExpectedShardSize;
 import static org.elasticsearch.cluster.routing.ExpectedShardSizeEstimator.shouldReserveSpaceForInitializingShard;
 import static org.elasticsearch.cluster.routing.ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE;
-import static org.elasticsearch.cluster.routing.UnassignedInfo.Reason.REINITIALIZED;
 
 public class ClusterInfoSimulator {
 
-    private static final Logger logger = LogManager.getLogger(ClusterInfoSimulator.class);
-
     private final RoutingAllocation allocation;
 
     private final Map<String, DiskUsage> leastAvailableSpaceUsage;
@@ -102,7 +95,7 @@ private static Map<String, DiskUsage> getAdjustedDiskSpace(RoutingAllocation all
      * Balance is later recalculated with a refreshed cluster info containing actual shards placement.
      */
     public void simulateShardStarted(ShardRouting shard) {
-        assert shard.initializing() : "expected an initializing shard, but got: " + shard;
+        assert shard.initializing();
 
         var project = allocation.metadata().projectFor(shard.index());
         var size = getExpectedShardSize(
@@ -129,36 +122,6 @@ public void simulateShardStarted(ShardRouting shard) {
         shardMovementWriteLoadSimulator.simulateShardStarted(shard);
     }
 
-    /**
-     * This method simulates starting an already started shard with an optional {@code sourceNodeId} in case of a relocation.
-     * @param startedShard The shard to simulate. Must be started already.
-     * @param sourceNodeId The source node ID if the shard started as a result of relocation. {@code null} otherwise.
-     */
-    public void simulateAlreadyStartedShard(ShardRouting startedShard, @Nullable String sourceNodeId) {
-        assert startedShard.started() : "expected an already started shard, but got: " + startedShard;
-        if (logger.isDebugEnabled()) {
-            logger.debug(
-                "simulated started shard {} on node [{}] as a {}",
-                startedShard.shardId(),
-                startedShard.currentNodeId(),
-                sourceNodeId != null ? "relocating shard from node [" + sourceNodeId + "]" : "new shard"
-            );
-        }
-        final long expectedShardSize = startedShard.getExpectedShardSize();
-        if (sourceNodeId != null) {
-            final var relocatingShard = startedShard.moveToUnassigned(new UnassignedInfo(REINITIALIZED, "simulation"))
-                .initialize(sourceNodeId, null, expectedShardSize)
-                .moveToStarted(expectedShardSize)
-                .relocate(startedShard.currentNodeId(), expectedShardSize)
-                .getTargetRelocatingShard();
-            simulateShardStarted(relocatingShard);
-        } else {
-            final var initializingShard = startedShard.moveToUnassigned(new UnassignedInfo(REINITIALIZED, "simulation"))
-                .initialize(startedShard.currentNodeId(), null, expectedShardSize);
-            simulateShardStarted(initializingShard);
-        }
-    }
-
     private void modifyDiskUsage(String nodeId, long freeDelta) {
         if (freeDelta == 0) {
             return;

server/src/main/java/org/elasticsearch/cluster/ClusterInfoSimulator.java

@@ -100,6 +100,13 @@ public class InternalClusterInfoService implements ClusterInfoService, ClusterSt
     private volatile TimeValue updateFrequency;
     private volatile TimeValue fetchTimeout;
 
+    private volatile Map<String, DiskUsage> leastAvailableSpaceUsages;
+    private volatile Map<String, DiskUsage> mostAvailableSpaceUsages;
+    private volatile Map<String, ByteSizeValue> maxHeapPerNode;
+    private volatile Map<String, Long> estimatedHeapUsagePerNode;
+    private volatile Map<String, NodeUsageStatsForThreadPools> nodeThreadPoolUsageStatsPerNode;
+    private volatile IndicesStatsSummary indicesStatsSummary;
+
     private final ThreadPool threadPool;
     private final Client client;
     private final Supplier<ClusterState> clusterStateSupplier;
@@ -113,8 +120,6 @@ public class InternalClusterInfoService implements ClusterInfoService, ClusterSt
     private AsyncRefresh currentRefresh;
     private RefreshScheduler refreshScheduler;
 
-    private volatile ClusterInfo currentClusterInfo = ClusterInfo.EMPTY;
-
     @SuppressWarnings("this-escape")
     public InternalClusterInfoService(
         Settings settings,
@@ -124,6 +129,12 @@ public InternalClusterInfoService(
         EstimatedHeapUsageCollector estimatedHeapUsageCollector,
         NodeUsageStatsForThreadPoolsCollector nodeUsageStatsForThreadPoolsCollector
     ) {
+        this.leastAvailableSpaceUsages = Map.of();
+        this.mostAvailableSpaceUsages = Map.of();
+        this.maxHeapPerNode = Map.of();
+        this.estimatedHeapUsagePerNode = Map.of();
+        this.nodeThreadPoolUsageStatsPerNode = Map.of();
+        this.indicesStatsSummary = IndicesStatsSummary.EMPTY;
         this.threadPool = threadPool;
         this.client = client;
         this.estimatedHeapUsageCollector = estimatedHeapUsageCollector;
@@ -197,13 +208,6 @@ public void clusterChanged(ClusterChangedEvent event) {
 
     private class AsyncRefresh {
 
-        private volatile Map<String, DiskUsage> leastAvailableSpaceUsages;
-        private volatile Map<String, DiskUsage> mostAvailableSpaceUsages;
-        private volatile Map<String, ByteSizeValue> maxHeapPerNode;
-        private volatile Map<String, Long> estimatedHeapUsagePerNode;
-        private volatile Map<String, NodeUsageStatsForThreadPools> nodeThreadPoolUsageStatsPerNode;
-        private volatile IndicesStatsSummary indicesStatsSummary;
-
         private final List<ActionListener<ClusterInfo>> thisRefreshListeners;
         private final RefCountingRunnable fetchRefs = new RefCountingRunnable(this::callListeners);
 
@@ -449,7 +453,7 @@ public void onFailure(Exception e) {
         private void callListeners() {
             try {
                 logger.trace("stats all received, computing cluster info and notifying listeners");
-                final ClusterInfo clusterInfo = updateAndGetCurrentClusterInfo();
+                final ClusterInfo clusterInfo = getClusterInfo();
                 boolean anyListeners = false;
                 for (final Consumer<ClusterInfo> listener : listeners) {
                     anyListeners = true;
@@ -469,32 +473,6 @@ private void callListeners() {
                 onRefreshComplete(this);
             }
         }
-
-        private ClusterInfo updateAndGetCurrentClusterInfo() {
-            final IndicesStatsSummary indicesStatsSummary = this.indicesStatsSummary; // single volatile read
-            final Map<String, EstimatedHeapUsage> estimatedHeapUsages = new HashMap<>();
-            final var currentMaxHeapPerNode = this.maxHeapPerNode; // Make sure we use a consistent view
-            currentMaxHeapPerNode.forEach((nodeId, maxHeapSize) -> {
-                final Long estimatedHeapUsage = estimatedHeapUsagePerNode.get(nodeId);
-                if (estimatedHeapUsage != null) {
-                    estimatedHeapUsages.put(nodeId, new EstimatedHeapUsage(nodeId, maxHeapSize.getBytes(), estimatedHeapUsage));
-                }
-            });
-            final var newClusterInfo = new ClusterInfo(
-                leastAvailableSpaceUsages,
-                mostAvailableSpaceUsages,
-                indicesStatsSummary.shardSizes,
-                indicesStatsSummary.shardDataSetSizes,
-                indicesStatsSummary.dataPath,
-                indicesStatsSummary.reservedSpace,
-                estimatedHeapUsages,
-                nodeThreadPoolUsageStatsPerNode,
-                indicesStatsSummary.shardWriteLoads(),
-                currentMaxHeapPerNode
-            );
-            currentClusterInfo = newClusterInfo;
-            return newClusterInfo;
-        }
     }
 
     private void onRefreshComplete(AsyncRefresh completedRefresh) {
@@ -559,7 +537,27 @@ private boolean shouldRefresh() {
 
     @Override
     public ClusterInfo getClusterInfo() {
-        return currentClusterInfo;
+        final IndicesStatsSummary indicesStatsSummary = this.indicesStatsSummary; // single volatile read
+        final Map<String, EstimatedHeapUsage> estimatedHeapUsages = new HashMap<>();
+        final var currentMaxHeapPerNode = this.maxHeapPerNode; // Make sure we use a consistent view
+        currentMaxHeapPerNode.forEach((nodeId, maxHeapSize) -> {
+            final Long estimatedHeapUsage = estimatedHeapUsagePerNode.get(nodeId);
+            if (estimatedHeapUsage != null) {
+                estimatedHeapUsages.put(nodeId, new EstimatedHeapUsage(nodeId, maxHeapSize.getBytes(), estimatedHeapUsage));
+            }
+        });
+        return new ClusterInfo(
+            leastAvailableSpaceUsages,
+            mostAvailableSpaceUsages,
+            indicesStatsSummary.shardSizes,
+            indicesStatsSummary.shardDataSetSizes,
+            indicesStatsSummary.dataPath,
+            indicesStatsSummary.reservedSpace,
+            estimatedHeapUsages,
+            nodeThreadPoolUsageStatsPerNode,
+            indicesStatsSummary.shardWriteLoads(),
+            currentMaxHeapPerNode
+        );
     }
 
     // allow tests to adjust the node stats on receipt

server/src/main/java/org/elasticsearch/cluster/InternalClusterInfoService.java

@@ -12,11 +12,7 @@
 import org.apache.logging.log4j.Level;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.cluster.ClusterInfo;
 import org.elasticsearch.cluster.ClusterInfoSimulator;
-import org.elasticsearch.cluster.node.DiscoveryNode;
-import org.elasticsearch.cluster.node.DiscoveryNodeRole;
-import org.elasticsearch.cluster.routing.RoutingNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.UnassignedInfo;
@@ -144,8 +140,6 @@ public DesiredBalance compute(
             return new DesiredBalance(desiredBalanceInput.index(), Map.of(), Map.of(), finishReason);
         }
 
-        maybeSimulateAlreadyStartedShards(desiredBalanceInput.routingAllocation().clusterInfo(), routingNodes, clusterInfoSimulator);
-
         // we assume that all ongoing recoveries will complete
         for (final var routingNode : routingNodes) {
             for (final var shardRouting : routingNode) {
@@ -489,73 +483,6 @@ public DesiredBalance compute(
         return new DesiredBalance(lastConvergedIndex, assignments, routingNodes.getBalanceWeightStatsPerNode(), finishReason);
     }
 
-    /**
-     * For shards started after initial polling of the ClusterInfo but before the next polling, we need to
-     * account for their impacts by simulating the events, either relocation or new shard start. This is done
-     * by comparing the current RoutingNodes against the shard allocation information from the ClusterInfo to
-     * find out the shard allocation changes. Note this approach is approximate in some edge cases:
-     * <ol>
-     * <li> If a shard is relocated twice from node A to B to C. It is considered as relocating from A to C directly
-     * for simulation purpose.</li>
-     * <li>If a shard has 2 replicas and they both relocate, replica 1 from A to X and replica 2 from B to Y. The
-     * simulation may see them as relocations A->X and B->Y. But it may also see them as A->Y and B->X. </li>
-     * </ol>
-     * In both cases, it should not really matter for simulation to account for resource changes.
-     */
-    static void maybeSimulateAlreadyStartedShards(
-        ClusterInfo clusterInfo,
-        RoutingNodes routingNodes,
-        ClusterInfoSimulator clusterInfoSimulator
-    ) {
-        // Find all shards that are started in RoutingNodes but have no data on corresponding node in ClusterInfo
-        final var startedShards = new ArrayList<ShardRouting>();
-        for (var routingNode : routingNodes) {
-            for (var shardRouting : routingNode.started()) {
-                if (clusterInfo.hasShardMoved(shardRouting)) {
-                    startedShards.add(shardRouting);
-                }
-            }
-        }
-        if (startedShards.isEmpty()) {
-            return;
-        }
-        logger.debug(
-            "Found [{}] started shards not accounted in ClusterInfo. The first one is {}",
-            startedShards.size(),
-            startedShards.getFirst()
-        );
-
-        // For started shards, attempt to find its source node. If found, it is a relocation, otherwise it is a new shard.
-        // The same shard on the same source node cannot be relocated twice to different nodes. So we exclude it once used.
-        final Map<ShardId, Set<String>> alreadySeenSourceNodes = new HashMap<>();
-        for (var startedShard : startedShards) {
-            // The source node is found by checking whether the ClusterInfo has a node hosting a shard with the same ShardId
-            // and has compatible node role. If multiple nodes are found, simply pick the first one.
-            final var sourceNodeId = clusterInfo.getNodeIdsForShard(startedShard.shardId())
-                .stream()
-                // Do not use the same source node twice for the same shard
-                .filter(nodeId -> alreadySeenSourceNodes.getOrDefault(startedShard.shardId(), Set.of()).contains(nodeId) == false)
-                .map(routingNodes::node)
-                // The source node must not currently host the shard
-                .filter(routingNode -> routingNode != null && routingNode.getByShardId(startedShard.shardId()) == null)
-                .map(RoutingNode::node)
-                // The source node must have compatible node roles
-                .filter(node -> node != null && switch (startedShard.role()) {
-                    case DEFAULT -> node.canContainData();
-                    case INDEX_ONLY -> node.getRoles().contains(DiscoveryNodeRole.INDEX_ROLE);
-                    case SEARCH_ONLY -> node.getRoles().contains(DiscoveryNodeRole.SEARCH_ROLE);
-                })
-                .map(DiscoveryNode::getId)
-                .findFirst()
-                .orElse(null);
-
-            if (sourceNodeId != null) {
-                alreadySeenSourceNodes.computeIfAbsent(startedShard.shardId(), k -> new HashSet<>()).add(sourceNodeId);
-            }
-            clusterInfoSimulator.simulateAlreadyStartedShard(startedShard, sourceNodeId);
-        }
-    }
-
     private void maybeLogAllocationExplainForUnassigned(
         DesiredBalance.ComputationFinishReason finishReason,
         RoutingNodes routingNodes,