[metrics] report ReportEvent request key counts by event type

Author: lucky-zzz

kv_cache_manager/metrics/kmonitor_metrics_reporter.cc

@@ -561,6 +561,9 @@ void KmonitorMetricsReporter::ReportPerQuery(MetricsCollector *collector) {
 
         const double error_code = p->GetErrorCodeSample();
         REPORT_METRICS_WHEN(event_report, error_qps, 1.0, !CommonUtil::IsZeroDouble(error_code));
+        if (p->HasRequestKeyCountSample()) {
+            REPORT_METRICS(manager, request_key_count, p->GetRequestKeyCountSample());
+        }
     } else if (dynamic_cast<DataStorageMetricsCollector *>(collector)) {
         const auto *p = dynamic_cast<DataStorageMetricsCollector *>(collector);
         const kmonitor::MetricsTags tags = ctx_->GetKmonitorTags(p->GetMetricsTags());

kv_cache_manager/metrics/metrics_collector.cc

@@ -195,11 +195,16 @@ bool ServiceMetricsCollector::Init() {
     REGISTER_METRICS_W_TAGS_COUNTER_(metrics_registry_, event_report, name, metrics_tags_)
 #define REGISTER_GAUGE_METRICS_FOR_EVENT_REPORT(name)                                                                  \
     REGISTER_METRICS_W_TAGS_GAUGE_(metrics_registry_, event_report, name, metrics_tags_)
+#define DEFINE_METRICS_NAME_FOR_EVENT_REPORT_MANAGER(name)                                                             \
+    DEFINE_METRICS_NAME_(EventReportMetricsCollector, manager, name)
+#define REGISTER_GAUGE_METRICS_FOR_EVENT_REPORT_MANAGER(name)                                                          \
+    REGISTER_METRICS_W_TAGS_GAUGE_(metrics_registry_, manager, name, metrics_tags_)
 
 DEFINE_METRICS_NAME_FOR_EVENT_REPORT(request_counter);
 DEFINE_METRICS_NAME_FOR_EVENT_REPORT(request_rt_us);
 DEFINE_METRICS_NAME_FOR_EVENT_REPORT(error_code);
 DEFINE_METRICS_NAME_FOR_EVENT_REPORT(error_counter);
+DEFINE_METRICS_NAME_FOR_EVENT_REPORT_MANAGER(request_key_count);
 
 EventReportMetricsCollector::EventReportMetricsCollector(std::shared_ptr<MetricsRegistry> metrics_registry,
                                                          MetricsTags metrics_tags) noexcept
@@ -211,6 +216,7 @@ EventReportMetricsCollector::EventReportMetricsCollector(const EventReportMetric
     event_report_request_rt_us_metrics_ = shared_collector.event_report_request_rt_us_metrics_;
     event_report_error_code_metrics_ = shared_collector.event_report_error_code_metrics_;
     event_report_error_counter_metrics_ = shared_collector.event_report_error_counter_metrics_;
+    manager_request_key_count_metrics_ = shared_collector.manager_request_key_count_metrics_;
 }
 
 bool EventReportMetricsCollector::Init() {
@@ -221,6 +227,7 @@ bool EventReportMetricsCollector::Init() {
     REGISTER_GAUGE_METRICS_FOR_EVENT_REPORT(request_rt_us);
     REGISTER_GAUGE_METRICS_FOR_EVENT_REPORT(error_code);
     REGISTER_COUNTER_METRICS_FOR_EVENT_REPORT(error_counter);
+    REGISTER_GAUGE_METRICS_FOR_EVENT_REPORT_MANAGER(request_key_count);
     return true;
 }
 

kv_cache_manager/metrics/metrics_collector.h

@@ -322,13 +322,14 @@ class ServiceMetricsCollector final : public MetricsCollector {
 };
 
 // Lightweight per-event observability for ReportEvent. This intentionally
-// avoids registering the manager/meta metric families carried by a full
-// ServiceMetricsCollector.
+// carries only the request key count from the manager metric family, rather
+// than registering every manager/meta metric from ServiceMetricsCollector.
 class EventReportMetricsCollector final : public MetricsCollector {
     KVCM_COUNTER_METRICS(event_report, request_counter)
     KVCM_GAUGE_METRICS(event_report, request_rt_us)
     KVCM_GAUGE_METRICS(event_report, error_code)
     KVCM_COUNTER_METRICS(event_report, error_counter)
+    KVCM_GAUGE_METRICS(manager, request_key_count)
 
 public:
     EventReportMetricsCollector() = delete;
@@ -341,7 +342,7 @@ class EventReportMetricsCollector final : public MetricsCollector {
     // Collector objects are request-local, while registry gauges with the
     // same event_type tag are shared. Synchronous reporters must consume this
     // local sample so concurrent requests cannot attribute one another's
-    // latency or error.
+    // latency, error, or request key count.
     void SetRequestSample(double request_rt_us, double error_code) noexcept {
         request_rt_us_sample_ = request_rt_us;
         error_code_sample_ = error_code;
@@ -353,11 +354,21 @@ class EventReportMetricsCollector final : public MetricsCollector {
     [[nodiscard]] double GetErrorCodeSample() const noexcept {
         return has_request_sample_ ? error_code_sample_ : get_event_report_error_code_metrics();
     }
+    void SetRequestKeyCountSample(double request_key_count) noexcept {
+        request_key_count_sample_ = request_key_count;
+        has_request_key_count_sample_ = true;
+    }
+    [[nodiscard]] bool HasRequestKeyCountSample() const noexcept { return has_request_key_count_sample_; }
+    [[nodiscard]] double GetRequestKeyCountSample() const noexcept {
+        return has_request_key_count_sample_ ? request_key_count_sample_ : get_manager_request_key_count_metrics();
+    }
 
 private:
     double request_rt_us_sample_ = 0.0;
     double error_code_sample_ = 0.0;
+    double request_key_count_sample_ = 0.0;
     bool has_request_sample_ = false;
+    bool has_request_key_count_sample_ = false;
 };
 
 /* ------------------ DataStorageMetricsCollector ------------------- */

kv_cache_manager/metrics/test/metrics_collector_test.cc

@@ -27,12 +27,13 @@ TEST_F(MetricsCollectorTest, EventReportMetricsTest) {
     auto collector = std::make_shared<EventReportMetricsCollector>(metrics_registry_, tags);
     ASSERT_TRUE(collector->Init());
     EXPECT_EQ(tags, collector->GetMetricsTags());
-    EXPECT_EQ(4, metrics_registry_->GetSize());
+    EXPECT_EQ(5, metrics_registry_->GetSize());
 
     EXPECT_EQ(0, GET(collector, event_report, request_counter));
     EXPECT_DOUBLE_EQ(0., GET(collector, event_report, request_rt_us));
     EXPECT_DOUBLE_EQ(0., GET(collector, event_report, error_code));
     EXPECT_EQ(0, GET(collector, event_report, error_counter));
+    EXPECT_DOUBLE_EQ(0., GET(collector, manager, request_key_count));
 
     SET_METRICS_(collector, event_report, request_rt_us, 123.);
     SET_METRICS_(collector, event_report, error_code, 10.);

kv_cache_manager/service/meta_service_metrics_base.cc

@@ -205,17 +205,36 @@ void MetaServiceMetricsBase::AttachReportEventTypeMetricsCollectors(const proto:
         return;
     }
 
+    static constexpr std::array<const char *, 7> kEventTypeTags = {
+        "unknown", "node_register", "block_add", "block_delete", "host_down", "heartbeat", "block_snapshot"};
     uint32_t event_type_mask = 0;
+    std::array<size_t, kEventTypeTags.size()> request_key_counts{};
     for (const auto &event : request.events()) {
         const int event_type = static_cast<int>(event.event_type());
-        event_type_mask |=
-            1U << ((event_type >= proto::meta::EVENT_NODE_REGISTER && event_type <= proto::meta::EVENT_BLOCK_SNAPSHOT)
-                       ? event_type
-                       : 0);
+        const int bounded_event_type =
+            (event_type >= proto::meta::EVENT_NODE_REGISTER && event_type <= proto::meta::EVENT_BLOCK_SNAPSHOT)
+                ? event_type
+                : 0;
+        event_type_mask |= 1U << bounded_event_type;
+        // Match request_key_count semantics used by the other manager APIs:
+        // count keys in the request payload, regardless of later validation,
+        // deduplication, or persistence outcomes.
+        switch (bounded_event_type) {
+        case proto::meta::EVENT_BLOCK_ADD:
+            request_key_counts[bounded_event_type] += event.has_block_add() ? 1 : 0;
+            break;
+        case proto::meta::EVENT_BLOCK_DELETE:
+            request_key_counts[bounded_event_type] += event.has_block_delete() ? 1 : 0;
+            break;
+        case proto::meta::EVENT_BLOCK_SNAPSHOT:
+            request_key_counts[bounded_event_type] +=
+                event.has_block_snapshot() ? static_cast<size_t>(event.block_snapshot().blocks_size()) : 0;
+            break;
+        default:
+            break;
+        }
     }
 
-    static constexpr std::array<const char *, 7> kEventTypeTags = {
-        "unknown", "node_register", "block_add", "block_delete", "host_down", "heartbeat", "block_snapshot"};
     for (size_t event_type = 0; event_type < kEventTypeTags.size(); ++event_type) {
         if ((event_type_mask & (1U << event_type)) == 0) {
             continue;
@@ -227,8 +246,12 @@ void MetaServiceMetricsBase::AttachReportEventTypeMetricsCollectors(const proto:
             // The cached object owns the registry handles. Each request gets a
             // lightweight view with the same handles but private sample state,
             // avoiding both registry re-registration and cross-request races.
-            request_context->GetMetricsCollectorsVehicle().AddMetricsCollector(
-                std::make_shared<EventReportMetricsCollector>(*event_collector));
+            auto request_collector = std::make_shared<EventReportMetricsCollector>(*event_collector);
+            if (event_type == proto::meta::EVENT_BLOCK_ADD || event_type == proto::meta::EVENT_BLOCK_DELETE ||
+                event_type == proto::meta::EVENT_BLOCK_SNAPSHOT) {
+                request_collector->SetRequestKeyCountSample(request_key_counts[event_type]);
+            }
+            request_context->GetMetricsCollectorsVehicle().AddMetricsCollector(std::move(request_collector));
         }
     }
 }

kv_cache_manager/service/test/meta_service_metrics_base_test.cc

@@ -155,7 +155,10 @@ TEST_F(MetaServiceMetricsBaseTest, AttachesOneCollectorPerDistinctEventTypeAndBo
     SeedInstance("inst1", "grp1");
     proto::meta::ReportEventRequest request;
     request.set_instance_id("inst1");
-    request.add_events()->set_event_type(proto::meta::EVENT_BLOCK_SNAPSHOT);
+    auto *snapshot_event = request.add_events();
+    snapshot_event->set_event_type(proto::meta::EVENT_BLOCK_SNAPSHOT);
+    snapshot_event->mutable_block_snapshot()->add_blocks()->set_block_key("1");
+    snapshot_event->mutable_block_snapshot()->add_blocks()->set_block_key("2");
     request.add_events()->set_event_type(proto::meta::EVENT_BLOCK_SNAPSHOT);
     request.add_events()->set_event_type(proto::meta::EVENT_HEARTBEAT);
     request.add_events()->set_event_type(static_cast<proto::meta::ReportEventType>(99));
@@ -167,8 +170,14 @@ TEST_F(MetaServiceMetricsBaseTest, AttachesOneCollectorPerDistinctEventTypeAndBo
     ASSERT_EQ(3, collectors.size());
     std::set<std::string> event_types;
     for (const auto &collector : collectors) {
-        ASSERT_NE(nullptr, dynamic_cast<EventReportMetricsCollector *>(collector.get()));
-        event_types.insert(collector->GetMetricsTags().at("event_type"));
+        auto *event_collector = dynamic_cast<EventReportMetricsCollector *>(collector.get());
+        ASSERT_NE(nullptr, event_collector);
+        const auto &event_type = collector->GetMetricsTags().at("event_type");
+        event_types.insert(event_type);
+        if (event_type == "block_snapshot") {
+            EXPECT_TRUE(event_collector->HasRequestKeyCountSample());
+            EXPECT_DOUBLE_EQ(2., event_collector->GetRequestKeyCountSample());
+        }
     }
     EXPECT_EQ((std::set<std::string>{"block_snapshot", "heartbeat", "unknown"}), event_types);
 }

kv_cache_manager/service/util/service_call_guard.cc

@@ -151,6 +151,12 @@ ServiceCallGuard::~ServiceCallGuard() {
                 SET_METRICS_(event_metrics_collector, event_report, request_rt_us, request_rt_us);
                 SET_METRICS_(event_metrics_collector, event_report, error_code, error_code);
                 event_metrics_collector->SetRequestSample(request_rt_us, error_code);
+                if (event_metrics_collector->HasRequestKeyCountSample()) {
+                    SET_METRICS_(event_metrics_collector,
+                                 manager,
+                                 request_key_count,
+                                 event_metrics_collector->GetRequestKeyCountSample());
+                }
             }
             metrics_reporter_->ReportPerQuery(mc.get());
         }