[Store] Add P2P local metrics (#1945)

* [Store] Add P2P local storage metrics for DataManager

Add LocalStorageMetric to track P2P client local storage operations:

LocalStorageMetric:
- put_requests_total, put_failures_total, put_bytes_total
- put_latency_us (histogram)
- get_requests_total, get_failures_total, get_misses_total
- get_bytes_total, get_latency_us (histogram)

Changes:
- Add LocalStorageMetric struct in client_metric.h
- Add SetMetrics() method to DataManager for metrics injection
- Record metrics in DataManager::Put() and DataManager::Get() methods
- Wire up metrics in P2PClientService initialization

* [Store] Improve local storage metrics accuracy and maintainability

- Add put_failures and get_failures to LocalStorageMetric::summary_metrics()
- Refactor PutViaMemcpy metrics recording into a helper lambda to avoid duplication
- Move Get metrics recording into async task handle for accurate latency measurement
- Record get_failures when async transfer fails (previously missed)

Co-Authored-By: Claude (antchat/GLM-5) <noreply@anthropic.com>

* [Store] Add Stopwatch class and simplify metrics timing code

- Add Stopwatch class in client_metric.h for elapsed time measurement
- Use Stopwatch to simplify latency recording in PutViaTe, PutViaMemcpy, and Get functions
- Method name elapsed_us() follows snake_case convention in codebase

Co-Authored-By: Claude (antchat/GLM-5) <noreply@anthropic.com>

* [Store] Move P2P local storage metrics from DataManager to P2PClientService

- Add P2PClientMetric struct for tracking local Get/Put operations
- Remove LocalStorageMetric integration from DataManager
- Record metrics in P2PClientService at CreateGetHandle/CreateLocalPutHandle
- Add GetSummaryMetrics() and SerializeMetrics() for P2P metrics export

Co-Authored-By: Claude (antchat/GLM-5) <noreply@anthropic.com>

* [Store] Fix P2P client metrics timing and recording positions

- Move latency measurement to Wait() calls for accurate timing
- Record requests at entry points (Put/Get/BatchPut/BatchGet)
- Record bytes only on successful operations
- Record failures at appropriate error points
- Add CalculateTotalBytes helper function
- Remove redundant metrics recording from CreateGetHandle/CreateLocalPutHandle

Co-Authored-By: Claude (antchat/GLM-5) <noreply@anthropic.com>

* [Store] Fix P2P client metrics timing to include handle creation

Move Stopwatch initialization before CreatePutHandle/CreateGetHandle calls
to accurately measure the full operation latency including handle creation time.
Also remove redundant CalculateTotalBytes helper in favor of CalculateSliceSize.

Co-Authored-By: Claude (antchat/GLM-5) <noreply@anthropic.com>

---------

Co-authored-by: wl177541 <wl177541@antgroup.com>
Co-authored-by: Claude (antchat/GLM-5) <noreply@anthropic.com>

Author: 3 people

mooncake-store/include/client_metric.h

@@ -22,6 +22,22 @@ const std::vector<double> kLatencyBucket = {
     // safeguards for long tails
     50000, 100000, 200000, 500000, 1000000};
 
+// Simple stopwatch for measuring elapsed time in microseconds
+class Stopwatch {
+   public:
+    Stopwatch() : start_time_(std::chrono::steady_clock::now()) {}
+
+    int64_t elapsed_us() const {
+        auto now = std::chrono::steady_clock::now();
+        return std::chrono::duration_cast<std::chrono::microseconds>(
+                   now - start_time_)
+            .count();
+    }
+
+   private:
+    std::chrono::steady_clock::time_point start_time_;
+};
+
 static inline std::string get_env_or_default(
     const char* env_var, const std::string& default_val = "") {
     const char* val = getenv(env_var);
@@ -276,6 +292,16 @@ struct ClientMetric {
     TransferMetric transfer_metric;
     MasterClientMetric master_client_metric;
 
+    /**
+     * @brief Check if metrics are enabled via environment variable
+     * @return true if enabled, false if disabled
+     *
+     * Environment variable:
+     * - MC_STORE_CLIENT_METRIC: Enable/disable metrics (enabled by default,
+     *   set to 0/false to disable)
+     */
+    static bool IsEnabled();
+
     /**
      * @brief Creates a ClientMetric instance based on environment variables
      * @return std::unique_ptr<ClientMetric> containing the instance if enabled,

mooncake-store/include/client_service.h

@@ -279,7 +279,7 @@ class ClientService {
         const std::vector<std::string>& keys) = 0;
 
     // For human-readable metrics
-    tl::expected<std::string, ErrorCode> GetSummaryMetrics() {
+    virtual tl::expected<std::string, ErrorCode> GetSummaryMetrics() {
         if (metrics_ == nullptr) {
             return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
         }
@@ -297,7 +297,7 @@ class ClientService {
     }
 
     // For Prometheus-style metrics
-    tl::expected<std::string, ErrorCode> SerializeMetrics() {
+    virtual tl::expected<std::string, ErrorCode> SerializeMetrics() {
         if (metrics_ == nullptr) {
             return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
         }

mooncake-store/include/p2p_client_metric.h

@@ -0,0 +1,29 @@
+#pragma once
+
+#include "client_metric.h"
+
+namespace mooncake {
+
+// P2P client metrics for local storage operations
+struct P2PClientMetric {
+    // Local Get metrics
+    ylt::metric::counter_t local_get_requests;
+    ylt::metric::counter_t local_get_hits;
+    ylt::metric::counter_t local_get_misses;
+    ylt::metric::counter_t local_get_failures;
+    ylt::metric::counter_t local_get_bytes;
+    ylt::metric::histogram_t local_get_latency;
+
+    // Local Put metrics
+    ylt::metric::counter_t local_put_requests;
+    ylt::metric::counter_t local_put_failures;
+    ylt::metric::counter_t local_put_bytes;
+    ylt::metric::histogram_t local_put_latency;
+
+    explicit P2PClientMetric(std::map<std::string, std::string> labels = {});
+
+    void serialize(std::string& str);
+    std::string summary_metrics();
+};
+
+}  // namespace mooncake

mooncake-store/include/p2p_client_service.h

@@ -17,6 +17,7 @@
 #include "client_rpc_service.h"
 #include "ha_recovery_manager.h"
 #include "peer_client.h"
+#include "p2p_client_metric.h"
 #include "p2p_master_client.h"
 #include "route_cache.h"
 #include "task_handle.h"
@@ -173,6 +174,9 @@ class P2PClientService final : public ClientService {
 
     std::string GetHealthStatus() const override;
 
+    tl::expected<std::string, ErrorCode> GetSummaryMetrics() override;
+    tl::expected<std::string, ErrorCode> SerializeMetrics() override;
+
    private:
     /**
      * @brief init TieredBackend and DataManager
@@ -352,6 +356,9 @@ class P2PClientService final : public ClientService {
 
     // HA recovery manager
     std::unique_ptr<HARecoveryManager> ha_manager_;
+
+    // P2P local storage metrics
+    std::unique_ptr<P2PClientMetric> p2p_metrics_;
 };
 
 }  // namespace mooncake

mooncake-store/src/CMakeLists.txt

@@ -10,6 +10,7 @@ set(MOONCAKE_STORE_SOURCES
     client_service.cpp
     centralized_client_service.cpp
     p2p_client_service.cpp
+    p2p_client_metric.cpp
     client_metric.cpp
     types.cpp
     master_client.cpp

mooncake-store/src/client_metric.cpp

@@ -67,6 +67,8 @@ ClientMetric::ClientMetric(uint64_t interval_seconds,
 
 ClientMetric::~ClientMetric() { StopMetricsReportingThread(); }
 
+bool ClientMetric::IsEnabled() { return parseMetricsEnabled(); }
+
 std::unique_ptr<ClientMetric> ClientMetric::Create(
     std::map<std::string, std::string> labels) {
     if (!parseMetricsEnabled()) {

mooncake-store/src/p2p_client_metric.cpp

@@ -0,0 +1,58 @@
+#include "p2p_client_metric.h"
+
+namespace mooncake {
+
+P2PClientMetric::P2PClientMetric(std::map<std::string, std::string> labels)
+    : local_get_requests("mooncake_p2p_local_get_requests_total",
+                         "Total number of local Get requests", labels),
+      local_get_hits("mooncake_p2p_local_get_hits_total",
+                     "Total number of local Get hits (found in local storage)",
+                     labels),
+      local_get_misses("mooncake_p2p_local_get_misses_total",
+                       "Total number of local Get misses (not found locally)",
+                       labels),
+      local_get_failures("mooncake_p2p_local_get_failures_total",
+                         "Total number of failed local Get requests", labels),
+      local_get_bytes("mooncake_p2p_local_get_bytes_total",
+                      "Total bytes read by local Get", labels),
+      local_get_latency("mooncake_p2p_local_get_latency_us",
+                        "Local Get latency (us)", kLatencyBucket, labels),
+      local_put_requests("mooncake_p2p_local_put_requests_total",
+                         "Total number of local Put requests", labels),
+      local_put_failures("mooncake_p2p_local_put_failures_total",
+                         "Total number of failed local Put requests", labels),
+      local_put_bytes("mooncake_p2p_local_put_bytes_total",
+                      "Total bytes written by local Put", labels),
+      local_put_latency("mooncake_p2p_local_put_latency_us",
+                        "Local Put latency (us)", kLatencyBucket, labels) {}
+
+void P2PClientMetric::serialize(std::string& str) {
+    local_get_requests.serialize(str);
+    local_get_hits.serialize(str);
+    local_get_misses.serialize(str);
+    local_get_failures.serialize(str);
+    local_get_bytes.serialize(str);
+    local_get_latency.serialize(str);
+    local_put_requests.serialize(str);
+    local_put_failures.serialize(str);
+    local_put_bytes.serialize(str);
+    local_put_latency.serialize(str);
+}
+
+std::string P2PClientMetric::summary_metrics() {
+    std::stringstream ss;
+    ss << "=== P2P Local Storage Metrics ===\n";
+
+    ss << "Local Get: " << local_get_requests.value() << " requests, "
+       << local_get_hits.value() << " hits, " << local_get_misses.value()
+       << " misses, " << local_get_failures.value() << " failures, "
+       << byte_size_to_string(local_get_bytes.value()) << " read\n";
+
+    ss << "Local Put: " << local_put_requests.value() << " requests, "
+       << local_put_failures.value() << " failures, "
+       << byte_size_to_string(local_put_bytes.value()) << " written\n";
+
+    return ss.str();
+}
+
+}  // namespace mooncake