enhance: add cache shard disk usage metric

include/cachinglayer/CacheSlot.h

@@ -26,6 +26,7 @@
 #include <memory>
 #include <numeric>
 #include <optional>
+#include <string>
 #include <type_traits>
 #include <utility>
 #include <vector>
@@ -68,6 +69,11 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
           storage_usage_tracking_enabled_(storage_usage_tracking_enabled),
           loading_timeout_(loading_timeout),
           warmup_loading_timeout_(warmup_loading_timeout) {
+        if (const auto& metric_attribution = translator_->meta()->metric_attribution;
+            metric_attribution && !metric_attribution->shard.empty()) {
+            shard_disk_usage_metric_ =
+                monitor::create_cache_shard_disk_usage_metric_handle(cell_data_type_, metric_attribution->shard);
+        }
         cells_.reserve(translator_->num_cells());
         for (cid_t i = 0; i < static_cast<cid_t>(translator_->num_cells()); ++i) {
             cells_.push_back(std::make_unique<CacheCell>(this, i));
@@ -636,6 +642,9 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
                         .Increment(loaded_size_.memory_bytes);
                     monitor::cache_loaded_bytes(slot_->cell_data_type_, StorageType::DISK)
                         .Increment(loaded_size_.file_bytes);
+                    if (slot_->shard_disk_usage_metric_ != nullptr && loaded_size_.file_bytes > 0) {
+                        slot_->shard_disk_usage_metric_->Increment(loaded_size_.file_bytes);
+                    }
                     monitor::cache_cell_loaded_count(slot_->cell_data_type_, slot_->storage_type_).Increment();
                 },
                 requesting_thread);
@@ -672,6 +681,9 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
                     .Decrement(loaded_size_.memory_bytes);
                 monitor::cache_loaded_bytes(slot_->cell_data_type_, StorageType::DISK)
                     .Decrement(loaded_size_.file_bytes);
+                if (slot_->shard_disk_usage_metric_ != nullptr && loaded_size_.file_bytes > 0) {
+                    slot_->shard_disk_usage_metric_->Decrement(loaded_size_.file_bytes);
+                }
                 LOG_TRACE("[MCL] CacheSlot Cell unloaded: key={}, size={}", key(), loaded_size_.ToString());
                 loaded_size_ = {0, 0};  // reset loaded_size_ to 0,0 to avoid double refund from dlist_
             }
@@ -691,6 +703,7 @@ class CacheSlot final : public std::enable_shared_from_this<CacheSlot<CellT>> {
     };
 
     const std::unique_ptr<Translator<CellT>> translator_;
+    std::unique_ptr<monitor::CacheShardDiskUsageMetricHandle> shard_disk_usage_metric_;
     folly::CancellationSource warmup_cancel_source_;
     // Each CacheCell's cid_t is its index in vector.
     // Using unique_ptr because CacheCell is non-movable (inherits from ListNode).

include/cachinglayer/Metrics.h

@@ -1,5 +1,10 @@
 #pragma once
 
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
 #include "cachinglayer/Utils.h"
 #include "common/PrometheusClient.h"
 
@@ -240,6 +245,7 @@ DECLARE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_loade
 DECLARE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_loading_bytes);
 DECLARE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_cell_loading_count);
 DECLARE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_cell_loaded_count);
+DECLARE_PROMETHEUS_GAUGE_FAMILY(internal_cache_shard_disk_usage_bytes);
 
 /* Metrics for Cache Cell Access */
 DECLARE_PROMETHEUS_COUNTER_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_access_event_total);
