Skip to content

[Store] Make NoF storage gauges projection-only #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Hubert-Zhu wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[Store] Make NoF storage gauges projection-only #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 2 additions & 11 deletions mooncake-store/include/master_metric_manager.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,14 +50,6 @@ class MasterMetricManager {
void inc_valid_get_nums(int64_t val = 1);
void inc_total_get_nums(int64_t val = 1);

// NoF segment Metrics
void inc_allocated_nof_size(const std::string& segment, int64_t val = 1);
void dec_allocated_nof_size(const std::string& segment, int64_t val = 1);
void reset_allocated_nof_size();
void inc_total_nof_capacity(const std::string& segment, int64_t val = 1);
void dec_total_nof_capacity(const std::string& segment, int64_t val = 1);
void reset_total_nof_capacity();

/**
* @brief Refresh storage gauges from authoritative allocator snapshots.
*
Expand Down Expand Up @@ -102,13 +94,12 @@ class MasterMetricManager {
// followed by client expiry / reaper cleanup).
void remove_segment_metrics(const std::string& segment);

// NoF segment Metrics
// NoF segment Metrics. Written only by project_storage_usage(); business
// code reads NoFSegmentManager::GetUsage() instead.
int64_t get_allocated_nof_size();
int64_t get_total_nof_capacity();
int64_t get_segment_allocated_nof_size(const std::string& segment);
int64_t get_segment_total_nof_capacity(const std::string& segment);
// Remove all per-segment NoF metric labels for the given segment.
void remove_nof_segment_metrics(const std::string& segment);

// File Storage Metrics
void inc_allocated_file_size(int64_t val = 1);
Expand Down
6 changes: 2 additions & 4 deletions mooncake-store/include/segment.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -264,15 +264,13 @@ class ScopedNoFSegmentAccess {
/**
* @brief Prepare to unmount a segment by deleting its allocator
*/
ErrorCode PrepareUnmountSegment(const UUID& segment_id,
size_t& metrics_dec_capacity);
ErrorCode PrepareUnmountSegment(const UUID& segment_id);

/**
* @brief Deleting the segment to complete the unmounting operation
*/
ErrorCode CommitUnmountSegment(const UUID& segment_id,
const UUID& client_id,
const size_t& metrics_dec_capacity);
const UUID& client_id);

