Allow allocating clones over low watermark (#137399)

A new shard which is the result of a clone operation starts out on the
same node as the shard being cloned and takes no additional space due to
hard-linking, so we can safely allocate it even if the node is over the
low watermark, and skip the checks that consider the additional space
taken by the shard.

Closes ES-13384

Author: DaveCTurner

docs/changelog/137399.yaml

@@ -0,0 +1,5 @@
+pr: 137399
+summary: Allow allocating clones over low watermark
+area: Allocation
+type: bug
+issues: []

server/src/internalClusterTest/java/org/elasticsearch/cluster/routing/allocation/decider/DiskThresholdDeciderIT.java

@@ -12,12 +12,18 @@
 import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteUtils;
 import org.elasticsearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
+import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
+import org.elasticsearch.action.admin.indices.shrink.ResizeRequest;
+import org.elasticsearch.action.admin.indices.shrink.ResizeType;
 import org.elasticsearch.action.admin.indices.stats.ShardStats;
 import org.elasticsearch.action.support.ActionTestUtils;
+import org.elasticsearch.action.support.ActiveShardCount;
+import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.cluster.ClusterInfoService;
 import org.elasticsearch.cluster.ClusterInfoServiceUtils;
 import org.elasticsearch.cluster.DiskUsageIntegTestCase;
 import org.elasticsearch.cluster.InternalClusterInfoService;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.routing.IndexRoutingTable;
 import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
 import org.elasticsearch.cluster.routing.ShardRouting;
@@ -53,6 +59,7 @@
 import static org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider.CLUSTER_ROUTING_REBALANCE_ENABLE_SETTING;
 import static org.elasticsearch.index.store.Store.INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
@@ -103,6 +110,62 @@ public void testHighWatermarkNotExceeded() throws Exception {
         assertBusyWithDiskUsageRefresh(dataNode0Id, indexName, contains(in(shardSizes.getSmallestShardIds())));
     }
 
+    public void testAllocateCloneIgnoresLowWatermark() throws Exception {
+        final var lowWatermarkBytes = randomLongBetween(WATERMARK_BYTES + 1, WATERMARK_BYTES * 5);
+
+        internalCluster().startMasterOnlyNode(
+            Settings.builder()
+                .put(DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_LOW_DISK_WATERMARK_SETTING.getKey(), lowWatermarkBytes + "b")
+                .build()
+        );
+        final var dataNodeName = internalCluster().startDataOnlyNode();
+        ensureStableCluster(2);
+
+        final InternalClusterInfoService clusterInfoService = getInternalClusterInfoService();
+        internalCluster().getCurrentMasterNodeInstance(ClusterService.class).addListener(event -> {
+            ClusterInfoServiceUtils.refresh(clusterInfoService);
+        });
+
+        final var sourceIndexName = "source-" + randomIdentifier();
+        createIndex(sourceIndexName, indexSettings(1, 0).put(INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING.getKey(), "0ms").build());
+        final var shardSizes = createReasonableSizedShards(sourceIndexName);
+
+        updateIndexSettings(Settings.builder().put("blocks.write", true), sourceIndexName);
+
+        final var totalSpace = randomLongBetween(
+            /* do not exceed the high watermark */
+            shardSizes.getSmallestShardSize() + WATERMARK_BYTES + 1,
+            /* but make it so that naively duplicating the shard would exceed the low watermark, or else it's not a meaningful test */
+            2 * shardSizes.getSmallestShardSize() + lowWatermarkBytes
+        );
+
+        getTestFileStore(dataNodeName).setTotalSpace(totalSpace);
+        refreshDiskUsage();
+
+        final var targetIndexName = "target-" + randomIdentifier();
+        final var resizeRequest = new ResizeRequest(targetIndexName, sourceIndexName);
+        resizeRequest.setResizeType(ResizeType.CLONE);
+        resizeRequest.masterNodeTimeout(TEST_REQUEST_TIMEOUT);
+        resizeRequest.ackTimeout(TEST_REQUEST_TIMEOUT);
+        resizeRequest.setWaitForActiveShards(ActiveShardCount.ALL);
+        resizeRequest.getTargetIndexRequest()
+            .settings(
+                Settings.builder().put(resizeRequest.getTargetIndexRequest().settings()).put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+            );
+
+        safeAwait(
+            SubscribableListener.<CreateIndexResponse>newForked(l -> indicesAdmin().resizeIndex(resizeRequest, l))
+                .andThenAccept(
+                    createIndexResponse -> assertThat(
+                        true,
+                        allOf(equalTo(createIndexResponse.isAcknowledged()), equalTo(createIndexResponse.isShardsAcknowledged()))
+                    )
+                )
+        );
+
+        ensureGreen(sourceIndexName, targetIndexName);
+    }
+
     public void testRestoreSnapshotAllocationDoesNotExceedWatermark() throws Exception {
         internalCluster().startMasterOnlyNode();
         internalCluster().startDataOnlyNode();

server/src/main/java/org/elasticsearch/cluster/routing/allocation/decider/DiskThresholdDecider.java

@@ -219,10 +219,11 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, Routing
         }
 
         // flag that determines whether the low threshold checks below can be skipped. We use this for a primary shard that is freshly
-        // allocated and empty.
-        boolean skipLowThresholdChecks = shardRouting.primary()
+        // allocated and either empty or the result of cloning another shard.
+        final var isNewCloneTarget = isNewCloneTarget(shardRouting, allocation);
+        final var skipLowThresholdChecks = shardRouting.primary()
             && shardRouting.active() == false
-            && shardRouting.recoverySource().getType() == RecoverySource.Type.EMPTY_STORE;
+            && (isNewCloneTarget || shardRouting.recoverySource().getType() == RecoverySource.Type.EMPTY_STORE);
 
         if (freeBytes < diskThresholdSettings.getFreeBytesThresholdLowStage(total).getBytes()) {
             if (skipLowThresholdChecks == false) {
@@ -283,6 +284,18 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, Routing
         }
 
         // Secondly, check that allocating the shard to this node doesn't put it above the high watermark
+
+        if (isNewCloneTarget) {
+            // The clone will be a hard-linked copy of the original shard so will not meaningfully increase disk usage
+            return allocation.decision(
+                Decision.YES,
+                NAME,
+                "enough disk for freshly-cloned shard on node, free: [%s], used: [%s]",
+                freeBytesValue,
+                Strings.format1Decimals(usedDiskPercentage, "%")
+            );
+        }
+
         final long shardSize = getExpectedShardSize(shardRouting, 0L, allocation);
         assert shardSize >= 0 : shardSize;
         long freeBytesAfterShard = freeBytes - shardSize;
@@ -326,6 +339,23 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, Routing
         );
     }
 
