forked from mooncake-track/Mooncake
feat(client): Add client-side metrics for transfer and RPC operations (#733)
* feat(client): Add RPC operation and tansfer metrics tracking * feat(client): Add client-side metrics for transfer and RPC operations This commit introduces a comprehensive metrics system for the client component, tracking transfer byte counts and operation latencies with both human-readable summaries and Prometheus-style serialization. Key features include: - New TransferMetric for tracking read/write bytes and latency histograms - MasterClientMetric for RPC call counting and latency tracking - Environment-controlled metrics reporting (MC_STORE_METRIC_REPORT) - Automatic periodic metrics collection thread - Enhanced test coverage for metrics validation - Unified metrics interface across all client operations The implementation provides detailed latency percentiles (P50/P95) and total byte tracking with automatic unit conversion (B/KB/MB/GB). * fix test * Update mooncake-store/src/client.cpp Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update mooncake-store/src/client.cpp Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update mooncake-store/src/client.cpp Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * fix format issue * refactor(client): make client metrics optional using ClientMetric class * fix lint --------- Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
JinYan Su
GitHub
parent
335d1a1b70
commit
57d1c71249
|
|
@ -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_;
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
Loading…
Reference in New Issue