Account for remaining recovery in disk allocator

docs/reference/cluster/nodes-stats.asciidoc

@@ -245,6 +245,18 @@ Total size of all shards assigned to the node.
 `size_in_bytes`::
 (integer)
 Total size, in bytes, of all shards assigned to the node.
+
+`reserved`::
+(<<byte-units,byte value>>)
+A prediction of how much larger the shard stores on this node will eventually
+grow due to ongoing peer recoveries, restoring snapshots, and similar
+activities. A value of `-1b` indicates that this is not available.
+
+`reserved_in_bytes`::
+(integer)
+A prediction, in bytes, of how much larger the shard stores on this node will
+eventually grow due to ongoing peer recoveries, restoring snapshots, and
+similar activities. A value of `-1` indicates that this is not available.
 =======
 
 `indexing`::

docs/reference/cluster/stats.asciidoc

@@ -231,6 +231,17 @@ Total size of all shards assigned to selected nodes.
 `size_in_bytes`::
 (integer)
 Total size, in bytes, of all shards assigned to selected nodes.
+
+`reserved`::
+(<<byte-units,byte value>>)
+A prediction of how much larger the shard stores will eventually grow due to
+ongoing peer recoveries, restoring snapshots, and similar activities.
+
+`reserved_in_bytes`::
+(integer)
+A prediction, in bytes, of how much larger the shard stores will eventually
+grow due to ongoing peer recoveries, restoring snapshots, and similar
+activities.
 =====
 
 `fielddata`::
@@ -1135,7 +1146,9 @@ The API returns the following response:
       },
       "store": {
          "size": "16.2kb",
-         "size_in_bytes": 16684
+         "size_in_bytes": 16684,
+         "reserved": "0b",
+         "reserved_in_bytes": 0
       },
       "fielddata": {
          "memory_size": "0b",

rest-api-spec/src/main/resources/rest-api-spec/test/nodes.stats/40_store_stats.yml

@@ -0,0 +1,22 @@
+---
+"Store stats":
+  - skip:
+      version: " - 7.99.99"
+      reason: "reserved_in_bytes field is not returned in prior versions"
+      features: [arbitrary_key]
+
+  - do:
+      nodes.info:
+        node_id: _master
+  - set:
+      nodes._arbitrary_key_: master
+
+  - do:
+      nodes.stats:
+        metric: [ indices ]
+        index_metric: [ store ]
+
+  - is_false: nodes.$master.discovery
+  - is_true:  nodes.$master.indices.store
+  - gte: { nodes.$master.indices.store.size_in_bytes: 0 }
+  - gte: { nodes.$master.indices.store.reserved_in_bytes: -1 }

server/src/internalClusterTest/java/org/elasticsearch/cluster/ClusterInfoServiceIT.java

@@ -29,6 +29,7 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.routing.RoutingTable;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -162,6 +163,8 @@ public void testClusterInfoServiceCollectsInformation() {
             IndexService indexService = indicesService.indexService(shard.index());
             IndexShard indexShard = indexService.getShardOrNull(shard.id());
             assertEquals(indexShard.shardPath().getRootDataPath().toString(), dataPath);
+
+            assertTrue(info.getReservedSpace(nodeId, dataPath).containsShardId(shard.shardId()));
         }
     }
 
@@ -232,6 +235,7 @@ public void testClusterInfoServiceInformationClearOnError() {
         assertThat(info.getNodeLeastAvailableDiskUsages().size(), equalTo(0));
         assertThat(info.getNodeMostAvailableDiskUsages().size(), equalTo(0));
         assertThat(info.shardSizes.size(), equalTo(0));
+        assertThat(info.reservedSpace.size(), equalTo(0));
 
         // check we recover
         blockingActionFilter.blockActions();
@@ -242,5 +246,10 @@ public void testClusterInfoServiceInformationClearOnError() {
         assertThat(info.getNodeMostAvailableDiskUsages().size(), equalTo(2));
         assertThat(info.shardSizes.size(), greaterThan(0));
 
+        RoutingTable routingTable = client().admin().cluster().prepareState().clear().setRoutingTable(true).get().getState().routingTable();
+        for (ShardRouting shard : routingTable.allShards()) {
+            assertTrue(info.getReservedSpace(shard.currentNodeId(), info.getDataPath(shard)).containsShardId(shard.shardId()));
+        }
+
     }
 }

