diff --git a/mooncake-store/include/client.h b/mooncake-store/include/client.h
index 7cf8fb48..c1b190ee 100644
--- a/mooncake-store/include/client.h
+++ b/mooncake-store/include/client.h
@@ -5,9 +5,11 @@
 #include <mutex>
 #include <optional>
 #include <string>
+#include <thread>
 #include <vector>
 #include <ylt/util/tl/expected.hpp>
 
+#include "client_metric.h"
 #include "ha_helper.h"
 #include "master_client.h"
 #include "storage_backend.h"
@@ -196,6 +198,24 @@ class Client {
     std::vector<tl::expected<bool, ErrorCode>> BatchIsExist(
         const std::vector<std::string>& keys);
 
+    // For human-readable metrics
+    tl::expected<std::string, ErrorCode> GetSummaryMetrics() {
+        if (metrics_ == nullptr) {
+            return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
+        }
+        return metrics_->summary_metrics();
+    }
+
+    // For Prometheus-style metrics
+    tl::expected<std::string, ErrorCode> SerializeMetrics() {
+        if (metrics_ == nullptr) {
+            return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
+        }
+        std::string str;
+        metrics_->serialize(str);
+        return str;
+    }
+
    private:
     /**
      * @brief Private constructor to enforce creation through Create() method
@@ -255,6 +275,9 @@ class Client {
     std::vector<tl::expected<void, ErrorCode>> CollectResults(
         const std::vector<PutOperation>& ops);
 
+    // Client-side metrics
+    std::unique_ptr<ClientMetric> metrics_;
+
     // Core components
     TransferEngine transfer_engine_;
     MasterClient master_client_;
diff --git a/mooncake-store/include/client_metric.h b/mooncake-store/include/client_metric.h
new file mode 100644
index 00000000..20882b2e
--- /dev/null
+++ b/mooncake-store/include/client_metric.h
@@ -0,0 +1,276 @@
+#pragma once
+
+#include <atomic>
+#include <sstream>
+#include <thread>
+#include <vector>
+#include <ylt/metric/counter.hpp>
+#include <ylt/metric/histogram.hpp>
+#include <ylt/metric/summary.hpp>
+#include "utils.h"
+
+namespace mooncake {
+
+// latency bucket is in microsecond
+// Tuned for RDMA: fine-grained in <1ms, with ms-scale tail up to 1s
+const std::vector<double> kLatencyBucket = {
+    // sub-ms to 1ms region
+    125, 150, 200, 250, 300, 400, 500, 750, 1000,
+    // ms-level tail for batch/occasional spikes
+    1500, 2000, 3000, 5000, 7000, 15000, 20000,
+    // safeguards for long tails
+    50000, 100000, 200000, 500000, 1000000};
+
+struct TransferMetric {
+    ylt::metric::counter_t total_read_bytes{"mooncake_transfer_read_bytes",
+                                            "Total bytes read"};
+    ylt::metric::counter_t total_write_bytes{"mooncake_transfer_write_bytes",
+                                             "Total bytes written"};
+    ylt::metric::histogram_t batch_put_latency_us{
+        "mooncake_transfer_batch_put_latency",
+        "Batch Put transfer latency (us)", kLatencyBucket};
+    ylt::metric::histogram_t batch_get_latency_us{
+        "mooncake_transfer_batch_get_latency",
+        "Batch Get transfer latency (us)", kLatencyBucket};
+    ylt::metric::histogram_t get_latency_us{"mooncake_transfer_get_latency",
+                                            "Get transfer latency (us)",
+                                            kLatencyBucket};
+    ylt::metric::histogram_t put_latency_us{"mooncake_transfer_put_latency",
+                                            "Put transfer latency (us)",
+                                            kLatencyBucket};
+
+    void serialize(std::string& str) {
+        total_read_bytes.serialize(str);
+        total_write_bytes.serialize(str);
+        batch_put_latency_us.serialize(str);
+        batch_get_latency_us.serialize(str);
+        get_latency_us.serialize(str);
+        put_latency_us.serialize(str);
+    }
+
+    std::string summary_metrics() {
+        std::stringstream ss;
+        ss << "=== Transfer Metrics Summary ===\n";
+
+        // Bytes transferred
+        auto read_bytes = total_read_bytes.value();
+        auto write_bytes = total_write_bytes.value();
+        ss << "Total Read: " << byte_size_to_string(read_bytes) << "\n";
+        ss << "Total Write: " << byte_size_to_string(write_bytes) << "\n";
+
+        // Latency summaries
+        ss << "\n=== Latency Summary (microseconds) ===\n";
+        ss << "Get: " << format_latency_summary(get_latency_us) << "\n";
+        ss << "Put: " << format_latency_summary(put_latency_us) << "\n";
+        ss << "Batch Get: " << format_latency_summary(batch_get_latency_us)
+           << "\n";
+        ss << "Batch Put: " << format_latency_summary(batch_put_latency_us)
+           << "\n";
+
+        return ss.str();
+    }
+
+   private:
+    std::string format_latency_summary(ylt::metric::histogram_t& hist) {
+        // Access the internal sum and bucket counts
+        auto sum_ptr =
+            const_cast<ylt::metric::histogram_t&>(hist).get_bucket_counts();
+        if (sum_ptr.empty()) {
+            return "No data";
+        }
+
+        // Calculate total count from all buckets
+        int64_t total_count = 0;
+        for (auto& bucket : sum_ptr) {
+            total_count += bucket->value();
+        }
+
+        if (total_count == 0) {
+            return "No data";
+        }
+
+        // Get sum from the histogram's internal sum gauge
+        // Note: We need to access the private sum_ member, which requires
+        // friendship or reflection For now, let's use a simpler approach
+        // showing just count
+        std::stringstream ss;
+        ss << "count=" << total_count;
+
+        // Find P95
+        int64_t p95_target = (total_count * 95) / 100;
+        int64_t cumulative = 0;
+        double p95_bucket = 0;
+
+        for (size_t i = 0; i < sum_ptr.size() && i < kLatencyBucket.size();
+             i++) {
+            cumulative += sum_ptr[i]->value();
+            if (cumulative >= p95_target && p95_bucket == 0) {
+                p95_bucket = kLatencyBucket[i];
+                break;
+            }
+        }
+
+        if (p95_bucket > 0) {
+            ss << ", p95<" << p95_bucket << "μs";
+        }
+
+        // Find max bucket (highest bucket with data)
+        double max_bucket = 0;
+        for (size_t i = sum_ptr.size(); i > 0; i--) {
+            size_t idx = i - 1;
+            if (idx < kLatencyBucket.size() && sum_ptr[idx]->value() > 0) {
+                max_bucket = kLatencyBucket[idx];
+                break;
+            }
+        }
+
+        if (max_bucket > 0) {
+            ss << ", max<" << max_bucket << "μs";
+        }
+
+        return ss.str();
+    }
+};
+
+struct MasterClientMetric {
+    std::array<std::string, 1> rpc_names = {"rpc_name"};
+
+    MasterClientMetric()
+        : rpc_count("mooncake_client_rpc_count",
+                    "Total number of RPC calls made by the client", rpc_names),
+          rpc_latency("mooncake_client_rpc_latency",
+                      "Latency of RPC calls made by the client (in us)",
+                      kLatencyBucket, rpc_names) {}
+
+    ylt::metric::dynamic_counter_1t rpc_count;
+    ylt::metric::dynamic_histogram_1t rpc_latency;
+    void serialize(std::string& str) {
+        rpc_count.serialize(str);
+        rpc_latency.serialize(str);
+    }
+
+    std::string summary_metrics() {
+        std::stringstream ss;
+        ss << "=== RPC Metrics Summary ===\n";
+
+        // For dynamic metrics, we need to check if there are any labels with
+        // data
+        if (rpc_count.label_value_count() == 0) {
+            ss << "No RPC calls recorded\n";
+            return ss.str();
+        }
+
+        // Get all available RPC names from the dynamic metrics
+        // We'll iterate through all possible RPC names instead of using a fixed
+        // list
+        std::vector<std::string> all_rpc_names = {"GetReplicaList",
+                                                  "PutStart",
+                                                  "PutEnd",
+                                                  "PutRevoke",
+                                                  "ExistKey",
+                                                  "Remove",
+                                                  "RemoveAll",
+                                                  "MountSegment",
+                                                  "UnmountSegment",
+                                                  "GetFsdir",
+                                                  "BatchGetReplicaList",
+                                                  "BatchPutStart",
+                                                  "BatchPutEnd",
+                                                  "BatchPutRevoke"};
+
+        bool found_any = false;
+        for (const auto& rpc_name : all_rpc_names) {
+            std::array<std::string, 1> label_array = {rpc_name};
+
+            // Check if this RPC has any data by trying to access bucket counts
+            auto bucket_counts = rpc_latency.get_bucket_counts();
+            int64_t total_count = 0;
+            for (auto& bucket : bucket_counts) {
+                total_count += bucket->value(label_array);
+            }
+
+            // Skip RPCs with zero count
+            if (total_count == 0) continue;
+
+            found_any = true;
+            ss << rpc_name << ": count=" << total_count;
+
+            // Find P95
+            int64_t p95_target = (total_count * 95) / 100;
+            int64_t cumulative = 0;
+            double p95_bucket = 0;
+
+            for (size_t i = 0;
+                 i < bucket_counts.size() && i < kLatencyBucket.size(); i++) {
+                cumulative += bucket_counts[i]->value(label_array);
+                if (cumulative >= p95_target && p95_bucket == 0) {
+                    p95_bucket = kLatencyBucket[i];
+                    break;
+                }
+            }
+
+            if (p95_bucket > 0) {
+                ss << ", p95<" << p95_bucket << "μs";
+            }
+
+            // Find max bucket (highest bucket with data)
+            double max_bucket = 0;
+            for (size_t i = bucket_counts.size(); i > 0; i--) {
+                size_t idx = i - 1;
+                if (idx < kLatencyBucket.size() &&
+                    bucket_counts[idx]->value(label_array) > 0) {
+                    max_bucket = kLatencyBucket[idx];
+                    break;
+                }
+            }
+
+            if (max_bucket > 0) {
+                ss << ", max<" << max_bucket << "μs";
+            }
+
+            ss << "\n";
+        }
+
+        if (!found_any) {
+            ss << "No RPC calls recorded\n";
+        }
+
+        return ss.str();
+    }
+};
+
+struct ClientMetric {
+    TransferMetric transfer_metric;
+    MasterClientMetric master_client_metric;
+
+    /**
+     * @brief Creates a ClientMetric instance based on environment variables
+     * @return std::unique_ptr<ClientMetric> containing the instance if enabled,
+     *         nullptr if disabled
+     *
+     * Environment variables:
+     * - MC_STORE_CLIENT_METRIC: Enable/disable metrics (enabled by default,
+     *   set to 0/false to disable)
+     * - MC_STORE_CLIENT_METRIC_INTERVAL: Reporting interval in seconds
+     *   (default: 0, 0 = collect but don't report)
+     */
+    static std::unique_ptr<ClientMetric> Create();
+
+    void serialize(std::string& str);
+    std::string summary_metrics();
+
+    uint64_t GetReportingInterval() const { return metrics_interval_seconds_; }
+
+    explicit ClientMetric(uint64_t interval_seconds = 0);
+    ~ClientMetric();
+
+   private:
+    // Metrics reporting thread management
+    std::jthread metrics_reporting_thread_;
+    std::atomic<bool> should_stop_metrics_thread_{false};
+    uint64_t metrics_interval_seconds_{0};
+
+    void StartMetricsReportingThread();
+    void StopMetricsReportingThread();
+};
+};  // namespace mooncake
\ No newline at end of file
diff --git a/mooncake-store/include/master_client.h b/mooncake-store/include/master_client.h
index a3e7404a..b576eeef 100644
--- a/mooncake-store/include/master_client.h
+++ b/mooncake-store/include/master_client.h
@@ -4,13 +4,10 @@
 #include <string>
 #include <vector>
 #include <ylt/coro_rpc/coro_rpc_client.hpp>
