Frozen autoscaling decider based on storage pct (#71756)

The frozen tier partially downloads shards only. This commit
introduces an autoscaling decider that scales the total storage
on the tier according to a configurable percentage relative to
the total data set size.

docs/reference/autoscaling/autoscaling-deciders.asciidoc

@@ -15,6 +15,10 @@ Available for policies governing hot data nodes.
 Estimates required memory capacity based on number of frozen shards.
 Available for policies governing frozen data nodes.
 
+<<autoscaling-frozen-storage-decider,Frozen storage decider>>::
+Estimates required storage capacity as a percentage of total frozen data set.
+Available for policies governing frozen data nodes.
+
 <<autoscaling-machine-learning-decider,Machine learning decider>>::
 Estimates required memory capacity based on machine learning jobs.
 Available for policies governing machine learning nodes.
@@ -25,5 +29,6 @@ Responds with a fixed required capacity. This decider is intended for testing on
 include::deciders/reactive-storage-decider.asciidoc[]
 include::deciders/proactive-storage-decider.asciidoc[]
 include::deciders/frozen-shards-decider.asciidoc[]
+include::deciders/frozen-storage-decider.asciidoc[]
 include::deciders/machine-learning-decider.asciidoc[]
 include::deciders/fixed-decider.asciidoc[]

docs/reference/autoscaling/deciders/frozen-storage-decider.asciidoc

@@ -0,0 +1,19 @@
+[role="xpack"]
+[[autoscaling-frozen-storage-decider]]
+=== Frozen storage decider
+
+The frozen storage decider (`frozen_storage`) calculates the local storage
+required to search the current set of frozen indices based on a percentage of
+the total data set size. It signals that additional storage capacity is
+necessary when existing capacity is less than the percentage multiplied by
+total data set size.
+
+The frozen storage decider is enabled for all policies governing frozen data
+nodes and has no configuration options.
+
+[[autoscaling-frozen-storage-decider-settings]]
+==== Configuration settings
+
+`percentage`::
+(Optional, number value)
+Percentage of local storage relative to the data set size. Defaults to 5.

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

@@ -28,6 +28,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.shard.IndexShard;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.store.Store;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.indices.SystemIndexDescriptor;
@@ -57,6 +58,7 @@ import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
 
 /**
@@ -151,8 +153,10 @@ public class ClusterInfoServiceIT extends ESIntegTestCase {
         ImmutableOpenMap<String, DiskUsage> leastUsages = info.getNodeLeastAvailableDiskUsages();
         ImmutableOpenMap<String, DiskUsage> mostUsages = info.getNodeMostAvailableDiskUsages();
         ImmutableOpenMap<String, Long> shardSizes = info.shardSizes;
+        ImmutableOpenMap<ShardId, Long> shardDataSetSizes = info.shardDataSetSizes;
         assertNotNull(leastUsages);
         assertNotNull(shardSizes);
+        assertNotNull(shardDataSetSizes);
         assertThat("some usages are populated", leastUsages.values().size(), Matchers.equalTo(2));
         assertThat("some shard sizes are populated", shardSizes.values().size(), greaterThan(0));
         for (ObjectCursor<DiskUsage> usage : leastUsages.values()) {
@@ -167,6 +171,10 @@ public class ClusterInfoServiceIT extends ESIntegTestCase {
             logger.info("--> shard size: {}", size.value);
             assertThat("shard size is greater than 0", size.value, greaterThanOrEqualTo(0L));
         }
+        for (ObjectCursor<Long> size : shardDataSetSizes.values()) {
+            assertThat("shard data set size is greater than 0", size.value, greaterThanOrEqualTo(0L));
+        }
+
         ClusterService clusterService = internalTestCluster.getInstance(ClusterService.class, internalTestCluster.getMasterName());
         ClusterState state = clusterService.state();
         for (ShardRouting shard : state.routingTable().allShards()) {
@@ -203,8 +211,11 @@ public class ClusterInfoServiceIT extends ESIntegTestCase {
         assertNotNull("failed to collect info", originalInfo);
         assertThat("some usages are populated", originalInfo.getNodeLeastAvailableDiskUsages().size(), Matchers.equalTo(2));
         assertThat("some shard sizes are populated", originalInfo.shardSizes.size(), greaterThan(0));
+        assertThat("some shard data set sizes are populated", originalInfo.shardDataSetSizes.size(), greaterThan(0));
         for (ShardRouting shardRouting : shardRoutings) {
             assertThat("size for shard " + shardRouting + " found", originalInfo.getShardSize(shardRouting), notNullValue());
+            assertThat("data set size for shard " + shardRouting + " found",
+                originalInfo.getShardDataSetSize(shardRouting.shardId()).isPresent(), is(true));
         }
 
         MockTransportService mockTransportService = (MockTransportService) internalCluster()
@@ -237,8 +248,12 @@ public class ClusterInfoServiceIT extends ESIntegTestCase {
         assertThat(infoAfterTimeout.getNodeMostAvailableDiskUsages().size(), equalTo(1));
         // indices stats from remote nodes will time out, but the local node's shard will be included
         assertThat(infoAfterTimeout.shardSizes.size(), greaterThan(0));
+        assertThat(infoAfterTimeout.shardDataSetSizes.size(), greaterThan(0));
         assertThat(shardRoutings.stream().filter(shardRouting -> infoAfterTimeout.getShardSize(shardRouting) != null)
                 .collect(Collectors.toList()), hasSize(1));
+        assertThat(shardRoutings.stream().map(ShardRouting::shardId).distinct()
+            .filter(shard -> infoAfterTimeout.getShardDataSetSize(shard).isPresent())
+            .collect(Collectors.toList()), hasSize(1));
 
         // now we cause an exception
         timeout.set(false);
@@ -258,6 +273,7 @@ public class ClusterInfoServiceIT extends ESIntegTestCase {
         assertThat(infoAfterException.getNodeLeastAvailableDiskUsages().size(), equalTo(0));
         assertThat(infoAfterException.getNodeMostAvailableDiskUsages().size(), equalTo(0));
         assertThat(infoAfterException.shardSizes.size(), equalTo(0));
+        assertThat(infoAfterException.shardDataSetSizes.size(), equalTo(0));
         assertThat(infoAfterException.reservedSpace.size(), equalTo(0));
 
         // check we recover
@@ -268,6 +284,7 @@ public class ClusterInfoServiceIT extends ESIntegTestCase {
         assertThat(infoAfterRecovery.getNodeLeastAvailableDiskUsages().size(), equalTo(2));
         assertThat(infoAfterRecovery.getNodeMostAvailableDiskUsages().size(), equalTo(2));
         assertThat(infoAfterRecovery.shardSizes.size(), greaterThan(0));
+        assertThat(infoAfterRecovery.shardDataSetSizes.size(), greaterThan(0));
         for (ShardRouting shardRouting : shardRoutings) {
             assertThat("size for shard " + shardRouting + " found", originalInfo.getShardSize(shardRouting), notNullValue());
         }

server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java

@@ -85,6 +85,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
+import java.util.Optional;
 import java.util.concurrent.BrokenBarrierException;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CyclicBarrier;
@@ -244,10 +245,14 @@ public class IndexShardIT extends ESSingleNodeTestCase {
         InternalClusterInfoService clusterInfoService = (InternalClusterInfoService) getInstanceFromNode(ClusterInfoService.class);
         ClusterInfoServiceUtils.refresh(clusterInfoService);
         ClusterState state = getInstanceFromNode(ClusterService.class).state();
-        Long test = clusterInfoService.getClusterInfo().getShardSize(state.getRoutingTable().index("test")
-            .getShards().get(0).primaryShard());
+        ShardRouting shardRouting = state.getRoutingTable().index("test").getShards().get(0).primaryShard();
+        Long test = clusterInfoService.getClusterInfo().getShardSize(shardRouting);
         assertNotNull(test);
         assertTrue(test > 0);
+
+        Optional<Long> dataSetSize = clusterInfoService.getClusterInfo().getShardDataSetSize(shardRouting.shardId());
+        assertTrue(dataSetSize.isPresent());
+        assertThat(dataSetSize.get(), greaterThan(0L));
     }
 
     public void testIndexCanChangeCustomDataPath() throws Exception {

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

@@ -11,6 +11,7 @@ package org.elasticsearch.cluster;
 import com.carrotsearch.hppc.ObjectHashSet;
 import com.carrotsearch.hppc.cursors.ObjectCursor;
 import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
+import org.elasticsearch.Version;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -24,6 +25,7 @@ import org.elasticsearch.index.store.StoreStats;
 import java.io.IOException;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 
 /**
  * ClusterInfo is an object representing a map of nodes to {@link DiskUsage}
@@ -32,15 +34,20 @@ import java.util.Objects;
  * for the key used in the shardSizes map
  */
 public class ClusterInfo implements ToXContentFragment, Writeable {
+
+    public static final Version DATA_SET_SIZE_SIZE_VERSION = Version.V_8_0_0; // todo: Version.V_7_13_0;
+
     private final ImmutableOpenMap<String, DiskUsage> leastAvailableSpaceUsage;
     private final ImmutableOpenMap<String, DiskUsage> mostAvailableSpaceUsage;
     final ImmutableOpenMap<String, Long> shardSizes;
+    final ImmutableOpenMap<ShardId, Long> shardDataSetSizes;
     public static final ClusterInfo EMPTY = new ClusterInfo();
     final ImmutableOpenMap<ShardRouting, String> routingToDataPath;
     final ImmutableOpenMap<NodeAndPath, ReservedSpace> reservedSpace;
 
     protected ClusterInfo() {
-       this(ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of());
+       this(ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of(),
+           ImmutableOpenMap.of());
     }
 
     /**
@@ -49,16 +56,18 @@ public class ClusterInfo implements ToXContentFragment, Writeable {
      * @param leastAvailableSpaceUsage a node id to disk usage mapping for the path that has the least available space on the node.
      * @param mostAvailableSpaceUsage  a node id to disk usage mapping for the path that has the most available space on the node.
      * @param shardSizes a shardkey to size in bytes mapping per shard.
+     * @param shardDataSetSizes a shard id to data set size in bytes mapping per shard
      * @param routingToDataPath the shard routing to datapath mapping
      * @param reservedSpace reserved space per shard broken down by node and data path
      * @see #shardIdentifierFromRouting
      */
     public ClusterInfo(ImmutableOpenMap<String, DiskUsage> leastAvailableSpaceUsage,
                        ImmutableOpenMap<String, DiskUsage> mostAvailableSpaceUsage, ImmutableOpenMap<String, Long> shardSizes,
-                       ImmutableOpenMap<ShardRouting, String> routingToDataPath,
+                       ImmutableOpenMap<ShardId, Long> shardDataSetSizes, ImmutableOpenMap<ShardRouting, String> routingToDataPath,
                        ImmutableOpenMap<NodeAndPath, ReservedSpace> reservedSpace) {
         this.leastAvailableSpaceUsage = leastAvailableSpaceUsage;
         this.shardSizes = shardSizes;
+        this.shardDataSetSizes = shardDataSetSizes;
         this.mostAvailableSpaceUsage = mostAvailableSpaceUsage;
         this.routingToDataPath = routingToDataPath;
         this.reservedSpace = reservedSpace;
@@ -68,6 +77,12 @@ public class ClusterInfo implements ToXContentFragment, Writeable {
         Map<String, DiskUsage> leastMap = in.readMap(StreamInput::readString, DiskUsage::new);
         Map<String, DiskUsage> mostMap = in.readMap(StreamInput::readString, DiskUsage::new);
         Map<String, Long> sizeMap = in.readMap(StreamInput::readString, StreamInput::readLong);
+        Map<ShardId, Long> dataSetSizeMap;
+        if (in.getVersion().onOrAfter(DATA_SET_SIZE_SIZE_VERSION)) {
+            dataSetSizeMap = in.readMap(ShardId::new, StreamInput::readLong);
+        } else {
+            dataSetSizeMap = Map.of();
+        }
         Map<ShardRouting, String> routingMap = in.readMap(ShardRouting::new, StreamInput::readString);
         Map<NodeAndPath, ReservedSpace> reservedSpaceMap;
         if (in.getVersion().onOrAfter(StoreStats.RESERVED_BYTES_VERSION)) {
@@ -82,6 +97,8 @@ public class ClusterInfo implements ToXContentFragment, Writeable {
         this.mostAvailableSpaceUsage = mostBuilder.putAll(mostMap).build();
         ImmutableOpenMap.Builder<String, Long> sizeBuilder = ImmutableOpenMap.builder();
         this.shardSizes = sizeBuilder.putAll(sizeMap).build();
+        ImmutableOpenMap.Builder<ShardId, Long> dataSetSizeBuilder = ImmutableOpenMap.builder();
+        this.shardDataSetSizes = dataSetSizeBuilder.putAll(dataSetSizeMap).build();
         ImmutableOpenMap.Builder<ShardRouting, String> routingBuilder = ImmutableOpenMap.builder();
         this.routingToDataPath = routingBuilder.putAll(routingMap).build();
         ImmutableOpenMap.Builder<NodeAndPath, ReservedSpace> reservedSpaceBuilder = ImmutableOpenMap.builder();
@@ -97,6 +114,9 @@ public class ClusterInfo implements ToXContentFragment, Writeable {
         }
         out.writeMap(this.mostAvailableSpaceUsage, StreamOutput::writeString, (o, v) -> v.writeTo(o));
         out.writeMap(this.shardSizes, StreamOutput::writeString, (o, v) -> out.writeLong(v == null ? -1 : v));
+        if (out.getVersion().onOrAfter(DATA_SET_SIZE_SIZE_VERSION)) {
+            out.writeMap(this.shardDataSetSizes, (o, s) -> s.writeTo(o), (o, v) -> out.writeLong(v));
+        }
         out.writeMap(this.routingToDataPath, (o, k) -> k.writeTo(o), StreamOutput::writeString);
         if (out.getVersion().onOrAfter(StoreStats.RESERVED_BYTES_VERSION)) {
             out.writeMap(this.reservedSpace);
@@ -130,6 +150,12 @@ public class ClusterInfo implements ToXContentFragment, Writeable {
             }
         }
         builder.endObject(); // end "shard_sizes"
+        builder.startObject("shard_data_set_sizes"); {
+            for (ObjectObjectCursor<ShardId, Long> c : this.shardDataSetSizes) {
+                builder.humanReadableField(c.key + "_bytes", c.key.toString(), new ByteSizeValue(c.value));
+            }
+        }
+        builder.endObject(); // end "shard_data_set_sizes"
         builder.startObject("shard_paths"); {
             for (ObjectObjectCursor<ShardRouting, String> c : this.routingToDataPath) {
                 builder.field(c.key.toString(), c.value);
@@ -188,6 +214,9 @@ public class ClusterInfo implements ToXContentFragment, Writeable {
         return shardSize == null ? defaultValue : shardSize;
     }
 
+    public Optional<Long> getShardDataSetSize(ShardId shardId) {
+        return Optional.ofNullable(shardDataSetSizes.get(shardId));
+    }
     /**
      * Returns the reserved space for each shard on the given node/path pair
      */

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

@@ -35,6 +35,7 @@ import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.CountDown;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.store.StoreStats;
 import org.elasticsearch.monitor.fs.FsInfo;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -230,18 +231,20 @@ public class InternalClusterInfoService implements ClusterInfoService, ClusterSt
 
                     final ShardStats[] stats = indicesStatsResponse.getShards();
                     final ImmutableOpenMap.Builder<String, Long> shardSizeByIdentifierBuilder = ImmutableOpenMap.builder();
+                    final ImmutableOpenMap.Builder<ShardId, Long> shardDataSetSizeBuilder = ImmutableOpenMap.builder();
                     final ImmutableOpenMap.Builder<ShardRouting, String> dataPathByShardRoutingBuilder = ImmutableOpenMap.builder();
                     final Map<ClusterInfo.NodeAndPath, ClusterInfo.ReservedSpace.Builder> reservedSpaceBuilders = new HashMap<>();
-                    buildShardLevelInfo(stats, shardSizeByIdentifierBuilder, dataPathByShardRoutingBuilder, reservedSpaceBuilders);
+                    buildShardLevelInfo(stats, shardSizeByIdentifierBuilder, shardDataSetSizeBuilder,
+                        dataPathByShardRoutingBuilder, reservedSpaceBuilders);
 
                     final ImmutableOpenMap.Builder<ClusterInfo.NodeAndPath, ClusterInfo.ReservedSpace> rsrvdSpace
                             = ImmutableOpenMap.builder();
                     reservedSpaceBuilders.forEach((nodeAndPath, builder) -> rsrvdSpace.put(nodeAndPath, builder.build()));
 
                     indicesStatsSummary = new IndicesStatsSummary(
-                            shardSizeByIdentifierBuilder.build(),
-                            dataPathByShardRoutingBuilder.build(),
-                            rsrvdSpace.build());
+                        shardSizeByIdentifierBuilder.build(), shardDataSetSizeBuilder.build(),
+                        dataPathByShardRoutingBuilder.build(),
+                        rsrvdSpace.build());
                 }
 
                 @Override
@@ -356,7 +359,8 @@ public class InternalClusterInfoService implements ClusterInfoService, ClusterSt
     public ClusterInfo getClusterInfo() {
         final IndicesStatsSummary indicesStatsSummary = this.indicesStatsSummary; // single volatile read
         return new ClusterInfo(leastAvailableSpaceUsages, mostAvailableSpaceUsages,
-            indicesStatsSummary.shardSizes, indicesStatsSummary.shardRoutingToDataPath, indicesStatsSummary.reservedSpace);
+            indicesStatsSummary.shardSizes, indicesStatsSummary.shardDataSetSizes,
+            indicesStatsSummary.shardRoutingToDataPath, indicesStatsSummary.reservedSpace);
     }
 
     // allow tests to adjust the node stats on receipt
@@ -380,6 +384,7 @@ public class InternalClusterInfoService implements ClusterInfoService, ClusterSt
     }
 
     static void buildShardLevelInfo(ShardStats[] stats, ImmutableOpenMap.Builder<String, Long> shardSizes,
+                                    ImmutableOpenMap.Builder<ShardId, Long> shardDataSetSizeBuilder,
                                     ImmutableOpenMap.Builder<ShardRouting, String> newShardRoutingToDataPath,
                                     Map<ClusterInfo.NodeAndPath, ClusterInfo.ReservedSpace.Builder> reservedSpaceByShard) {
         for (ShardStats s : stats) {
@@ -391,12 +396,15 @@ public class InternalClusterInfoService implements ClusterInfoService, ClusterSt
                 continue;
             }
             final long size = storeStats.sizeInBytes();
+            final long dataSetSize = storeStats.totalDataSetSizeInBytes();
             final long reserved = storeStats.getReservedSize().getBytes();
 
             final String shardIdentifier = ClusterInfo.shardIdentifierFromRouting(shardRouting);
             logger.trace("shard: {} size: {} reserved: {}", shardIdentifier, size, reserved);
             shardSizes.put(shardIdentifier, size);
-
+            if (dataSetSize > shardDataSetSizeBuilder.getOrDefault(shardRouting.shardId(), -1L)) {
+                shardDataSetSizeBuilder.put(shardRouting.shardId(), dataSetSize);
+            }
             if (reserved != StoreStats.UNKNOWN_RESERVED_BYTES) {
                 final ClusterInfo.ReservedSpace.Builder reservedSpaceBuilder = reservedSpaceByShard.computeIfAbsent(
                     new ClusterInfo.NodeAndPath(shardRouting.currentNodeId(), s.getDataPath()),
@@ -466,16 +474,19 @@ public class InternalClusterInfoService implements ClusterInfoService, ClusterSt
 
     private static class IndicesStatsSummary {
         static final IndicesStatsSummary EMPTY
-            = new IndicesStatsSummary(ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of());
+            = new IndicesStatsSummary(ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of(), ImmutableOpenMap.of());
 
         final ImmutableOpenMap<String, Long> shardSizes;
+        final ImmutableOpenMap<ShardId, Long> shardDataSetSizes;
         final ImmutableOpenMap<ShardRouting, String> shardRoutingToDataPath;
         final ImmutableOpenMap<ClusterInfo.NodeAndPath, ClusterInfo.ReservedSpace> reservedSpace;
 
         IndicesStatsSummary(ImmutableOpenMap<String, Long> shardSizes,
+                            ImmutableOpenMap<ShardId, Long> shardDataSetSizes,
                             ImmutableOpenMap<ShardRouting, String> shardRoutingToDataPath,
                             ImmutableOpenMap<ClusterInfo.NodeAndPath, ClusterInfo.ReservedSpace> reservedSpace) {
             this.shardSizes = shardSizes;
+            this.shardDataSetSizes = shardDataSetSizes;
             this.shardRoutingToDataPath = shardRoutingToDataPath;
             this.reservedSpace = reservedSpace;
         }

server/src/main/java/org/elasticsearch/index/store/StoreStats.java

@@ -100,6 +100,10 @@ public class StoreStats implements Writeable, ToXContentFragment {
         return totalDataSetSize();
     }
 
+    public long totalDataSetSizeInBytes() {
+        return totalDataSetSizeInBytes;
+    }
+
     /**
      * A prediction of how much larger this store will eventually grow. For instance, if we are currently doing a peer recovery or restoring
      * a snapshot into this store then we can account for the rest of the recovery using this field. A value of {@code -1B} indicates that

server/src/test/java/org/elasticsearch/cluster/ClusterInfoTests.java

@@ -19,7 +19,7 @@ public class ClusterInfoTests extends ESTestCase {
 
     public void testSerialization() throws Exception {
         ClusterInfo clusterInfo = new ClusterInfo(
-                randomDiskUsage(), randomDiskUsage(), randomShardSizes(), randomRoutingToDataPath(),
+                randomDiskUsage(), randomDiskUsage(), randomShardSizes(), randomDataSetSizes(), randomRoutingToDataPath(),
                 randomReservedSpace());
         BytesStreamOutput output = new BytesStreamOutput();
         clusterInfo.writeTo(output);
@@ -28,6 +28,7 @@ public class ClusterInfoTests extends ESTestCase {
         assertEquals(clusterInfo.getNodeLeastAvailableDiskUsages(), result.getNodeLeastAvailableDiskUsages());
         assertEquals(clusterInfo.getNodeMostAvailableDiskUsages(), result.getNodeMostAvailableDiskUsages());
         assertEquals(clusterInfo.shardSizes, result.shardSizes);
+        assertEquals(clusterInfo.shardDataSetSizes, result.shardDataSetSizes);
         assertEquals(clusterInfo.routingToDataPath, result.routingToDataPath);
         assertEquals(clusterInfo.reservedSpace, result.reservedSpace);
     }
@@ -57,6 +58,17 @@ public class ClusterInfoTests extends ESTestCase {
         return builder.build();
     }
 
+    private static ImmutableOpenMap<ShardId, Long> randomDataSetSizes() {
+        int numEntries = randomIntBetween(0, 128);
+        ImmutableOpenMap.Builder<ShardId, Long> builder = ImmutableOpenMap.builder(numEntries);
+        for (int i = 0; i < numEntries; i++) {
+            ShardId key = new ShardId(randomAlphaOfLength(10), randomAlphaOfLength(10), between(0, Integer.MAX_VALUE));
+            long shardSize = randomIntBetween(0, Integer.MAX_VALUE);
+            builder.put(key, shardSize);
+        }
+        return builder.build();
+    }
+
     private static ImmutableOpenMap<ShardRouting, String> randomRoutingToDataPath() {
         int numEntries = randomIntBetween(0, 128);
         ImmutableOpenMap.Builder<ShardRouting, String> builder = ImmutableOpenMap.builder(numEntries);

server/src/test/java/org/elasticsearch/cluster/DiskUsageTests.java

@@ -94,28 +94,38 @@ public class DiskUsageTests extends ESTestCase {
         test_0 = ShardRoutingHelper.moveToStarted(test_0);
         Path test0Path = createTempDir().resolve("indices").resolve(index.getUUID()).resolve("0");
         CommonStats commonStats0 = new CommonStats();
-        commonStats0.store = new StoreStats(100, 100, 0L);
+        commonStats0.store = new StoreStats(100, 101, 0L);
         ShardRouting test_1 = ShardRouting.newUnassigned(new ShardId(index, 1), false, PeerRecoverySource.INSTANCE,
             new UnassignedInfo(UnassignedInfo.Reason.INDEX_CREATED, "foo"));
         test_1 = ShardRoutingHelper.initialize(test_1, "node2");
         test_1 = ShardRoutingHelper.moveToStarted(test_1);
         Path test1Path = createTempDir().resolve("indices").resolve(index.getUUID()).resolve("1");
         CommonStats commonStats1 = new CommonStats();
-        commonStats1.store = new StoreStats(1000, 1000, 0L);
+        commonStats1.store = new StoreStats(1000, 1001, 0L);
+        CommonStats commonStats2 = new CommonStats();
+        commonStats2.store = new StoreStats(1000, 999, 0L);
         ShardStats[] stats  = new ShardStats[] {
                 new ShardStats(test_0, new ShardPath(false, test0Path, test0Path, test_0.shardId()), commonStats0 , null, null, null),
-                new ShardStats(test_1, new ShardPath(false, test1Path, test1Path, test_1.shardId()), commonStats1 , null, null, null)
+                new ShardStats(test_1, new ShardPath(false, test1Path, test1Path, test_1.shardId()), commonStats1 , null, null, null),
+                new ShardStats(test_1, new ShardPath(false, test1Path, test1Path, test_1.shardId()), commonStats2 , null, null, null)
         };
         ImmutableOpenMap.Builder<String, Long> shardSizes = ImmutableOpenMap.builder();
+        ImmutableOpenMap.Builder<ShardId, Long> shardDataSetSizes = ImmutableOpenMap.builder();
         ImmutableOpenMap.Builder<ShardRouting, String> routingToPath = ImmutableOpenMap.builder();
         ClusterState state = ClusterState.builder(new ClusterName("blarg")).version(0).build();
-        InternalClusterInfoService.buildShardLevelInfo(stats, shardSizes, routingToPath, new HashMap<>());
+        InternalClusterInfoService.buildShardLevelInfo(stats, shardSizes, shardDataSetSizes, routingToPath, new HashMap<>());
         assertEquals(2, shardSizes.size());
         assertTrue(shardSizes.containsKey(ClusterInfo.shardIdentifierFromRouting(test_0)));
         assertTrue(shardSizes.containsKey(ClusterInfo.shardIdentifierFromRouting(test_1)));
         assertEquals(100L, shardSizes.get(ClusterInfo.shardIdentifierFromRouting(test_0)).longValue());
         assertEquals(1000L, shardSizes.get(ClusterInfo.shardIdentifierFromRouting(test_1)).longValue());
 
+        assertEquals(2, shardDataSetSizes.size());
+        assertTrue(shardDataSetSizes.containsKey(test_0.shardId()));
+        assertTrue(shardDataSetSizes.containsKey(test_1.shardId()));
+        assertEquals(101L, shardDataSetSizes.get(test_0.shardId()).longValue());
+        assertEquals(1001L, shardDataSetSizes.get(test_1.shardId()).longValue());
+
         assertEquals(2, routingToPath.size());
         assertTrue(routingToPath.containsKey(test_0));
         assertTrue(routingToPath.containsKey(test_1));

server/src/test/java/org/elasticsearch/cluster/routing/allocation/DiskThresholdMonitorTests.java

@@ -631,7 +631,7 @@ public class DiskThresholdMonitorTests extends ESAllocationTestCase {
 
     private static ClusterInfo clusterInfo(ImmutableOpenMap<String, DiskUsage> diskUsages,
                                            ImmutableOpenMap<ClusterInfo.NodeAndPath, ClusterInfo.ReservedSpace> reservedSpace) {
-        return new ClusterInfo(diskUsages, null, null, null, reservedSpace);
+        return new ClusterInfo(diskUsages, null, null, null, null, reservedSpace);
     }
 
     private static DiscoveryNode newFrozenOnlyNode(String nodeId) {

server/src/test/java/org/elasticsearch/cluster/routing/allocation/decider/DiskThresholdDeciderTests.java

@@ -1263,7 +1263,7 @@ public class DiskThresholdDeciderTests extends ESAllocationTestCase {
         DevNullClusterInfo(ImmutableOpenMap<String, DiskUsage> leastAvailableSpaceUsage,
                            ImmutableOpenMap<String, DiskUsage> mostAvailableSpaceUsage,
                            ImmutableOpenMap<String, Long> shardSizes, ImmutableOpenMap<NodeAndPath, ReservedSpace> reservedSpace) {
-            super(leastAvailableSpaceUsage, mostAvailableSpaceUsage, shardSizes, null, reservedSpace);
+            super(leastAvailableSpaceUsage, mostAvailableSpaceUsage, shardSizes, null, null, reservedSpace);
         }
 
         @Override

server/src/test/java/org/elasticsearch/cluster/routing/allocation/decider/DiskThresholdDeciderUnitTests.java

@@ -93,7 +93,7 @@ public class DiskThresholdDeciderUnitTests extends ESAllocationTestCase {
         ImmutableOpenMap.Builder<String, Long> shardSizes = ImmutableOpenMap.builder();
         shardSizes.put("[test][0][p]", 10L); // 10 bytes
         final ClusterInfo clusterInfo = new ClusterInfo(leastAvailableUsages.build(),
-            mostAvailableUsage.build(), shardSizes.build(), ImmutableOpenMap.of(),  ImmutableOpenMap.of());
+            mostAvailableUsage.build(), shardSizes.build(), null, ImmutableOpenMap.of(),  ImmutableOpenMap.of());
         RoutingAllocation allocation = new RoutingAllocation(new AllocationDeciders(Collections.singleton(decider)),
             clusterState.getRoutingNodes(), clusterState, clusterInfo, null, System.nanoTime());
         allocation.debugDecision(true);
@@ -148,7 +148,7 @@ public class DiskThresholdDeciderUnitTests extends ESAllocationTestCase {
         final long shardSize = randomIntBetween(110, 1000);
         shardSizes.put("[test][0][p]", shardSize);
         ClusterInfo clusterInfo = new ClusterInfo(leastAvailableUsages.build(), mostAvailableUsage.build(),
-            shardSizes.build(), ImmutableOpenMap.of(),  ImmutableOpenMap.of());
+            shardSizes.build(), null, ImmutableOpenMap.of(),  ImmutableOpenMap.of());
         RoutingAllocation allocation = new RoutingAllocation(new AllocationDeciders(Collections.singleton(decider)),
             clusterState.getRoutingNodes(), clusterState, clusterInfo, null, System.nanoTime());
         allocation.debugDecision(true);
@@ -229,7 +229,7 @@ public class DiskThresholdDeciderUnitTests extends ESAllocationTestCase {
         shardSizes.put("[test][2][p]", 10L);
 
         final ClusterInfo clusterInfo = new ClusterInfo(leastAvailableUsages.build(), mostAvailableUsage.build(),
-            shardSizes.build(), shardRoutingMap.build(), ImmutableOpenMap.of());
+            shardSizes.build(), null, shardRoutingMap.build(), ImmutableOpenMap.of());
         RoutingAllocation allocation = new RoutingAllocation(new AllocationDeciders(Collections.singleton(decider)),
             clusterState.getRoutingNodes(), clusterState, clusterInfo, null, System.nanoTime());
         allocation.debugDecision(true);
@@ -504,7 +504,8 @@ public class DiskThresholdDeciderUnitTests extends ESAllocationTestCase {
         ImmutableOpenMap.Builder<String, Long> shardSizes = ImmutableOpenMap.builder();
         shardSizes.put("[test][0][p]", 10L); // 10 bytes
         final ImmutableOpenMap<String, DiskUsage> usages = allFullUsages.build();
-        final ClusterInfo clusterInfo = new ClusterInfo(usages, usages, shardSizes.build(), ImmutableOpenMap.of(), ImmutableOpenMap.of());
+        final ClusterInfo clusterInfo = new ClusterInfo(usages, usages, shardSizes.build(), null, ImmutableOpenMap.of(),
+            ImmutableOpenMap.of());
         RoutingAllocation allocation = new RoutingAllocation(new AllocationDeciders(Collections.singleton(decider)),
                 clusterState.getRoutingNodes(), clusterState, clusterInfo, null, System.nanoTime());
         allocation.debugDecision(true);

test/framework/src/main/java/org/elasticsearch/cluster/MockInternalClusterInfoService.java

@@ -81,7 +81,7 @@ public class MockInternalClusterInfoService extends InternalClusterInfoService {
     class SizeFakingClusterInfo extends ClusterInfo {
         SizeFakingClusterInfo(ClusterInfo delegate) {
             super(delegate.getNodeLeastAvailableDiskUsages(), delegate.getNodeMostAvailableDiskUsages(),
-                delegate.shardSizes, delegate.routingToDataPath, delegate.reservedSpace);
+                delegate.shardSizes, delegate.shardDataSetSizes, delegate.routingToDataPath, delegate.reservedSpace);
         }
 
         @Override

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

@@ -0,0 +1,125 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling;
+
+import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.snapshots.AbstractSnapshotIntegTestCase;
+import org.elasticsearch.snapshots.SnapshotInfo;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.xpack.autoscaling.action.GetAutoscalingCapacityAction;
+import org.elasticsearch.xpack.autoscaling.action.PutAutoscalingPolicyAction;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingCapacity;
+import org.elasticsearch.xpack.autoscaling.shards.LocalStateAutoscalingAndSearchableSnapshots;
+import org.elasticsearch.xpack.core.DataTier;
+import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotAction;
+import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
+import org.elasticsearch.xpack.searchablesnapshots.cache.shared.FrozenCacheService;
+import org.junit.Before;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.elasticsearch.index.IndexSettings.INDEX_SOFT_DELETES_SETTING;
+import static org.elasticsearch.license.LicenseService.SELF_GENERATED_LICENSE_TYPE;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.equalTo;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
+public abstract class AbstractFrozenAutoscalingIntegTestCase extends AbstractSnapshotIntegTestCase {
+
+    protected final String indexName = "index";
+    protected final String restoredIndexName = "restored";
+    protected final String fsRepoName = randomAlphaOfLength(10);
+    protected final String snapshotName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
+    protected final String policyName = "frozen";
+
+    @Override
+    protected boolean addMockInternalEngine() {
+        return false;
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return List.of(LocalStateAutoscalingAndSearchableSnapshots.class);
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
+        Settings.Builder builder = Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal, otherSettings))
+            .put(SELF_GENERATED_LICENSE_TYPE.getKey(), "trial");
+        if (DiscoveryNode.canContainData(otherSettings)) {
+            builder.put(FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey(), new ByteSizeValue(10, ByteSizeUnit.MB));
+        }
+        return builder.build();
+    }
+
+    @Before
+    public void setupPolicyAndMountedIndex() throws Exception {
+        createRepository(fsRepoName, "fs");
+        putAutoscalingPolicy();
+        assertAcked(prepareCreate(indexName, Settings.builder().put(INDEX_SOFT_DELETES_SETTING.getKey(), true)));
+
+        indexRandom(
+            randomBoolean(),
+            IntStream.range(0, 10).mapToObj(i -> client().prepareIndex(indexName).setSource()).collect(Collectors.toList())
+        );
+
+        final SnapshotInfo snapshotInfo = createFullSnapshot(fsRepoName, snapshotName);
+
+        final AutoscalingCapacity.AutoscalingResources total = capacity().results().get("frozen").requiredCapacity().total();
+        assertThat(total.memory(), equalTo(ByteSizeValue.ZERO));
+        assertThat(total.storage(), equalTo(ByteSizeValue.ZERO));
+
+        final MountSearchableSnapshotRequest req = new MountSearchableSnapshotRequest(
+            restoredIndexName,
+            fsRepoName,
+            snapshotInfo.snapshotId().getName(),
+            indexName,
+            Settings.EMPTY,
+            Strings.EMPTY_ARRAY,
+            true,
+            MountSearchableSnapshotRequest.Storage.SHARED_CACHE
+        );
+        final RestoreSnapshotResponse restoreSnapshotResponse = client().execute(MountSearchableSnapshotAction.INSTANCE, req).get();
+        assertThat(restoreSnapshotResponse.getRestoreInfo().failedShards(), equalTo(0));
+    }
+
+    protected GetAutoscalingCapacityAction.Response capacity() {
+        GetAutoscalingCapacityAction.Request request = new GetAutoscalingCapacityAction.Request();
+        return client().execute(GetAutoscalingCapacityAction.INSTANCE, request).actionGet();
+    }
+
+    private void putAutoscalingPolicy() {
+        // randomly set the setting to verify it can be set.
+        final Settings settings = randomBoolean() ? Settings.EMPTY : addDeciderSettings(Settings.builder()).build();
+        final PutAutoscalingPolicyAction.Request request = new PutAutoscalingPolicyAction.Request(
+            policyName,
+            new TreeSet<>(Set.of(DataTier.DATA_FROZEN)),
+            new TreeMap<>(Map.of(deciderName(), settings))
+        );
+        assertAcked(client().execute(PutAutoscalingPolicyAction.INSTANCE, request).actionGet());
+    }
+
+    protected abstract String deciderName();
+
+    protected abstract Settings.Builder addDeciderSettings(Settings.Builder builder);
+}

x-pack/plugin/autoscaling/src/internalClusterTest/java/org/elasticsearch/xpack/autoscaling/shards/FrozenShardsDeciderIT.java

@@ -7,121 +7,31 @@
 
 package org.elasticsearch.xpack.autoscaling.shards;
 
-import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
-import org.elasticsearch.cluster.node.DiscoveryNode;
-import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.unit.ByteSizeUnit;
-import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.plugins.Plugin;
-import org.elasticsearch.snapshots.AbstractSnapshotIntegTestCase;
-import org.elasticsearch.snapshots.SnapshotInfo;
-import org.elasticsearch.xpack.autoscaling.action.GetAutoscalingCapacityAction;
-import org.elasticsearch.xpack.autoscaling.action.PutAutoscalingPolicyAction;
-import org.elasticsearch.xpack.core.DataTier;
-import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotAction;
-import org.elasticsearch.xpack.core.searchablesnapshots.MountSearchableSnapshotRequest;
-import org.elasticsearch.xpack.searchablesnapshots.cache.shared.FrozenCacheService;
 
-import java.util.Collection;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.TreeSet;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-import static org.elasticsearch.index.IndexSettings.INDEX_SOFT_DELETES_SETTING;
-import static org.elasticsearch.license.LicenseService.SELF_GENERATED_LICENSE_TYPE;
-import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.hamcrest.Matchers.equalTo;
 
-public class FrozenShardsDeciderIT extends AbstractSnapshotIntegTestCase {
-
-    @Override
-    protected boolean addMockInternalEngine() {
-        return false;
-    }
-
-    @Override
-    protected Collection<Class<? extends Plugin>> nodePlugins() {
-        return List.of(LocalStateAutoscalingAndSearchableSnapshots.class);
-    }
-
-    @Override
-    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
-        Settings.Builder builder = Settings.builder()
-            .put(super.nodeSettings(nodeOrdinal, otherSettings))
-            .put(SELF_GENERATED_LICENSE_TYPE.getKey(), "trial");
-        if (DiscoveryNode.canContainData(otherSettings)) {
-            builder.put(FrozenCacheService.SNAPSHOT_CACHE_SIZE_SETTING.getKey(), new ByteSizeValue(10, ByteSizeUnit.MB));
-        }
-        return builder.build();
-    }
+public class FrozenShardsDeciderIT extends org.elasticsearch.xpack.autoscaling.AbstractFrozenAutoscalingIntegTestCase {
 
     @Override
     protected int numberOfShards() {
         return 1;
     }
 
-    public void testScale() throws Exception {
-        final String indexName = "index";
-        final String restoredIndexName = "restored";
-        final String fsRepoName = randomAlphaOfLength(10);
-        final String snapshotName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);
-
-        createRepository(fsRepoName, "fs");
-        putAutoscalingPolicy("frozen");
-        assertAcked(prepareCreate(indexName, Settings.builder().put(INDEX_SOFT_DELETES_SETTING.getKey(), true)));
-
-        indexRandom(
-            randomBoolean(),
-            IntStream.range(0, 10).mapToObj(i -> client().prepareIndex(indexName).setSource()).collect(Collectors.toList())
-        );
-
-        final SnapshotInfo snapshotInfo = createFullSnapshot(fsRepoName, snapshotName);
-
-        assertThat(capacity().results().get("frozen").requiredCapacity().total().memory(), equalTo(ByteSizeValue.ZERO));
-
-        final MountSearchableSnapshotRequest req = new MountSearchableSnapshotRequest(
-            restoredIndexName,
-            fsRepoName,
-            snapshotInfo.snapshotId().getName(),
-            indexName,
-            Settings.EMPTY,
-            Strings.EMPTY_ARRAY,
-            true,
-            MountSearchableSnapshotRequest.Storage.SHARED_CACHE
-        );
-        final RestoreSnapshotResponse restoreSnapshotResponse = client().execute(MountSearchableSnapshotAction.INSTANCE, req).get();
-        assertThat(restoreSnapshotResponse.getRestoreInfo().failedShards(), equalTo(0));
-
+    public void testScale() {
         assertThat(
             capacity().results().get("frozen").requiredCapacity().total().memory(),
             equalTo(FrozenShardsDeciderService.DEFAULT_MEMORY_PER_SHARD)
         );
     }
 
-    private GetAutoscalingCapacityAction.Response capacity() {
-        GetAutoscalingCapacityAction.Request request = new GetAutoscalingCapacityAction.Request();
-        return client().execute(GetAutoscalingCapacityAction.INSTANCE, request).actionGet();
+    @Override
+    protected String deciderName() {
+        return FrozenShardsDeciderService.NAME;
     }
 
-    private void putAutoscalingPolicy(String policyName) {
-        // randomly set the setting to verify it can be set.
-        Settings settings = randomBoolean()
-            ? Settings.EMPTY
-            : Settings.builder()
-                .put(FrozenShardsDeciderService.MEMORY_PER_SHARD.getKey(), FrozenShardsDeciderService.DEFAULT_MEMORY_PER_SHARD)
-                .build();
-        final PutAutoscalingPolicyAction.Request request = new PutAutoscalingPolicyAction.Request(
-            policyName,
-            new TreeSet<>(Set.of(DataTier.DATA_FROZEN)),
-            new TreeMap<>(Map.of(FrozenShardsDeciderService.NAME, settings))
-        );
-        assertAcked(client().execute(PutAutoscalingPolicyAction.INSTANCE, request).actionGet());
+    @Override
+    protected Settings.Builder addDeciderSettings(Settings.Builder builder) {
+        return builder.put(FrozenShardsDeciderService.MEMORY_PER_SHARD.getKey(), FrozenShardsDeciderService.DEFAULT_MEMORY_PER_SHARD);
     }
-
 }

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

@@ -0,0 +1,50 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.storage;
+
+import org.elasticsearch.action.admin.indices.stats.IndicesStatsRequest;
+import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
+import org.elasticsearch.cluster.ClusterInfoService;
+import org.elasticsearch.cluster.ClusterInfoServiceUtils;
+import org.elasticsearch.cluster.InternalClusterInfoService;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.xpack.autoscaling.AbstractFrozenAutoscalingIntegTestCase;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class FrozenStorageDeciderIT extends AbstractFrozenAutoscalingIntegTestCase {
+
+    public void testScale() {
+        IndicesStatsResponse statsResponse = client().admin()
+            .indices()
+            .stats(new IndicesStatsRequest().indices(restoredIndexName))
+            .actionGet();
+        final ClusterInfoService clusterInfoService = internalCluster().getCurrentMasterNodeInstance(ClusterInfoService.class);
+        ClusterInfoServiceUtils.refresh(((InternalClusterInfoService) clusterInfoService));
+        assertThat(
+            capacity().results().get("frozen").requiredCapacity().total().storage(),
+            equalTo(
+                ByteSizeValue.ofBytes(
+                    (long) (statsResponse.getPrimaries().store.totalDataSetSize().getBytes()
+                        * FrozenStorageDeciderService.DEFAULT_PERCENTAGE) / 100
+                )
+            )
+        );
+    }
+
+    @Override
+    protected String deciderName() {
+        return FrozenStorageDeciderService.NAME;
+    }
+
+    @Override
+    protected Settings.Builder addDeciderSettings(Settings.Builder builder) {
+        return builder.put(FrozenStorageDeciderService.PERCENTAGE.getKey(), FrozenStorageDeciderService.DEFAULT_PERCENTAGE);
+    }
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/Autoscaling.java

@@ -54,6 +54,7 @@ import org.elasticsearch.xpack.autoscaling.rest.RestGetAutoscalingCapacityHandle
 import org.elasticsearch.xpack.autoscaling.rest.RestGetAutoscalingPolicyHandler;
 import org.elasticsearch.xpack.autoscaling.rest.RestPutAutoscalingPolicyHandler;
 import org.elasticsearch.xpack.autoscaling.shards.FrozenShardsDeciderService;
+import org.elasticsearch.xpack.autoscaling.storage.FrozenStorageDeciderService;
 import org.elasticsearch.xpack.autoscaling.storage.ProactiveStorageDeciderService;
 import org.elasticsearch.xpack.autoscaling.storage.ReactiveStorageDeciderService;
 
@@ -170,6 +171,11 @@ public class Autoscaling extends Plugin implements ActionPlugin, ExtensiblePlugi
                 AutoscalingDeciderResult.Reason.class,
                 FrozenShardsDeciderService.NAME,
                 FrozenShardsDeciderService.FrozenShardsReason::new
+            ),
+            new NamedWriteableRegistry.Entry(
+                AutoscalingDeciderResult.Reason.class,
+                FrozenStorageDeciderService.NAME,
+                FrozenStorageDeciderService.FrozenReason::new
             )
         );
     }
@@ -201,7 +207,8 @@ public class Autoscaling extends Plugin implements ActionPlugin, ExtensiblePlugi
                 clusterService.get().getClusterSettings(),
                 allocationDeciders.get()
             ),
-            new FrozenShardsDeciderService()
+            new FrozenShardsDeciderService(),
+            new FrozenStorageDeciderService()
         );
     }
 

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/shards/FrozenShardsDeciderService.java

@@ -20,8 +20,7 @@ import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingCapacity;
 import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderContext;
 import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderResult;
 import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderService;
-import org.elasticsearch.xpack.cluster.routing.allocation.DataTierAllocationDecider;
-import org.elasticsearch.xpack.core.DataTier;
+import org.elasticsearch.xpack.autoscaling.util.FrozenUtils;
 
 import java.io.IOException;
 import java.util.List;
@@ -60,22 +59,11 @@ public class FrozenShardsDeciderService implements AutoscalingDeciderService {
 
     static int countFrozenShards(Metadata metadata) {
         return StreamSupport.stream(metadata.spliterator(), false)
-            .filter(imd -> isFrozenIndex(imd.getSettings()))
+            .filter(imd -> FrozenUtils.isFrozenIndex(imd.getSettings()))
             .mapToInt(IndexMetadata::getTotalNumberOfShards)
             .sum();
     }
 
-    static boolean isFrozenIndex(Settings indexSettings) {
-        String tierPreference = DataTierAllocationDecider.INDEX_ROUTING_PREFER_SETTING.get(indexSettings);
-        String[] preferredTiers = DataTierAllocationDecider.parseTierList(tierPreference);
-        if (preferredTiers.length >= 1 && preferredTiers[0].equals(DataTier.DATA_FROZEN)) {
-            assert preferredTiers.length == 1 : "frozen tier preference must be frozen only";
-            return true;
-        } else {
-            return false;
-        }
-    }
-
     @Override
     public List<Setting<?>> deciderSettings() {
         return List.of(MEMORY_PER_SHARD);

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/FrozenStorageDeciderService.java

@@ -0,0 +1,127 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.storage;
+
+import org.elasticsearch.cluster.ClusterInfo;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.cluster.node.DiscoveryNodeRole;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingCapacity;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderContext;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderResult;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderService;
+import org.elasticsearch.xpack.autoscaling.util.FrozenUtils;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.StreamSupport;
+
+public class FrozenStorageDeciderService implements AutoscalingDeciderService {
+    public static final String NAME = "frozen_storage";
+
+    static final double DEFAULT_PERCENTAGE = 5.0d;
+    public static final Setting<Double> PERCENTAGE = Setting.doubleSetting("percentage", DEFAULT_PERCENTAGE, 0.0d);
+
+    @Override
+    public String name() {
+        return NAME;
+    }
+
+    @Override
+    public AutoscalingDeciderResult scale(Settings configuration, AutoscalingDeciderContext context) {
+        Metadata metadata = context.state().metadata();
+        long dataSetSize = StreamSupport.stream(metadata.spliterator(), false)
+            .filter(imd -> FrozenUtils.isFrozenIndex(imd.getSettings()))
+            .mapToLong(imd -> estimateSize(imd, context.info()))
+            .sum();
+
+        long storageSize = (long) (PERCENTAGE.get(configuration) * dataSetSize) / 100;
+        return new AutoscalingDeciderResult(AutoscalingCapacity.builder().total(storageSize, null).build(), new FrozenReason(dataSetSize));
+    }
+
+    static long estimateSize(IndexMetadata imd, ClusterInfo info) {
+        int copies = imd.getNumberOfReplicas() + 1;
+        long sum = 0;
+        for (int i = 0; i < imd.getNumberOfShards(); ++i) {
+            ShardId shardId = new ShardId(imd.getIndex(), i);
+            long size = info.getShardDataSetSize(shardId).orElse(0L);
+            sum += size * copies;
+        }
+        return sum;
+    }
+
+    @Override
+    public List<Setting<?>> deciderSettings() {
+        return List.of(PERCENTAGE);
+    }
+
+    @Override
+    public List<DiscoveryNodeRole> roles() {
+        return List.of(DiscoveryNodeRole.DATA_FROZEN_NODE_ROLE);
+    }
+
+    public static class FrozenReason implements AutoscalingDeciderResult.Reason {
+        private final long totalDataSetSize;
+
+        public FrozenReason(long totalDataSetSize) {
+            assert totalDataSetSize >= 0;
+            this.totalDataSetSize = totalDataSetSize;
+        }
+
+        public FrozenReason(StreamInput in) throws IOException {
+            this.totalDataSetSize = in.readLong();
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            builder.field("total_data_set_size", totalDataSetSize);
+            builder.endObject();
+            return builder;
+        }
+
+        @Override
+        public String getWriteableName() {
+            return NAME;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            out.writeLong(totalDataSetSize);
+        }
+
+        @Override
+        public String summary() {
+            return "total data set size [" + totalDataSetSize + "]";
+        }
+
+        public long totalDataSetSize() {
+            return totalDataSetSize;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            FrozenReason that = (FrozenReason) o;
+            return totalDataSetSize == that.totalDataSetSize;
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(totalDataSetSize);
+        }
+    }
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/ReactiveStorageDeciderService.java

@@ -39,6 +39,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.snapshots.SnapshotShardSizeInfo;
 import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingCapacity;
 import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderContext;
@@ -560,7 +561,14 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
             private final ClusterInfo delegate;
 
             private ExtendedClusterInfo(ImmutableOpenMap<String, Long> extraShardSizes, ClusterInfo info) {
-                super(info.getNodeLeastAvailableDiskUsages(), info.getNodeMostAvailableDiskUsages(), extraShardSizes, null, null);
+                super(
+                    info.getNodeLeastAvailableDiskUsages(),
+                    info.getNodeMostAvailableDiskUsages(),
+                    extraShardSizes,
+                    ImmutableOpenMap.of(),
+                    null,
+                    null
+                );
                 this.delegate = info;
             }
 
@@ -584,6 +592,11 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
                 }
             }
 
+            @Override
+            public Optional<Long> getShardDataSetSize(ShardId shardId) {
+                return delegate.getShardDataSetSize(shardId);
+            }
+
             @Override
             public String getDataPath(ShardRouting shardRouting) {
                 return delegate.getDataPath(shardRouting);

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/util/FrozenUtils.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.util;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.xpack.cluster.routing.allocation.DataTierAllocationDecider;
+import org.elasticsearch.xpack.core.DataTier;
+
+public class FrozenUtils {
+    public static boolean isFrozenIndex(Settings indexSettings) {
+        String tierPreference = DataTierAllocationDecider.INDEX_ROUTING_PREFER_SETTING.get(indexSettings);
+        String[] preferredTiers = DataTierAllocationDecider.parseTierList(tierPreference);
+        if (preferredTiers.length >= 1 && preferredTiers[0].equals(DataTier.DATA_FROZEN)) {
+            assert preferredTiers.length == 1 : "frozen tier preference must be frozen only";
+            return true;
+        } else {
+            return false;
+        }
+    }
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/capacity/AutoscalingCalculateCapacityServiceTests.java

@@ -249,7 +249,7 @@ public class AutoscalingCalculateCapacityServiceTests extends AutoscalingTestCas
         state = ClusterState.builder(ClusterName.DEFAULT).nodes(nodes).build();
         ImmutableOpenMap<String, DiskUsage> leastUsages = leastUsagesBuilder.build();
         ImmutableOpenMap<String, DiskUsage> mostUsages = mostUsagesBuilder.build();
-        info = new ClusterInfo(leastUsages, mostUsages, null, null, null);
+        info = new ClusterInfo(leastUsages, mostUsages, null, null, null, null);
         context = new AutoscalingCalculateCapacityService.DefaultAutoscalingDeciderContext(roleNames, state, info, null, n -> memory);
 
         assertThat(context.nodes(), equalTo(expectedNodes));
@@ -288,7 +288,7 @@ public class AutoscalingCalculateCapacityServiceTests extends AutoscalingTestCas
                 )
             );
 
-            info = new ClusterInfo(leastUsages, mostUsagesBuilder.build(), null, null, null);
+            info = new ClusterInfo(leastUsages, mostUsagesBuilder.build(), null, null, null, null);
             context = new AutoscalingCalculateCapacityService.DefaultAutoscalingDeciderContext(roleNames, state, info, null, n -> memory);
             assertThat(context.nodes(), equalTo(expectedNodes));
             if (hasDataRole) {

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/shards/FrozenShardsDeciderServiceTests.java

@@ -7,8 +7,6 @@
 
 package org.elasticsearch.xpack.autoscaling.shards;
 
-import joptsimple.internal.Strings;
-import org.elasticsearch.Version;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
@@ -16,30 +14,20 @@ import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.index.IndexModule;
 import org.elasticsearch.xpack.autoscaling.AutoscalingTestCase;
 import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderContext;
 import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderResult;
-import org.elasticsearch.xpack.cluster.routing.allocation.DataTierAllocationDecider;
+import org.elasticsearch.xpack.autoscaling.util.FrozenUtilsTests;
 import org.elasticsearch.xpack.core.DataTier;
-import org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsConstants;
 
 import java.util.Objects;
 
 import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.is;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 public class FrozenShardsDeciderServiceTests extends AutoscalingTestCase {
 
-    public void testIsFrozenIndex() {
-        assertThat(FrozenShardsDeciderService.isFrozenIndex(indexSettings(DataTier.DATA_FROZEN)), is(true));
-        assertThat(FrozenShardsDeciderService.isFrozenIndex(indexSettings(null)), is(false));
-        String notFrozenAlone = randomNonFrozenTierPreference();
-        assertThat(FrozenShardsDeciderService.isFrozenIndex(indexSettings(notFrozenAlone)), is(false));
-    }
-
     public void testCountFrozenShards() {
         final Metadata.Builder builder = Metadata.builder();
         int count = 0;
@@ -95,22 +83,10 @@ public class FrozenShardsDeciderServiceTests extends AutoscalingTestCase {
     }
 
     private String randomNonFrozenTierPreference() {
-        return randomValueOtherThanMany(
-            tiers -> tiers.contains(DataTier.DATA_FROZEN),
-            () -> Strings.join(randomSubsetOf(DataTier.ALL_DATA_TIERS), ",")
-        );
+        return FrozenUtilsTests.randomNonFrozenTierPreference();
     }
 
-    private Settings indexSettings(String tierPreference) {
-        Settings.Builder settings = Settings.builder()
-            .put(randomAlphaOfLength(10), randomLong())
-            .put(DataTierAllocationDecider.INDEX_ROUTING_PREFER, tierPreference)
-            .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT);
-        // pass setting validator.
-        if (Objects.equals(tierPreference, DataTier.DATA_FROZEN)) {
-            settings.put(SearchableSnapshotsConstants.SNAPSHOT_PARTIAL_SETTING.getKey(), true)
-                .put(IndexModule.INDEX_STORE_TYPE_SETTING.getKey(), SearchableSnapshotsConstants.SNAPSHOT_DIRECTORY_FACTORY_KEY);
-        }
-        return settings.build();
+    private static Settings indexSettings(String tierPreference) {
+        return FrozenUtilsTests.indexSettings(tierPreference);
     }
 }

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/storage/FrozenStorageDeciderReasonWireSerializationTests.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.storage;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+
+public class FrozenStorageDeciderReasonWireSerializationTests extends AbstractWireSerializingTestCase<
+    FrozenStorageDeciderService.FrozenReason> {
+    @Override
+    protected Writeable.Reader<FrozenStorageDeciderService.FrozenReason> instanceReader() {
+        return FrozenStorageDeciderService.FrozenReason::new;
+    }
+
+    @Override
+    protected FrozenStorageDeciderService.FrozenReason mutateInstance(FrozenStorageDeciderService.FrozenReason instance) {
+        return new FrozenStorageDeciderService.FrozenReason(
+            randomValueOtherThan(instance.totalDataSetSize(), () -> randomLongBetween(0, Long.MAX_VALUE))
+        );
+    }
+
+    @Override
+    protected FrozenStorageDeciderService.FrozenReason createTestInstance() {
+        return new FrozenStorageDeciderService.FrozenReason(randomLongBetween(0, Long.MAX_VALUE));
+    }
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/storage/FrozenStorageDeciderServiceTests.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.storage;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.ClusterInfo;
+import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.metadata.Metadata;
+import org.elasticsearch.common.collect.ImmutableOpenMap;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.index.Index;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.xpack.autoscaling.AutoscalingTestCase;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderContext;
+import org.elasticsearch.xpack.autoscaling.capacity.AutoscalingDeciderResult;
+import org.elasticsearch.xpack.autoscaling.util.FrozenUtilsTests;
+import org.elasticsearch.xpack.core.DataTier;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.nullValue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class FrozenStorageDeciderServiceTests extends AutoscalingTestCase {
+
+    public void testEstimateSize() {
+        final int shards = between(1, 10);
+        final int replicas = between(0, 9);
+        final IndexMetadata indexMetadata = IndexMetadata.builder(randomAlphaOfLength(5))
+            .settings(Settings.builder().put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT))
+            .numberOfShards(shards)
+            .numberOfReplicas(replicas)
+            .build();
+        final Tuple<Long, ClusterInfo> sizeAndClusterInfo = sizeAndClusterInfo(indexMetadata);
+        final long expected = sizeAndClusterInfo.v1();
+        final ClusterInfo info = sizeAndClusterInfo.v2();
+        assertThat(FrozenStorageDeciderService.estimateSize(indexMetadata, info), equalTo(expected));
+        assertThat(FrozenStorageDeciderService.estimateSize(indexMetadata, ClusterInfo.EMPTY), equalTo(0L));
+    }
+
+    public void testScale() {
+        FrozenStorageDeciderService service = new FrozenStorageDeciderService();
+
+        int shards = between(1, 3);
+        int replicas = between(0, 2);
+        Metadata metadata = Metadata.builder()
+            .put(
+                IndexMetadata.builder("index")
+                    .settings(FrozenUtilsTests.indexSettings(DataTier.DATA_FROZEN))
+                    .numberOfShards(shards)
+                    .numberOfReplicas(replicas)
+            )
+            .build();
+        ClusterState state = ClusterState.builder(ClusterName.DEFAULT).metadata(metadata).build();
+        AutoscalingDeciderContext context = mock(AutoscalingDeciderContext.class);
+        when(context.state()).thenReturn(state);
+        final Tuple<Long, ClusterInfo> sizeAndClusterInfo = sizeAndClusterInfo(metadata.index("index"));
+        final long dataSetSize = sizeAndClusterInfo.v1();
+        final ClusterInfo info = sizeAndClusterInfo.v2();
+        when(context.info()).thenReturn(info);
+
+        AutoscalingDeciderResult defaultSettingsResult = service.scale(Settings.EMPTY, context);
+        assertThat(
+            defaultSettingsResult.requiredCapacity().total().storage(),
+            equalTo(ByteSizeValue.ofBytes((long) (FrozenStorageDeciderService.DEFAULT_PERCENTAGE * dataSetSize) / 100))
+        );
+        assertThat(defaultSettingsResult.requiredCapacity().total().memory(), nullValue());
+        assertThat(defaultSettingsResult.reason().summary(), equalTo("total data set size [" + dataSetSize + "]"));
+
+        // the percentage is not the cache size, rather the node size. So someone could want 101% just as much as 100% so we do not have
+        // a real upper bound. Therefore testing to 200%.
+        double percentage = randomDoubleBetween(0.0d, 200.0d, true);
+        AutoscalingDeciderResult overrideSettingsResult = service.scale(
+            Settings.builder().put(FrozenStorageDeciderService.PERCENTAGE.getKey(), percentage).build(),
+            context
+        );
+        assertThat(
+            overrideSettingsResult.requiredCapacity().total().storage(),
+            equalTo(ByteSizeValue.ofBytes((long) (percentage * dataSetSize) / 100))
+        );
+    }
+
+    public Tuple<Long, ClusterInfo> sizeAndClusterInfo(IndexMetadata indexMetadata) {
+        long totalSize = 0;
+        ImmutableOpenMap.Builder<ShardId, Long> sizesBuilder = ImmutableOpenMap.builder();
+        Index index = indexMetadata.getIndex();
+        Index otherIndex = randomValueOtherThan(index, () -> new Index(randomAlphaOfLength(5), randomAlphaOfLength(5)));
+        int shards = indexMetadata.getNumberOfShards();
+        int replicas = indexMetadata.getNumberOfReplicas();
+        for (int i = 0; i < shards; ++i) {
+            long size = randomLongBetween(0, Integer.MAX_VALUE);
+            totalSize += size * (replicas + 1);
+            sizesBuilder.put(new ShardId(index, i), size);
+            // add other index shards.
+            sizesBuilder.put(new ShardId(otherIndex, i), randomLongBetween(0, Integer.MAX_VALUE));
+        }
+        for (int i = shards; i < shards + between(0, 3); ++i) {
+            // add irrelevant shards noise for completeness (should not happen IRL).
+            sizesBuilder.put(new ShardId(index, i), randomLongBetween(0, Integer.MAX_VALUE));
+        }
+        ClusterInfo info = new ClusterInfo(null, null, null, sizesBuilder.build(), null, null);
+        return Tuple.tuple(totalSize, info);
+    }
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/storage/ProactiveStorageDeciderServiceTests.java

@@ -355,7 +355,7 @@ public class ProactiveStorageDeciderServiceTests extends AutoscalingTestCase {
     }
 
     private ClusterInfo randomClusterInfo(ClusterState state) {
-        Map<String, Long> collect = state.routingTable()
+        Map<String, Long> shardSizes = state.routingTable()
             .allShards()
             .stream()
             .map(ClusterInfo::shardIdentifierFromRouting)
@@ -369,7 +369,8 @@ public class ProactiveStorageDeciderServiceTests extends AutoscalingTestCase {
         return new ClusterInfo(
             diskUsage,
             diskUsage,
-            ImmutableOpenMap.<String, Long>builder().putAll(collect).build(),
+            ImmutableOpenMap.<String, Long>builder().putAll(shardSizes).build(),
+            ImmutableOpenMap.of(),
             ImmutableOpenMap.of(),
             ImmutableOpenMap.of()
         );

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/storage/ReactiveStorageDeciderServiceTests.java

@@ -179,7 +179,7 @@ public class ReactiveStorageDeciderServiceTests extends AutoscalingTestCase {
         ImmutableOpenMap.Builder<String, Long> shardSizeBuilder,
         long expected
     ) {
-        ClusterInfo info = new ClusterInfo(null, null, shardSizeBuilder.build(), null, null);
+        ClusterInfo info = new ClusterInfo(null, null, shardSizeBuilder.build(), null, null, null);
         ReactiveStorageDeciderService.AllocationState allocationState = new ReactiveStorageDeciderService.AllocationState(
             clusterState,
             null,
@@ -343,7 +343,7 @@ public class ReactiveStorageDeciderServiceTests extends AutoscalingTestCase {
         if (shardsWithSizes.isEmpty() == false) {
             shardSizeBuilder.put(shardIdentifier(randomFrom(shardsWithSizes)), ByteSizeUnit.KB.toBytes(minShardSize));
         }
-        ClusterInfo info = new ClusterInfo(diskUsages, diskUsages, shardSizeBuilder.build(), null, null);
+        ClusterInfo info = new ClusterInfo(diskUsages, diskUsages, shardSizeBuilder.build(), null, null, null);
 
         ReactiveStorageDeciderService.AllocationState allocationState = new ReactiveStorageDeciderService.AllocationState(
             clusterState,

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/util/FrozenUtilsTests.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.util;
+
+import joptsimple.internal.Strings;
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.IndexModule;
+import org.elasticsearch.xpack.autoscaling.AutoscalingTestCase;
+import org.elasticsearch.xpack.cluster.routing.allocation.DataTierAllocationDecider;
+import org.elasticsearch.xpack.core.DataTier;
+import org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsConstants;
+
+import java.util.Objects;
+
+import static org.hamcrest.Matchers.is;
+
+public class FrozenUtilsTests extends AutoscalingTestCase {
+
+    public void testIsFrozenIndex() {
+        assertThat(FrozenUtils.isFrozenIndex(indexSettings(DataTier.DATA_FROZEN)), is(true));
+        assertThat(FrozenUtils.isFrozenIndex(indexSettings(null)), is(false));
+        String notFrozenAlone = randomNonFrozenTierPreference();
+        assertThat(FrozenUtils.isFrozenIndex(indexSettings(notFrozenAlone)), is(false));
+    }
+
+    public static String randomNonFrozenTierPreference() {
+        return randomValueOtherThanMany(
+            tiers -> tiers.contains(DataTier.DATA_FROZEN),
+            () -> Strings.join(randomSubsetOf(DataTier.ALL_DATA_TIERS), ",")
+        );
+    }
+
+    public static Settings indexSettings(String tierPreference) {
+        Settings.Builder settings = Settings.builder()
+            .put(randomAlphaOfLength(10), randomLong())
+            .put(DataTierAllocationDecider.INDEX_ROUTING_PREFER, tierPreference)
+            .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT);
+        // pass setting validator.
+        if (Objects.equals(tierPreference, DataTier.DATA_FROZEN)) {
+            settings.put(SearchableSnapshotsConstants.SNAPSHOT_PARTIAL_SETTING.getKey(), true)
+                .put(IndexModule.INDEX_STORE_TYPE_SETTING.getKey(), SearchableSnapshotsConstants.SNAPSHOT_DIRECTORY_FACTORY_KEY);
+        }
+        return settings.build();
+    }
+}