/**
* @brief Get all the segments of a client
Expand Down
21 changes: 0 additions & 21 deletions mooncake-store/src/allocator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,9 +219,6 @@ CachelibBufferAllocator::~CachelibBufferAllocator() {
if (replica_type_ == ReplicaType::MEMORY) {
MasterMetricManager::instance().dec_allocated_mem_size(segment_name_,
size());
} else if (replica_type_ == ReplicaType::NOF_SSD) {
MasterMetricManager::instance().dec_allocated_nof_size(segment_name_,
size());
}
};

Expand Down Expand Up @@ -251,9 +248,6 @@ std::unique_ptr<AllocatedBuffer> CachelibBufferAllocator::allocate(
if (replica_type_ == ReplicaType::MEMORY) {
MasterMetricManager::instance().inc_allocated_mem_size(segment_name_,
size);
} else if (replica_type_ == ReplicaType::NOF_SSD) {
MasterMetricManager::instance().inc_allocated_nof_size(segment_name_,
size);
}
return std::make_unique<AllocatedBuffer>(shared_from_this(), buffer, size);
}
Expand All @@ -271,9 +265,6 @@ void CachelibBufferAllocator::deallocate(AllocatedBuffer* handle) {
if (replica_type_ == ReplicaType::MEMORY) {
MasterMetricManager::instance().dec_allocated_mem_size(
segment_name_, freed_size);
} else if (replica_type_ == ReplicaType::NOF_SSD) {
MasterMetricManager::instance().dec_allocated_nof_size(
segment_name_, freed_size);
}
VLOG(1) << "deallocation_succeeded address=" << handle->buffer_ptr_
<< " size=" << freed_size << " segment=" << segment_name_;
Expand All @@ -290,9 +281,6 @@ std::unique_ptr<AllocatedBuffer> CachelibBufferAllocator::adoptImportedBuffer(
if (replica_type_ == ReplicaType::MEMORY) {
MasterMetricManager::instance().inc_allocated_mem_size(
segment_name_, allocation.requested_size);
} else if (replica_type_ == ReplicaType::NOF_SSD) {
MasterMetricManager::instance().inc_allocated_nof_size(
segment_name_, allocation.requested_size);
}
return std::make_unique<AllocatedBuffer>(
shared_from_this(),
Expand Down Expand Up @@ -388,9 +376,6 @@ OffsetBufferAllocator::~OffsetBufferAllocator() {
if (replica_type_ == ReplicaType::MEMORY) {
MasterMetricManager::instance().dec_allocated_mem_size(segment_name_,
size());
} else if (replica_type_ == ReplicaType::NOF_SSD) {
MasterMetricManager::instance().dec_allocated_nof_size(segment_name_,
size());
}
};

Expand Down Expand Up @@ -432,9 +417,6 @@ std::unique_ptr<AllocatedBuffer> OffsetBufferAllocator::allocate(size_t size) {
if (replica_type_ == ReplicaType::MEMORY) {
MasterMetricManager::instance().inc_allocated_mem_size(segment_name_,
size);
} else if (replica_type_ == ReplicaType::NOF_SSD) {
MasterMetricManager::instance().inc_allocated_nof_size(segment_name_,
size);
}
return allocated_buffer;
}
Expand All @@ -449,9 +431,6 @@ void OffsetBufferAllocator::deallocate(AllocatedBuffer* handle) {
if (replica_type_ == ReplicaType::MEMORY) {
MasterMetricManager::instance().dec_allocated_mem_size(
segment_name_, freed_size);
} else if (replica_type_ == ReplicaType::NOF_SSD) {
MasterMetricManager::instance().dec_allocated_nof_size(
segment_name_, freed_size);
}
VLOG(1) << "deallocation_succeeded address=" << handle->data()
<< " size=" << freed_size << " segment=" << segment_name_;
Expand Down
38 changes: 0 additions & 38 deletions mooncake-store/src/master_metric_manager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -718,38 +718,6 @@ void MasterMetricManager::remove_segment_metrics(const std::string& segment) {
}

// NoF segment Metrics
void MasterMetricManager::inc_allocated_nof_size(const std::string& segment,
int64_t val) {
nof_allocated_size_.inc(val);
if (!segment.empty()) nof_allocated_size_per_segment_.inc({segment}, val);
}

void MasterMetricManager::dec_allocated_nof_size(const std::string& segment,
int64_t val) {
nof_allocated_size_.dec(val);
if (!segment.empty()) nof_allocated_size_per_segment_.dec({segment}, val);
}

void MasterMetricManager::reset_allocated_nof_size() {
nof_allocated_size_.reset();
}

void MasterMetricManager::inc_total_nof_capacity(const std::string& segment,
int64_t val) {
nof_total_capacity_.inc(val);
if (!segment.empty()) nof_total_capacity_per_segment_.inc({segment}, val);
}

void MasterMetricManager::dec_total_nof_capacity(const std::string& segment,
int64_t val) {
nof_total_capacity_.dec(val);
if (!segment.empty()) nof_total_capacity_per_segment_.dec({segment}, val);
}

void MasterMetricManager::reset_total_nof_capacity() {
nof_total_capacity_.reset();
}

int64_t MasterMetricManager::get_allocated_nof_size() {
return nof_allocated_size_.value();
}
Expand Down Expand Up @@ -808,12 +776,6 @@ int64_t MasterMetricManager::get_segment_total_nof_capacity(
return nof_total_capacity_per_segment_.value({segment});
}

void MasterMetricManager::remove_nof_segment_metrics(
const std::string& segment) {
nof_allocated_size_per_segment_.remove_label_value({{"segment", segment}});
nof_total_capacity_per_segment_.remove_label_value({{"segment", segment}});
}

// File Storage Metrics
void MasterMetricManager::inc_allocated_file_size(int64_t val) {
file_allocated_size_.inc(val);
Expand Down
15 changes: 5 additions & 10 deletions mooncake-store/src/master_service.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2857,8 +2857,6 @@ auto MasterService::UnmountNoFSegment(const UUID& segment_id,
<< ", error=nof_pool_disabled";
return tl::make_unexpected(ErrorCode::UNAVAILABLE_IN_CURRENT_MODE);
#else
size_t metrics_dec_capacity = 0; // to update the metrics

std::shared_lock<std::shared_mutex> client_lock(client_mutex_);
std::shared_lock<std::shared_mutex> shared_lock(snapshot_mutex_);
auto alive_clients = ok_client_;
Expand All @@ -2868,8 +2866,7 @@ auto MasterService::UnmountNoFSegment(const UUID& segment_id,
{
ScopedNoFSegmentAccess segment_access =
nof_segment_manager_.getNoFSegmentAccess();
ErrorCode err = segment_access.PrepareUnmountSegment(
segment_id, metrics_dec_capacity);
ErrorCode err = segment_access.PrepareUnmountSegment(segment_id);
if (err == ErrorCode::SEGMENT_NOT_FOUND) {
// Return OK because this is an idempotent operation
return {};
Expand All @@ -2886,8 +2883,7 @@ auto MasterService::UnmountNoFSegment(const UUID& segment_id,
// 3. Commit the unmount operation
ScopedNoFSegmentAccess segment_access =
nof_segment_manager_.getNoFSegmentAccess();
auto err = segment_access.CommitUnmountSegment(segment_id, client_id,
metrics_dec_capacity);
auto err = segment_access.CommitUnmountSegment(segment_id, client_id);
if (err != ErrorCode::OK) {
return tl::make_unexpected(err);
}
Expand Down Expand Up @@ -11282,15 +11278,14 @@ bool MasterService::ProbeNoFSegment(const std::string& te_endpoint,
bool MasterService::TryUnmountNoFSegmentByHeartbeat(
const MountedNoFSegmentSnapshot& snapshot,
const std::string& error_reason) {
size_t metrics_dec_capacity = 0;
std::shared_lock<std::shared_mutex> client_lock(client_mutex_);
std::shared_lock<std::shared_mutex> snapshot_lock(snapshot_mutex_);
auto alive_clients = ok_client_;
client_lock.unlock();
{
auto nof_segment_access = nof_segment_manager_.getNoFSegmentAccess();
ErrorCode err = nof_segment_access.PrepareUnmountSegment(
snapshot.segment_id, metrics_dec_capacity);
ErrorCode err =
nof_segment_access.PrepareUnmountSegment(snapshot.segment_id);
if (err == ErrorCode::SEGMENT_NOT_FOUND ||
err == ErrorCode::UNAVAILABLE_IN_CURRENT_STATUS) {
std::lock_guard<std::mutex> lock(nof_heartbeat_mutex_);
Expand All @@ -11315,7 +11310,7 @@ bool MasterService::TryUnmountNoFSegmentByHeartbeat(
{
auto nof_segment_access = nof_segment_manager_.getNoFSegmentAccess();
ErrorCode err = nof_segment_access.CommitUnmountSegment(
snapshot.segment_id, snapshot.client_id, metrics_dec_capacity);
snapshot.segment_id, snapshot.client_id);
if (err != ErrorCode::OK && err != ErrorCode::SEGMENT_NOT_FOUND) {
LOG(ERROR) << "segment_id=" << snapshot.segment_id
<< ", segment_name=" << snapshot.segment.name
Expand Down
15 changes: 4 additions & 11 deletions mooncake-store/src/segment.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1212,7 +1212,6 @@ ErrorCode ScopedNoFSegmentAccess::MountSegment(const NoFSegment& segment,
nof_segment_manager_->mounted_segments_[segment.id] = {
segment, client_id, SegmentStatus::OK, std::move(allocator)};
nof_segment_manager_->client_by_name_[segment.name] = client_id;
MasterMetricManager::instance().inc_total_nof_capacity(segment.name, size);

return ErrorCode::OK;
}
Expand Down Expand Up @@ -1243,7 +1242,7 @@ ErrorCode ScopedNoFSegmentAccess::ReMountSegment(
}

ErrorCode ScopedNoFSegmentAccess::PrepareUnmountSegment(
const UUID& segment_id, size_t& metrics_dec_capacity) {
const UUID& segment_id) {
auto it = nof_segment_manager_->mounted_segments_.find(segment_id);
if (it == nof_segment_manager_->mounted_segments_.end()) {
LOG(WARNING) << "NoF segment unmount: segment_id=" << segment_id
Expand All @@ -1258,7 +1257,6 @@ ErrorCode ScopedNoFSegmentAccess::PrepareUnmountSegment(

auto& mounted_segment = it->second;
auto& segment = mounted_segment.segment;
metrics_dec_capacity = segment.size;

std::shared_ptr<BufferAllocatorBase> allocator =
mounted_segment.buf_allocator;
Expand All @@ -1274,8 +1272,7 @@ ErrorCode ScopedNoFSegmentAccess::PrepareUnmountSegment(
}

ErrorCode ScopedNoFSegmentAccess::CommitUnmountSegment(
const UUID& segment_id, const UUID& client_id,
const size_t& metrics_dec_capacity) {
const UUID& segment_id, const UUID& client_id) {
bool found_in_client_segments = false;
auto client_it = nof_segment_manager_->client_segments_.find(client_id);
if (client_it != nof_segment_manager_->client_segments_.end()) {
Expand All @@ -1295,17 +1292,13 @@ ErrorCode ScopedNoFSegmentAccess::CommitUnmountSegment(
<< ", error=segment_not_found_in_client_segments";
}

std::string segment_name;
auto segment_it = nof_segment_manager_->mounted_segments_.find(segment_id);
if (segment_it != nof_segment_manager_->mounted_segments_.end()) {
segment_name = segment_it->second.segment.name;
nof_segment_manager_->client_by_name_.erase(segment_name);
nof_segment_manager_->client_by_name_.erase(
segment_it->second.segment.name);
}

nof_segment_manager_->mounted_segments_.erase(segment_id);
MasterMetricManager::instance().dec_total_nof_capacity(
segment_name, metrics_dec_capacity);
MasterMetricManager::instance().remove_nof_segment_metrics(segment_name);

return ErrorCode::OK;
}
Expand Down
80 changes: 69 additions & 11 deletions mooncake-store/tests/segment_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -461,9 +461,9 @@ TEST_F(SegmentTest, NoFUsageSnapshotSurvivesMetricsReset) {
EXPECT_EQ(usage.capacity_bytes, kSegmentSize);
EXPECT_DOUBLE_EQ(usage.used_ratio(), 0.25);

// Zero the gauges by projecting an empty snapshot, as a standby does.
auto& metrics = MasterMetricManager::instance();
metrics.reset_allocated_nof_size();
metrics.reset_total_nof_capacity();
metrics.project_storage_usage({});
EXPECT_EQ(metrics.get_allocated_nof_size(), 0);
EXPECT_EQ(metrics.get_total_nof_capacity(), 0);

Expand All @@ -476,26 +476,84 @@ TEST_F(SegmentTest, NoFUsageSnapshotSurvivesMetricsReset) {
EXPECT_EQ(usage.capacity_bytes, kSegmentSize);
EXPECT_DOUBLE_EQ(usage.used_ratio(), 0.25);

// Restore global gauges before teardown because allocator and segment
// cleanup still emit their matching decrements.
metrics.inc_allocated_nof_size("", kAllocationSize);
metrics.inc_total_nof_capacity("", kSegmentSize);
buffer.reset();
{
auto segment_access = segment_manager.getNoFSegmentAccess();
size_t metrics_dec_capacity = 0;
ASSERT_EQ(segment_access.PrepareUnmountSegment(segment.id,
metrics_dec_capacity),
ASSERT_EQ(segment_access.PrepareUnmountSegment(segment.id),
ErrorCode::OK);
ASSERT_EQ(segment_access.CommitUnmountSegment(segment.id, client_id,
metrics_dec_capacity),
ASSERT_EQ(segment_access.CommitUnmountSegment(segment.id, client_id),
ErrorCode::OK);
}
allocator.reset();
EXPECT_EQ(segment_manager.GetUsage().used_bytes, 0u);
EXPECT_EQ(segment_manager.GetUsage().capacity_bytes, 0u);
}

// The NoF gauges are a projection: only project_storage_usage() writes them.
TEST_F(SegmentTest, NoFGaugesMoveOnlyWhenProjected) {
NoFSegmentManager segment_manager(BufferAllocatorType::OFFSET);
constexpr size_t kSegmentSize = 16 * 1024 * 1024;
constexpr size_t kAllocationSize = 4 * 1024 * 1024;
constexpr int64_t kProjectedUsed = 111;
constexpr int64_t kProjectedCapacity = 222;

auto& metrics = MasterMetricManager::instance();
TieredStorageUsageSnapshot projected;
projected.nof.used_bytes = static_cast<size_t>(kProjectedUsed);
projected.nof.capacity_bytes = static_cast<size_t>(kProjectedCapacity);
metrics.project_storage_usage(projected);
ASSERT_EQ(metrics.get_allocated_nof_size(), kProjectedUsed);
ASSERT_EQ(metrics.get_total_nof_capacity(), kProjectedCapacity);

NoFSegment segment;
segment.id = generate_uuid();
segment.name = "nof_projection_only_segment";
segment.size = kSegmentSize;
segment.base = 0x340000000;
segment.te_endpoint = "nof_projection_only_endpoint";
UUID client_id = generate_uuid();
{
auto segment_access = segment_manager.getNoFSegmentAccess();
ASSERT_EQ(segment_access.MountSegment(segment, client_id),
ErrorCode::OK);
}
auto allocator = GetNoFAllocatorForTesting(segment_manager, segment.id);
ASSERT_NE(allocator, nullptr);
auto buffer = allocator->allocate(kAllocationSize);
ASSERT_NE(buffer, nullptr);

// Domain state moved; the gauges did not.
EXPECT_EQ(segment_manager.GetUsage().used_bytes, kAllocationSize);
EXPECT_EQ(metrics.get_allocated_nof_size(), kProjectedUsed);
EXPECT_EQ(metrics.get_total_nof_capacity(), kProjectedCapacity);

projected.nof = segment_manager.GetUsageSnapshot();
metrics.project_storage_usage(projected);
EXPECT_EQ(metrics.get_allocated_nof_size(),
static_cast<int64_t>(kAllocationSize));
EXPECT_EQ(metrics.get_total_nof_capacity(),
static_cast<int64_t>(kSegmentSize));
const std::string label = "segment=\"" + segment.name + "\"";
EXPECT_NE(metrics.serialize_metrics().find(label), std::string::npos);

buffer.reset();
{
auto segment_access = segment_manager.getNoFSegmentAccess();
ASSERT_EQ(segment_access.PrepareUnmountSegment(segment.id),
ErrorCode::OK);
ASSERT_EQ(segment_access.CommitUnmountSegment(segment.id, client_id),
ErrorCode::OK);
}
allocator.reset();

// The projection, not the unmount, retires the per-segment labels.
projected.nof = segment_manager.GetUsageSnapshot();
metrics.project_storage_usage(projected);
EXPECT_EQ(metrics.get_allocated_nof_size(), 0);
EXPECT_EQ(metrics.get_total_nof_capacity(), 0);
EXPECT_EQ(metrics.serialize_metrics().find(label), std::string::npos);
}

// MountSegmentDuplicate Tests:
// 1. MountSegment with the same segment id. The second mount operation return
// SEGMENT_ALREADY_EXISTS.
Expand Down
Loading