[Store] Revive HTTP metadata cleanup from #1219 - #2256
Open
stmatengss wants to merge 2 commits into
Open
stmatengss wants to merge 2 commits into
stmatengss wants to merge 2 commits into
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request refactors the mooncake-store configuration and service architecture, introducing a simplified leader election helper and master service supervisor while removing obsolete features such as snapshotting, promotion-on-hit, and several unused RPC endpoints. Additionally, the separate MasterAdminServer has been replaced by integrating HTTP metrics directly into WrappedMasterService. The review feedback highlights critical issues regarding thread management and shutdown responsiveness. Specifically, the reviewer points out a potential thread leak in the supervisor's startup failure path and notes that background threads in both HttpMetadataServer and WrappedMasterService use blocking sleeps that delay shutdown. Implementing the suggested condition variables and proper thread joining will ensure clean, responsive termination and prevent undefined behavior.
Comment on lines
+174
to
+177
| if (!http_metadata_server->is_running()) { | ||
| LOG(ERROR) << "Failed to start HTTP metadata server"; | ||
| return -1; | ||
| } |
Contributor
There was a problem hiding this comment.
If the HTTP metadata server fails to start, the function returns -1 immediately without stopping or joining keep_leader_thread. Since keep_leader_thread captures local variables server and mv_helper by reference, this leads to a thread leak and undefined behavior/crash when the thread accesses these destroyed objects. We should cancel the keep-alive and join the thread before returning.
Suggested change
| if (!http_metadata_server->is_running()) { | |
| LOG(ERROR) << "Failed to start HTTP metadata server"; | |
| return -1; | |
| } | |
| if (!http_metadata_server->is_running()) { | |
| LOG(ERROR) << "Failed to start HTTP metadata server"; | |
| EtcdHelper::CancelKeepAlive(lease_id); | |
| keep_leader_thread.join(); | |
| return -1; | |
| } |
Comment on lines
138
to
+228
| return; | ||
| } | ||
|
|
||
| // Stop health monitoring thread | ||
| health_monitor_running_ = false; | ||
| if (health_monitor_thread_.joinable()) { | ||
| health_monitor_thread_.join(); | ||
| } | ||
|
|
||
| server_->stop(); | ||
| running_ = false; | ||
| LOG(INFO) << "HTTP metadata server stopped"; | ||
| } | ||
|
|
||
| void HttpMetadataServer::health_monitor_thread_func() { | ||
| while (health_monitor_running_) { | ||
| check_and_cleanup_metadata(); | ||
| std::this_thread::sleep_for( | ||
| std::chrono::milliseconds(kHealthMonitorSleepMs)); | ||
| } | ||
| } | ||
|
|
||
| void HttpMetadataServer::check_and_cleanup_metadata() { | ||
| if (!wrapped_master_service_) { | ||
| return; | ||
| } | ||
|
|
||
| // Get all segments once from master service | ||
| auto segments_result = wrapped_master_service_->GetAllSegments(); | ||
| if (!segments_result.has_value()) { | ||
| LOG(WARNING) << "Failed to get all segments for metadata cleanup"; | ||
| return; | ||
| } | ||
| const auto& all_segments = segments_result.value(); | ||
|
|
||
| // Convert to unordered_set for O(1) lookup | ||
| std::unordered_set<std::string> segment_set(all_segments.begin(), | ||
| all_segments.end()); | ||
|
|
||
| // Get all current keys from the metadata store | ||
| std::vector<std::string> keys_to_check; | ||
| { | ||
| std::lock_guard<std::mutex> lock(store_mutex_); | ||
| for (const auto& pair : store_) { | ||
| keys_to_check.push_back(pair.first); | ||
| } | ||
| } | ||
|
|
||
| // Check each key to see if it corresponds to segment metadata | ||
| // that should be cleaned up | ||
| for (const auto& key : keys_to_check) { | ||
| // Check if this key corresponds to segment metadata (e.g., keys | ||
| // containing "segment") | ||
| if (key.find("segment") != std::string::npos) { | ||
| std::string segment_name = key; | ||
| // Extract segment name from key if needed | ||
| if (key.find("rpc_meta_") == 0) { | ||
| segment_name = key.substr(9); // Remove "rpc_meta_" prefix | ||
| } | ||
| if (!is_segment_healthy(segment_name, segment_set)) { | ||
| cleanup_segment_metadata(segment_name); | ||
| } | ||
| } | ||
| // Note: Removed client health check as it has side effects. | ||
| // Client metadata cleanup should be handled by the master service | ||
| // based on actual client liveness tracking. | ||
| } | ||
| } | ||
|
|
||
| bool HttpMetadataServer::is_segment_healthy( | ||
| const std::string& segment_name, | ||
| const std::unordered_set<std::string>& all_segments) { | ||
| // Check if the segment exists in the provided set | ||
| return all_segments.find(segment_name) != all_segments.end(); | ||
| } | ||
|
|
||
| void HttpMetadataServer::cleanup_segment_metadata( | ||
| const std::string& segment_name) { | ||
| std::lock_guard<std::mutex> lock(store_mutex_); | ||
|
|
||
| // Find and remove all metadata entries related to this segment | ||
| for (auto it = store_.begin(); it != store_.end();) { | ||
| if (it->first.find(segment_name) != std::string::npos) { | ||
| LOG(INFO) << "Cleaning up metadata for segment: " << segment_name | ||
| << ", key: " << it->first; | ||
| it = store_.erase(it); | ||
| } else { | ||
| ++it; | ||
| } | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
Using std::this_thread::sleep_for with a 10-minute interval (kHealthMonitorSleepMs) causes the health monitor thread to block shutdown for up to 10 minutes when stop() is called. Additionally, the original check_and_cleanup_metadata implementation locks store_mutex_ multiple times and performs
return;
}
// Stop health monitoring thread
{
std::lock_guard<std::mutex> lock(health_monitor_mutex_);
health_monitor_running_ = false;
}
health_monitor_cv_.notify_all();
if (health_monitor_thread_.joinable()) {
health_monitor_thread_.join();
}
server_->stop();
running_ = false;
LOG(INFO) << "HTTP metadata server stopped";
}
void HttpMetadataServer::health_monitor_thread_func() {
std::unique_lock<std::mutex> lock(health_monitor_mutex_);
while (health_monitor_running_) {
lock.unlock();
check_and_cleanup_metadata();
lock.lock();
if (!health_monitor_running_) {
break;
}
health_monitor_cv_.wait_for(lock, std::chrono::milliseconds(kHealthMonitorSleepMs), [this] { return !health_monitor_running_; });
}
}
void HttpMetadataServer::check_and_cleanup_metadata() {
if (!wrapped_master_service_) {
return;
}
// Get all segments once from master service
auto segments_result = wrapped_master_service_->GetAllSegments();
if (!segments_result.has_value()) {
LOG(WARNING) << "Failed to get all segments for metadata cleanup";
return;
}
const auto& all_segments = segments_result.value();
// Convert to unordered_set for O(1) lookup
std::unordered_set<std::string> segment_set(all_segments.begin(),
all_segments.end());
std::lock_guard<std::mutex> lock(store_mutex_);
for (auto it = store_.begin(); it != store_.end();) {
const auto& key = it->first;
if (key.find("segment") != std::string::npos) {
std::string segment_name = key;
if (key.find("rpc_meta_") == 0) {
segment_name = key.substr(9); // Remove "rpc_meta_" prefix
}
if (segment_set.find(segment_name) == segment_set.end()) {
LOG(INFO) << "Cleaning up metadata for segment: " << segment_name
<< ", key: " << key;
it = store_.erase(it);
continue;
}
}
++it;
}
}
bool HttpMetadataServer::is_segment_healthy(
const std::string& segment_name,
const std::unordered_set<std::string>& all_segments) {
return all_segments.find(segment_name) != all_segments.end();
}
void HttpMetadataServer::cleanup_segment_metadata(
const std::string& segment_name) {
std::lock_guard<std::mutex> lock(store_mutex_);
for (auto it = store_.begin(); it != store_.end();) {
if (it->first.find(segment_name) != std::string::npos) {
LOG(INFO) << "Cleaning up metadata for segment: " << segment_name
<< ", key: " << it->first;
it = store_.erase(it);
} else {
++it;
}
}
}
Comment on lines
29
to
47
| WrappedMasterService::WrappedMasterService( | ||
| const WrappedMasterServiceConfig& config) | ||
| : master_service_(MasterServiceConfig(config)) {} | ||
|
|
||
| WrappedMasterService::~WrappedMasterService() = default; | ||
|
|
||
| MasterAdminServer::MasterAdminServer(uint16_t http_port, | ||
| bool enable_metric_reporting) | ||
| : http_port_(http_port), | ||
| enable_metric_reporting_(enable_metric_reporting), | ||
| http_server_(4, http_port) {} | ||
|
|
||
| MasterAdminServer::~MasterAdminServer() { Stop(); } | ||
| : master_service_(MasterServiceConfig(config)), | ||
| http_server_(4, config.http_port), | ||
| metric_report_running_(config.enable_metric_reporting) { | ||
| init_http_server(); | ||
|
|
||
| bool MasterAdminServer::Start() { | ||
| HAMetricManager::Init(); | ||
| InitHttpServer(); | ||
|
|
||
| auto ec = http_server_.async_start(); | ||
| if (ec.hasResult()) { | ||
| LOG(ERROR) << "Failed to start master admin server on port " | ||
| << http_port_; | ||
| return false; | ||
| } | ||
|
|
||
| started_.store(true); | ||
| if (enable_metric_reporting_) { | ||
| metric_report_running_.store(true); | ||
| if (config.enable_metric_reporting) { | ||
| metric_report_thread_ = std::thread([this]() { | ||
| while (metric_report_running_.load()) { | ||
| const auto snapshot = SnapshotState(); | ||
| std::ostringstream log_stream; | ||
| log_stream << "Master Admin Metrics: role=" | ||
| << ha::MasterRuntimeRoleToString(snapshot.state) | ||
| << ", state=" | ||
| << ha::MasterRuntimeStateToString(snapshot.state) | ||
| << ", service_ready=" | ||
| << (snapshot.service_available ? "true" : "false") | ||
| << ", master={" | ||
| << MasterMetricManager::instance() | ||
| .get_summary_string_and_update_snapshot() | ||
| << "}" | ||
| << ", ha={" | ||
| << HAMetricManager::instance().get_summary_string() | ||
| << "}"; | ||
| if (snapshot.leader_view.has_value()) { | ||
| log_stream | ||
| << ", leader=" << snapshot.leader_view->leader_address | ||
| << ", view_version=" | ||
| << snapshot.leader_view->view_version; | ||
| } | ||
| LOG(INFO) << log_stream.str(); | ||
| while (metric_report_running_) { | ||
| std::string metrics_summary = | ||
| MasterMetricManager::instance().get_summary_string(); | ||
| LOG(INFO) << "Master Metrics: " << metrics_summary; | ||
| std::this_thread::sleep_for( | ||
| std::chrono::seconds(kMetricReportIntervalSeconds)); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| LOG(INFO) << "Master admin server started on port " << http_server_.port(); | ||
| return true; | ||
| } |
Contributor
There was a problem hiding this comment.
Using std::this_thread::sleep_for with a 10-second interval (kMetricReportIntervalSeconds) blocks the destructor of WrappedMasterService for up to 10 seconds during shutdown. We should use a condition variable to allow immediate wake-up.
Suggested change
| WrappedMasterService::WrappedMasterService( | |
| const WrappedMasterServiceConfig& config) | |
| : master_service_(MasterServiceConfig(config)) {} | |
| WrappedMasterService::~WrappedMasterService() = default; | |
| MasterAdminServer::MasterAdminServer(uint16_t http_port, | |
| bool enable_metric_reporting) | |
| : http_port_(http_port), | |
| enable_metric_reporting_(enable_metric_reporting), | |
| http_server_(4, http_port) {} | |
| MasterAdminServer::~MasterAdminServer() { Stop(); } | |
| : master_service_(MasterServiceConfig(config)), | |
| http_server_(4, config.http_port), | |
| metric_report_running_(config.enable_metric_reporting) { | |
| init_http_server(); | |
| bool MasterAdminServer::Start() { | |
| HAMetricManager::Init(); | |
| InitHttpServer(); | |
| auto ec = http_server_.async_start(); | |
| if (ec.hasResult()) { | |
| LOG(ERROR) << "Failed to start master admin server on port " | |
| << http_port_; | |
| return false; | |
| } | |
| started_.store(true); | |
| if (enable_metric_reporting_) { | |
| metric_report_running_.store(true); | |
| if (config.enable_metric_reporting) { | |
| metric_report_thread_ = std::thread([this]() { | |
| while (metric_report_running_.load()) { | |
| const auto snapshot = SnapshotState(); | |
| std::ostringstream log_stream; | |
| log_stream << "Master Admin Metrics: role=" | |
| << ha::MasterRuntimeRoleToString(snapshot.state) | |
| << ", state=" | |
| << ha::MasterRuntimeStateToString(snapshot.state) | |
| << ", service_ready=" | |
| << (snapshot.service_available ? "true" : "false") | |
| << ", master={" | |
| << MasterMetricManager::instance() | |
| .get_summary_string_and_update_snapshot() | |
| << "}" | |
| << ", ha={" | |
| << HAMetricManager::instance().get_summary_string() | |
| << "}"; | |
| if (snapshot.leader_view.has_value()) { | |
| log_stream | |
| << ", leader=" << snapshot.leader_view->leader_address | |
| << ", view_version=" | |
| << snapshot.leader_view->view_version; | |
| } | |
| LOG(INFO) << log_stream.str(); | |
| while (metric_report_running_) { | |
| std::string metrics_summary = | |
| MasterMetricManager::instance().get_summary_string(); | |
| LOG(INFO) << "Master Metrics: " << metrics_summary; | |
| std::this_thread::sleep_for( | |
| std::chrono::seconds(kMetricReportIntervalSeconds)); | |
| } | |
| }); | |
| } | |
| LOG(INFO) << "Master admin server started on port " << http_server_.port(); | |
| return true; | |
| } | |
| WrappedMasterService::WrappedMasterService( | |
| const WrappedMasterServiceConfig& config) | |
| : master_service_(MasterServiceConfig(config)), | |
| http_server_(4, config.http_port), | |
| metric_report_running_(config.enable_metric_reporting) { | |
| init_http_server(); | |
| if (config.enable_metric_reporting) { | |
| metric_report_thread_ = std::thread([this]() { | |
| std::unique_lock<std::mutex> lock(metric_report_mutex_); | |
| while (metric_report_running_) { | |
| std::string metrics_summary = | |
| MasterMetricManager::instance().get_summary_string(); | |
| LOG(INFO) << "Master Metrics: " << metrics_summary; | |
| metric_report_cv_.wait_for(lock, std::chrono::seconds(kMetricReportIntervalSeconds), [this] { return !metric_report_running_; }); | |
| } | |
| }); | |
| } | |
| } |
Comment on lines
49
to
55
| WrappedMasterService::~WrappedMasterService() { | ||
| metric_report_running_ = false; | ||
| if (metric_report_thread_.joinable()) { | ||
| metric_report_thread_.join(); | ||
| } | ||
| if (started_.exchange(false)) { | ||
| http_server_.stop(); | ||
| } | ||
| } | ||
|
|
||
| void MasterAdminServer::SetRuntimeState(ha::MasterRuntimeState state) { | ||
| std::lock_guard<std::mutex> lock(state_mutex_); | ||
| state_ = state; | ||
| } | ||
|
|
||
| void MasterAdminServer::SetObservedLeader( | ||
| const std::optional<ha::MasterView>& leader_view) { | ||
| std::lock_guard<std::mutex> lock(state_mutex_); | ||
| leader_view_ = leader_view; | ||
| } | ||
|
|
||
| void MasterAdminServer::SetServiceDelegate( | ||
| std::shared_ptr<WrappedMasterService> service) { | ||
| std::lock_guard<std::mutex> lock(state_mutex_); | ||
| service_ = std::move(service); | ||
| if (!service_) { | ||
| service_available_ = false; | ||
| } | ||
| } | ||
|
|
||
| void MasterAdminServer::SetServiceAvailable(bool available) { | ||
| std::lock_guard<std::mutex> lock(state_mutex_); | ||
| service_available_ = available && service_ != nullptr; | ||
| } | ||
|
|
||
| MasterAdminServer::RuntimeSnapshot MasterAdminServer::SnapshotState() const { | ||
| std::lock_guard<std::mutex> lock(state_mutex_); | ||
| return RuntimeSnapshot{ | ||
| .state = state_, | ||
| .leader_view = leader_view_, | ||
| .service = service_, | ||
| .service_available = service_available_, | ||
| }; | ||
| } | ||
|
|
||
| std::string MasterAdminServer::BuildMetricsText() const { | ||
| return AppendMetricSections( | ||
| MasterMetricManager::instance().serialize_metrics(), | ||
| HAMetricManager::instance().serialize_metrics()); | ||
| } | ||
|
|
||
| std::string MasterAdminServer::BuildMetricsSummaryText() const { | ||
| const auto snapshot = SnapshotState(); | ||
| std::ostringstream oss; | ||
| oss << "role=" << ha::MasterRuntimeRoleToString(snapshot.state) | ||
| << ", state=" << ha::MasterRuntimeStateToString(snapshot.state) | ||
| << ", service_ready=" << (snapshot.service_available ? "true" : "false") | ||
| << ", master={" << MasterMetricManager::instance().get_summary_string() | ||
| << "}, ha={" << HAMetricManager::instance().get_summary_string() << "}"; | ||
| if (snapshot.leader_view.has_value()) { | ||
| oss << ", leader=" << snapshot.leader_view->leader_address | ||
| << ", view_version=" << snapshot.leader_view->view_version; | ||
| } | ||
| return oss.str(); | ||
| } | ||
|
|
||
| std::string MasterAdminServer::BuildHealthJson() const { | ||
| const auto snapshot = SnapshotState(); | ||
| std::ostringstream oss; | ||
| oss << "{\"status\":\"ok\",\"role\":\"" | ||
| << ha::MasterRuntimeRoleToString(snapshot.state) << "\",\"ha_state\":\"" | ||
| << ha::MasterRuntimeStateToString(snapshot.state) | ||
| << "\",\"service_ready\":" | ||
| << (snapshot.service_available ? "true" : "false"); | ||
| if (snapshot.leader_view.has_value()) { | ||
| oss << ",\"leader_address\":\"" | ||
| << EscapeJson(snapshot.leader_view->leader_address) | ||
| << "\",\"view_version\":" << snapshot.leader_view->view_version; | ||
| } | ||
| oss << "}"; | ||
| return oss.str(); | ||
| } | ||
|
|
||
| std::string MasterAdminServer::BuildLeaderJson() const { | ||
| const auto snapshot = SnapshotState(); | ||
| if (!snapshot.leader_view.has_value()) { | ||
| return "{\"present\":false}"; | ||
| } | ||
|
|
||
| std::ostringstream oss; | ||
| oss << "{\"present\":true,\"leader_address\":\"" | ||
| << EscapeJson(snapshot.leader_view->leader_address) | ||
| << "\",\"view_version\":" << snapshot.leader_view->view_version << "}"; | ||
| return oss.str(); | ||
| } | ||
|
|
||
| std::shared_ptr<WrappedMasterService> MasterAdminServer::GetActiveService() | ||
| const { | ||
| const auto snapshot = SnapshotState(); | ||
| if (!snapshot.service_available) { | ||
| return nullptr; | ||
| } | ||
| return snapshot.service; | ||
| http_server_.stop(); | ||
| } |
Contributor
There was a problem hiding this comment.
Update the destructor to notify the condition variable and join the metric reporting thread immediately.
WrappedMasterService::~WrappedMasterService() {
{
std::lock_guard<std::mutex> lock(metric_report_mutex_);
metric_report_running_ = false;
}
metric_report_cv_.notify_all();
if (metric_report_thread_.joinable()) {
metric_report_thread_.join();
}
http_server_.stop();
}
Comment on lines
4
to
+15
| #include <csignal> | ||
| #include <string> | ||
| #include <unordered_map> | ||
| #include <unordered_set> | ||
| #include <mutex> | ||
| #include <thread> | ||
| #include <atomic> | ||
| #include <chrono> | ||
|
|
||
| #include <ylt/coro_http/coro_http_server.hpp> | ||
| #include "types.h" | ||
| #include "rpc_service.h" |
Contributor
There was a problem hiding this comment.
Include <condition_variable> to support condition variable-based sleep in the health monitor thread, which avoids blocking the server shutdown.
Suggested change
| #include <csignal> | |
| #include <string> | |
| #include <unordered_map> | |
| #include <unordered_set> | |
| #include <mutex> | |
| #include <thread> | |
| #include <atomic> | |
| #include <chrono> | |
| #include <ylt/coro_http/coro_http_server.hpp> | |
| #include "types.h" | |
| #include "rpc_service.h" | |
| #include <csignal> | |
| #include <string> | |
| #include <unordered_map> | |
| #include <unordered_set> | |
| #include <mutex> | |
| #include <thread> | |
| #include <atomic> | |
| #include <chrono> | |
| #include <condition_variable> | |
| #include <ylt/coro_http/coro_http_server.hpp> | |
| #include "types.h" | |
| #include "rpc_service.h" |
Comment on lines
51
to
72
| private: | ||
| void init_server(); | ||
| void health_monitor_thread_func(); | ||
| void check_and_cleanup_metadata(); | ||
| bool is_segment_healthy( | ||
| const std::string& segment_name, | ||
| const std::unordered_set<std::string>& all_segments); | ||
| void cleanup_segment_metadata(const std::string& segment_name); | ||
|
|
||
| uint16_t port_; | ||
| std::string host_; | ||
| std::unique_ptr<coro_http::coro_http_server> server_; | ||
| std::unordered_map<std::string, std::string> store_; | ||
| mutable std::mutex store_mutex_; | ||
| bool running_; | ||
|
|
||
| // Health monitoring | ||
| std::shared_ptr<WrappedMasterService> wrapped_master_service_; | ||
| std::thread health_monitor_thread_; | ||
| std::atomic<bool> health_monitor_running_{false}; | ||
| static constexpr uint64_t kHealthMonitorSleepMs = 600000; // 10 minutes | ||
| }; |
Contributor
There was a problem hiding this comment.
Add health_monitor_mutex_ and health_monitor_cv_ to support responsive thread shutdown without blocking for up to 10 minutes.
private:
void init_server();
void health_monitor_thread_func();
void check_and_cleanup_metadata();
bool is_segment_healthy(
const std::string& segment_name,
const std::unordered_set<std::string>& all_segments);
void cleanup_segment_metadata(const std::string& segment_name);
uint16_t port_;
std::string host_;
std::unique_ptr<coro_http::coro_http_server> server_;
std::unordered_map<std::string, std::string> store_;
mutable std::mutex store_mutex_;
bool running_;
// Health monitoring
std::shared_ptr<WrappedMasterService> wrapped_master_service_;
std::thread health_monitor_thread_;
std::atomic<bool> health_monitor_running_{false};
std::mutex health_monitor_mutex_;
std::condition_variable health_monitor_cv_;
static constexpr uint64_t kHealthMonitorSleepMs = 600000; // 10 minutes
};
Comment on lines
2
to
16
|
|
||
| #include <csignal> | ||
| #include <atomic> | ||
| #include <memory> | ||
| #include <mutex> | ||
| #include <optional> | ||
| #include <string> | ||
| #include <boost/functional/hash.hpp> | ||
| #include <cstdint> | ||
| #include <thread> | ||
| #include <ylt/coro_http/coro_http_server.hpp> | ||
| #include <ylt/coro_rpc/coro_rpc_server.hpp> | ||
| #include <ylt/util/tl/expected.hpp> | ||
|
|
||
| #include "ha/ha_types.h" | ||
| #include "master_service.h" | ||
| #include "types.h" | ||
| #include "rpc_types.h" | ||
| #include "master_config.h" | ||
| #include "segment.h" | ||
|
|
Contributor
There was a problem hiding this comment.
Include <mutex> and <condition_variable> to support responsive shutdown of the metric reporting thread.
Suggested change
| #include <csignal> | |
| #include <atomic> | |
| #include <memory> | |
| #include <mutex> | |
| #include <optional> | |
| #include <string> | |
| #include <boost/functional/hash.hpp> | |
| #include <cstdint> | |
| #include <thread> | |
| #include <ylt/coro_http/coro_http_server.hpp> | |
| #include <ylt/coro_rpc/coro_rpc_server.hpp> | |
| #include <ylt/util/tl/expected.hpp> | |
| #include "ha/ha_types.h" | |
| #include "master_service.h" | |
| #include "types.h" | |
| #include "rpc_types.h" | |
| #include "master_config.h" | |
| #include "segment.h" | |
| #include <csignal> | |
| #include <atomic> | |
| #include <mutex> | |
| #include <condition_variable> | |
| #include <boost/functional/hash.hpp> | |
| #include <cstdint> | |
| #include <thread> | |
| #include <ylt/coro_http/coro_http_server.hpp> | |
| #include <ylt/coro_rpc/coro_rpc_server.hpp> | |
| #include <ylt/util/tl/expected.hpp> | |
| #include "master_service.h" | |
| #include "types.h" | |
| #include "rpc_types.h" | |
| #include "master_config.h" |
Comment on lines
152
to
157
| private: | ||
| MasterService master_service_; | ||
| }; | ||
|
|
||
| class MasterAdminServer { | ||
| public: | ||
| MasterAdminServer(uint16_t http_port, bool enable_metric_reporting); | ||
|
|
||
| ~MasterAdminServer(); | ||
|
|
||
| bool Start(); | ||
|
|
||
| void Stop(); | ||
|
|
||
| void SetRuntimeState(ha::MasterRuntimeState state); | ||
|
|
||
| void SetObservedLeader(const std::optional<ha::MasterView>& leader_view); | ||
|
|
||
| void SetServiceDelegate(std::shared_ptr<WrappedMasterService> service); | ||
|
|
||
| void SetServiceAvailable(bool available); | ||
|
|
||
| private: | ||
| struct RuntimeSnapshot { | ||
| ha::MasterRuntimeState state = ha::MasterRuntimeState::kStarting; | ||
| std::optional<ha::MasterView> leader_view; | ||
| std::shared_ptr<WrappedMasterService> service; | ||
| bool service_available = false; | ||
| }; | ||
|
|
||
| RuntimeSnapshot SnapshotState() const; | ||
|
|
||
| std::string BuildMetricsText() const; | ||
|
|
||
| std::string BuildMetricsSummaryText() const; | ||
|
|
||
| std::string BuildHealthJson() const; | ||
|
|
||
| std::string BuildLeaderJson() const; | ||
|
|
||
| std::shared_ptr<WrappedMasterService> GetActiveService() const; | ||
|
|
||
| void InitHttpServer(); | ||
|
|
||
| uint16_t http_port_; | ||
| bool enable_metric_reporting_ = false; | ||
| coro_http::coro_http_server http_server_; | ||
| std::thread metric_report_thread_; | ||
| std::atomic<bool> metric_report_running_{false}; | ||
| std::atomic<bool> started_{false}; | ||
| mutable std::mutex state_mutex_; | ||
| ha::MasterRuntimeState state_{ha::MasterRuntimeState::kStarting}; | ||
| std::optional<ha::MasterView> leader_view_; | ||
| std::shared_ptr<WrappedMasterService> service_; | ||
| bool service_available_ = false; | ||
| coro_http::coro_http_server http_server_; | ||
| std::atomic<bool> metric_report_running_; | ||
| }; |
Contributor
There was a problem hiding this comment.
Add metric_report_mutex_ and metric_report_cv_ to allow the metric reporting thread to wake up immediately during shutdown.
private:
MasterService master_service_;
std::thread metric_report_thread_;
coro_http::coro_http_server http_server_;
std::atomic<bool> metric_report_running_;
std::mutex metric_report_mutex_;
std::condition_variable metric_report_cv_;
};
stmatengss
force-pushed
the
store/revive-pr1219-v2
branch
from
June 7, 2026 17:58
6ad1eef to
ea84b18
Compare
The WrappedMasterService class exposes GetAllSegmentsForAdmin() (which wraps MasterService::GetAllSegments()), not GetAllSegments() directly. This fixes the compile error in http_metadata_server.cpp. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
|
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
30 tasks
32 tasks
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR revives and reapplies the HTTP metadata cleanup work from #1219 on top of current
main.Context
mainNotes