From 599a7ca83bd3685eba04200b045e4dc32940b1fb Mon Sep 17 00:00:00 2001 From: sunby Date: Wed, 24 Jun 2026 12:15:47 +0800 Subject: [PATCH 1/3] add cache shard disk metric Signed-off-by: sunby --- include/cachinglayer/CacheSlot.h | 13 ++++ include/cachinglayer/Metrics.h | 6 ++ include/cachinglayer/Translator.h | 9 ++- src/cachinglayer/Metrics.cpp | 29 ++++++++ test/test_cachinglayer/test_cache_slot.cpp | 85 ++++++++++++++++++++++ 5 files changed, 140 insertions(+), 2 deletions(-) diff --git a/include/cachinglayer/CacheSlot.h b/include/cachinglayer/CacheSlot.h index 3af0faa..eef98ce 100644 --- a/include/cachinglayer/CacheSlot.h +++ b/include/cachinglayer/CacheSlot.h @@ -26,6 +26,7 @@ #include #include #include +#include #include #include #include @@ -62,12 +63,16 @@ class CacheSlot final : public std::enable_shared_from_this> { cell_id_mapping_mode_(translator_->meta()->cell_id_mapping_mode), cell_data_type_(translator_->meta()->cell_data_type), storage_type_(translator_->meta()->storage_type), + shard_(translator_->meta()->shard), dlist_(dlist), evictable_(evictable), self_reserve_(self_reserve), storage_usage_tracking_enabled_(storage_usage_tracking_enabled), loading_timeout_(loading_timeout), warmup_loading_timeout_(warmup_loading_timeout) { + if (!shard_.empty()) { + shard_disk_usage_gauge_ = &monitor::cache_shard_disk_usage_bytes(cell_data_type_, shard_); + } cells_.reserve(translator_->num_cells()); for (cid_t i = 0; i < static_cast(translator_->num_cells()); ++i) { cells_.push_back(std::make_unique(this, i)); @@ -636,6 +641,9 @@ class CacheSlot final : public std::enable_shared_from_this> { .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_gauge_ != nullptr && loaded_size_.file_bytes > 0) { + slot_->shard_disk_usage_gauge_->Increment(loaded_size_.file_bytes); + } monitor::cache_cell_loaded_count(slot_->cell_data_type_, slot_->storage_type_).Increment(); }, requesting_thread); @@ -672,6 +680,9 @@ class CacheSlot final : public std::enable_shared_from_this> { .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_gauge_ != nullptr && loaded_size_.file_bytes > 0) { + slot_->shard_disk_usage_gauge_->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_ } @@ -699,6 +710,8 @@ class CacheSlot final : public std::enable_shared_from_this> { CellIdMappingMode cell_id_mapping_mode_; CellDataType cell_data_type_; StorageType storage_type_; + std::string shard_; + prometheus::Gauge* shard_disk_usage_gauge_{nullptr}; internal::DList* dlist_; const bool evictable_; const bool self_reserve_; diff --git a/include/cachinglayer/Metrics.h b/include/cachinglayer/Metrics.h index a6daa14..defa6cf 100644 --- a/include/cachinglayer/Metrics.h +++ b/include/cachinglayer/Metrics.h @@ -1,5 +1,7 @@ #pragma once +#include + #include "cachinglayer/Utils.h" #include "common/PrometheusClient.h" @@ -240,6 +242,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 +268,9 @@ 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); +prometheus::Gauge& +cache_shard_disk_usage_bytes(CellDataType t, 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); diff --git a/include/cachinglayer/Translator.h b/include/cachinglayer/Translator.h index 8f75d6f..bbf5a21 100644 --- a/include/cachinglayer/Translator.h +++ b/include/cachinglayer/Translator.h @@ -13,6 +13,7 @@ #include #include +#include #include #include @@ -43,15 +44,19 @@ 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 loading_overhead; + // Optional shard identifier for per-shard cache disk usage metrics. + // Empty means this translator is not attributed to any shard. + std::string shard; explicit Meta(StorageType storage_type, CellIdMappingMode cell_id_mapping_mode, CellDataType cell_data_type, CacheWarmupPolicy cache_warmup_policy, bool support_eviction, - std::optional loading_overhead = std::nullopt) + std::optional loading_overhead = std::nullopt, std::string shard = "") : 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)), + shard(std::move(shard)) { } }; diff --git a/src/cachinglayer/Metrics.cpp b/src/cachinglayer/Metrics.cpp index 22a1d9b..8368714 100644 --- a/src/cachinglayer/Metrics.cpp +++ b/src/cachinglayer/Metrics.cpp @@ -1,5 +1,7 @@ #include "cachinglayer/Metrics.h" +#include + namespace milvus::cachinglayer::monitor { DEFINE_LABEL_MAP_WITH_DATA_TYPE_AND_LOCATION(vector_field, memory); @@ -34,6 +36,7 @@ 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]cache 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 +60,30 @@ DEFINE_PROMETHEUS_HISTOGRAM_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_ce milvus::monitor::secondsBuckets, "[cpp]cache cell lifetime seconds"); +namespace { + +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"; + } + return "unknown"; +} + +} // namespace + +prometheus::Gauge& +cache_shard_disk_usage_bytes(CellDataType type, const std::string& shard) { + return internal_cache_shard_disk_usage_bytes_family.Add({{"data_type", CellDataTypeLabel(type)}, {"shard", shard}}); +} + } // namespace milvus::cachinglayer::monitor diff --git a/test/test_cachinglayer/test_cache_slot.cpp b/test/test_cachinglayer/test_cache_slot.cpp index 626512f..ef12fb3 100644 --- a/test/test_cachinglayer/test_cache_slot.cpp +++ b/test/test_cachinglayer/test_cache_slot.cpp @@ -18,6 +18,7 @@ #include #include "cachinglayer/CacheSlot.h" +#include "cachinglayer/Metrics.h" #include "cachinglayer/Translator.h" #include "cachinglayer/Utils.h" #include "cachinglayer/lrucache/DList.h" @@ -226,6 +227,65 @@ class MockTranslator : public Translator { bool for_concurrent_test_ = false; }; +class DiskShardTranslator : public Translator { + public: + DiskShardTranslator(std::string key, std::string shard, std::vector cell_sizes) + : key_(std::move(key)), + cell_sizes_(std::move(cell_sizes)), + meta_(StorageType::DISK, CellIdMappingMode::IDENTICAL, CellDataType::SCALAR_FIELD, + CacheWarmupPolicy::CacheWarmupPolicy_Disable, true, std::nullopt, std::move(shard)) { + } + + size_t + num_cells() const override { + return cell_sizes_.size(); + } + + cid_t + cell_id_of(cl_uid_t uid) const override { + return uid; + } + + std::pair + estimated_byte_size_of_cell(cid_t cid) const override { + return {cell_sizes_.at(cid), {}}; + } + + int64_t + cells_storage_bytes(const std::vector& cids) const override { + int64_t total_bytes = 0; + for (const auto& cid : cids) { + total_bytes += cell_sizes_.at(cid).file_bytes; + } + return total_bytes; + } + + const std::string& + key() const override { + return key_; + } + + Meta* + meta() override { + return &meta_; + } + + std::vector>> + get_cells(milvus::OpContext*, const std::vector& cids) override { + std::vector>> result; + result.reserve(cids.size()); + for (const auto& cid : cids) { + result.emplace_back(cid, std::make_unique(static_cast(cid), cid, cell_sizes_.at(cid))); + } + return result; + } + + private: + std::string key_; + std::vector cell_sizes_; + Meta meta_; +}; + class CacheSlotTest : public ::testing::Test { protected: std::shared_ptr dlist_; @@ -354,6 +414,31 @@ TEST_F(CacheSlotTest, PinMultipleCellsSuccess) { } } +TEST(CacheSlotShardDiskUsageTest, RecordsLoadedDiskBytesByShard) { + constexpr auto kShard = "cache-slot-shard-disk-usage-test"; + auto before = monitor::cache_shard_disk_usage_bytes(CellDataType::SCALAR_FIELD, kShard).Value(); + + auto limit = ResourceUsage{0, 1024}; + auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); + auto translator = std::make_unique("cache_slot_shard_disk_usage", kShard, + std::vector{{0, 128}, {0, 256}}); + auto cache_slot = + std::make_shared>(std::move(translator), dlist.get(), true, true, true, + std::chrono::milliseconds(100000), std::chrono::milliseconds(0)); + + { + auto op_ctx = std::make_unique(); + auto accessor = cache_slot->PinCellsDirect(op_ctx.get(), {0, 1}); + ASSERT_NE(accessor, nullptr); + ASSERT_NE(accessor->get_ith_cell(0), nullptr); + ASSERT_NE(accessor->get_ith_cell(1), nullptr); + EXPECT_EQ(monitor::cache_shard_disk_usage_bytes(CellDataType::SCALAR_FIELD, kShard).Value(), before + 384); + } + + ASSERT_TRUE(cache_slot->ManualEvictAll()); + EXPECT_EQ(monitor::cache_shard_disk_usage_bytes(CellDataType::SCALAR_FIELD, kShard).Value(), before); +} + TEST_F(CacheSlotTest, PinMultipleUidsMappingToSameCid) { std::vector target_uids = {30, 50, 31, 51, 32}; std::vector expected_unique_cids = {2, 4}; From e779d86ab6df25ad23a8cb00e290fc6f8eb485e0 Mon Sep 17 00:00:00 2001 From: sunby Date: Wed, 24 Jun 2026 17:31:48 +0800 Subject: [PATCH 2/3] fix cache shard metric lifecycle Signed-off-by: sunby --- include/cachinglayer/CacheSlot.h | 18 ++-- include/cachinglayer/Metrics.h | 43 +++++++- include/cachinglayer/Translator.h | 17 ++- src/cachinglayer/Metrics.cpp | 115 ++++++++++++++++++++- test/test_cachinglayer/test_cache_slot.cpp | 87 +++++++++++++++- 5 files changed, 255 insertions(+), 25 deletions(-) diff --git a/include/cachinglayer/CacheSlot.h b/include/cachinglayer/CacheSlot.h index eef98ce..3e3eb76 100644 --- a/include/cachinglayer/CacheSlot.h +++ b/include/cachinglayer/CacheSlot.h @@ -63,15 +63,16 @@ class CacheSlot final : public std::enable_shared_from_this> { cell_id_mapping_mode_(translator_->meta()->cell_id_mapping_mode), cell_data_type_(translator_->meta()->cell_data_type), storage_type_(translator_->meta()->storage_type), - shard_(translator_->meta()->shard), dlist_(dlist), evictable_(evictable), self_reserve_(self_reserve), storage_usage_tracking_enabled_(storage_usage_tracking_enabled), loading_timeout_(loading_timeout), warmup_loading_timeout_(warmup_loading_timeout) { - if (!shard_.empty()) { - shard_disk_usage_gauge_ = &monitor::cache_shard_disk_usage_bytes(cell_data_type_, shard_); + 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(translator_->num_cells()); ++i) { @@ -641,8 +642,8 @@ class CacheSlot final : public std::enable_shared_from_this> { .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_gauge_ != nullptr && loaded_size_.file_bytes > 0) { - slot_->shard_disk_usage_gauge_->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(); }, @@ -680,8 +681,8 @@ class CacheSlot final : public std::enable_shared_from_this> { .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_gauge_ != nullptr && loaded_size_.file_bytes > 0) { - slot_->shard_disk_usage_gauge_->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_ @@ -702,6 +703,7 @@ class CacheSlot final : public std::enable_shared_from_this> { }; const std::unique_ptr> translator_; + std::unique_ptr 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). @@ -710,8 +712,6 @@ class CacheSlot final : public std::enable_shared_from_this> { CellIdMappingMode cell_id_mapping_mode_; CellDataType cell_data_type_; StorageType storage_type_; - std::string shard_; - prometheus::Gauge* shard_disk_usage_gauge_{nullptr}; internal::DList* dlist_; const bool evictable_; const bool self_reserve_; diff --git a/include/cachinglayer/Metrics.h b/include/cachinglayer/Metrics.h index defa6cf..3fae886 100644 --- a/include/cachinglayer/Metrics.h +++ b/include/cachinglayer/Metrics.h @@ -1,5 +1,7 @@ #pragma once +#include +#include #include #include "cachinglayer/Utils.h" @@ -268,8 +270,45 @@ 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); -prometheus::Gauge& -cache_shard_disk_usage_bytes(CellDataType t, const std::string& shard); +// RAII handle for one {data_type, shard} cache-slot disk usage series. +// Destroying the last handle removes the dynamic Prometheus time series. +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 + create_cache_shard_disk_usage_metric_handle(CellDataType type, const std::string& shard); + + CacheShardDiskUsageMetricHandle(std::string data_type_label, std::string shard); + + std::string data_type_label_; + std::string shard_; +}; + +// Returns nullptr for an empty shard, leaving the slot unattributed. +std::unique_ptr +create_cache_shard_disk_usage_metric_handle(CellDataType type, const std::string& shard); + +// Returns nullopt without creating a series. +std::optional +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 diff --git a/include/cachinglayer/Translator.h b/include/cachinglayer/Translator.h index bbf5a21..742e42e 100644 --- a/include/cachinglayer/Translator.h +++ b/include/cachinglayer/Translator.h @@ -28,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 CacheSlot-owned metric handle removes the time series when 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. @@ -44,19 +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 loading_overhead; - // Optional shard identifier for per-shard cache disk usage metrics. - // Empty means this translator is not attributed to any shard. - std::string shard; + std::optional 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 loading_overhead = std::nullopt, std::string shard = "") + std::optional loading_overhead = std::nullopt, + std::optional 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)), - shard(std::move(shard)) { + metric_attribution(std::move(metric_attribution)) { } }; diff --git a/src/cachinglayer/Metrics.cpp b/src/cachinglayer/Metrics.cpp index 8368714..d549e46 100644 --- a/src/cachinglayer/Metrics.cpp +++ b/src/cachinglayer/Metrics.cpp @@ -1,6 +1,10 @@ #include "cachinglayer/Metrics.h" +#include +#include +#include #include +#include namespace milvus::cachinglayer::monitor { @@ -36,7 +40,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]cache loaded disk usage bytes by shard"); +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, @@ -62,6 +67,16 @@ DEFINE_PROMETHEUS_HISTOGRAM_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_ce namespace { +using ShardDiskUsageMetricKey = std::pair; + +struct ShardDiskUsageMetricEntry { + prometheus::Gauge* gauge{nullptr}; + size_t ref_count{0}; +}; + +std::mutex shard_disk_usage_mutex; +std::map shard_disk_usage_metrics; + const char* CellDataTypeLabel(CellDataType type) { switch (type) { @@ -76,14 +91,104 @@ CellDataTypeLabel(CellDataType type) { case CellDataType::OTHER: return "other"; } - return "unknown"; + ThrowInfo(ErrorCode::UnexpectedError, "Unknown CellDataType"); +} + +ShardDiskUsageMetricKey +MakeShardDiskUsageMetricKey(CellDataType type, const std::string& shard) { + return {CellDataTypeLabel(type), shard}; +} + +std::map +MakeShardDiskUsageLabels(const ShardDiskUsageMetricKey& key) { + return {{"data_type", key.first}, {"shard", key.second}}; +} + +void +ReleaseShardDiskUsageMetricHandle(const ShardDiskUsageMetricKey& key) { + std::lock_guard lock(shard_disk_usage_mutex); + auto it = shard_disk_usage_metrics.find(key); + if (it == shard_disk_usage_metrics.end()) { + return; + } + if (--it->second.ref_count > 0) { + return; + } + internal_cache_shard_disk_usage_bytes_family.Remove(it->second.gauge); + shard_disk_usage_metrics.erase(it); } } // namespace -prometheus::Gauge& -cache_shard_disk_usage_bytes(CellDataType type, const std::string& shard) { - return internal_cache_shard_disk_usage_bytes_family.Add({{"data_type", CellDataTypeLabel(type)}, {"shard", shard}}); +CacheShardDiskUsageMetricHandle::CacheShardDiskUsageMetricHandle(std::string data_type_label, std::string shard) + : data_type_label_(std::move(data_type_label)), shard_(std::move(shard)) { +} + +CacheShardDiskUsageMetricHandle::~CacheShardDiskUsageMetricHandle() { + ReleaseShardDiskUsageMetricHandle({data_type_label_, shard_}); +} + +void +CacheShardDiskUsageMetricHandle::Increment(double value) { + std::lock_guard lock(shard_disk_usage_mutex); + auto it = shard_disk_usage_metrics.find({data_type_label_, shard_}); + if (it != shard_disk_usage_metrics.end()) { + it->second.gauge->Increment(value); + } +} + +void +CacheShardDiskUsageMetricHandle::Decrement(double value) { + std::lock_guard lock(shard_disk_usage_mutex); + auto it = shard_disk_usage_metrics.find({data_type_label_, shard_}); + if (it != shard_disk_usage_metrics.end()) { + it->second.gauge->Decrement(value); + } +} + +double +CacheShardDiskUsageMetricHandle::Value() const { + std::lock_guard lock(shard_disk_usage_mutex); + auto it = shard_disk_usage_metrics.find({data_type_label_, shard_}); + if (it == shard_disk_usage_metrics.end()) { + return 0; + } + return it->second.gauge->Value(); +} + +std::unique_ptr +create_cache_shard_disk_usage_metric_handle(CellDataType type, const std::string& shard) { + auto key = MakeShardDiskUsageMetricKey(type, shard); + if (key.second.empty()) { + return nullptr; + } + auto handle = + std::unique_ptr(new CacheShardDiskUsageMetricHandle(key.first, key.second)); + + std::lock_guard lock(shard_disk_usage_mutex); + auto it = shard_disk_usage_metrics.find(key); + if (it != shard_disk_usage_metrics.end()) { + ++it->second.ref_count; + return handle; + } + + auto& gauge = internal_cache_shard_disk_usage_bytes_family.Add(MakeShardDiskUsageLabels(key)); + shard_disk_usage_metrics.emplace(std::move(key), ShardDiskUsageMetricEntry{&gauge, 1}); + return handle; +} + +std::optional +cache_shard_disk_usage_bytes_value(CellDataType type, const std::string& shard) { + auto key = MakeShardDiskUsageMetricKey(type, shard); + if (key.second.empty()) { + return std::nullopt; + } + std::lock_guard lock(shard_disk_usage_mutex); + auto it = shard_disk_usage_metrics.find(key); + if (it == shard_disk_usage_metrics.end()) { + return std::nullopt; + } + return it->second.gauge->Value(); } } // namespace milvus::cachinglayer::monitor diff --git a/test/test_cachinglayer/test_cache_slot.cpp b/test/test_cachinglayer/test_cache_slot.cpp index ef12fb3..a389fb5 100644 --- a/test/test_cachinglayer/test_cache_slot.cpp +++ b/test/test_cachinglayer/test_cache_slot.cpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -233,7 +234,8 @@ class DiskShardTranslator : public Translator { : key_(std::move(key)), cell_sizes_(std::move(cell_sizes)), meta_(StorageType::DISK, CellIdMappingMode::IDENTICAL, CellDataType::SCALAR_FIELD, - CacheWarmupPolicy::CacheWarmupPolicy_Disable, true, std::nullopt, std::move(shard)) { + CacheWarmupPolicy::CacheWarmupPolicy_Disable, true, std::nullopt, + MetricAttribution{.shard = std::move(shard)}) { } size_t @@ -286,6 +288,15 @@ class DiskShardTranslator : public Translator { Meta meta_; }; +namespace { + +std::optional +ScalarFieldShardDiskUsage(const std::string& shard) { + return monitor::cache_shard_disk_usage_bytes_value(CellDataType::SCALAR_FIELD, shard); +} + +} // namespace + class CacheSlotTest : public ::testing::Test { protected: std::shared_ptr dlist_; @@ -416,7 +427,7 @@ TEST_F(CacheSlotTest, PinMultipleCellsSuccess) { TEST(CacheSlotShardDiskUsageTest, RecordsLoadedDiskBytesByShard) { constexpr auto kShard = "cache-slot-shard-disk-usage-test"; - auto before = monitor::cache_shard_disk_usage_bytes(CellDataType::SCALAR_FIELD, kShard).Value(); + auto before = ScalarFieldShardDiskUsage(kShard).value_or(0); auto limit = ResourceUsage{0, 1024}; auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); @@ -432,11 +443,79 @@ TEST(CacheSlotShardDiskUsageTest, RecordsLoadedDiskBytesByShard) { ASSERT_NE(accessor, nullptr); ASSERT_NE(accessor->get_ith_cell(0), nullptr); ASSERT_NE(accessor->get_ith_cell(1), nullptr); - EXPECT_EQ(monitor::cache_shard_disk_usage_bytes(CellDataType::SCALAR_FIELD, kShard).Value(), before + 384); + EXPECT_EQ(ScalarFieldShardDiskUsage(kShard), before + 384); } ASSERT_TRUE(cache_slot->ManualEvictAll()); - EXPECT_EQ(monitor::cache_shard_disk_usage_bytes(CellDataType::SCALAR_FIELD, kShard).Value(), before); + EXPECT_EQ(ScalarFieldShardDiskUsage(kShard), before); +} + +TEST(CacheSlotShardDiskUsageTest, RemovingOneOfTwoSlotsKeepsSharedShardUsage) { + constexpr auto kShard = "cache-slot-shard-disk-usage-shared-test"; + EXPECT_EQ(ScalarFieldShardDiskUsage(kShard), std::nullopt); + + auto limit = ResourceUsage{0, 1024}; + auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); + auto translator1 = std::make_unique("cache_slot_shard_disk_usage_shared_1", kShard, + std::vector{{0, 128}}); + auto translator2 = std::make_unique("cache_slot_shard_disk_usage_shared_2", kShard, + std::vector{{0, 256}}); + auto cache_slot1 = + std::make_shared>(std::move(translator1), dlist.get(), true, true, true, + std::chrono::milliseconds(100000), std::chrono::milliseconds(0)); + auto cache_slot2 = + std::make_shared>(std::move(translator2), dlist.get(), true, true, true, + std::chrono::milliseconds(100000), std::chrono::milliseconds(0)); + + { + auto op_ctx = std::make_unique(); + auto accessor1 = cache_slot1->PinCellsDirect(op_ctx.get(), {0}); + auto accessor2 = cache_slot2->PinCellsDirect(op_ctx.get(), {0}); + ASSERT_NE(accessor1, nullptr); + ASSERT_NE(accessor2, nullptr); + ASSERT_NE(accessor1->get_ith_cell(0), nullptr); + ASSERT_NE(accessor2->get_ith_cell(0), nullptr); + EXPECT_EQ(ScalarFieldShardDiskUsage(kShard), 384); + } + + cache_slot1.reset(); + EXPECT_EQ(ScalarFieldShardDiskUsage(kShard), 256); + + cache_slot2.reset(); + EXPECT_EQ(ScalarFieldShardDiskUsage(kShard), std::nullopt); +} + +TEST(CacheSlotShardDiskUsageTest, DestroyingLoadedSlotRefundsAndRemovesShardUsage) { + constexpr auto kShard = "cache-slot-shard-disk-usage-destroy-test"; + EXPECT_EQ(ScalarFieldShardDiskUsage(kShard), std::nullopt); + + auto limit = ResourceUsage{0, 1024}; + auto dlist = std::make_shared(true, limit, limit, limit, EvictionConfig{10, true, 600}); + auto translator = std::make_unique("cache_slot_shard_disk_usage_destroy", kShard, + std::vector{{0, 512}}); + auto cache_slot = + std::make_shared>(std::move(translator), dlist.get(), true, true, true, + std::chrono::milliseconds(100000), std::chrono::milliseconds(0)); + + { + auto op_ctx = std::make_unique(); + auto accessor = cache_slot->PinCellsDirect(op_ctx.get(), {0}); + ASSERT_NE(accessor, nullptr); + ASSERT_NE(accessor->get_ith_cell(0), nullptr); + EXPECT_EQ(ScalarFieldShardDiskUsage(kShard), 512); + } + + cache_slot.reset(); + EXPECT_EQ(ScalarFieldShardDiskUsage(kShard), std::nullopt); +} + +TEST(CacheSlotShardDiskUsageTest, RejectsUnknownCellDataType) { + EXPECT_THROW( + { + static_cast(monitor::create_cache_shard_disk_usage_metric_handle( + static_cast(-1), "cache-slot-shard-disk-usage-unknown-type-test")); + }, + milvus::SegcoreError); } TEST_F(CacheSlotTest, PinMultipleUidsMappingToSameCid) { From e22838add828a10430a3fcbd879a6bd809655de8 Mon Sep 17 00:00:00 2001 From: sunby Date: Thu, 25 Jun 2026 15:09:00 +0800 Subject: [PATCH 3/3] adjust cache shard metric cleanup Signed-off-by: sunby --- include/cachinglayer/Metrics.h | 20 +++- include/cachinglayer/Translator.h | 2 +- src/cachinglayer/Metrics.cpp | 113 +++++++++++++-------- test/test_cachinglayer/test_cache_slot.cpp | 8 +- 4 files changed, 93 insertions(+), 50 deletions(-) diff --git a/include/cachinglayer/Metrics.h b/include/cachinglayer/Metrics.h index 3fae886..9ce41a3 100644 --- a/include/cachinglayer/Metrics.h +++ b/include/cachinglayer/Metrics.h @@ -3,6 +3,7 @@ #include #include #include +#include #include "cachinglayer/Utils.h" #include "common/PrometheusClient.h" @@ -270,8 +271,16 @@ 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. -// Destroying the last handle removes the dynamic Prometheus time series. +// The collector removes the dynamic Prometheus time series after the last handle is gone. class CacheShardDiskUsageMetricHandle { public: ~CacheShardDiskUsageMetricHandle(); @@ -296,16 +305,19 @@ class CacheShardDiskUsageMetricHandle { friend std::unique_ptr create_cache_shard_disk_usage_metric_handle(CellDataType type, const std::string& shard); - CacheShardDiskUsageMetricHandle(std::string data_type_label, std::string shard); + explicit CacheShardDiskUsageMetricHandle(std::shared_ptr entry); - std::string data_type_label_; - std::string shard_; + std::shared_ptr entry_; }; // Returns nullptr for an empty shard, leaving the slot unattributed. std::unique_ptr 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 +collect_cache_shard_disk_usage_stats(); + // Returns nullopt without creating a series. std::optional cache_shard_disk_usage_bytes_value(CellDataType type, const std::string& shard); diff --git a/include/cachinglayer/Translator.h b/include/cachinglayer/Translator.h index 742e42e..074ea70 100644 --- a/include/cachinglayer/Translator.h +++ b/include/cachinglayer/Translator.h @@ -31,7 +31,7 @@ struct LoadingOverheadConfig { 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 CacheSlot-owned metric handle removes the time series when the last slot for this label is gone. + // 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; }; diff --git a/src/cachinglayer/Metrics.cpp b/src/cachinglayer/Metrics.cpp index d549e46..9293be0 100644 --- a/src/cachinglayer/Metrics.cpp +++ b/src/cachinglayer/Metrics.cpp @@ -1,10 +1,12 @@ #include "cachinglayer/Metrics.h" #include +#include #include #include #include #include +#include namespace milvus::cachinglayer::monitor { @@ -67,15 +69,26 @@ DEFINE_PROMETHEUS_HISTOGRAM_METRIC_WITH_DATA_TYPE_AND_LOCATION(internal_cache_ce namespace { -using ShardDiskUsageMetricKey = std::pair; +struct ShardDiskUsageMetricKey { + CellDataType cell_data_type; + std::string shard; -struct ShardDiskUsageMetricEntry { + bool + operator<(const ShardDiskUsageMetricKey& other) const { + if (cell_data_type != other.cell_data_type) { + return static_cast(cell_data_type) < static_cast(other.cell_data_type); + } + return shard < other.shard; + } +}; + +struct ShardDiskUsageMetricValue { + std::weak_ptr entry; prometheus::Gauge* gauge{nullptr}; - size_t ref_count{0}; }; std::mutex shard_disk_usage_mutex; -std::map shard_disk_usage_metrics; +std::map shard_disk_usage_metrics; const char* CellDataTypeLabel(CellDataType type) { @@ -96,98 +109,110 @@ CellDataTypeLabel(CellDataType type) { ShardDiskUsageMetricKey MakeShardDiskUsageMetricKey(CellDataType type, const std::string& shard) { - return {CellDataTypeLabel(type), shard}; + static_cast(CellDataTypeLabel(type)); + return {type, shard}; } std::map MakeShardDiskUsageLabels(const ShardDiskUsageMetricKey& key) { - return {{"data_type", key.first}, {"shard", key.second}}; -} - -void -ReleaseShardDiskUsageMetricHandle(const ShardDiskUsageMetricKey& key) { - std::lock_guard lock(shard_disk_usage_mutex); - auto it = shard_disk_usage_metrics.find(key); - if (it == shard_disk_usage_metrics.end()) { - return; - } - if (--it->second.ref_count > 0) { - return; - } - internal_cache_shard_disk_usage_bytes_family.Remove(it->second.gauge); - shard_disk_usage_metrics.erase(it); + return {{"data_type", CellDataTypeLabel(key.cell_data_type)}, {"shard", key.shard}}; } } // namespace -CacheShardDiskUsageMetricHandle::CacheShardDiskUsageMetricHandle(std::string data_type_label, std::string shard) - : data_type_label_(std::move(data_type_label)), shard_(std::move(shard)) { -} +struct CacheShardDiskUsageMetricEntry { + ShardDiskUsageMetricKey key; + prometheus::Gauge* gauge{nullptr}; +}; -CacheShardDiskUsageMetricHandle::~CacheShardDiskUsageMetricHandle() { - ReleaseShardDiskUsageMetricHandle({data_type_label_, shard_}); +CacheShardDiskUsageMetricHandle::CacheShardDiskUsageMetricHandle(std::shared_ptr entry) + : entry_(std::move(entry)) { } +CacheShardDiskUsageMetricHandle::~CacheShardDiskUsageMetricHandle() = default; + void CacheShardDiskUsageMetricHandle::Increment(double value) { std::lock_guard lock(shard_disk_usage_mutex); - auto it = shard_disk_usage_metrics.find({data_type_label_, shard_}); - if (it != shard_disk_usage_metrics.end()) { - it->second.gauge->Increment(value); + if (entry_->gauge != nullptr) { + entry_->gauge->Increment(value); } } void CacheShardDiskUsageMetricHandle::Decrement(double value) { std::lock_guard lock(shard_disk_usage_mutex); - auto it = shard_disk_usage_metrics.find({data_type_label_, shard_}); - if (it != shard_disk_usage_metrics.end()) { - it->second.gauge->Decrement(value); + if (entry_->gauge != nullptr) { + entry_->gauge->Decrement(value); } } double CacheShardDiskUsageMetricHandle::Value() const { std::lock_guard lock(shard_disk_usage_mutex); - auto it = shard_disk_usage_metrics.find({data_type_label_, shard_}); - if (it == shard_disk_usage_metrics.end()) { + if (entry_->gauge == nullptr) { return 0; } - return it->second.gauge->Value(); + return entry_->gauge->Value(); } std::unique_ptr create_cache_shard_disk_usage_metric_handle(CellDataType type, const std::string& shard) { - auto key = MakeShardDiskUsageMetricKey(type, shard); - if (key.second.empty()) { + if (shard.empty()) { return nullptr; } - auto handle = - std::unique_ptr(new CacheShardDiskUsageMetricHandle(key.first, key.second)); + auto key = MakeShardDiskUsageMetricKey(type, shard); std::lock_guard lock(shard_disk_usage_mutex); auto it = shard_disk_usage_metrics.find(key); if (it != shard_disk_usage_metrics.end()) { - ++it->second.ref_count; - return handle; + if (auto entry = it->second.entry.lock()) { + return std::unique_ptr(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)); - shard_disk_usage_metrics.emplace(std::move(key), ShardDiskUsageMetricEntry{&gauge, 1}); - return handle; + auto entry = std::make_shared(CacheShardDiskUsageMetricEntry{key, &gauge}); + shard_disk_usage_metrics.emplace(std::move(key), ShardDiskUsageMetricValue{entry, &gauge}); + return std::unique_ptr(new CacheShardDiskUsageMetricHandle(std::move(entry))); +} + +std::vector +collect_cache_shard_disk_usage_stats() { + std::lock_guard lock(shard_disk_usage_mutex); + + std::vector 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 cache_shard_disk_usage_bytes_value(CellDataType type, const std::string& shard) { - auto key = MakeShardDiskUsageMetricKey(type, shard); - if (key.second.empty()) { + if (shard.empty()) { return std::nullopt; } + auto key = MakeShardDiskUsageMetricKey(type, shard); std::lock_guard 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(); } diff --git a/test/test_cachinglayer/test_cache_slot.cpp b/test/test_cachinglayer/test_cache_slot.cpp index a389fb5..3e6e4d5 100644 --- a/test/test_cachinglayer/test_cache_slot.cpp +++ b/test/test_cachinglayer/test_cache_slot.cpp @@ -292,7 +292,13 @@ namespace { std::optional ScalarFieldShardDiskUsage(const std::string& shard) { - return monitor::cache_shard_disk_usage_bytes_value(CellDataType::SCALAR_FIELD, shard); + const auto stats = monitor::collect_cache_shard_disk_usage_stats(); + for (const auto& stat : stats) { + if (stat.cell_data_type == CellDataType::SCALAR_FIELD && stat.shard == shard) { + return stat.disk_bytes; + } + } + return std::nullopt; } } // namespace