zbtrs2
/
Mooncake
forked from mooncake-track/Mooncake
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[Store] Add per-segment memory usage metrics (#1030)

This commit is contained in:
sinpo828 2025-11-10 16:34:38 +08:00 committed by GitHub
parent 6d05c9344b
commit 196af0d4a9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 108 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
mooncake-store/include/master_metric_manager.h
View File

@ -19,16 +19,19 @@ class MasterMetricManager {
MasterMetricManager(MasterMetricManager&&) = delete;
MasterMetricManager& operator=(MasterMetricManager&&) = delete;
// Memory Storage Metrics
void inc_allocated_mem_size(int64_t val = 1);
void dec_allocated_mem_size(int64_t val = 1);
// Memory Storage Metrics(global & segment)
void inc_allocated_mem_size(const std::string& segment, int64_t val = 1);
void dec_allocated_mem_size(const std::string& segment, int64_t val = 1);
void reset_allocated_mem_size();
void inc_total_mem_capacity(int64_t val = 1);
void dec_total_mem_capacity(int64_t val = 1);
void inc_total_mem_capacity(const std::string& segment, int64_t val = 1);
void dec_total_mem_capacity(const std::string& segment, int64_t val = 1);
void reset_total_mem_capacity();
double get_global_mem_used_ratio(void);
int64_t get_allocated_mem_size();
int64_t get_total_mem_capacity();
double get_global_mem_used_ratio(void);
double get_segment_mem_used_ratio(const std::string& segment);
int64_t get_segment_allocated_mem_size(const std::string& segment);
int64_t get_segment_total_mem_capacity(const std::string& segment);
// File Storage Metrics
void inc_allocated_file_size(int64_t val = 1);
@ -196,8 +199,16 @@ class MasterMetricManager {
// --- Metric Members ---
// Memory Storage Metrics
ylt::metric::gauge_t mem_allocated_size_; // Use update for gauge
ylt::metric::gauge_t mem_total_capacity_; // Use update for gauge
ylt::metric::gauge_t
mem_allocated_size_; // Overall memory usage update for gauge
ylt::metric::gauge_t
mem_total_capacity_; // Overall memory capacity update for gauge
ylt::metric::dynamic_gauge_1t
mem_allocated_size_per_segment_; // Segment memory usage update for
// gauge
ylt::metric::dynamic_gauge_1t
mem_total_capacity_per_segment_; // Segment memory capacity update for
// gauge
// File Storage Metrics
ylt::metric::gauge_t file_allocated_size_;

26
mooncake-store/src/allocator.cpp
View File

@ -18,13 +18,13 @@ std::string AllocatedBuffer::getSegmentName() const noexcept {
}
AllocatedBuffer::~AllocatedBuffer() {
// Note: This is an edge case. If the 'weak_ptr' is released, the segment
// has already been deallocated at this point, and its memory usage details
// (capacity/allocated) no longer need to be maintained.
auto alloc = allocator_.lock();
if (alloc) {
alloc->deallocate(this);
VLOG(1) << "buf_handle_deallocated size=" << size_;
} else {
MasterMetricManager::instance().dec_allocated_mem_size(size_);
VLOG(1) << "allocator=expired_or_null in buf_handle_destructor";
}
}
@ -92,7 +92,10 @@ CachelibBufferAllocator::CachelibBufferAllocator(std::string segment_name,
<< static_cast<int>(pool_id_);
}
CachelibBufferAllocator::~CachelibBufferAllocator() = default;
CachelibBufferAllocator::~CachelibBufferAllocator() {
MasterMetricManager::instance().dec_allocated_mem_size(segment_name_,
cur_size_);
};
std::unique_ptr<AllocatedBuffer> CachelibBufferAllocator::allocate(
size_t size) {
@ -117,7 +120,7 @@ std::unique_ptr<AllocatedBuffer> CachelibBufferAllocator::allocate(
VLOG(1) << "allocation_succeeded size=" << size
<< " segment=" << segment_name_ << " address=" << buffer;
cur_size_.fetch_add(size);
MasterMetricManager::instance().inc_allocated_mem_size(size);
MasterMetricManager::instance().inc_allocated_mem_size(segment_name_, size);
return std::make_unique<AllocatedBuffer>(shared_from_this(), buffer, size);
}
@ -128,7 +131,8 @@ void CachelibBufferAllocator::deallocate(AllocatedBuffer* handle) {
size_t freed_size =
handle->size_; // Store size before handle might become invalid
cur_size_.fetch_sub(freed_size);
MasterMetricManager::instance().dec_allocated_mem_size(freed_size);
MasterMetricManager::instance().dec_allocated_mem_size(segment_name_,
freed_size);
VLOG(1) << "deallocation_succeeded address=" << handle->buffer_ptr_
<< " size=" << freed_size << " segment=" << segment_name_;
} catch (const std::exception& e) {
@ -180,7 +184,10 @@ OffsetBufferAllocator::OffsetBufferAllocator(std::string segment_name,
}
}
OffsetBufferAllocator::~OffsetBufferAllocator() = default;
OffsetBufferAllocator::~OffsetBufferAllocator() {
MasterMetricManager::instance().dec_allocated_mem_size(segment_name_,
cur_size_);
};
std::unique_ptr<AllocatedBuffer> OffsetBufferAllocator::allocate(size_t size) {
if (!offset_allocator_) {
@ -217,7 +224,7 @@ std::unique_ptr<AllocatedBuffer> OffsetBufferAllocator::allocate(size_t size) {
}
cur_size_.fetch_add(size);
MasterMetricManager::instance().inc_allocated_mem_size(size);
MasterMetricManager::instance().inc_allocated_mem_size(segment_name_, size);
return allocated_buffer;
}
@ -228,7 +235,8 @@ void OffsetBufferAllocator::deallocate(AllocatedBuffer* handle) {
size_t freed_size = handle->size();
handle->offset_handle_.reset();
cur_size_.fetch_sub(freed_size);
MasterMetricManager::instance().dec_allocated_mem_size(freed_size);
MasterMetricManager::instance().dec_allocated_mem_size(segment_name_,
freed_size);
VLOG(1) << "deallocation_succeeded address=" << handle->data()
<< " size=" << freed_size << " segment=" << segment_name_;
} catch (const std::exception& e) {

50
mooncake-store/src/master_metric_manager.cpp
View File

@ -23,11 +23,17 @@ MasterMetricManager::MasterMetricManager()
"Total memory bytes currently allocated across all segments"),
mem_total_capacity_("master_total_capacity_bytes",
"Total memory capacity across all mounted segments"),
file_total_capacity_("master_total_file_capacity_bytes",
"Total capacity for file storage in 3fs/nfs"),
mem_allocated_size_per_segment_(
"segment_allocated_bytes",
"Total memory bytes currently allocated of the segment", {"segment"}),
mem_total_capacity_per_segment_(
"segment_total_capacity_bytes",
"Total memory capacity of the mounted segment", {"segment"}),
file_allocated_size_(
"master_allocated_file_size_bytes",
"Total bytes currently allocated for file storage in 3fs/nfs"),
file_total_capacity_("master_total_file_capacity_bytes",
"Total capacity for file storage in 3fs/nfs"),
key_count_("master_key_count",
"Total number of keys managed by the master"),
soft_pin_key_count_(
@ -204,22 +210,32 @@ MasterMetricManager::MasterMetricManager()
// --- Metric Interface Methods ---
// Memory Storage Metrics
void MasterMetricManager::inc_allocated_mem_size(int64_t val) {
void MasterMetricManager::inc_allocated_mem_size(const std::string& segment,
int64_t val) {
mem_allocated_size_.inc(val);
if (!segment.empty()) mem_allocated_size_per_segment_.inc({segment}, val);
}
void MasterMetricManager::dec_allocated_mem_size(int64_t val) {
void MasterMetricManager::dec_allocated_mem_size(const std::string& segment,
int64_t val) {
mem_allocated_size_.dec(val);
if (!segment.empty()) mem_allocated_size_per_segment_.dec({segment}, val);
}
void MasterMetricManager::reset_allocated_mem_size() {
mem_allocated_size_.reset();
}
void MasterMetricManager::inc_total_mem_capacity(int64_t val) {
void MasterMetricManager::inc_total_mem_capacity(const std::string& segment,
int64_t val) {
mem_total_capacity_.inc(val);
if (!segment.empty()) mem_total_capacity_per_segment_.inc({segment}, val);
}
void MasterMetricManager::dec_total_mem_capacity(int64_t val) {
void MasterMetricManager::dec_total_mem_capacity(const std::string& segment,
int64_t val) {
mem_total_capacity_.dec(val);
if (!segment.empty()) mem_total_capacity_per_segment_.dec({segment}, val);
}
void MasterMetricManager::reset_total_mem_capacity() {
@ -243,6 +259,26 @@ double MasterMetricManager::get_global_mem_used_ratio(void) {
return allocated / capacity;
}
int64_t MasterMetricManager::get_segment_allocated_mem_size(
const std::string& segment) {
return mem_allocated_size_per_segment_.value({segment});
}
int64_t MasterMetricManager::get_segment_total_mem_capacity(
const std::string& segment) {
return mem_total_capacity_per_segment_.value({segment});
}
double MasterMetricManager::get_segment_mem_used_ratio(
const std::string& segment) {
double allocated = get_segment_allocated_mem_size(segment);
double capacity = get_segment_total_mem_capacity(segment);
if (capacity == 0) {
return 0.0;
}
return allocated / capacity;
}
// File Storage Metrics
void MasterMetricManager::inc_allocated_file_size(int64_t val) {
file_allocated_size_.inc(val);
@ -736,6 +772,8 @@ std::string MasterMetricManager::serialize_metrics() {
// Serialize Gauges
serialize_metric(mem_allocated_size_);
serialize_metric(mem_total_capacity_);
serialize_metric(mem_allocated_size_per_segment_);
serialize_metric(mem_total_capacity_per_segment_);
serialize_metric(file_allocated_size_);
serialize_metric(file_total_capacity_);
serialize_metric(key_count_);

10
mooncake-store/src/segment.cpp
View File

@ -81,7 +81,7 @@ ErrorCode ScopedSegmentAccess::MountSegment(const Segment& segment,
segment_manager_->mounted_segments_[segment.id] = {
segment, SegmentStatus::OK, std::move(allocator)};
MasterMetricManager::instance().inc_total_mem_capacity(size);
MasterMetricManager::instance().inc_total_mem_capacity(segment.name, size);
return ErrorCode::OK;
}
@ -199,12 +199,18 @@ ErrorCode ScopedSegmentAccess::CommitUnmountSegment(
<< ", error=segment_not_found_in_client_segments";
}
// segment_id -> segment_name
std::string segment_name;
auto&& segment = segment_manager_->mounted_segments_.find(segment_id);
if (segment != segment_manager_->mounted_segments_.end()) {
segment_name = segment->second.segment.name;
}
// Remove from mounted_segments_
segment_manager_->mounted_segments_.erase(segment_id);
// Decrease the total capacity
MasterMetricManager::instance().dec_total_mem_capacity(
metrics_dec_capacity);
segment_name, metrics_dec_capacity);
return ErrorCode::OK;
}

20
mooncake-store/tests/master_metrics_test.cpp
View File

@ -129,6 +129,10 @@ TEST_F(MasterMetricsTest, BasicRequestTest) {
ASSERT_EQ(metrics.get_allocated_mem_size(), 0);
ASSERT_EQ(metrics.get_total_mem_capacity(), kSegmentSize);
ASSERT_DOUBLE_EQ(metrics.get_global_mem_used_ratio(), 0.0);
ASSERT_EQ(metrics.get_segment_allocated_mem_size(segment.name), 0);
ASSERT_EQ(metrics.get_segment_total_mem_capacity(segment.name),
kSegmentSize);
ASSERT_DOUBLE_EQ(metrics.get_segment_mem_used_ratio(segment.name), 0.0);
ASSERT_EQ(metrics.get_mount_segment_requests(), 1);
ASSERT_EQ(metrics.get_mount_segment_failures(), 0);
@ -138,6 +142,8 @@ TEST_F(MasterMetricsTest, BasicRequestTest) {
ASSERT_TRUE(put_start_result1.has_value());
ASSERT_EQ(metrics.get_key_count(), 1);
ASSERT_EQ(metrics.get_allocated_mem_size(), value_length);
ASSERT_EQ(metrics.get_segment_allocated_mem_size(segment.name),
value_length);
ASSERT_EQ(metrics.get_put_start_requests(), 1);
ASSERT_EQ(metrics.get_put_start_failures(), 0);
auto put_revoke_result =
@ -145,6 +151,7 @@ TEST_F(MasterMetricsTest, BasicRequestTest) {
ASSERT_TRUE(put_revoke_result.has_value());
ASSERT_EQ(metrics.get_key_count(), 0);
ASSERT_EQ(metrics.get_allocated_mem_size(), 0);
ASSERT_EQ(metrics.get_segment_allocated_mem_size(segment.name), 0);
ASSERT_EQ(metrics.get_put_revoke_requests(), 1);
ASSERT_EQ(metrics.get_put_revoke_failures(), 0);
@ -154,12 +161,16 @@ TEST_F(MasterMetricsTest, BasicRequestTest) {
ASSERT_TRUE(put_start_result2.has_value());
ASSERT_EQ(metrics.get_key_count(), 1);
ASSERT_EQ(metrics.get_allocated_mem_size(), value_length);
ASSERT_EQ(metrics.get_segment_allocated_mem_size(segment.name),
value_length);
ASSERT_EQ(metrics.get_put_start_requests(), 2);
ASSERT_EQ(metrics.get_put_start_failures(), 0);
auto put_end_result = service_.PutEnd(client_id, key, ReplicaType::MEMORY);
ASSERT_TRUE(put_end_result.has_value());
ASSERT_EQ(metrics.get_key_count(), 1);
ASSERT_EQ(metrics.get_allocated_mem_size(), value_length);
ASSERT_EQ(metrics.get_segment_allocated_mem_size(segment.name),
value_length);
ASSERT_EQ(metrics.get_put_end_requests(), 1);
ASSERT_EQ(metrics.get_put_end_failures(), 0);
@ -184,6 +195,7 @@ TEST_F(MasterMetricsTest, BasicRequestTest) {
ASSERT_EQ(metrics.get_remove_failures(), 0);
ASSERT_EQ(metrics.get_key_count(), 0);
ASSERT_EQ(metrics.get_allocated_mem_size(), 0);
ASSERT_EQ(metrics.get_segment_allocated_mem_size(segment.name), 0);
// Test RemoveAll request
auto put_start_result3 =
@ -197,6 +209,7 @@ TEST_F(MasterMetricsTest, BasicRequestTest) {
ASSERT_EQ(metrics.get_remove_all_failures(), 0);
ASSERT_EQ(metrics.get_key_count(), 0);
ASSERT_EQ(metrics.get_allocated_mem_size(), 0);
ASSERT_EQ(metrics.get_segment_allocated_mem_size(segment.name), 0);
// Test UnmountSegment request
auto put_start_result4 =
@ -212,6 +225,13 @@ TEST_F(MasterMetricsTest, BasicRequestTest) {
ASSERT_EQ(metrics.get_allocated_mem_size(), 0);
ASSERT_EQ(metrics.get_total_mem_capacity(), 0);
ASSERT_DOUBLE_EQ(metrics.get_global_mem_used_ratio(), 0.0);
ASSERT_EQ(metrics.get_segment_allocated_mem_size(segment.name), 0);
ASSERT_EQ(metrics.get_segment_total_mem_capacity(segment.name), 0);
ASSERT_DOUBLE_EQ(metrics.get_segment_mem_used_ratio(segment.name), 0.0);
// check segment mem used ratio for non-existent segment
ASSERT_DOUBLE_EQ(metrics.get_segment_mem_used_ratio(""), 0.0);
ASSERT_DOUBLE_EQ(metrics.get_segment_mem_used_ratio("xxxxxx_segment"), 0.0);
}
TEST_F(MasterMetricsTest, BatchRequestTest) {