+    private static boolean isNewCloneTarget(ShardRouting shardRouting, RoutingAllocation allocation) {
+        if (shardRouting.unassigned() == false
+            || shardRouting.primary() == false
+            || shardRouting.recoverySource() != RecoverySource.LocalShardsRecoverySource.INSTANCE) {
+            return false;
+        }
+
+        final var targetMetadata = allocation.metadata().indexMetadata(shardRouting.index());
+        final var sourceIndex = targetMetadata.getResizeSourceIndex();
+        if (sourceIndex == null) {
+            return false;
+        }
+
+        final var sourceMetadata = allocation.metadata().indexMetadata(sourceIndex);
+        return sourceMetadata != null && sourceMetadata.getNumberOfShards() == targetMetadata.getNumberOfShards();
+    }
+
     @Override
     public Decision canForceAllocateDuringReplace(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
         Map<String, DiskUsage> usages = allocation.clusterInfo().getNodeMostAvailableDiskUsages();

x-pack/plugin/autoscaling/src/internalClusterTest/java/org/elasticsearch/xpack/autoscaling/storage/ReactiveStorageIT.java

@@ -26,7 +26,6 @@
 import org.elasticsearch.node.Node;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.NodeRoles;
-import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.autoscaling.action.GetAutoscalingCapacityAction;
 import org.elasticsearch.xpack.autoscaling.action.PutAutoscalingPolicyAction;
@@ -400,13 +399,7 @@ public void testScaleWhileShrinking() throws Exception {
         );
     }
 
-    @TestLogging(
-        reason = "Debugging https://github.com/elastic/elasticsearch/issues/96764",
-        value = "org.elasticsearch.cluster.InternalClusterInfoService:TRACE"
-            + ",org.elasticsearch.xpack.autoscaling.action:TRACE"
-            + ",org.elasticsearch.cluster.routing.allocation:DEBUG"
-    )
-    public void testScaleDuringSplitOrClone() throws Exception {
+    public void testScaleDuringSplit() throws Exception {
         internalCluster().startMasterOnlyNode();
         final String dataNode1Name = internalCluster().startDataOnlyNode();
 
@@ -460,9 +453,8 @@ public void testScaleDuringSplitOrClone() throws Exception {
         );
 
         updateIndexSettings(Settings.builder().put("index.blocks.write", true), indexName);
-        ResizeType resizeType = randomFrom(ResizeType.CLONE, ResizeType.SPLIT);
         String cloneName = "clone-" + indexName;
-        int resizedShardCount = resizeType == ResizeType.CLONE ? 1 : between(2, 10);
+        int resizedShardCount = between(2, 10);
         assertAcked(
             indicesAdmin().prepareResizeIndex(indexName, cloneName)
                 .setSettings(
@@ -472,7 +464,7 @@ public void testScaleDuringSplitOrClone() throws Exception {
                         .build()
                 )
                 .setWaitForActiveShards(ActiveShardCount.NONE)
-                .setResizeType(resizeType)
+                .setResizeType(ResizeType.SPLIT)
         );
 
         // * 2 since worst case is no hard links, see DiskThresholdDecider.getExpectedShardSize.