@@ -265,6 +271,57 @@ DEFINE_METRIC_HELPER_WITH_DATA_TYPE_AND_LOCATION(prometheus::Gauge, cache_cell_l
 DEFINE_METRIC_HELPER_WITH_DATA_TYPE_AND_LOCATION(prometheus::Gauge, cache_loaded_bytes);
 DEFINE_METRIC_HELPER_WITH_DATA_TYPE_AND_LOCATION(prometheus::Gauge, cache_cell_loaded_count);
 
+struct CacheShardDiskUsageStats {
+    CellDataType cell_data_type;
+    std::string shard;
+    double disk_bytes;
+};
+
+class CacheShardDiskUsageMetricEntry;
+
+// RAII handle for one {data_type, shard} cache-slot disk usage series.
+// The collector removes the dynamic Prometheus time series after the last handle is gone.
+class CacheShardDiskUsageMetricHandle {
+ public:
+    ~CacheShardDiskUsageMetricHandle();
+
+    CacheShardDiskUsageMetricHandle(const CacheShardDiskUsageMetricHandle&) = delete;
+    CacheShardDiskUsageMetricHandle&
+    operator=(const CacheShardDiskUsageMetricHandle&) = delete;
+    CacheShardDiskUsageMetricHandle(CacheShardDiskUsageMetricHandle&&) = delete;
+    CacheShardDiskUsageMetricHandle&
+    operator=(CacheShardDiskUsageMetricHandle&&) = delete;
+
+    void
+    Increment(double value);
+
+    void
+    Decrement(double value);
+
+    [[nodiscard]] double
+    Value() const;
+
+ private:
+    friend std::unique_ptr<CacheShardDiskUsageMetricHandle>
+    create_cache_shard_disk_usage_metric_handle(CellDataType type, const std::string& shard);
+
+    explicit CacheShardDiskUsageMetricHandle(std::shared_ptr<CacheShardDiskUsageMetricEntry> entry);
+
+    std::shared_ptr<CacheShardDiskUsageMetricEntry> entry_;
+};
+
+// Returns nullptr for an empty shard, leaving the slot unattributed.
+std::unique_ptr<CacheShardDiskUsageMetricHandle>
+create_cache_shard_disk_usage_metric_handle(CellDataType type, const std::string& shard);
+
+// Returns live shard disk usage stats and lazily removes expired metric series.
+std::vector<CacheShardDiskUsageStats>
+collect_cache_shard_disk_usage_stats();
+
+// Returns nullopt without creating a series.
+std::optional<double>
+cache_shard_disk_usage_bytes_value(CellDataType type, const std::string& shard);
+
 DEFINE_METRIC_HELPER_WITH_DATA_TYPE_AND_LOCATION(prometheus::Counter, cache_access_event_total);
 // ignore cache_access_cells_total since we can parse it from the sum of cache_access_hit/miss_bytes_total
 DEFINE_METRIC_HELPER_WITH_DATA_TYPE_AND_LOCATION(prometheus::Counter, cache_cell_access_hit_bytes_total);

include/cachinglayer/Translator.h

@@ -13,6 +13,7 @@
 
 #include <memory>
 #include <optional>
+#include <string>
 #include <utility>
 #include <vector>
 
@@ -27,6 +28,14 @@ struct LoadingOverheadConfig {
     std::string group;
 };
 
+struct MetricAttribution {
+    // Optional stable shard/channel label for attributed cache-slot disk usage metrics.
+    // Keep this value bounded in cardinality, such as loaded shards/channels, not request/user IDs.
+    // The shard disk usage collector removes the time series after the last slot for this label is gone.
+    // Empty means this translator is unattributed and no shard metric is emitted.
+    std::string shard;
+};
+
 struct Meta {
     // This storage type is currently used only by metrics to distinguish the slot type.
     // In actual resource reservation, we use the actual size of the cell to determine the type.
@@ -43,15 +52,18 @@ struct Meta {
     // loads happen. The real resource usage is bounded by loading_pool_size * cell_size.
     // If not set, no capping is applied (existing behavior).
     std::optional<LoadingOverheadConfig> loading_overhead;
+    std::optional<MetricAttribution> metric_attribution;
     explicit Meta(StorageType storage_type, CellIdMappingMode cell_id_mapping_mode, CellDataType cell_data_type,
                   CacheWarmupPolicy cache_warmup_policy, bool support_eviction,
-                  std::optional<LoadingOverheadConfig> loading_overhead = std::nullopt)
+                  std::optional<LoadingOverheadConfig> loading_overhead = std::nullopt,
+                  std::optional<MetricAttribution> metric_attribution = std::nullopt)
         : storage_type(storage_type),
           cell_id_mapping_mode(cell_id_mapping_mode),
           cell_data_type(cell_data_type),
           cache_warmup_policy(cache_warmup_policy),
           support_eviction(support_eviction),
-          loading_overhead(std::move(loading_overhead)) {
+          loading_overhead(std::move(loading_overhead)),
+          metric_attribution(std::move(metric_attribution)) {
     }
 };
 

src/cachinglayer/Metrics.cpp

@@ -1,5 +1,13 @@
 #include "cachinglayer/Metrics.h"
 
+#include <map>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
 namespace milvus::cachinglayer::monitor {
 
 DEFINE_LABEL_MAP_WITH_DATA_TYPE_AND_LOCATION(vector_field, memory);
@@ -34,6 +42,8 @@ DEFINE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_cell_l
                                                            "[cpp]cache cell loading count");
 DEFINE_PROMETHEUS_GAUGE_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_cell_loaded_count,
                                                            "[cpp]cache cell loaded count");
+DEFINE_PROMETHEUS_GAUGE_FAMILY(internal_cache_shard_disk_usage_bytes,
+                               "[cpp]attributed cache-slot loaded disk usage bytes by shard");
 
 /* Metrics for Cache Cell Access */
 DEFINE_PROMETHEUS_COUNTER_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_access_event_total,
@@ -57,4 +67,153 @@ DEFINE_PROMETHEUS_HISTOGRAM_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_ce
                                                                milvus::monitor::secondsBuckets,
                                                                "[cpp]cache cell lifetime seconds");
 
