mooncake-track
/
Mooncake
0
0
Fork 12
Code Issues Pull Requests 13 Actions Packages Projects Releases Wiki Activity

[CCF Archive] Store object type eviction policy submission #3

Closed
kancel wants to merge 382 commits from kancel:ccf-archive-pr2746 into main
pull from: kancel:ccf-archive-pr2746
merge into: mooncake-track:main
Conversation 0 Commits 382 Files Changed 654 +264 -7
9 changed files with 264 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 788c1c737e - Show all commits

2
mooncake-store/include/client_service.h
View File

@ -804,6 +804,8 @@ class Client {
std::thread task_poll_thread_;
std::atomic<bool> task_poll_running_{false};
std::atomic<bool> last_ping_success_{false};
std::atomic<bool> segment_desc_publish_pending_{false};
std::atomic<bool> rpc_meta_publish_pending_{false};
ErrorCode SwitchLeader(const ha::MasterView& target_view);
void LeaderMonitorThreadMain();
void StorageHeartbeatThreadMain();

9
mooncake-store/include/file_storage.h
View File

@ -118,6 +118,13 @@ class FileStorage {
void ClientBufferGCThreadFunc();
/**
* @brief Re-registers all offloaded objects with the master.
* Called after master restart recovery to sync SSD object metadata.
* This is the same logic as the ScanMeta step in Init().
*/
tl::expected<void, ErrorCode> ReRegisterOffloadedObjects();
std::shared_ptr<Client> client_;
SsdMetric* ssd_metric_{nullptr};
std::string local_rpc_addr_;
@ -136,6 +143,8 @@ class FileStorage {
std::thread heartbeat_thread_;
std::atomic<bool> client_buffer_gc_running_;
std::thread client_buffer_gc_thread_;
std::future<void> rescan_future_;
std::atomic<bool> metadata_resync_pending_{false};
};
} // namespace mooncake

10
mooncake-store/include/storage_backend.h
View File

@ -267,6 +267,11 @@ class StorageBackendInterface {
const std::vector<std::string>& keys,
std::vector<StorageObjectMetadata>& metadatas)>& handler) = 0;
// Reset internal scan iterator so that the next ScanMeta() call
// starts from the beginning. Required for backends that use
// cursor-based iteration (e.g. BucketStorageBackend).
virtual void ResetScanIterator() {}
// Test-only: Set predicate to force failures for specific keys in
// BatchOffload. Default implementation does nothing (no failures injected).
// Concrete backends can override to provide test failure injection.
@ -781,6 +786,11 @@ class BucketStorageBackend : public StorageBackendInterface {
const std::vector<std::string>& keys,
std::vector<StorageObjectMetadata>& metadatas)>& handler) override;
void ResetScanIterator() override {
MutexLocker locker(&iterator_mutex_);
next_bucket_ = -1;
}
/**
* @brief Checks whether the backend is allowed to continue offloading.
* @return tl::expected<bool, ErrorCode>

72
mooncake-store/src/client_service.cpp
View File

@ -3039,6 +3039,43 @@ void Client::StorageHeartbeatThreadMain() {
ErrorCode err = remount_result.error();
LOG(ERROR) << "Failed to remount segments: " << err;
}
// Re-publish Transfer Engine segment descriptors to the HTTP
// metadata server. When Master (which hosts the HTTP metadata
// server in the same process) is killed and restarted, all
// in-memory KV entries are lost. ReMountSegment above only
// restores Master-side allocation state; it does NOT write back
// the transport-level segment descriptors. Without this, remote
// peers get HTTP 404 when querying our segment descriptor and
// data transfers fail.
auto metadata = transfer_engine_->getMetadata();
if (metadata) {
int rc = metadata->updateLocalSegmentDesc();
if (rc != 0) {
LOG(ERROR) << "Failed to re-publish segment descriptor "
<< "to metadata server, rc=" << rc
<< ", will retry in next heartbeat cycle";
segment_desc_publish_pending_.store(true);
} else {
segment_desc_publish_pending_.store(false);
}
// Also re-publish RPC meta entry (mooncake/rpc_meta/<hostname>).
// Remote peers need this to locate our RDMA RPC port for
// handshake. Like segment descriptors, this entry is lost
// when the HTTP metadata server is cleared on Master restart.
rc = metadata->rePublishRpcMetaEntry(local_hostname_);
if (rc != 0) {
LOG(ERROR) << "Failed to re-publish RPC meta entry "
<< "to metadata server, rc=" << rc
<< ", will retry in next heartbeat cycle";
rpc_meta_publish_pending_.store(true);
} else {
rpc_meta_publish_pending_.store(false);
}
}
// Note: LOCAL_DISK segment remount is NOT done here.
// It is handled by FileStorage::Heartbeat() when it detects
// SEGMENT_NOT_FOUND, which also triggers ScanMeta to
// re-register offloaded object metadata.
};
// Use another thread to remount segments to avoid blocking the ping
// thread
@ -3064,6 +3101,41 @@ void Client::StorageHeartbeatThreadMain() {
// Ensure at most one remount segment thread is running
remount_segment_future =
std::async(std::launch::async, remount_segment);
} else if (segment_desc_publish_pending_.load() &&
!remount_segment_future.valid()) {
// Previous remount succeeded but updateLocalSegmentDesc()
// failed (e.g. transient HTTP error). Retry it directly
// without re-running ReMountSegment.
auto metadata = transfer_engine_->getMetadata();
if (metadata) {
int rc = metadata->updateLocalSegmentDesc();
if (rc != 0) {
LOG(ERROR)
<< "Retry: failed to re-publish segment "
<< "descriptor to metadata server, rc=" << rc;
} else {
LOG(INFO) << "Retry: successfully re-published "
<< "segment descriptor to metadata server";
segment_desc_publish_pending_.store(false);
}
}
} else if (rpc_meta_publish_pending_.load() &&
!remount_segment_future.valid()) {
// Previous remount succeeded but rePublishRpcMetaEntry()
// failed. Retry it directly.
auto metadata = transfer_engine_->getMetadata();
if (metadata) {
int rc = metadata->rePublishRpcMetaEntry(local_hostname_);
if (rc != 0) {
LOG(ERROR)
<< "Retry: failed to re-publish RPC "
<< "meta entry to metadata server, rc=" << rc;
} else {
LOG(INFO) << "Retry: successfully re-published "
<< "RPC meta entry to metadata server";
rpc_meta_publish_pending_.store(false);
}
}
}
std::this_thread::sleep_for(

123
mooncake-store/src/file_storage.cpp
View File

@ -501,6 +501,26 @@ tl::expected<void, ErrorCode> FileStorage::Heartbeat() {
return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
}
// Join previous rescan if completed
if (rescan_future_.valid() && rescan_future_.wait_for(std::chrono::seconds(
0)) == std::future_status::ready) {
rescan_future_ = std::future<void>();
}
// Retry metadata resync if previous attempt failed.
if (metadata_resync_pending_.load() && !rescan_future_.valid()) {
LOG(INFO) << "Retrying background metadata rescan";
rescan_future_ = std::async(std::launch::async, [this]() {
auto result = ReRegisterOffloadedObjects();
if (!result) {
LOG(ERROR) << "Background metadata rescan retry "
<< "failed: " << result.error();
} else {
metadata_resync_pending_.store(false);
}
});
}
std::unordered_map<std::string, int64_t>
offloading_objects; // Objects selected for offloading
@ -510,9 +530,52 @@ tl::expected<void, ErrorCode> FileStorage::Heartbeat() {
auto heartbeat_result = client_->OffloadObjectHeartbeat(
enable_offloading_, offloading_objects);
if (!heartbeat_result) {
LOG(ERROR) << "Failed to send heartbeat with error: "
<< heartbeat_result.error();
return heartbeat_result;
ErrorCode err = heartbeat_result.error();
if (err == ErrorCode::SEGMENT_NOT_FOUND) {
// Master lost our LOCAL_DISK segment (likely restarted).
// Re-register the segment, retry the heartbeat, and
// trigger async ScanMeta to re-register object metadata.
LOG(WARNING) << "OffloadObjectHeartbeat returned "
<< "SEGMENT_NOT_FOUND, attempting to "
<< "re-register local disk segment and "
<< "re-register object metadata";
auto remount_result =
client_->MountLocalDiskSegment(enable_offloading_);
if (remount_result) {
heartbeat_result = client_->OffloadObjectHeartbeat(
enable_offloading_, offloading_objects);
if (!heartbeat_result) {
LOG(ERROR) << "Heartbeat failed after re-registration: "
<< heartbeat_result.error();
return heartbeat_result;
}
// Master lost all object metadata on restart.
// Trigger async ScanMeta to re-register them,
// same as what Init() does on startup.
if (!rescan_future_.valid()) {
LOG(INFO) << "Triggering background metadata rescan "
<< "after LOCAL_DISK segment re-registration";
metadata_resync_pending_.store(true);
rescan_future_ =
std::async(std::launch::async, [this]() {
auto result = ReRegisterOffloadedObjects();
if (!result) {
LOG(ERROR) << "Background metadata rescan "
<< "failed: " << result.error();
} else {
metadata_resync_pending_.store(false);
}
});
}
} else {
LOG(ERROR) << "Failed to re-register local disk segment: "
<< remount_result.error();
return tl::make_unexpected(remount_result.error());
}
} else {
LOG(ERROR) << "Failed to send heartbeat with error: " << err;
return heartbeat_result;
}
}
}
@ -863,4 +926,58 @@ bool FileStorage::ReleaseBuffer(uint64_t batch_id) {
return false;
}
tl::expected<void, ErrorCode> FileStorage::ReRegisterOffloadedObjects() {
LOG(INFO) << "ReRegisterOffloadedObjects: starting ScanMeta to re-register "
<< "offloaded objects with master";
int total_keys = 0;
int total_batches = 0;
int total_failures = 0;
// Reset the scan iterator so ScanMeta starts from the beginning.
// BucketStorageBackend uses cursor-based iteration (next_bucket_);
// after Init() completes the cursor is 0 and HasNext() returns false,
// which would make ScanMeta skip all buckets.
storage_backend_->ResetScanIterator();
LOG(INFO) << "ReRegisterOffloadedObjects: about to call "
"storage_backend_->ScanMeta()";
auto scan_meta_result =
storage_backend_->ScanMeta(
[this, &total_keys, &total_batches, &total_failures](
const std::vector<std::string>& keys,
std::vector<StorageObjectMetadata>& metadatas) {
total_batches++;
total_keys += keys.size();
for (auto& metadata : metadatas) {
metadata.transport_endpoint = local_rpc_addr_;
}
auto add_object_result =
client_->NotifyOffloadSuccess(keys, metadatas);
if (!add_object_result) {
total_failures++;
LOG(ERROR)
<< "ReRegisterOffloadedObjects: NotifyOffloadSuccess "
<< "failed for batch " << total_batches << " with "
<< keys.size()
<< " keys, error: " << add_object_result.error();
return add_object_result.error();
}
LOG(INFO) << "ReRegisterOffloadedObjects: NotifyOffloadSuccess "
<< "succeeded for batch " << total_batches << " with "
<< keys.size() << " keys";
return ErrorCode::OK;
});
LOG(INFO) << "ReRegisterOffloadedObjects: ScanMeta returned. success="
<< scan_meta_result.has_value();
if (!scan_meta_result) {
LOG(ERROR) << "ReRegisterOffloadedObjects: ScanMeta failed: "
<< scan_meta_result.error();
return scan_meta_result;
}
LOG(INFO) << "ReRegisterOffloadedObjects: completed. "
<< "total_keys=" << total_keys
<< " total_batches=" << total_batches
<< " total_failures=" << total_failures;
return {};
}
} // namespace mooncake

14
mooncake-store/src/master_service.cpp
View File

@ -2617,6 +2617,20 @@ auto MasterService::MountLocalDiskSegment(const UUID& client_id,
} else if (err != ErrorCode::OK) {
return tl::make_unexpected(err);
}
// Notify the client monitor thread to start tracking this client's TTL.
// Without this, a client that only mounts a LOCAL_DISK segment (and
// doesn't ping) would be considered expired by ClientMonitorFunc, which
// would then clear all its LOCAL_DISK replicas.
PodUUID pod_client_id;
pod_client_id.first = client_id.first;
pod_client_id.second = client_id.second;
if (!client_ping_queue_.push(pod_client_id)) {
LOG(ERROR) << "client_id=" << client_id
<< ", error=client_ping_queue_full";
return tl::make_unexpected(ErrorCode::INTERNAL_ERROR);
}
return {};
}

8
mooncake-store/src/real_client.cpp
View File

@ -4557,10 +4557,10 @@ RealClient::batch_get_into_internal(const std::vector<std::string> &keys,
std::chrono::duration_cast<std::chrono::microseconds>(
end_time - start_read_store_time)
.count();
LOG(INFO) << "Time taken for batch_get_into: " << elapsed_time
<< "us, read store: " << read_store_time
<< "us, with memory key count: " << valid_operations.size()
<< ", offload key count: " << offload_object_count;
// LOG(INFO) << "Time taken for batch_get_into: " << elapsed_time
// << "us, read store: " << read_store_time
// << "us, with memory key count: " << valid_operations.size()
// << ", offload key count: " << offload_object_count;
return results;
}

3
mooncake-transfer-engine/include/transfer_metadata.h
View File

@ -178,6 +178,9 @@ class TransferMetadata {
int removeRpcMetaEntry(const std::string &server_name);
// Re-publish the local RPC meta entry to the HTTP metadata server.
int rePublishRpcMetaEntry(const std::string &server_name);
int getRpcMetaEntry(const std::string &server_name, RpcMetaDesc &desc);
int getNotifies(std::vector<NotifyDesc> &notifies);

30
mooncake-transfer-engine/src/transfer_metadata.cpp
View File

@ -1147,6 +1147,36 @@ int TransferMetadata::removeRpcMetaEntry(const std::string &server_name) {
return 0;
}
int TransferMetadata::rePublishRpcMetaEntry(const std::string &server_name) {
if (p2p_handshake_mode_) {
return 0;
}
const std::string full_key = rpc_meta_prefix_ + server_name;
Json::Value existing;
if (storage_plugin_->get(full_key, existing)) {
Json::Value desired;
desired["ip_or_host_name"] = local_rpc_meta_.ip_or_host_name;
desired["rpc_port"] =
static_cast<Json::UInt64>(local_rpc_meta_.rpc_port);
if (existing == desired) {
return 0;
}
storage_plugin_->remove(full_key);
}
LOG(INFO) << "Re-publishing RPC meta entry for " << server_name;
Json::Value rpcMetaJSON;
rpcMetaJSON["ip_or_host_name"] = local_rpc_meta_.ip_or_host_name;
rpcMetaJSON["rpc_port"] =
static_cast<Json::UInt64>(local_rpc_meta_.rpc_port);
if (!storage_plugin_->set(full_key, rpcMetaJSON)) {
LOG(ERROR) << "Failed to re-publish RPC meta entry for " << server_name;
return ERR_METADATA;
}
return 0;
}
int TransferMetadata::getRpcMetaEntry(const std::string &server_name,
RpcMetaDesc &desc) {
{