feat: Add new tier/used/capacity metric to coordinator that reports actual loaded bytes across a tier

docs/operations/metrics.md

@@ -469,8 +469,11 @@ These metrics are emitted by the Druid Coordinator in every run of the correspon
 |`segment/availableDeepStorageOnly/count`|Number of unique segments that are only available for querying directly from deep storage.|`dataSource`|Varies|
 |`tier/historical/count`|Number of available historical nodes in each tier. The `tierAlias` dimension is emitted only when the tier belongs to an alias configured via [`historicalTierAliases`](../configuration/index.md#dynamic-configuration), and can be used to aggregate metrics across the tiers in an alias.|`tier`, `tierAlias`|Varies|
 |`tier/replication/factor`|Configured maximum replication factor in each tier. The `tierAlias` dimension is emitted only when the tier belongs to an alias configured via [`historicalTierAliases`](../configuration/index.md#dynamic-configuration).|`tier`, `tierAlias`|Varies|
-|`tier/required/capacity`|Total capacity in bytes required in each tier. The `tierAlias` dimension is emitted only when the tier belongs to an alias configured via [`historicalTierAliases`](../configuration/index.md#dynamic-configuration).|`tier`, `tierAlias`|Varies|
-|`tier/total/capacity`|Total capacity in bytes available in each tier. The `tierAlias` dimension is emitted only when the tier belongs to an alias configured via [`historicalTierAliases`](../configuration/index.md#dynamic-configuration).|`tier`, `tierAlias`|Varies|
+|`tier/required/capacity`|Total capacity in bytes that the load rules require in each tier, counting each replica at its full segment size. Compare against `tier/assignable/capacity` to see how heavily the tier is subscribed. The `tierAlias` dimension is emitted only when the tier belongs to an alias configured via [`historicalTierAliases`](../configuration/index.md#dynamic-configuration).|`tier`, `tierAlias`|Varies|
+|`tier/assignable/capacity`|Total capacity in bytes that each tier advertises for segment assignment. Can be greater than the physical disk reported by `tier/storage/capacity` if using virtual storage, so use that metric rather than this one to monitor physical capacity. The `tierAlias` dimension is emitted only when the tier belongs to an alias configured via [`historicalTierAliases`](../configuration/index.md#dynamic-configuration).|`tier`, `tierAlias`|Varies|
+|`tier/storage/capacity`|Total physical size in bytes of the segment cache locations configured on the historicals in each tier. The `tierAlias` dimension is emitted only when the tier belongs to an alias configured via [`historicalTierAliases`](../configuration/index.md#dynamic-configuration).|`tier`, `tierAlias`|Varies|
+|`tier/storage/used`|Bytes occupied on disk in each tier by the segments its historicals have announced. Partially loaded segments count only the bytes the historical reported loading, not their full size. Compare against `tier/storage/capacity` to see how full the tier's disks are. The `tierAlias` dimension is emitted only when the tier belongs to an alias configured via [`historicalTierAliases`](../configuration/index.md#dynamic-configuration).|`tier`, `tierAlias`|Varies|
+|`tier/total/capacity`|Deprecated. Use `tier/assignable/capacity` instead, which reports the same value under a name that does not imply physical disk size. This metric will be removed in a future release.|`tier`, `tierAlias`|Varies|
 |`compact/task/count`|Number of tasks issued in the auto compaction run.| |Varies|
 |`compactTask/maxSlot/count`|Maximum number of task slots available for auto compaction tasks in the auto compaction run.| |Varies|
 |`compactTask/availableSlot/count`|Number of currently vacant task slots out of the total slots allocated for auto compaction tasks. This value is computed as the difference between the total number of task slots allocated for auto compaction and the estimated number of task slots currently occupied by running compaction tasks. The number of sub-tasks of each compaction task is estimated to be `maxNumConcurrentSubTasks`.| |Varies|

embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryVirtualStorageTest.java

@@ -300,10 +300,10 @@ void testQueryPartials()
         CACHE_SIZE,
         coordinatorEmitter.getLatestMetricEventValue(Stats.Tier.STORAGE_CAPACITY.getMetricName())
     );
-    coordinatorEmitter.waitForEvent(event -> event.hasMetricName(Stats.Tier.TOTAL_CAPACITY.getMetricName()));
+    coordinatorEmitter.waitForEvent(event -> event.hasMetricName(Stats.Tier.ASSIGNABLE_CAPACITY.getMetricName()));
     Assertions.assertEquals(
         MAX_SIZE,
-        coordinatorEmitter.getLatestMetricEventValue(Stats.Tier.TOTAL_CAPACITY.getMetricName())
+        coordinatorEmitter.getLatestMetricEventValue(Stats.Tier.ASSIGNABLE_CAPACITY.getMetricName())
     );
   }
 

extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json

@@ -190,6 +190,9 @@
   "tier/replication/factor" : { "dimensions" : ["tier", "tierAlias"], "type" : "count", "help": "Configured maximum replication factor in each tier."},
   "tier/required/capacity" : { "dimensions" : ["tier", "tierAlias"], "type" : "count", "help": "Total capacity in bytes required in each tier."},
   "tier/total/capacity" : { "dimensions" : ["tier", "tierAlias"], "type" : "count", "help": "Total capacity in bytes available in each tier."},
+  "tier/assignable/capacity": { "dimensions" : ["tier", "tierAlias"], "type" : "count", "help": "Total capacity in bytes available for assignment in each tier."},
+  "tier/storage/capacity": { "dimensions" : ["tier", "tierAlias"], "type" : "count", "help": "Total physical capacity in bytes available in each tier."},
+  "tier/storage/used": { "dimensions" : ["tier", "tierAlias"], "type" : "count", "help": "Total bytes used in each tier."},
   "compact/task/count" : { "dimensions" : [], "type" : "gauge", "help": "Number of tasks issued in the auto compaction run."},
   "compactTask/maxSlot/count" : { "dimensions" : [], "type" : "gauge", "help": "Maximum number of task slots available for auto compaction tasks in the auto compaction run."},
   "compactTask/availableSlot/count" : { "dimensions" : [], "type" : "gauge", "help": "Number of available task slots that can be used for auto compaction tasks in the auto compaction run."},

extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json

@@ -185,6 +185,9 @@
   "tier/required/capacity" : { "dimensions" : ["tier", "tierAlias"], "type" : "gauge" },
   "tier/total/capacity" : { "dimensions" : ["tier", "tierAlias"], "type" : "gauge" },
   "tier/replication/factor" : { "dimensions" : ["tier", "tierAlias"], "type" : "gauge" },
+  "tier/assignable/capacity": { "dimensions" : ["tier", "tierAlias"], "type" : "gauge"},
+  "tier/storage/capacity": { "dimensions" : ["tier", "tierAlias"], "type" : "gauge"},
+  "tier/storage/used": { "dimensions" : ["tier", "tierAlias"], "type" : "gauge"},
   "tier/historical/count" : { "dimensions" : ["tier", "tierAlias"], "type" : "count" },
 
   "compact/task/count" : { "dimensions" : [], "type" : "count" },

server/src/main/java/org/apache/druid/server/coordinator/duty/PrepareBalancerAndLoadQueues.java

@@ -184,20 +184,27 @@ private void collectHistoricalStats(
                             : RowKey.with(Dimension.TIER, tier).and(Dimension.TIER_ALIAS, alias);
       stats.add(Stats.Tier.HISTORICAL_COUNT, rowKey, historicals.size());
 
-      long totalCapacity = 0;
+      long assignableCapacity = 0;
       long storageCapacity = 0;
+      long usedStorage = 0;
       long cloneCount = 0;
       for (ServerHolder holder : historicals) {
         if (holder.isUnmanaged()) {
           cloneCount += 1;
         } else {
-          totalCapacity += holder.getMaxSize();
+          assignableCapacity += holder.getMaxSize();
           storageCapacity += holder.getStorageSize();
+          // getCurrSize(), not getSizeUsed(): queued loads are counted at full segment size, which would overstate
+          // disk usage on a tier serving partial loads for the duration of every load
+          usedStorage += holder.getServer().getCurrSize();
         }
       }
       stats.add(Stats.Tier.CLONE_COUNT, rowKey, cloneCount);
-      stats.add(Stats.Tier.TOTAL_CAPACITY, rowKey, totalCapacity);
+      stats.add(Stats.Tier.ASSIGNABLE_CAPACITY, rowKey, assignableCapacity);
+      // Deprecated alias of tier/assignable/capacity, emitted until the deprecation period is over
+      stats.add(Stats.Tier.TOTAL_CAPACITY, rowKey, assignableCapacity);
       stats.add(Stats.Tier.STORAGE_CAPACITY, rowKey, storageCapacity);
+      stats.add(Stats.Tier.USED_STORAGE, rowKey, usedStorage);
     });
   }
 

server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java

@@ -106,12 +106,34 @@ public static class SegmentQueue
 
   public static class Tier
   {
+    /**
+     * The demand the rules place on the tier to full load all segments.
+     */
     public static final CoordinatorStat REQUIRED_CAPACITY
         = CoordinatorStat.toDebugAndEmit("reqdCap", "tier/required/capacity");
+    /**
+     * The total assignable capacity of the tier defined by summing the configured max size of all servers in the tier.
+     */
+    public static final CoordinatorStat ASSIGNABLE_CAPACITY
+        = CoordinatorStat.toDebugAndEmit("assignableCap", "tier/assignable/capacity");
+    /**
+     * Superseded by ASSIGNABLE_CAPACITY, which carries the same value under a name that says it is an assignment
+     * budget rather than physical disk. Still emitted so existing dashboards keep working; remove after deprecation.
+     */
+    @Deprecated
     public static final CoordinatorStat TOTAL_CAPACITY
         = CoordinatorStat.toDebugAndEmit("totalCap", "tier/total/capacity");
+    /**
+     * Aggregate physical disk capacity of the tier.
+     */
     public static final CoordinatorStat STORAGE_CAPACITY
         = CoordinatorStat.toDebugAndEmit("storageCap", "tier/storage/capacity");
+    /**
+     * Actual physical disk consumed by the tier. Only reports realized bytes on disk, so partial loads when using
+     * virtual storage report their true loaded size, which can be less than the full segment size.
+     */
+    public static final CoordinatorStat USED_STORAGE
+        = CoordinatorStat.toDebugAndEmit("usedStorage", "tier/storage/used");
     public static final CoordinatorStat REPLICATION_FACTOR
         = CoordinatorStat.toDebugAndEmit("maxRepFactor", "tier/replication/factor");
     public static final CoordinatorStat HISTORICAL_COUNT

server/src/test/java/org/apache/druid/server/coordinator/simulate/HistoricalTierAliasTest.java

@@ -38,6 +38,7 @@
 public class HistoricalTierAliasTest extends CoordinatorSimulationBaseTest
 {
   private static final long SIZE_1TB = 1_000_000;
+  private static final long SEGMENT_SIZE = 500_000_000;
   private static final String ALIAS = "hot";
 
   private DruidServer historicalT1;
@@ -79,18 +80,22 @@ public void testCapacityMetricsAreTaggedWithAlias()
 
     final long expectedCapacity = SIZE_1TB << 20;
 
-    // tier/total/capacity is emitted per physical tier AND tagged with the alias
+    // tier/assignable/capacity is emitted per physical tier AND tagged with the alias
     verifyValue(
-        Stats.Tier.TOTAL_CAPACITY.getMetricName(),
-        Map.of(Dimension.TIER.reportedName(), Tier.T1, Dimension.TIER_ALIAS.reportedName(), ALIAS),
+        Stats.Tier.ASSIGNABLE_CAPACITY.getMetricName(),
+        aliasedTier(Tier.T1),
         expectedCapacity
     );
     verifyValue(
-        Stats.Tier.TOTAL_CAPACITY.getMetricName(),
-        Map.of(Dimension.TIER.reportedName(), Tier.T2, Dimension.TIER_ALIAS.reportedName(), ALIAS),
+        Stats.Tier.ASSIGNABLE_CAPACITY.getMetricName(),
+        aliasedTier(Tier.T2),
         expectedCapacity
     );
 
+    // The deprecated tier/total/capacity carries the same value while it is still emitted
+    verifyValue(Stats.Tier.TOTAL_CAPACITY.getMetricName(), aliasedTier(Tier.T1), expectedCapacity);
+    verifyValue(Stats.Tier.TOTAL_CAPACITY.getMetricName(), aliasedTier(Tier.T2), expectedCapacity);
+
     // tier/historical/count carries the alias too, so it can be summed across the pair
     verifyValue(
         Stats.Tier.HISTORICAL_COUNT.getMetricName(),
@@ -104,6 +109,43 @@ public void testCapacityMetricsAreTaggedWithAlias()
     );
   }
 
+  @Test
+  public void testUsedStorageReportsOnlyBytesLoadedAndAnnounced()
+  {
+    // tier/storage/used does not report bytes consumed until historicals have actually loaded and announced the segments. It only reports the actual realized bytes of loaded segments.
+    final CoordinatorSimulation sim =
+        CoordinatorSimulation.builder()
+                             .withSegments(Segments.WIKI_10X1D)
+                             .withServers(historicalT1, historicalT2)
+                             .withRules(datasource, Load.on(ALIAS, 1).forever())
+                             .withDynamicConfig(
+                                 CoordinatorDynamicConfig.builder()
+                                                         .withHistoricalTierAliases(
+                                                             Map.of(ALIAS, Set.of(Tier.T1, Tier.T2))
+                                                         )
+                                                         .withSmartSegmentLoading(true)
+                                                         .build()
+                             )
+                             .withImmediateSegmentLoading(true)
+                             .build();
+
+    startSimulation(sim);
+
+    runCoordinatorCycle();
+    verifyValue(Stats.Tier.USED_STORAGE.getMetricName(), aliasedTier(Tier.T1), 0L);
+    verifyValue(Stats.Tier.USED_STORAGE.getMetricName(), aliasedTier(Tier.T2), 0L);
+
+    runCoordinatorCycle();
+    final long loadedBytes = 10 * SEGMENT_SIZE;
+    verifyValue(Stats.Tier.USED_STORAGE.getMetricName(), aliasedTier(Tier.T1), loadedBytes);
+    verifyValue(Stats.Tier.USED_STORAGE.getMetricName(), aliasedTier(Tier.T2), loadedBytes);
+  }
+
+  private static Map<String, Object> aliasedTier(String tier)
+  {
+    return Map.of(Dimension.TIER.reportedName(), tier, Dimension.TIER_ALIAS.reportedName(), ALIAS);
+  }
+
   @Test
   public void testCapacityMetricsHaveNoAliasWhenNotConfigured()
   {
@@ -127,12 +169,12 @@ public void testCapacityMetricsHaveNoAliasWhenNotConfigured()
 
     // Without an alias configured, capacity is reported against the physical tier only
     verifyValue(
-        Stats.Tier.TOTAL_CAPACITY.getMetricName(),
+        Stats.Tier.ASSIGNABLE_CAPACITY.getMetricName(),
         filterByTier(Tier.T1),
         expectedCapacity
     );
     verifyValue(
-        Stats.Tier.TOTAL_CAPACITY.getMetricName(),
+        Stats.Tier.ASSIGNABLE_CAPACITY.getMetricName(),
         filterByTier(Tier.T2),
         expectedCapacity
     );