+namespace {
+
+struct ShardDiskUsageMetricKey {
+    CellDataType cell_data_type;
+    std::string shard;
+
+    bool
+    operator<(const ShardDiskUsageMetricKey& other) const {
+        if (cell_data_type != other.cell_data_type) {
+            return static_cast<int>(cell_data_type) < static_cast<int>(other.cell_data_type);
+        }
+        return shard < other.shard;
+    }
+};
+
+struct ShardDiskUsageMetricValue {
+    std::weak_ptr<CacheShardDiskUsageMetricEntry> entry;
+    prometheus::Gauge* gauge{nullptr};
+};
+
+std::mutex shard_disk_usage_mutex;
+std::map<ShardDiskUsageMetricKey, ShardDiskUsageMetricValue> shard_disk_usage_metrics;
+
+const char*
+CellDataTypeLabel(CellDataType type) {
+    switch (type) {
+        case CellDataType::VECTOR_FIELD:
+            return "vector_field";
+        case CellDataType::VECTOR_INDEX:
+            return "vector_index";
+        case CellDataType::SCALAR_FIELD:
+            return "scalar_field";
+        case CellDataType::SCALAR_INDEX:
+            return "scalar_index";
+        case CellDataType::OTHER:
+            return "other";
+    }
+    ThrowInfo(ErrorCode::UnexpectedError, "Unknown CellDataType");
+}
+
+ShardDiskUsageMetricKey
+MakeShardDiskUsageMetricKey(CellDataType type, const std::string& shard) {
+    static_cast<void>(CellDataTypeLabel(type));
+    return {type, shard};
+}
+
+std::map<std::string, std::string>
+MakeShardDiskUsageLabels(const ShardDiskUsageMetricKey& key) {
+    return {{"data_type", CellDataTypeLabel(key.cell_data_type)}, {"shard", key.shard}};
+}
+
+}  // namespace
+
+struct CacheShardDiskUsageMetricEntry {
+    ShardDiskUsageMetricKey key;
+    prometheus::Gauge* gauge{nullptr};
+};
+
+CacheShardDiskUsageMetricHandle::CacheShardDiskUsageMetricHandle(std::shared_ptr<CacheShardDiskUsageMetricEntry> entry)
+    : entry_(std::move(entry)) {
+}
+
+CacheShardDiskUsageMetricHandle::~CacheShardDiskUsageMetricHandle() = default;
+
+void
+CacheShardDiskUsageMetricHandle::Increment(double value) {
+    std::lock_guard<std::mutex> lock(shard_disk_usage_mutex);
+    if (entry_->gauge != nullptr) {
+        entry_->gauge->Increment(value);
+    }
+}
+
+void
+CacheShardDiskUsageMetricHandle::Decrement(double value) {
+    std::lock_guard<std::mutex> lock(shard_disk_usage_mutex);
+    if (entry_->gauge != nullptr) {
+        entry_->gauge->Decrement(value);
+    }
+}
+
+double
+CacheShardDiskUsageMetricHandle::Value() const {
+    std::lock_guard<std::mutex> lock(shard_disk_usage_mutex);
+    if (entry_->gauge == nullptr) {
+        return 0;
+    }
+    return entry_->gauge->Value();
+}
+
+std::unique_ptr<CacheShardDiskUsageMetricHandle>
+create_cache_shard_disk_usage_metric_handle(CellDataType type, const std::string& shard) {
+    if (shard.empty()) {
+        return nullptr;
+    }
+    auto key = MakeShardDiskUsageMetricKey(type, shard);
+
+    std::lock_guard<std::mutex> lock(shard_disk_usage_mutex);
+    auto it = shard_disk_usage_metrics.find(key);
+    if (it != shard_disk_usage_metrics.end()) {
+        if (auto entry = it->second.entry.lock()) {
+            return std::unique_ptr<CacheShardDiskUsageMetricHandle>(new CacheShardDiskUsageMetricHandle(entry));
+        }
+        internal_cache_shard_disk_usage_bytes_family.Remove(it->second.gauge);
+        shard_disk_usage_metrics.erase(it);
+    }
+
+    auto& gauge = internal_cache_shard_disk_usage_bytes_family.Add(MakeShardDiskUsageLabels(key));
+    auto entry = std::make_shared<CacheShardDiskUsageMetricEntry>(CacheShardDiskUsageMetricEntry{key, &gauge});
+    shard_disk_usage_metrics.emplace(std::move(key), ShardDiskUsageMetricValue{entry, &gauge});
+    return std::unique_ptr<CacheShardDiskUsageMetricHandle>(new CacheShardDiskUsageMetricHandle(std::move(entry)));
+}
+
+std::vector<CacheShardDiskUsageStats>
+collect_cache_shard_disk_usage_stats() {
+    std::lock_guard<std::mutex> lock(shard_disk_usage_mutex);
+
+    std::vector<CacheShardDiskUsageStats> stats;
+    for (auto it = shard_disk_usage_metrics.begin(); it != shard_disk_usage_metrics.end();) {
+        auto entry = it->second.entry.lock();
+        if (entry == nullptr) {
+            internal_cache_shard_disk_usage_bytes_family.Remove(it->second.gauge);
+            it = shard_disk_usage_metrics.erase(it);
+            continue;
+        }
+        stats.push_back(CacheShardDiskUsageStats{entry->key.cell_data_type, entry->key.shard, entry->gauge->Value()});
+        ++it;
+    }
+    return stats;
+}
+
+std::optional<double>
+cache_shard_disk_usage_bytes_value(CellDataType type, const std::string& shard) {
+    if (shard.empty()) {
+        return std::nullopt;
+    }
+    auto key = MakeShardDiskUsageMetricKey(type, shard);
+    std::lock_guard<std::mutex> lock(shard_disk_usage_mutex);
+    auto it = shard_disk_usage_metrics.find(key);
+    if (it == shard_disk_usage_metrics.end()) {
+        return std::nullopt;
+    }
+    if (it->second.entry.expired()) {
+        internal_cache_shard_disk_usage_bytes_family.Remove(it->second.gauge);
+        shard_disk_usage_metrics.erase(it);
+        return std::nullopt;
+    }
+    return it->second.gauge->Value();
+}
+
 }  // namespace milvus::cachinglayer::monitor