server/src/internalClusterTest/java/org/elasticsearch/indices/recovery/IndexRecoveryIT.java

@@ -83,7 +83,9 @@
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.store.Store;
+import org.elasticsearch.index.store.StoreStats;
 import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.indices.NodeIndicesStats;
 import org.elasticsearch.indices.analysis.AnalysisModule;
 import org.elasticsearch.indices.flush.SyncedFlushUtil;
 import org.elasticsearch.indices.recovery.RecoveryState.Stage;
@@ -146,6 +148,7 @@
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.everyItem;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
@@ -1834,4 +1837,59 @@ public void testCancelRecoveryWithAutoExpandReplicas() throws Exception {
             }
         });
     }
+
+    public void testReservesBytesDuringPeerRecoveryPhaseOne() throws Exception {
+        internalCluster().startNode();
+        List<String> dataNodes = internalCluster().startDataOnlyNodes(2);
+        String indexName = "test-index";
+        createIndex(indexName, Settings.builder()
+            .put("index.number_of_shards", 1).put("index.number_of_replicas", 0)
+            .put("index.routing.allocation.include._name", String.join(",", dataNodes)).build());
+        ensureGreen(indexName);
+        final List<IndexRequestBuilder> indexRequests = IntStream.range(0, between(10, 500))
+            .mapToObj(n -> client().prepareIndex(indexName, "_doc").setSource("foo", "bar"))
+            .collect(Collectors.toList());
+        indexRandom(randomBoolean(), true, true, indexRequests);
+        assertThat(client().admin().indices().prepareFlush(indexName).get().getFailedShards(), equalTo(0));
+
+        ClusterState clusterState = client().admin().cluster().prepareState().get().getState();
+        DiscoveryNode nodeWithPrimary = clusterState.nodes().get(clusterState.routingTable()
+            .index(indexName).shard(0).primaryShard().currentNodeId());
+        MockTransportService transportService = (MockTransportService) internalCluster()
+            .getInstance(TransportService.class, nodeWithPrimary.getName());
+
+        final AtomicBoolean fileInfoIntercepted = new AtomicBoolean();
+        final AtomicBoolean fileChunkIntercepted = new AtomicBoolean();
+        transportService.addSendBehavior((connection, requestId, action, request, options) -> {
+            if (action.equals(PeerRecoveryTargetService.Actions.FILES_INFO)) {
+                if (fileInfoIntercepted.compareAndSet(false, true)) {
+                    final NodeIndicesStats nodeIndicesStats = client().admin().cluster().prepareNodesStats(connection.getNode().getId())
+                        .clear().setIndices(new CommonStatsFlags(CommonStatsFlags.Flag.Store)).get().getNodes().get(0).getIndices();
+                    assertThat(nodeIndicesStats.getStore().getReservedSize().getBytes(), equalTo(0L));
+                    assertThat(nodeIndicesStats.getShardStats(clusterState.metadata().index(indexName).getIndex())
+                        .stream().flatMap(s -> Arrays.stream(s.getShards())).map(s -> s.getStats().getStore().getReservedSize().getBytes())
+                        .collect(Collectors.toList()),
+                        everyItem(equalTo(StoreStats.UNKNOWN_RESERVED_BYTES)));
+                }
+            } else if (action.equals(PeerRecoveryTargetService.Actions.FILE_CHUNK)) {
+                if (fileChunkIntercepted.compareAndSet(false, true)) {
+                    assertThat(client().admin().cluster().prepareNodesStats(connection.getNode().getId()).clear()
+                            .setIndices(new CommonStatsFlags(CommonStatsFlags.Flag.Store)).get().getNodes().get(0)
+                            .getIndices().getStore().getReservedSize().getBytes(),
+                        greaterThan(0L));
+                }
+            }
+            connection.sendRequest(requestId, action, request, options);
+        });
+
+        assertAcked(client().admin().indices().prepareUpdateSettings(indexName)
+            .setSettings(Settings.builder().put("index.number_of_replicas", 1)));
+        ensureGreen();
+        assertTrue(fileInfoIntercepted.get());
+        assertTrue(fileChunkIntercepted.get());
+
+        assertThat(client().admin().cluster().prepareNodesStats().get().getNodes().stream()
+            .mapToLong(n -> n.getIndices().getStore().getReservedSize().getBytes()).sum(), equalTo(0L));
+    }
+
 }