+#include "client_metric.h"
 
-#include "rpc_service.h"
 #include "types.h"
 
-using namespace async_simple;
-using namespace coro_rpc;
-
 namespace mooncake {
 
 static const std::string kDefaultMasterAddress = "localhost:50051";
@@ -20,7 +17,7 @@ static const std::string kDefaultMasterAddress = "localhost:50051";
  */
 class MasterClient {
    public:
-    MasterClient();
+    MasterClient(MasterClientMetric* metrics = nullptr) : metrics_(metrics) {}
     ~MasterClient();
 
     MasterClient(const MasterClient&) = delete;
@@ -219,22 +216,25 @@ class MasterClient {
      */
     class RpcClientAccessor {
        public:
-        void SetClient(std::shared_ptr<coro_rpc_client> client) {
+        void SetClient(std::shared_ptr<coro_rpc::coro_rpc_client> client) {
             std::lock_guard<std::shared_mutex> lock(client_mutex_);
             client_ = client;
         }
 
-        std::shared_ptr<coro_rpc_client> GetClient() {
+        std::shared_ptr<coro_rpc::coro_rpc_client> GetClient() {
             std::shared_lock<std::shared_mutex> lock(client_mutex_);
             return client_;
         }
 
        private:
         mutable std::shared_mutex client_mutex_;
-        std::shared_ptr<coro_rpc_client> client_;
+        std::shared_ptr<coro_rpc::coro_rpc_client> client_;
     };
     RpcClientAccessor client_accessor_;
 
+    // Metrics for tracking RPC operations
+    MasterClientMetric* metrics_;
+
     // Mutex to insure the Connect function is atomic.
     mutable Mutex connect_mutex_;
     // The address which is passed to the coro_rpc_client
diff --git a/mooncake-store/include/transfer_task.h b/mooncake-store/include/transfer_task.h
index 753fef76..423317b9 100644
--- a/mooncake-store/include/transfer_task.h
+++ b/mooncake-store/include/transfer_task.h
@@ -2,6 +2,7 @@
 
 #include <atomic>
 #include <condition_variable>
+#include <cstdint>
 #include <cstring>
 #include <memory>
 #include <mutex>
@@ -16,6 +17,7 @@
 #include "transport/transport.h"
 #include "types.h"
 #include "storage_backend.h"
+#include "client_metric.h"
 
 namespace mooncake {
 
@@ -349,7 +351,8 @@ class TransferSubmitter {
    public:
     explicit TransferSubmitter(TransferEngine& engine,
                                const std::string& local_hostname,
-                               std::shared_ptr<StorageBackend>& backend);
+                               std::shared_ptr<StorageBackend>& backend,
+                               TransferMetric* transfer_metric = nullptr);
 
     /**
      * @brief Submit an asynchronous transfer operation
@@ -374,6 +377,7 @@ class TransferSubmitter {
     std::unique_ptr<MemcpyWorkerPool> memcpy_pool_;
     std::unique_ptr<FilereadWorkerPool> fileread_pool_;
     bool memcpy_enabled_;
+    TransferMetric* transfer_metric_;
 
     /**
      * @brief Select the optimal transfer strategy
@@ -412,6 +416,12 @@ class TransferSubmitter {
     std::optional<TransferFuture> submitFileReadOperation(
         const Replica::Descriptor& replica, std::vector<Slice>& slices,
         Transport::TransferRequest::OpCode op_code);
+
+    /**
+     * @brief Calculate total bytes for transfer operation and update metrics
+     */
+    void updateTransferMetrics(const std::vector<Slice>& slices,
+                               Transport::TransferRequest::OpCode op);
 };
 
 }  // namespace mooncake
\ No newline at end of file
diff --git a/mooncake-store/src/CMakeLists.txt b/mooncake-store/src/CMakeLists.txt
index fab7df13..812cc741 100644
--- a/mooncake-store/src/CMakeLists.txt
+++ b/mooncake-store/src/CMakeLists.txt
@@ -5,6 +5,7 @@ set(MOONCAKE_STORE_SOURCES
     allocator.cpp
     master_service.cpp
     client.cpp
+    client_metric.cpp
     types.cpp
     master_client.cpp
     utils.cpp
diff --git a/mooncake-store/src/client.cpp b/mooncake-store/src/client.cpp
index 6e73df11..b190d48c 100644
--- a/mooncake-store/src/client.cpp
+++ b/mooncake-store/src/client.cpp
@@ -6,8 +6,10 @@
 #include <cassert>
 #include <chrono>
 #include <cstdint>
+#include <cstdlib>
 #include <optional>
 #include <ranges>
+#include <thread>
 
 #include "transfer_engine.h"
 #include "transfer_task.h"
@@ -36,12 +38,28 @@ namespace mooncake {
 Client::Client(const std::string& local_hostname,
                const std::string& metadata_connstring,
                const std::string& storage_root_dir)
-    : local_hostname_(local_hostname),
+    : metrics_(ClientMetric::Create()),
+      master_client_(metrics_ ? &metrics_->master_client_metric : nullptr),
+      local_hostname_(local_hostname),
       metadata_connstring_(metadata_connstring),
       storage_root_dir_(storage_root_dir),
       write_thread_pool_(2) {
     client_id_ = generate_uuid();
     LOG(INFO) << "client_id=" << client_id_;
+
+    if (metrics_) {
+        if (metrics_->GetReportingInterval() > 0) {
+            LOG(INFO) << "Client metrics enabled with reporting thread started "
+                         "(interval: "
+                      << metrics_->GetReportingInterval() << "s)";
+        } else {
+            LOG(INFO)
+                << "Client metrics enabled but reporting disabled (interval=0)";
+        }
+    } else {
+        LOG(INFO) << "Client metrics disabled (set MC_STORE_CLIENT_METRIC=1 to "
+                     "enable)";
+    }
 }
 
 Client::~Client() {
@@ -232,7 +250,8 @@ ErrorCode Client::InitTransferEngine(const std::string& local_hostname,
 
     // Initialize TransferSubmitter after transfer engine is ready
     transfer_submitter_ = std::make_unique<TransferSubmitter>(
-        transfer_engine_, local_hostname, storage_backend_);
+        transfer_engine_, local_hostname, storage_backend_,
+        metrics_ ? &metrics_->transfer_metric : nullptr);
 
     return ErrorCode::OK;
 }
@@ -400,7 +419,15 @@ tl::expected<void, ErrorCode> Client::Get(
         return tl::unexpected(err);
     }
 
+    auto t0_get = std::chrono::steady_clock::now();
     err = TransferRead(replica, slices);
+    auto us_get = std::chrono::duration_cast<std::chrono::microseconds>(
+                      std::chrono::steady_clock::now() - t0_get)
+                      .count();
+    if (metrics_) {
+        metrics_->transfer_metric.get_latency_us.observe(us_get);
+    }
+
     if (err != ErrorCode::OK) {
         LOG(ERROR) << "transfer_read_failed key=" << object_key;
         return tl::unexpected(err);
@@ -439,6 +466,8 @@ std::vector<tl::expected<void, ErrorCode>> Client::BatchGet(
     std::vector<std::tuple<size_t, std::string, TransferFuture>>
         pending_transfers;
     std::vector<tl::expected<void, ErrorCode>> results(object_keys.size());
+    // Record batch get transfer latency (Submit + Wait)
+    auto t0_batch_get = std::chrono::steady_clock::now();
 
     // Submit all transfers in parallel
     for (size_t i = 0; i < object_keys.size(); ++i) {
@@ -492,6 +521,13 @@ std::vector<tl::expected<void, ErrorCode>> Client::BatchGet(
         }
     }
 
+    auto us_batch_get = std::chrono::duration_cast<std::chrono::microseconds>(
+                            std::chrono::steady_clock::now() - t0_batch_get)
+                            .count();
+    if (metrics_) {
+        metrics_->transfer_metric.batch_get_latency_us.observe(us_batch_get);
+    }
+
     VLOG(1) << "BatchGet completed for " << object_keys.size() << " keys";
     return results;
 }
@@ -517,6 +553,9 @@ tl::expected<void, ErrorCode> Client::Put(const ObjectKey& key,
         return tl::unexpected(err);
     }
 
+    // Record Put transfer latency (all replicas)
+    auto t0_put = std::chrono::steady_clock::now();
+
     // Transfer data using allocated handles from all replicas
     for (const auto& replica : start_result.value()) {
         ErrorCode transfer_err = TransferWrite(replica, slices);
@@ -531,6 +570,13 @@ tl::expected<void, ErrorCode> Client::Put(const ObjectKey& key,
         }
     }
 
+    auto us_put = std::chrono::duration_cast<std::chrono::microseconds>(
+                      std::chrono::steady_clock::now() - t0_put)
+                      .count();
+    if (metrics_) {
+        metrics_->transfer_metric.put_latency_us.observe(us_put);
+    }
+
     // End put operation
     auto end_result = master_client_.PutEnd(key);
     if (!end_result) {
@@ -933,8 +979,17 @@ std::vector<tl::expected<void, ErrorCode>> Client::BatchPut(
     const ReplicateConfig& config) {
     std::vector<PutOperation> ops = CreatePutOperations(keys, batched_slices);
     StartBatchPut(ops, config);
+
+    auto t0 = std::chrono::steady_clock::now();
     SubmitTransfers(ops);
     WaitForTransfers(ops);
+    auto us = std::chrono::duration_cast<std::chrono::microseconds>(
+                  std::chrono::steady_clock::now() - t0)
+                  .count();
+    if (metrics_) {
+        metrics_->transfer_metric.batch_put_latency_us.observe(us);
+    }
+
     FinalizeBatchPut(ops);
     BatchPuttoLocalFile(ops);
     return CollectResults(ops);
diff --git a/mooncake-store/src/client_metric.cpp b/mooncake-store/src/client_metric.cpp
new file mode 100644
index 00000000..4678bf7b
--- /dev/null
+++ b/mooncake-store/src/client_metric.cpp
@@ -0,0 +1,133 @@
+#include "client_metric.h"
+
+#include <glog/logging.h>
+#include <algorithm>
+#include <cctype>
+#include <chrono>
+#include <cstdlib>
+#include <thread>
+
+namespace mooncake {
+
+namespace {
+
+std::string toLower(const std::string& str) {
+    std::string result = str;
+    std::transform(result.begin(), result.end(), result.begin(),
+                   [](unsigned char c) { return std::tolower(c); });
+    return result;
+}
+
+bool parseMetricsEnabled() {
+    const char* metric_env = std::getenv("MC_STORE_CLIENT_METRIC");
+    if (!metric_env) {
+        return true;
+    }
+    std::string value = toLower(metric_env);
+    return (value == "1" || value == "true" || value == "yes" ||
+            value == "on" || value == "enable");
+}
+
+uint64_t parseMetricsInterval() {
+    const char* interval_env = std::getenv("MC_STORE_CLIENT_METRIC_INTERVAL");
+    if (!interval_env) {
+        // Default to disabled
+        return 0;
+    }
+
+    try {
+        uint64_t interval = std::stoull(interval_env);
+        if (interval == 0) {
+            LOG(INFO) << "Client metrics reporting disabled (interval=0) via "
+                         "MC_STORE_CLIENT_METRIC_INTERVAL";
+        } else {
+            LOG(INFO) << "Client metrics interval set to " << interval
+                      << "s via MC_STORE_CLIENT_METRIC_INTERVAL";
+        }
+        return interval;
+    } catch (const std::exception& e) {
+        LOG(WARNING) << "Failed to parse MC_STORE_CLIENT_METRIC_INTERVAL: "
+                     << interval_env << ", disabling metrics reporting";
+        return 0;
+    }
+}
+
+}  // anonymous namespace
+
+ClientMetric::ClientMetric(uint64_t interval_seconds)
+    : should_stop_metrics_thread_(false),
+      metrics_interval_seconds_(interval_seconds) {
+    if (metrics_interval_seconds_ > 0) {
+        StartMetricsReportingThread();
+    }
+}
+
+ClientMetric::~ClientMetric() { StopMetricsReportingThread(); }
+
+std::unique_ptr<ClientMetric> ClientMetric::Create() {
+    if (!parseMetricsEnabled()) {
+        LOG(INFO) << "Client metrics disabled (set MC_STORE_CLIENT_METRIC=0 to "
+                     "disable)";
+        return nullptr;
+    }
+
+    uint64_t interval = parseMetricsInterval();
+
+    LOG(INFO) << "Client metrics enabled (default enabled)";
+
+    return std::make_unique<ClientMetric>(interval);
+}
+
+void ClientMetric::serialize(std::string& str) {
+    transfer_metric.serialize(str);
+    master_client_metric.serialize(str);
+}
+
+std::string ClientMetric::summary_metrics() {
+    std::stringstream ss;
+    ss << "Client Metrics Summary\n";
+    ss << transfer_metric.summary_metrics();
+    ss << "\n";
+    ss << master_client_metric.summary_metrics();
+    return ss.str();
+}
+
+void ClientMetric::StartMetricsReportingThread() {
+    should_stop_metrics_thread_ = false;
+    metrics_reporting_thread_ = std::jthread([this](
+                                                 std::stop_token stop_token) {
+        LOG(INFO) << "Client metrics reporting thread started (interval: "
+                  << metrics_interval_seconds_ << "s)";
+
+        while (!stop_token.stop_requested() && !should_stop_metrics_thread_) {
+            // Sleep for the interval, checking periodically for stop signal
+            for (uint64_t i = 0;
+                 i < metrics_interval_seconds_ &&
+                 !stop_token.stop_requested() && !should_stop_metrics_thread_;
+                 ++i) {
+                std::this_thread::sleep_for(std::chrono::seconds(1));
+            }
+
+            if (stop_token.stop_requested() || should_stop_metrics_thread_) {
+                break;  // Exit if stopped during sleep
+            }
+
+            // Print metrics summary
+            std::string summary = summary_metrics();
+            LOG(INFO) << "Client Metrics Report:\n" << summary;
+        }
+        LOG(INFO) << "Client metrics reporting thread stopped";
+    });
+}
+
+void ClientMetric::StopMetricsReportingThread() {
+    should_stop_metrics_thread_ = true;  // Signal the thread to stop
+    if (metrics_reporting_thread_.joinable()) {
+        LOG(INFO) << "Waiting for client metrics reporting thread to join...";
+        metrics_reporting_thread_.request_stop();
+        metrics_reporting_thread_.join();  // Wait for the thread to finish
+        LOG(INFO) << "Client metrics reporting thread joined";
+    }
+}
+
+}  // namespace mooncake
\ No newline at end of file
diff --git a/mooncake-store/src/master_client.cpp b/mooncake-store/src/master_client.cpp
index eb9b5693..3287f71b 100644
--- a/mooncake-store/src/master_client.cpp
+++ b/mooncake-store/src/master_client.cpp
@@ -14,10 +14,97 @@
 #include "types.h"
 #include "utils/scoped_vlog_timer.h"
 
+#include <source_location>
+
 namespace mooncake {
 
-using namespace coro_rpc;
-using namespace async_simple::coro;
+template <auto Method>
+struct RpcNameTraits;
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::ExistKey> {
+    static constexpr const char* value = "ExistKey";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::BatchExistKey> {
+    static constexpr const char* value = "BatchExistKey";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::GetReplicaList> {
+    static constexpr const char* value = "GetReplicaList";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::BatchGetReplicaList> {
+    static constexpr const char* value = "BatchGetReplicaList";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::PutStart> {
+    static constexpr const char* value = "PutStart";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::BatchPutStart> {
+    static constexpr const char* value = "BatchPutStart";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::PutEnd> {
+    static constexpr const char* value = "PutEnd";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::BatchPutEnd> {
+    static constexpr const char* value = "BatchPutEnd";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::PutRevoke> {
+    static constexpr const char* value = "PutRevoke";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::BatchPutRevoke> {
+    static constexpr const char* value = "BatchPutRevoke";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::Remove> {
+    static constexpr const char* value = "Remove";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::RemoveAll> {
+    static constexpr const char* value = "RemoveAll";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::MountSegment> {
+    static constexpr const char* value = "MountSegment";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::ReMountSegment> {
+    static constexpr const char* value = "ReMountSegment";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::UnmountSegment> {
+    static constexpr const char* value = "UnmountSegment";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::Ping> {
+    static constexpr const char* value = "Ping";
+};
+
+template <>
+struct RpcNameTraits<&WrappedMasterService::GetFsdir> {
+    static constexpr const char* value = "GetFsdir";
+};
 
 template <auto ServiceMethod, typename ReturnType, typename... Args>
 tl::expected<ReturnType, ErrorCode> MasterClient::invoke_rpc(Args&&... args) {
@@ -27,15 +114,30 @@ tl::expected<ReturnType, ErrorCode> MasterClient::invoke_rpc(Args&&... args) {
         return tl::make_unexpected(ErrorCode::RPC_FAIL);
     }
 
+    // Increment RPC counter
+    if (metrics_) {
+        metrics_->rpc_count.inc({RpcNameTraits<ServiceMethod>::value});
+    }
+
+    auto start_time = std::chrono::steady_clock::now();
     auto request_result =
         client->send_request<ServiceMethod>(std::forward<Args>(args)...);
-    return coro::syncAwait(
-        [&]() -> coro::Lazy<tl::expected<ReturnType, ErrorCode>> {
+
+    return async_simple::coro::syncAwait(
+        [&]() -> async_simple::coro::Lazy<tl::expected<ReturnType, ErrorCode>> {
             auto result = co_await co_await request_result;
             if (!result) {
                 LOG(ERROR) << "RPC call failed: " << result.error().msg;
                 co_return tl::make_unexpected(ErrorCode::RPC_FAIL);
             }
+            if (metrics_) {
+                auto end_time = std::chrono::steady_clock::now();
+                auto latency =
+                    std::chrono::duration_cast<std::chrono::microseconds>(
+                        end_time - start_time);
+                metrics_->rpc_latency.observe(
+                    {RpcNameTraits<ServiceMethod>::value}, latency.count());
+            }
             co_return result->result();
         }());
 }
@@ -50,10 +152,17 @@ std::vector<tl::expected<ResultType, ErrorCode>> MasterClient::invoke_batch_rpc(
             input_size, tl::make_unexpected(ErrorCode::RPC_FAIL));
     }
 
+    // Increment RPC counter
+    if (metrics_) {
+        metrics_->rpc_count.inc({RpcNameTraits<ServiceMethod>::value});
+    }
+
+    auto start_time = std::chrono::steady_clock::now();
     auto request_result =
         client->send_request<ServiceMethod>(std::forward<Args>(args)...);
-    return coro::syncAwait(
-        [&]() -> coro::Lazy<std::vector<tl::expected<ResultType, ErrorCode>>> {
+    return async_simple::coro::syncAwait(
+        [&]() -> async_simple::coro::Lazy<
+                  std::vector<tl::expected<ResultType, ErrorCode>>> {
             auto result = co_await co_await request_result;
             if (!result) {
                 LOG(ERROR) << "Batch RPC call failed: " << result.error().msg;
@@ -65,21 +174,33 @@ std::vector<tl::expected<ResultType, ErrorCode>> MasterClient::invoke_batch_rpc(
                 }
                 co_return error_results;
             }
+            if (metrics_) {
+                auto end_time = std::chrono::steady_clock::now();
+                auto latency =
+                    std::chrono::duration_cast<std::chrono::microseconds>(
+                        end_time - start_time);
+                metrics_->rpc_latency.observe(
+                    {RpcNameTraits<ServiceMethod>::value}, latency.count());
+            }
             co_return result->result();
         }());
 }
 
-MasterClient::MasterClient() = default;
 MasterClient::~MasterClient() = default;
 
 ErrorCode MasterClient::Connect(const std::string& master_addr) {
     ScopedVLogTimer timer(1, "MasterClient::Connect");
     timer.LogRequest("master_addr=", master_addr);
 
+    auto location = std::source_location::current();
+    auto name = location.function_name();
+    LOG(INFO) << "Connecting to master at " << master_addr << " from " << name;
+
     MutexLocker lock(&connect_mutex_);
     if (client_addr_param_ == master_addr) {
         auto client = client_accessor_.GetClient();
-        auto result = coro::syncAwait(client->connect(master_addr));
+        auto result =
+            async_simple::coro::syncAwait(client->connect(master_addr));
         if (result.val() != 0) {
             LOG(ERROR) << "Failed to connect to master: " << result.message();
             timer.LogResponse("error_code=", ErrorCode::RPC_FAIL);
@@ -91,8 +212,9 @@ ErrorCode MasterClient::Connect(const std::string& master_addr) {
         // Once connected to address A, the coro_rpc_client does not support
         // connect to a new address B. So we need to create a new
         // coro_rpc_client if the address is different from the current one.
-        auto client = std::make_shared<coro_rpc_client>();
-        auto result = coro::syncAwait(client->connect(master_addr));
+        auto client = std::make_shared<coro_rpc::coro_rpc_client>();
+        auto result =
+            async_simple::coro::syncAwait(client->connect(master_addr));
         if (result.val() != 0) {
             LOG(ERROR) << "Failed to connect to master: " << result.message();
             timer.LogResponse("error_code=", ErrorCode::RPC_FAIL);
diff --git a/mooncake-store/src/transfer_task.cpp b/mooncake-store/src/transfer_task.cpp
index f35b84c5..98861420 100644
--- a/mooncake-store/src/transfer_task.cpp
+++ b/mooncake-store/src/transfer_task.cpp
@@ -5,8 +5,6 @@
 #include <algorithm>
 #include <cstdlib>
 
-#include "utils.h"
-
 namespace mooncake {
 
 // ============================================================================
@@ -356,11 +354,13 @@ TransferStrategy TransferFuture::strategy() const {
 
 TransferSubmitter::TransferSubmitter(TransferEngine& engine,
                                      const std::string& local_hostname,
-                                     std::shared_ptr<StorageBackend>& backend)
+                                     std::shared_ptr<StorageBackend>& backend,
+                                     TransferMetric* transfer_metric)
     : engine_(engine),
       local_hostname_(local_hostname),
       memcpy_pool_(std::make_unique<MemcpyWorkerPool>()),
-      fileread_pool_(std::make_unique<FilereadWorkerPool>(backend)) {
+      fileread_pool_(std::make_unique<FilereadWorkerPool>(backend)),
+      transfer_metric_(transfer_metric) {
     if (local_hostname_.empty()) {
         LOG(ERROR) << "Local hostname cannot be empty";
         throw std::invalid_argument("Local hostname cannot be empty");
@@ -395,6 +395,8 @@ TransferSubmitter::TransferSubmitter(TransferEngine& engine,
 std::optional<TransferFuture> TransferSubmitter::submit(
     const Replica::Descriptor& replica, std::vector<Slice>& slices,
     Transport::TransferRequest::OpCode op_code) {
+    std::optional<TransferFuture> future;
+
     if (replica.is_memory_replica()) {
         std::vector<AllocatedBuffer::Descriptor> handles;
         auto& mem_desc = replica.get_memory_descriptor();
@@ -408,16 +410,26 @@ std::optional<TransferFuture> TransferSubmitter::submit(
 
         switch (strategy) {
             case TransferStrategy::LOCAL_MEMCPY:
-                return submitMemcpyOperation(handles, slices, op_code);
+                future = submitMemcpyOperation(handles, slices, op_code);
+                break;
             case TransferStrategy::TRANSFER_ENGINE:
-                return submitTransferEngineOperation(handles, slices, op_code);
+                future =
+                    submitTransferEngineOperation(handles, slices, op_code);
+                break;
             default:
                 LOG(ERROR) << "Unknown transfer strategy: " << strategy;
                 return std::nullopt;
         }
     } else {
-        return submitFileReadOperation(replica, slices, op_code);
+        future = submitFileReadOperation(replica, slices, op_code);
     }
+
+    // Update metrics on successful submission
+    if (future.has_value()) {
+        updateTransferMetrics(slices, op_code);
+    }
+
+    return future;
 }
 
 std::optional<TransferFuture> TransferSubmitter::submitMemcpyOperation(
@@ -592,4 +604,24 @@ bool TransferSubmitter::validateTransferParams(
     return true;
 }
 
+void TransferSubmitter::updateTransferMetrics(
+    const std::vector<Slice>& slices,
+    Transport::TransferRequest::OpCode op_code) {
+    size_t total_bytes = 0;
+    for (const auto& slice : slices) {
+        total_bytes += slice.size;
+    }
+
+    if (transfer_metric_ == nullptr) {
+        return;
+    }
+
+    if (op_code == Transport::TransferRequest::READ) {
+        transfer_metric_->total_read_bytes.inc(total_bytes);
+
+    } else if (op_code == Transport::TransferRequest::WRITE) {
+        transfer_metric_->total_write_bytes.inc(total_bytes);
+    }
+}
+
 }  // namespace mooncake
diff --git a/mooncake-store/tests/CMakeLists.txt b/mooncake-store/tests/CMakeLists.txt
index f85b5b28..0e1ca581 100644
--- a/mooncake-store/tests/CMakeLists.txt
+++ b/mooncake-store/tests/CMakeLists.txt
@@ -136,4 +136,15 @@ target_link_libraries(pybind_client_test PUBLIC
 )
 add_test(NAME pybind_client_test COMMAND pybind_client_test)
 
+add_executable(client_metrics_test client_metrics_test.cpp)
+target_link_libraries(client_metrics_test PUBLIC
+    mooncake_store
+    cachelib_memory_allocator
+    glog
+    gtest
+    gtest_main
+    pthread
+)
+add_test(NAME client_metrics_test COMMAND client_metrics_test)
+
 add_subdirectory(e2e)
diff --git a/mooncake-store/tests/client_metrics_test.cpp b/mooncake-store/tests/client_metrics_test.cpp
new file mode 100644
index 00000000..ac49904f
--- /dev/null
+++ b/mooncake-store/tests/client_metrics_test.cpp
@@ -0,0 +1,178 @@
+#include <glog/logging.h>
+#include <gtest/gtest.h>
+
+#include <cstdlib>
+#include <string>
+
+#include "client_metric.h"
+
+namespace mooncake::test {
+
+class ClientMetricsTest : public ::testing::Test {
+   protected:
+    void SetUp() override {
+        google::InitGoogleLogging("ClientMetricsTest");
+        FLAGS_logtostderr = true;
+    }
+
+    void TearDown() override { google::ShutdownGoogleLogging(); }
+};
+
+TEST_F(ClientMetricsTest, TransferMetricsSummaryTest) {
+    TransferMetric metrics;
+
+    // Test empty metrics
+    std::string summary = metrics.summary_metrics();
+    EXPECT_TRUE(summary.find("Total Read: 0 B") != std::string::npos);
+    EXPECT_TRUE(summary.find("Total Write: 0 B") != std::string::npos);
+    EXPECT_TRUE(summary.find("Get: No data") != std::string::npos);
+    EXPECT_TRUE(summary.find("Put: No data") != std::string::npos);
+
+    // Add some data
+    metrics.total_read_bytes.inc(1024);              // 1KB
+    metrics.total_write_bytes.inc(2 * 1024 * 1024);  // 2MB
+
+    // Add latency observations
+    metrics.get_latency_us.observe(150);  // 150 microseconds
+    metrics.get_latency_us.observe(200);  // 200 microseconds
+    metrics.get_latency_us.observe(300);  // 300 microseconds
+
+    metrics.put_latency_us.observe(500);  // 500 microseconds
+    metrics.put_latency_us.observe(750);  // 750 microseconds
+
+    summary = metrics.summary_metrics();
+
+    // Check byte formatting
+    EXPECT_TRUE(summary.find("Total Read: 1.00 KB") != std::string::npos);
+    EXPECT_TRUE(summary.find("Total Write: 2.00 MB") != std::string::npos);
+
+    // Check latency summaries
+    EXPECT_TRUE(summary.find("Get: count=3") != std::string::npos);
+    EXPECT_TRUE(summary.find("Put: count=2") != std::string::npos);
+
+    // Check percentiles are present
+    EXPECT_TRUE(summary.find("p95<") != std::string::npos);
+    EXPECT_TRUE(summary.find("max<") != std::string::npos);
+
+    std::cout << "Transfer Metrics Summary:\n" << summary << std::endl;
+}
+
+TEST_F(ClientMetricsTest, MasterClientMetricsSummaryTest) {
+    MasterClientMetric metrics;
+
+    // Test empty metrics
+    std::string summary = metrics.summary_metrics();
+    EXPECT_TRUE(summary.find("No RPC calls recorded") != std::string::npos);
+
+    // Add some RPC calls
+    std::array<std::string, 1> get_replica_label = {"GetReplicaList"};
+    std::array<std::string, 1> mount_segment_label = {"MountSegment"};
+    std::array<std::string, 1> unmount_segment_label = {"UnmountSegment"};
+
+    // Simulate RPC calls
+    metrics.rpc_count.inc(get_replica_label);
+    metrics.rpc_count.inc(get_replica_label);
+    metrics.rpc_count.inc(mount_segment_label);
+    metrics.rpc_count.inc(unmount_segment_label);
+
+    // Add latency observations
+    metrics.rpc_latency.observe(get_replica_label, 200);  // 200 microseconds
+    metrics.rpc_latency.observe(get_replica_label, 250);  // 250 microseconds
+    metrics.rpc_latency.observe(mount_segment_label, 37789);   // 37.789 ms
+    metrics.rpc_latency.observe(unmount_segment_label, 7536);  // 7.536 ms
+
+    summary = metrics.summary_metrics();
+
+    // Check that RPC calls are recorded
+    EXPECT_TRUE(summary.find("GetReplicaList: count=2") != std::string::npos);
+    EXPECT_TRUE(summary.find("MountSegment: count=1") != std::string::npos);
+    EXPECT_TRUE(summary.find("UnmountSegment: count=1") != std::string::npos);
+
+    // Check percentiles are present for RPCs with data
+    EXPECT_TRUE(summary.find("p95<") != std::string::npos);
+    EXPECT_TRUE(summary.find("max<") != std::string::npos);
+
+    std::cout << "Master Client Metrics Summary:\n" << summary << std::endl;
+}
+
+TEST_F(ClientMetricsTest, ClientMetricsSummaryTest) {
+    ClientMetric metrics;
+
+    // Add some transfer data
+    metrics.transfer_metric.total_read_bytes.inc(5 * 1024 * 1024);    // 5MB
+    metrics.transfer_metric.total_write_bytes.inc(10 * 1024 * 1024);  // 10MB
+
+    metrics.transfer_metric.batch_get_latency_us.observe(1500);  // 1.5ms
+    metrics.transfer_metric.batch_put_latency_us.observe(2000);  // 2ms
+
+    // Add some RPC data
+    std::array<std::string, 1> exist_key_label = {"ExistKey"};
+    metrics.master_client_metric.rpc_count.inc(exist_key_label);
+    metrics.master_client_metric.rpc_latency.observe(exist_key_label, 180);
+
+    std::string summary = metrics.summary_metrics();
+
+    // Should contain both transfer and RPC metrics
+    EXPECT_TRUE(summary.find("Transfer Metrics Summary") != std::string::npos);
+    EXPECT_TRUE(summary.find("RPC Metrics Summary") != std::string::npos);
+    EXPECT_TRUE(summary.find("Total Read: 5.00 MB") != std::string::npos);
+    EXPECT_TRUE(summary.find("Total Write: 10.00 MB") != std::string::npos);
+    EXPECT_TRUE(summary.find("ExistKey: count=1") != std::string::npos);
+
+    std::cout << "Full Client Metrics Summary:\n" << summary << std::endl;
+}
+
+TEST_F(ClientMetricsTest, ByteFormattingTest) {
+    TransferMetric metrics;
+
+    // Test different byte sizes
+    metrics.total_read_bytes.inc(512);  // 512 B
+    std::string summary = metrics.summary_metrics();
+    EXPECT_TRUE(summary.find("512 B") != std::string::npos);
+
+    metrics.total_read_bytes.inc(1024 - 512);  // Total 1024 B = 1 KB
+    summary = metrics.summary_metrics();
+    EXPECT_TRUE(summary.find("1.00 KB") != std::string::npos);
+
+    metrics.total_read_bytes.inc(1024 * 1024 - 1024);  // Total 1 MB
+    summary = metrics.summary_metrics();
+    EXPECT_TRUE(summary.find("1.00 MB") != std::string::npos);
+
+    metrics.total_read_bytes.inc(1024LL * 1024 * 1024 -
+                                 1024 * 1024);  // Total 1 GB
+    summary = metrics.summary_metrics();
+    EXPECT_TRUE(summary.find("1.00 GB") != std::string::npos);
+}
+
+TEST_F(ClientMetricsTest, CompareWithSerializedMetrics) {
+    ClientMetric metrics;
+
+    // Add some data
+    metrics.transfer_metric.total_read_bytes.inc(1024 * 1024);
+    metrics.transfer_metric.get_latency_us.observe(200);
+
+    std::array<std::string, 1> get_replica_label = {"GetReplicaList"};
+    metrics.master_client_metric.rpc_count.inc(get_replica_label);
+    metrics.master_client_metric.rpc_latency.observe(get_replica_label, 250);
+
+    // Get both summary and full serialized metrics
+    std::string summary = metrics.summary_metrics();
+    std::string serialized;
+    metrics.serialize(serialized);
+
+    std::cout << "\n=== Summary Metrics ===" << std::endl;
+    std::cout << summary << std::endl;
+
+    std::cout << "\n=== Full Serialized Metrics ===" << std::endl;
+    std::cout << serialized << std::endl;
+
+    // Summary should be much shorter and more readable
+    EXPECT_LT(summary.length(), serialized.length());
+    EXPECT_TRUE(summary.find("count=") != std::string::npos);
+    EXPECT_TRUE(summary.find("p95<") != std::string::npos ||
+                summary.find("No data") != std::string::npos);
+    EXPECT_TRUE(summary.find("max<") != std::string::npos ||
+                summary.find("No data") != std::string::npos);
+}
+
+}  // namespace mooncake::test