From 9fa7d40306cff0553c3c9398c591b88d8c1c9b9b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Tue, 14 Jul 2026 17:53:02 +0800 Subject: [PATCH 1/6] feat(store): add tier-aware replica placement policy model --- .../include/replica_placement_policy.h | 99 ++++++ mooncake-store/src/CMakeLists.txt | 1 + .../src/replica_placement_policy.cpp | 136 ++++++++ mooncake-store/tests/CMakeLists.txt | 1 + .../tests/replica_placement_policy_test.cpp | 292 ++++++++++++++++++ 5 files changed, 529 insertions(+) create mode 100644 mooncake-store/include/replica_placement_policy.h create mode 100644 mooncake-store/src/replica_placement_policy.cpp create mode 100644 mooncake-store/tests/replica_placement_policy_test.cpp diff --git a/mooncake-store/include/replica_placement_policy.h b/mooncake-store/include/replica_placement_policy.h new file mode 100644 index 0000000000..4e7904d98d --- /dev/null +++ b/mooncake-store/include/replica_placement_policy.h @@ -0,0 +1,99 @@ +// Copyright 2026 Mooncake Authors + +#pragma once + +#include +#include +#include +#include + +namespace mooncake { + +enum class ReplicaPlacementTier : uint8_t { + MEMORY = 0, + LOCAL_DISK, + NOF_SSD, + REMOTE_STORE, + COUNT, +}; + +enum class ReplicaTemperature : uint8_t { + COLD = 0, + WARM, + HOT, + COUNT, +}; + +inline constexpr size_t kReplicaPlacementTierCount = + static_cast(ReplicaPlacementTier::COUNT); +inline constexpr size_t kReplicaTemperatureCount = + static_cast(ReplicaTemperature::COUNT); + +struct ReplicaTierTarget { + uint32_t desired{0}; + bool required{false}; +}; + +struct ReplicaPlacementPolicyConfig { + std::array, + kReplicaTemperatureCount> + targets{}; + uint32_t min_complete_replicas{1}; + uint32_t max_total_replicas{16}; +}; + +enum class ReplicaTierSignalState : uint8_t { + UNKNOWN = 0, + AVAILABLE, + UNAVAILABLE, +}; + +struct ReplicaTierObservation { + uint32_t complete{0}; + uint32_t pending{0}; + ReplicaTierSignalState allocation_state{ReplicaTierSignalState::UNKNOWN}; + ReplicaTierSignalState health_state{ReplicaTierSignalState::UNKNOWN}; +}; + +enum class ReplicaPlacementDegradedReason : uint8_t { + REQUIRED_TIER_UNAVAILABLE = 0, + REQUIRED_TIER_UNHEALTHY, + REQUIRED_TIER_SIGNAL_UNKNOWN, + MIN_COMPLETE_UNSATISFIED, +}; + +struct ReplicaTierAdjustment { + ReplicaPlacementTier tier{ReplicaPlacementTier::MEMORY}; + uint32_t add{0}; + uint32_t remove{0}; + uint32_t effective_target{0}; +}; + +struct ReplicaPlacementPlan { + std::array adjustments{}; + std::array + degraded_reasons{}; + size_t degraded_reason_count{0}; + + [[nodiscard]] bool degraded() const noexcept { + return degraded_reason_count != 0; + } +}; + +// Pure, immutable policy. It produces intent only; callers remain responsible +// for admission, source/target selection, task submission, and transactional +// metadata changes. +class ReplicaPlacementPolicy final { + public: + explicit ReplicaPlacementPolicy(ReplicaPlacementPolicyConfig config); + + [[nodiscard]] ReplicaPlacementPlan Plan( + ReplicaTemperature temperature, + std::span + observations) const; + + private: + ReplicaPlacementPolicyConfig config_; +}; + +} // namespace mooncake diff --git a/mooncake-store/src/CMakeLists.txt b/mooncake-store/src/CMakeLists.txt index 1ffff9ac3d..5d7f80ec75 100644 --- a/mooncake-store/src/CMakeLists.txt +++ b/mooncake-store/src/CMakeLists.txt @@ -73,6 +73,7 @@ set(MOONCAKE_STORE_SOURCES store_c.cpp memory_alloc.cpp ssd_register_client.cpp + replica_placement_policy.cpp engram/engram_store.cpp) set(EXTRA_LIBS "") diff --git a/mooncake-store/src/replica_placement_policy.cpp b/mooncake-store/src/replica_placement_policy.cpp new file mode 100644 index 0000000000..6118315840 --- /dev/null +++ b/mooncake-store/src/replica_placement_policy.cpp @@ -0,0 +1,136 @@ +// Copyright 2026 Mooncake Authors + +#include "replica_placement_policy.h" + +#include +#include +#include + +namespace mooncake { +namespace { + +size_t TemperatureIndex(ReplicaTemperature temperature) { + return static_cast(temperature); +} + +uint32_t SaturatingAdd(uint32_t lhs, uint32_t rhs) { + if (rhs > std::numeric_limits::max() - lhs) { + return std::numeric_limits::max(); + } + return lhs + rhs; +} + +} // namespace + +ReplicaPlacementPolicy::ReplicaPlacementPolicy( + ReplicaPlacementPolicyConfig config) + : config_(std::move(config)) { + if (config_.min_complete_replicas == 0 || + config_.max_total_replicas < config_.min_complete_replicas) { + throw std::invalid_argument("invalid replica placement safety bounds"); + } + for (const auto& temperature_targets : config_.targets) { + uint64_t total = 0; + for (const auto& target : temperature_targets) { + if (target.required && target.desired == 0) { + throw std::invalid_argument( + "required replica placement tier has zero target"); + } + total += target.desired; + } + if (total < config_.min_complete_replicas || + total > config_.max_total_replicas) { + throw std::invalid_argument( + "replica placement target violates total replica bounds"); + } + } +} + +ReplicaPlacementPlan ReplicaPlacementPolicy::Plan( + ReplicaTemperature temperature, + std::span + observations) const { + const size_t temperature_index = TemperatureIndex(temperature); + if (temperature_index >= kReplicaTemperatureCount) { + throw std::invalid_argument("invalid replica temperature"); + } + + ReplicaPlacementPlan plan; + uint64_t projected_complete = 0; + for (size_t i = 0; i < kReplicaPlacementTierCount; ++i) { + const auto tier = static_cast(i); + const auto& observation = observations[i]; + const auto& target = config_.targets[temperature_index][i]; + auto& adjustment = plan.adjustments[i]; + adjustment.tier = tier; + + const uint32_t present = + SaturatingAdd(observation.complete, observation.pending); + const bool allocation_known_available = + observation.allocation_state == ReplicaTierSignalState::AVAILABLE; + const bool health_known_available = + observation.health_state == ReplicaTierSignalState::AVAILABLE; + const bool health_known_unavailable = + observation.health_state == ReplicaTierSignalState::UNAVAILABLE; + const bool can_add = + allocation_known_available && health_known_available; + adjustment.effective_target = target.desired; + if (target.desired > present && !can_add) { + adjustment.effective_target = present; + } + if (target.required && + plan.degraded_reason_count < plan.degraded_reasons.size()) { + if (health_known_unavailable) { + plan.degraded_reasons[plan.degraded_reason_count++] = + ReplicaPlacementDegradedReason::REQUIRED_TIER_UNHEALTHY; + } else if (observation.health_state == + ReplicaTierSignalState::UNKNOWN) { + plan.degraded_reasons[plan.degraded_reason_count++] = + ReplicaPlacementDegradedReason:: + REQUIRED_TIER_SIGNAL_UNKNOWN; + } else if (target.desired > present && + observation.allocation_state == + ReplicaTierSignalState::UNAVAILABLE) { + plan.degraded_reasons[plan.degraded_reason_count++] = + ReplicaPlacementDegradedReason::REQUIRED_TIER_UNAVAILABLE; + } else if (target.desired > present && + observation.allocation_state == + ReplicaTierSignalState::UNKNOWN) { + plan.degraded_reasons[plan.degraded_reason_count++] = + ReplicaPlacementDegradedReason:: + REQUIRED_TIER_SIGNAL_UNKNOWN; + } + } + + if (adjustment.effective_target > present) { + adjustment.add = adjustment.effective_target - present; + } else if (adjustment.effective_target < observation.complete && + observation.pending == 0 && health_known_available) { + adjustment.remove = + observation.complete - adjustment.effective_target; + } + // The safety floor is explicitly about replicas that are COMPLETE at + // planning time. Pending replicas and add intents suppress duplicate + // additions, but they cannot justify deleting the last readable copy: + // either may still fail before becoming COMPLETE. + if (health_known_available) { + projected_complete += observation.complete - adjustment.remove; + } + } + + // Never emit destructive intent when the available/addable result cannot + // preserve the configured minimum. Existing replicas are safer than a + // policy target under partial tier failure. + if (projected_complete < config_.min_complete_replicas) { + for (auto& adjustment : plan.adjustments) { + adjustment.remove = 0; + } + if (plan.degraded_reason_count < plan.degraded_reasons.size()) { + plan.degraded_reasons[plan.degraded_reason_count++] = + ReplicaPlacementDegradedReason::MIN_COMPLETE_UNSATISFIED; + } + } + return plan; +} + +} // namespace mooncake diff --git a/mooncake-store/tests/CMakeLists.txt b/mooncake-store/tests/CMakeLists.txt index e7e66c9fbb..ae41a41690 100644 --- a/mooncake-store/tests/CMakeLists.txt +++ b/mooncake-store/tests/CMakeLists.txt @@ -44,6 +44,7 @@ add_test( set_tests_properties( replica_selection_env_opt_in_test PROPERTIES ENVIRONMENT "MC_STORE_REPLICA_SCORING=1") +add_store_test(replica_placement_policy_test replica_placement_policy_test.cpp) add_store_test(eviction_strategy_test eviction_strategy_test.cpp) add_store_test(deadline_scheduler_test deadline_scheduler_test.cpp) add_store_test(kv_event_publisher_test kv_event_publisher_test.cpp) diff --git a/mooncake-store/tests/replica_placement_policy_test.cpp b/mooncake-store/tests/replica_placement_policy_test.cpp new file mode 100644 index 0000000000..d2723c5a6c --- /dev/null +++ b/mooncake-store/tests/replica_placement_policy_test.cpp @@ -0,0 +1,292 @@ +// Copyright 2026 Mooncake Authors + +#include "replica_placement_policy.h" + +#include +#include +#include +#include +#include + +#include + +namespace mooncake { +namespace { + +size_t Index(ReplicaPlacementTier tier) { return static_cast(tier); } + +ReplicaTierObservation Available(uint32_t complete = 0, uint32_t pending = 0) { + return {.complete = complete, + .pending = pending, + .allocation_state = ReplicaTierSignalState::AVAILABLE, + .health_state = ReplicaTierSignalState::AVAILABLE}; +} + +ReplicaPlacementPolicyConfig TestConfig() { + ReplicaPlacementPolicyConfig config; + config.targets[static_cast(ReplicaTemperature::COLD)] = { + ReplicaTierTarget{0, false}, ReplicaTierTarget{1, true}, + ReplicaTierTarget{0, false}, ReplicaTierTarget{1, true}}; + config.targets[static_cast(ReplicaTemperature::WARM)] = { + ReplicaTierTarget{1, true}, ReplicaTierTarget{1, false}, + ReplicaTierTarget{1, false}, ReplicaTierTarget{1, true}}; + config.targets[static_cast(ReplicaTemperature::HOT)] = { + ReplicaTierTarget{2, true}, ReplicaTierTarget{1, false}, + ReplicaTierTarget{1, false}, ReplicaTierTarget{1, true}}; + config.min_complete_replicas = 1; + config.max_total_replicas = 8; + return config; +} + +TEST(ReplicaPlacementPolicyTest, ProducesDeterministicHotTierDiff) { + ReplicaPlacementPolicy policy(TestConfig()); + std::array observed{{ + Available(1), + Available(2), + Available(), + Available(1), + }}; + + const auto plan = policy.Plan(ReplicaTemperature::HOT, observed); + + EXPECT_FALSE(plan.degraded()); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::MEMORY)].add, 1); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::LOCAL_DISK)].remove, + 1); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::NOF_SSD)].add, 1); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::REMOTE_STORE)].add, + 0); +} + +TEST(ReplicaPlacementPolicyTest, PendingReplicaCountsTowardTarget) { + ReplicaPlacementPolicy policy(TestConfig()); + std::array observed{{ + Available(1, 1), + Available(1), + Available(0, 1), + Available(1), + }}; + + const auto plan = policy.Plan(ReplicaTemperature::HOT, observed); + + for (const auto& adjustment : plan.adjustments) { + EXPECT_EQ(adjustment.add, 0); + EXPECT_EQ(adjustment.remove, 0); + } +} + +TEST(ReplicaPlacementPolicyTest, DefersRemovalWhileTierHasPendingReplica) { + ReplicaPlacementPolicy policy(TestConfig()); + std::array observed{{ + Available(3, 1), + Available(1), + Available(1), + Available(1), + }}; + + const auto plan = policy.Plan(ReplicaTemperature::HOT, observed); + + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::MEMORY)].add, 0); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::MEMORY)].remove, 0); +} + +TEST(ReplicaPlacementPolicyTest, DoesNotRemoveFromUnhealthyTier) { + ReplicaPlacementPolicy policy(TestConfig()); + std::array observed{{ + {.complete = 3, + .allocation_state = ReplicaTierSignalState::AVAILABLE, + .health_state = ReplicaTierSignalState::UNAVAILABLE}, + Available(1), + Available(1), + Available(1), + }}; + + const auto plan = policy.Plan(ReplicaTemperature::HOT, observed); + + EXPECT_TRUE(plan.degraded()); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::MEMORY)].remove, 0); +} + +TEST(ReplicaPlacementPolicyTest, RequiredUnavailableTierDegradesWithoutAdd) { + ReplicaPlacementPolicy policy(TestConfig()); + std::array observed{{ + {.allocation_state = ReplicaTierSignalState::UNAVAILABLE, + .health_state = ReplicaTierSignalState::AVAILABLE}, + Available(1), + Available(1), + Available(1), + }}; + + const auto plan = policy.Plan(ReplicaTemperature::HOT, observed); + + ASSERT_TRUE(plan.degraded()); + ASSERT_EQ(plan.degraded_reason_count, 1); + EXPECT_EQ(plan.degraded_reasons[0], + ReplicaPlacementDegradedReason::REQUIRED_TIER_UNAVAILABLE); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::MEMORY)].add, 0); +} + +TEST(ReplicaPlacementPolicyTest, RequiredUnhealthyTierHasDistinctReason) { + ReplicaPlacementPolicy policy(TestConfig()); + std::array observed{{ + {.allocation_state = ReplicaTierSignalState::AVAILABLE, + .health_state = ReplicaTierSignalState::UNAVAILABLE}, + Available(1), + Available(1), + Available(1), + }}; + + const auto plan = policy.Plan(ReplicaTemperature::HOT, observed); + + ASSERT_EQ(plan.degraded_reason_count, 1); + EXPECT_EQ(plan.degraded_reasons[0], + ReplicaPlacementDegradedReason::REQUIRED_TIER_UNHEALTHY); +} + +TEST(ReplicaPlacementPolicyTest, InsufficientTargetRejectsConfiguration) { + auto config = TestConfig(); + config.targets[static_cast(ReplicaTemperature::COLD)] = {}; + EXPECT_THROW(ReplicaPlacementPolicy(std::move(config)), + std::invalid_argument); +} + +TEST(ReplicaPlacementPolicyTest, RequiredUnknownSignalIsNotFabricated) { + ReplicaPlacementPolicy policy(TestConfig()); + std::array observed{{ + {}, + Available(1), + Available(1), + Available(1), + }}; + + const auto plan = policy.Plan(ReplicaTemperature::HOT, observed); + + ASSERT_EQ(plan.degraded_reason_count, 1); + EXPECT_EQ(plan.degraded_reasons[0], + ReplicaPlacementDegradedReason::REQUIRED_TIER_SIGNAL_UNKNOWN); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::MEMORY)].add, 0); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::MEMORY)].remove, 0); +} + +TEST(ReplicaPlacementPolicyTest, ExcessiveTargetRejectsConfiguration) { + auto config = TestConfig(); + config + .targets[static_cast(ReplicaTemperature::HOT)] + [Index(ReplicaPlacementTier::MEMORY)] + .desired = 9; + EXPECT_THROW(ReplicaPlacementPolicy(std::move(config)), + std::invalid_argument); +} + +TEST(ReplicaPlacementPolicyTest, RequiredZeroTargetRejectsConfiguration) { + auto config = TestConfig(); + config + .targets[static_cast(ReplicaTemperature::HOT)] + [Index(ReplicaPlacementTier::NOF_SSD)] + .required = true; + config + .targets[static_cast(ReplicaTemperature::HOT)] + [Index(ReplicaPlacementTier::NOF_SSD)] + .desired = 0; + EXPECT_THROW(ReplicaPlacementPolicy(std::move(config)), + std::invalid_argument); +} + +TEST(ReplicaPlacementPolicyTest, InvalidTemperatureIsRejected) { + ReplicaPlacementPolicy policy(TestConfig()); + std::array observed{}; + EXPECT_THROW( + (void)policy.Plan(static_cast(255), observed), + std::invalid_argument); +} + +TEST(ReplicaPlacementPolicyTest, + DoesNotRemoveLastReplicaWhenTargetUnavailable) { + auto config = TestConfig(); + config.min_complete_replicas = 2; + ReplicaPlacementPolicy policy(std::move(config)); + std::array observed{{ + {.complete = 1, + .allocation_state = ReplicaTierSignalState::UNAVAILABLE, + .health_state = ReplicaTierSignalState::AVAILABLE}, + Available(1), + {.allocation_state = ReplicaTierSignalState::UNAVAILABLE, + .health_state = ReplicaTierSignalState::AVAILABLE}, + {.allocation_state = ReplicaTierSignalState::UNAVAILABLE, + .health_state = ReplicaTierSignalState::AVAILABLE}, + }}; + + const auto plan = policy.Plan(ReplicaTemperature::COLD, observed); + + EXPECT_TRUE(plan.degraded()); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::MEMORY)].remove, 0); +} + +TEST(ReplicaPlacementPolicyTest, + PendingReplicaCannotSatisfyCompletedReplicaSafetyFloor) { + ReplicaPlacementPolicy policy(TestConfig()); + std::array observed{{ + Available(1), + Available(0, 1), + Available(), + Available(), + }}; + + const auto plan = policy.Plan(ReplicaTemperature::COLD, observed); + + EXPECT_TRUE(plan.degraded()); + EXPECT_EQ(plan.adjustments[Index(ReplicaPlacementTier::MEMORY)].remove, 0); + ASSERT_GT(plan.degraded_reason_count, 0); + EXPECT_EQ(plan.degraded_reasons[plan.degraded_reason_count - 1], + ReplicaPlacementDegradedReason::MIN_COMPLETE_UNSATISFIED); +} + +TEST(ReplicaPlacementPolicyTest, ConcurrentPlanningIsDeterministic) { + const ReplicaPlacementPolicy policy(TestConfig()); + const std::array + observed{{ + Available(1), + Available(2), + Available(), + Available(1), + }}; + const auto expected = policy.Plan(ReplicaTemperature::HOT, observed); + std::atomic mismatches{0}; + constexpr size_t kThreadCount = 32; + constexpr size_t kIterationsPerThread = 10'000; + std::vector threads; + threads.reserve(kThreadCount); + + for (size_t thread = 0; thread < kThreadCount; ++thread) { + threads.emplace_back([&]() { + for (size_t iteration = 0; iteration < kIterationsPerThread; + ++iteration) { + const auto actual = + policy.Plan(ReplicaTemperature::HOT, observed); + if (actual.degraded_reason_count != + expected.degraded_reason_count) { + mismatches.fetch_add(1, std::memory_order_relaxed); + continue; + } + for (size_t i = 0; i < kReplicaPlacementTierCount; ++i) { + const auto& lhs = actual.adjustments[i]; + const auto& rhs = expected.adjustments[i]; + if (lhs.tier != rhs.tier || lhs.add != rhs.add || + lhs.remove != rhs.remove || + lhs.effective_target != rhs.effective_target) { + mismatches.fetch_add(1, std::memory_order_relaxed); + break; + } + } + } + }); + } + for (auto& thread : threads) { + thread.join(); + } + + EXPECT_EQ(mismatches.load(std::memory_order_relaxed), 0); +} + +} // namespace +} // namespace mooncake From 20e5df20fc71f7e241a2a54df7f242f5deacfafd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Tue, 14 Jul 2026 19:34:30 +0800 Subject: [PATCH 2/6] feat(store): add non-actuating replica placement shadow evaluator --- .../include/replica_placement_shadow.h | 145 ++++++++ mooncake-store/src/CMakeLists.txt | 1 + .../src/replica_placement_shadow.cpp | 200 +++++++++++ mooncake-store/tests/CMakeLists.txt | 1 + .../tests/replica_placement_shadow_test.cpp | 312 ++++++++++++++++++ 5 files changed, 659 insertions(+) create mode 100644 mooncake-store/include/replica_placement_shadow.h create mode 100644 mooncake-store/src/replica_placement_shadow.cpp create mode 100644 mooncake-store/tests/replica_placement_shadow_test.cpp diff --git a/mooncake-store/include/replica_placement_shadow.h b/mooncake-store/include/replica_placement_shadow.h new file mode 100644 index 0000000000..28996c311c --- /dev/null +++ b/mooncake-store/include/replica_placement_shadow.h @@ -0,0 +1,145 @@ +// Copyright 2026 Mooncake Authors + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "count_min_sketch.h" +#include "replica.h" +#include "replica_placement_policy.h" + +namespace mooncake { + +class ReplicaPlacementSignalClock { + public: + using TimePoint = std::chrono::steady_clock::time_point; + + virtual ~ReplicaPlacementSignalClock() = default; + [[nodiscard]] virtual TimePoint Now() const noexcept = 0; +}; + +struct ReplicaTierCounts { + uint32_t complete{0}; + uint32_t pending{0}; +}; + +using ReplicaPlacementInventory = + std::array; + +[[nodiscard]] ReplicaPlacementInventory BuildReplicaPlacementInventory( + std::span replicas) noexcept; + +struct ReplicaPlacementTierSignals { + ReplicaTierSignalState allocation_state{ReplicaTierSignalState::UNKNOWN}; + ReplicaTierSignalState health_state{ReplicaTierSignalState::UNKNOWN}; +}; + +struct ReplicaPlacementSignalSnapshot { + uint64_t generation{0}; + bool ready{false}; + ReplicaPlacementSignalClock::TimePoint observed_at{}; + std::array tiers{}; +}; + +enum class ReplicaPlacementSignalPublishStatus : uint8_t { + PUBLISHED = 0, + INVALID_SNAPSHOT, + GENERATION_NOT_INCREASING, +}; + +enum class ReplicaPlacementShadowSignalStatus : uint8_t { + READY = 0, + MISSING_SNAPSHOT, + SOURCE_UNAVAILABLE, + STALE_SNAPSHOT, + COUNT, +}; + +inline constexpr size_t kReplicaPlacementShadowSignalStatusCount = + static_cast(ReplicaPlacementShadowSignalStatus::COUNT); +inline constexpr size_t kReplicaPlacementDegradedReasonCount = + static_cast( + ReplicaPlacementDegradedReason::MIN_COMPLETE_UNSATISFIED) + + 1; + +struct ReplicaPlacementShadowConfig { + ReplicaPlacementPolicyConfig policy; + uint8_t warm_threshold{2}; + uint8_t hot_threshold{8}; + std::chrono::nanoseconds signal_ttl{std::chrono::seconds(30)}; + size_t sketch_width{4096}; + size_t sketch_depth{4}; +}; + +struct ReplicaPlacementShadowResult { + uint8_t estimated_frequency{0}; + ReplicaTemperature temperature{ReplicaTemperature::COLD}; + ReplicaPlacementShadowSignalStatus signal_status{ + ReplicaPlacementShadowSignalStatus::MISSING_SNAPSHOT}; + ReplicaPlacementInventory inventory{}; + ReplicaPlacementPlan plan; +}; + +struct ReplicaPlacementShadowCountersSnapshot { + std::array + observations{}; + std::array + add_intents{}; + std::array + remove_intents{}; + std::array degraded{}; +}; + +// Opt-in SHADOW evaluator. It never submits Copy/Move tasks and never mutates +// replica metadata. Keys are used only by CountMinSketch and never retained or +// exposed as metric labels. +class ReplicaPlacementShadowEvaluator final { + public: + explicit ReplicaPlacementShadowEvaluator( + ReplicaPlacementShadowConfig config, + std::shared_ptr clock = nullptr); + + [[nodiscard]] ReplicaPlacementSignalPublishStatus PublishSignalSnapshot( + ReplicaPlacementSignalSnapshot snapshot); + + [[nodiscard]] uint64_t CurrentGeneration() const noexcept; + + [[nodiscard]] ReplicaPlacementShadowResult Observe( + const std::string& tenant_scoped_key, + std::span replicas); + + [[nodiscard]] ReplicaPlacementShadowCountersSnapshot Counters() + const noexcept; + + private: + ReplicaPlacementShadowConfig config_; + ReplicaPlacementPolicy policy_; + CountMinSketch sketch_; + std::shared_ptr clock_; + std::atomic> + snapshot_; + mutable std::mutex publish_mutex_; + std::array, + kReplicaTemperatureCount * + kReplicaPlacementShadowSignalStatusCount> + observations_{}; + std::array, + kReplicaTemperatureCount * kReplicaPlacementTierCount> + add_intents_{}; + std::array, + kReplicaTemperatureCount * kReplicaPlacementTierCount> + remove_intents_{}; + std::array, kReplicaPlacementDegradedReasonCount> + degraded_{}; +}; + +} // namespace mooncake diff --git a/mooncake-store/src/CMakeLists.txt b/mooncake-store/src/CMakeLists.txt index 5d7f80ec75..fae54c22c1 100644 --- a/mooncake-store/src/CMakeLists.txt +++ b/mooncake-store/src/CMakeLists.txt @@ -74,6 +74,7 @@ set(MOONCAKE_STORE_SOURCES memory_alloc.cpp ssd_register_client.cpp replica_placement_policy.cpp + replica_placement_shadow.cpp engram/engram_store.cpp) set(EXTRA_LIBS "") diff --git a/mooncake-store/src/replica_placement_shadow.cpp b/mooncake-store/src/replica_placement_shadow.cpp new file mode 100644 index 0000000000..319631f91a --- /dev/null +++ b/mooncake-store/src/replica_placement_shadow.cpp @@ -0,0 +1,200 @@ +// Copyright 2026 Mooncake Authors + +#include "replica_placement_shadow.h" + +#include +#include +#include + +namespace mooncake { +namespace { + +class SteadyReplicaPlacementClock final : public ReplicaPlacementSignalClock { + public: + [[nodiscard]] TimePoint Now() const noexcept override { + return std::chrono::steady_clock::now(); + } +}; + +bool IsValidSignalState(ReplicaTierSignalState state) noexcept { + return state == ReplicaTierSignalState::UNKNOWN || + state == ReplicaTierSignalState::AVAILABLE || + state == ReplicaTierSignalState::UNAVAILABLE; +} + +uint32_t SaturatingIncrement(uint32_t value) noexcept { + return value == std::numeric_limits::max() ? value : value + 1; +} + +size_t TierIndex(const Replica::Descriptor& replica) noexcept { + if (replica.is_memory_replica()) { + return static_cast(ReplicaPlacementTier::MEMORY); + } + if (replica.is_local_disk_replica()) { + return static_cast(ReplicaPlacementTier::LOCAL_DISK); + } + if (replica.is_nof_replica()) { + return static_cast(ReplicaPlacementTier::NOF_SSD); + } + return static_cast(ReplicaPlacementTier::REMOTE_STORE); +} + +size_t ObservationIndex(ReplicaTemperature temperature, + ReplicaPlacementShadowSignalStatus status) noexcept { + return static_cast(temperature) * + kReplicaPlacementShadowSignalStatusCount + + static_cast(status); +} + +size_t IntentIndex(ReplicaTemperature temperature, size_t tier_index) noexcept { + return static_cast(temperature) * kReplicaPlacementTierCount + + tier_index; +} + +} // namespace + +ReplicaPlacementInventory BuildReplicaPlacementInventory( + std::span replicas) noexcept { + ReplicaPlacementInventory inventory{}; + for (const auto& replica : replicas) { + const size_t tier = TierIndex(replica); + if (replica.status == ReplicaStatus::COMPLETE) { + inventory[tier].complete = + SaturatingIncrement(inventory[tier].complete); + } else if (replica.status == ReplicaStatus::INITIALIZED || + replica.status == ReplicaStatus::PROCESSING) { + inventory[tier].pending = + SaturatingIncrement(inventory[tier].pending); + } + } + return inventory; +} + +ReplicaPlacementShadowEvaluator::ReplicaPlacementShadowEvaluator( + ReplicaPlacementShadowConfig config, + std::shared_ptr clock) + : config_(std::move(config)), + policy_(config_.policy), + sketch_(config_.sketch_width, config_.sketch_depth), + clock_(clock ? std::move(clock) + : std::make_shared()), + snapshot_(nullptr) { + if (config_.warm_threshold == 0 || + config_.hot_threshold <= config_.warm_threshold || + config_.signal_ttl <= std::chrono::nanoseconds::zero() || + config_.sketch_width == 0 || config_.sketch_depth == 0) { + throw std::invalid_argument("invalid replica placement shadow config"); + } +} + +ReplicaPlacementSignalPublishStatus +ReplicaPlacementShadowEvaluator::PublishSignalSnapshot( + ReplicaPlacementSignalSnapshot snapshot) { + if (snapshot.generation == 0 || snapshot.observed_at > clock_->Now()) { + return ReplicaPlacementSignalPublishStatus::INVALID_SNAPSHOT; + } + for (const auto& tier : snapshot.tiers) { + if (!IsValidSignalState(tier.allocation_state) || + !IsValidSignalState(tier.health_state)) { + return ReplicaPlacementSignalPublishStatus::INVALID_SNAPSHOT; + } + if (!snapshot.ready && + (tier.allocation_state != ReplicaTierSignalState::UNKNOWN || + tier.health_state != ReplicaTierSignalState::UNKNOWN)) { + return ReplicaPlacementSignalPublishStatus::INVALID_SNAPSHOT; + } + } + + std::lock_guard lock(publish_mutex_); + const auto current = snapshot_.load(std::memory_order_acquire); + if (current && snapshot.generation <= current->generation) { + return ReplicaPlacementSignalPublishStatus::GENERATION_NOT_INCREASING; + } + snapshot_.store(std::make_shared( + std::move(snapshot)), + std::memory_order_release); + return ReplicaPlacementSignalPublishStatus::PUBLISHED; +} + +uint64_t ReplicaPlacementShadowEvaluator::CurrentGeneration() const noexcept { + const auto snapshot = snapshot_.load(std::memory_order_acquire); + return snapshot ? snapshot->generation : 0; +} + +ReplicaPlacementShadowResult ReplicaPlacementShadowEvaluator::Observe( + const std::string& tenant_scoped_key, + std::span replicas) { + ReplicaPlacementShadowResult result; + result.estimated_frequency = sketch_.increment(tenant_scoped_key); + result.temperature = result.estimated_frequency >= config_.hot_threshold + ? ReplicaTemperature::HOT + : result.estimated_frequency >= config_.warm_threshold + ? ReplicaTemperature::WARM + : ReplicaTemperature::COLD; + result.inventory = BuildReplicaPlacementInventory(replicas); + + std::array + observations{}; + for (size_t i = 0; i < kReplicaPlacementTierCount; ++i) { + observations[i].complete = result.inventory[i].complete; + observations[i].pending = result.inventory[i].pending; + } + + const auto snapshot = snapshot_.load(std::memory_order_acquire); + if (!snapshot) { + result.signal_status = + ReplicaPlacementShadowSignalStatus::MISSING_SNAPSHOT; + } else if (!snapshot->ready) { + result.signal_status = + ReplicaPlacementShadowSignalStatus::SOURCE_UNAVAILABLE; + } else if (clock_->Now() - snapshot->observed_at > config_.signal_ttl) { + result.signal_status = + ReplicaPlacementShadowSignalStatus::STALE_SNAPSHOT; + } else { + result.signal_status = ReplicaPlacementShadowSignalStatus::READY; + for (size_t i = 0; i < kReplicaPlacementTierCount; ++i) { + observations[i].allocation_state = + snapshot->tiers[i].allocation_state; + observations[i].health_state = snapshot->tiers[i].health_state; + } + } + + result.plan = policy_.Plan(result.temperature, observations); + observations_[ObservationIndex(result.temperature, result.signal_status)] + .fetch_add(1, std::memory_order_relaxed); + for (size_t i = 0; i < kReplicaPlacementTierCount; ++i) { + const size_t index = IntentIndex(result.temperature, i); + add_intents_[index].fetch_add(result.plan.adjustments[i].add, + std::memory_order_relaxed); + remove_intents_[index].fetch_add(result.plan.adjustments[i].remove, + std::memory_order_relaxed); + } + for (size_t i = 0; i < result.plan.degraded_reason_count; ++i) { + const size_t reason = + static_cast(result.plan.degraded_reasons[i]); + if (reason < degraded_.size()) { + degraded_[reason].fetch_add(1, std::memory_order_relaxed); + } + } + return result; +} + +ReplicaPlacementShadowCountersSnapshot +ReplicaPlacementShadowEvaluator::Counters() const noexcept { + ReplicaPlacementShadowCountersSnapshot result; + for (size_t i = 0; i < observations_.size(); ++i) { + result.observations[i] = + observations_[i].load(std::memory_order_relaxed); + } + for (size_t i = 0; i < add_intents_.size(); ++i) { + result.add_intents[i] = add_intents_[i].load(std::memory_order_relaxed); + result.remove_intents[i] = + remove_intents_[i].load(std::memory_order_relaxed); + } + for (size_t i = 0; i < degraded_.size(); ++i) { + result.degraded[i] = degraded_[i].load(std::memory_order_relaxed); + } + return result; +} + +} // namespace mooncake diff --git a/mooncake-store/tests/CMakeLists.txt b/mooncake-store/tests/CMakeLists.txt index ae41a41690..3783200a2e 100644 --- a/mooncake-store/tests/CMakeLists.txt +++ b/mooncake-store/tests/CMakeLists.txt @@ -45,6 +45,7 @@ set_tests_properties( replica_selection_env_opt_in_test PROPERTIES ENVIRONMENT "MC_STORE_REPLICA_SCORING=1") add_store_test(replica_placement_policy_test replica_placement_policy_test.cpp) +add_store_test(replica_placement_shadow_test replica_placement_shadow_test.cpp) add_store_test(eviction_strategy_test eviction_strategy_test.cpp) add_store_test(deadline_scheduler_test deadline_scheduler_test.cpp) add_store_test(kv_event_publisher_test kv_event_publisher_test.cpp) diff --git a/mooncake-store/tests/replica_placement_shadow_test.cpp b/mooncake-store/tests/replica_placement_shadow_test.cpp new file mode 100644 index 0000000000..3931a13ca6 --- /dev/null +++ b/mooncake-store/tests/replica_placement_shadow_test.cpp @@ -0,0 +1,312 @@ +// Copyright 2026 Mooncake Authors + +#include "replica_placement_shadow.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace mooncake { +namespace { + +using namespace std::chrono_literals; + +class FakeClock final : public ReplicaPlacementSignalClock { + public: + [[nodiscard]] TimePoint Now() const noexcept override { + return TimePoint(std::chrono::nanoseconds(now_ns_.load())); + } + + void Set(std::chrono::nanoseconds now) { + now_ns_.store(now.count(), std::memory_order_relaxed); + } + + private: + std::atomic now_ns_{0}; +}; + +size_t Tier(ReplicaPlacementTier tier) { return static_cast(tier); } + +size_t Observation(ReplicaTemperature temperature, + ReplicaPlacementShadowSignalStatus status) { + return static_cast(temperature) * + kReplicaPlacementShadowSignalStatusCount + + static_cast(status); +} + +size_t Intent(ReplicaTemperature temperature, ReplicaPlacementTier tier) { + return static_cast(temperature) * kReplicaPlacementTierCount + + Tier(tier); +} + +ReplicaPlacementPolicyConfig PolicyConfig() { + ReplicaPlacementPolicyConfig config; + config.targets[static_cast(ReplicaTemperature::COLD)] = { + ReplicaTierTarget{0, false}, ReplicaTierTarget{1, true}, + ReplicaTierTarget{0, false}, ReplicaTierTarget{1, true}}; + config.targets[static_cast(ReplicaTemperature::WARM)] = { + ReplicaTierTarget{1, true}, ReplicaTierTarget{1, false}, + ReplicaTierTarget{1, false}, ReplicaTierTarget{1, true}}; + config.targets[static_cast(ReplicaTemperature::HOT)] = { + ReplicaTierTarget{2, true}, ReplicaTierTarget{1, false}, + ReplicaTierTarget{1, false}, ReplicaTierTarget{1, true}}; + config.min_complete_replicas = 1; + config.max_total_replicas = 8; + return config; +} + +ReplicaPlacementShadowConfig ShadowConfig() { + ReplicaPlacementShadowConfig config; + config.policy = PolicyConfig(); + config.warm_threshold = 2; + config.hot_threshold = 3; + config.signal_ttl = 10s; + config.sketch_width = 1024; + config.sketch_depth = 4; + return config; +} + +Replica::Descriptor Memory(ReplicaID id, ReplicaStatus status) { + Replica::Descriptor descriptor; + descriptor.id = id; + descriptor.descriptor_variant = MemoryDescriptor{}; + descriptor.status = status; + return descriptor; +} + +Replica::Descriptor LocalDisk(ReplicaID id, ReplicaStatus status) { + Replica::Descriptor descriptor; + descriptor.id = id; + descriptor.descriptor_variant = LocalDiskDescriptor{}; + descriptor.status = status; + return descriptor; +} + +Replica::Descriptor NoF(ReplicaID id, ReplicaStatus status) { + Replica::Descriptor descriptor; + descriptor.id = id; + descriptor.descriptor_variant = NoFDescriptor{}; + descriptor.status = status; + return descriptor; +} + +Replica::Descriptor RemoteStore(ReplicaID id, ReplicaStatus status) { + Replica::Descriptor descriptor; + descriptor.id = id; + descriptor.descriptor_variant = DiskDescriptor{}; + descriptor.status = status; + return descriptor; +} + +ReplicaPlacementSignalSnapshot AvailableSnapshot( + uint64_t generation, ReplicaPlacementSignalClock::TimePoint observed_at) { + ReplicaPlacementSignalSnapshot snapshot; + snapshot.generation = generation; + snapshot.ready = true; + snapshot.observed_at = observed_at; + for (auto& tier : snapshot.tiers) { + tier.allocation_state = ReplicaTierSignalState::AVAILABLE; + tier.health_state = ReplicaTierSignalState::AVAILABLE; + } + return snapshot; +} + +TEST(ReplicaPlacementShadowTest, + BuildsFourTierInventoryAndIgnoresTerminalNoise) { + const std::vector replicas = { + Memory(1, ReplicaStatus::COMPLETE), + Memory(2, ReplicaStatus::PROCESSING), + LocalDisk(3, ReplicaStatus::COMPLETE), + NoF(4, ReplicaStatus::INITIALIZED), + RemoteStore(5, ReplicaStatus::COMPLETE), + RemoteStore(6, ReplicaStatus::FAILED), + LocalDisk(7, ReplicaStatus::REMOVED), + }; + + const auto inventory = BuildReplicaPlacementInventory(replicas); + + EXPECT_EQ(inventory[Tier(ReplicaPlacementTier::MEMORY)].complete, 1); + EXPECT_EQ(inventory[Tier(ReplicaPlacementTier::MEMORY)].pending, 1); + EXPECT_EQ(inventory[Tier(ReplicaPlacementTier::LOCAL_DISK)].complete, 1); + EXPECT_EQ(inventory[Tier(ReplicaPlacementTier::NOF_SSD)].pending, 1); + EXPECT_EQ(inventory[Tier(ReplicaPlacementTier::REMOTE_STORE)].complete, 1); +} + +TEST(ReplicaPlacementShadowTest, RejectsInvalidShadowConfiguration) { + auto clock = std::make_shared(); + auto config = ShadowConfig(); + config.hot_threshold = config.warm_threshold; + EXPECT_THROW(ReplicaPlacementShadowEvaluator(std::move(config), clock), + std::invalid_argument); +} + +TEST(ReplicaPlacementShadowTest, MissingSnapshotNeverFabricatesSignals) { + auto clock = std::make_shared(); + ReplicaPlacementShadowEvaluator evaluator(ShadowConfig(), clock); + const std::vector replicas = { + LocalDisk(1, ReplicaStatus::COMPLETE), + RemoteStore(2, ReplicaStatus::COMPLETE), + }; + + const auto result = evaluator.Observe("tenant/key", replicas); + + EXPECT_EQ(result.temperature, ReplicaTemperature::COLD); + EXPECT_EQ(result.signal_status, + ReplicaPlacementShadowSignalStatus::MISSING_SNAPSHOT); + EXPECT_TRUE(result.plan.degraded()); + EXPECT_EQ(result.plan.adjustments[Tier(ReplicaPlacementTier::MEMORY)].add, + 0); + const auto counters = evaluator.Counters(); + EXPECT_EQ(counters.observations[Observation( + ReplicaTemperature::COLD, + ReplicaPlacementShadowSignalStatus::MISSING_SNAPSHOT)], + 1); +} + +TEST(ReplicaPlacementShadowTest, ValidatesUnavailableAndGenerationSnapshots) { + auto clock = std::make_shared(); + clock->Set(10s); + ReplicaPlacementShadowEvaluator evaluator(ShadowConfig(), clock); + + ReplicaPlacementSignalSnapshot unavailable; + unavailable.generation = 1; + unavailable.observed_at = clock->Now(); + EXPECT_EQ(evaluator.PublishSignalSnapshot(unavailable), + ReplicaPlacementSignalPublishStatus::PUBLISHED); + EXPECT_EQ(evaluator.CurrentGeneration(), 1); + + auto invalid = unavailable; + invalid.generation = 2; + invalid.tiers[0].health_state = ReplicaTierSignalState::AVAILABLE; + EXPECT_EQ(evaluator.PublishSignalSnapshot(invalid), + ReplicaPlacementSignalPublishStatus::INVALID_SNAPSHOT); + EXPECT_EQ(evaluator.CurrentGeneration(), 1); + + unavailable.generation = 1; + EXPECT_EQ(evaluator.PublishSignalSnapshot(unavailable), + ReplicaPlacementSignalPublishStatus::GENERATION_NOT_INCREASING); +} + +TEST(ReplicaPlacementShadowTest, FreshSignalsDriveColdWarmHotPlansAndCounters) { + auto clock = std::make_shared(); + clock->Set(10s); + ReplicaPlacementShadowEvaluator evaluator(ShadowConfig(), clock); + ASSERT_EQ( + evaluator.PublishSignalSnapshot(AvailableSnapshot(1, clock->Now())), + ReplicaPlacementSignalPublishStatus::PUBLISHED); + const std::vector replicas = { + LocalDisk(1, ReplicaStatus::COMPLETE), + RemoteStore(2, ReplicaStatus::COMPLETE), + }; + + const auto cold = evaluator.Observe("tenant/hot-key", replicas); + const auto warm = evaluator.Observe("tenant/hot-key", replicas); + const auto hot = evaluator.Observe("tenant/hot-key", replicas); + + EXPECT_EQ(cold.temperature, ReplicaTemperature::COLD); + EXPECT_EQ(warm.temperature, ReplicaTemperature::WARM); + EXPECT_EQ(hot.temperature, ReplicaTemperature::HOT); + EXPECT_EQ(warm.plan.adjustments[Tier(ReplicaPlacementTier::MEMORY)].add, 1); + EXPECT_EQ(hot.plan.adjustments[Tier(ReplicaPlacementTier::MEMORY)].add, 2); + EXPECT_EQ(hot.plan.adjustments[Tier(ReplicaPlacementTier::NOF_SSD)].add, 1); + + const auto counters = evaluator.Counters(); + EXPECT_EQ(counters.add_intents[Intent(ReplicaTemperature::WARM, + ReplicaPlacementTier::MEMORY)], + 1); + EXPECT_EQ(counters.add_intents[Intent(ReplicaTemperature::HOT, + ReplicaPlacementTier::MEMORY)], + 2); +} + +TEST(ReplicaPlacementShadowTest, StaleAndUnavailableSnapshotsStayNonActuating) { + auto clock = std::make_shared(); + clock->Set(10s); + ReplicaPlacementShadowEvaluator evaluator(ShadowConfig(), clock); + ASSERT_EQ( + evaluator.PublishSignalSnapshot(AvailableSnapshot(1, clock->Now())), + ReplicaPlacementSignalPublishStatus::PUBLISHED); + clock->Set(21s); + const auto stale = evaluator.Observe( + "tenant/key-a", std::span{}); + EXPECT_EQ(stale.signal_status, + ReplicaPlacementShadowSignalStatus::STALE_SNAPSHOT); + for (const auto& adjustment : stale.plan.adjustments) { + EXPECT_EQ(adjustment.add, 0); + EXPECT_EQ(adjustment.remove, 0); + } + + ReplicaPlacementSignalSnapshot unavailable; + unavailable.generation = 2; + unavailable.observed_at = clock->Now(); + ASSERT_EQ(evaluator.PublishSignalSnapshot(unavailable), + ReplicaPlacementSignalPublishStatus::PUBLISHED); + const auto down = evaluator.Observe("tenant/key-b", + std::span{}); + EXPECT_EQ(down.signal_status, + ReplicaPlacementShadowSignalStatus::SOURCE_UNAVAILABLE); + for (const auto& adjustment : down.plan.adjustments) { + EXPECT_EQ(adjustment.add, 0); + EXPECT_EQ(adjustment.remove, 0); + } +} + +TEST(ReplicaPlacementShadowTest, ConcurrentObserveAndPublishKeepsExactCount) { + auto clock = std::make_shared(); + clock->Set(10s); + ReplicaPlacementShadowEvaluator evaluator(ShadowConfig(), clock); + ASSERT_EQ( + evaluator.PublishSignalSnapshot(AvailableSnapshot(1, clock->Now())), + ReplicaPlacementSignalPublishStatus::PUBLISHED); + const std::vector replicas = { + Memory(1, ReplicaStatus::COMPLETE), + LocalDisk(2, ReplicaStatus::COMPLETE), + RemoteStore(3, ReplicaStatus::COMPLETE), + }; + constexpr size_t kThreadCount = 8; + constexpr size_t kIterations = 2'000; + std::atomic publish_failures{0}; + std::vector threads; + threads.reserve(kThreadCount + 1); + + for (size_t thread = 0; thread < kThreadCount; ++thread) { + threads.emplace_back([&, thread]() { + const std::string key = "tenant/key-" + std::to_string(thread); + for (size_t i = 0; i < kIterations; ++i) { + (void)evaluator.Observe(key, replicas); + } + }); + } + threads.emplace_back([&]() { + for (uint64_t generation = 2; generation <= 101; ++generation) { + if (evaluator.PublishSignalSnapshot( + AvailableSnapshot(generation, clock->Now())) != + ReplicaPlacementSignalPublishStatus::PUBLISHED) { + publish_failures.fetch_add(1, std::memory_order_relaxed); + } + } + }); + for (auto& thread : threads) { + thread.join(); + } + + const auto counters = evaluator.Counters(); + uint64_t total = 0; + for (uint64_t count : counters.observations) { + total += count; + } + EXPECT_EQ(total, kThreadCount * kIterations); + EXPECT_EQ(publish_failures.load(std::memory_order_relaxed), 0); + EXPECT_EQ(evaluator.CurrentGeneration(), 101); +} + +} // namespace +} // namespace mooncake From 62b4dcc7f07eb76f90e4f94726d32dca9d2d3175 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Wed, 15 Jul 2026 10:25:03 +0800 Subject: [PATCH 3/6] feat(store): wire opt-in replica placement shadow --- mooncake-store/include/master_config.h | 17 ++ mooncake-store/include/master_service.h | 18 ++ .../include/replica_placement_shadow.h | 1 + mooncake-store/src/master.cpp | 112 ++++++++- mooncake-store/src/master_service.cpp | 69 ++++++ mooncake-store/tests/CMakeLists.txt | 2 + .../replica_placement_master_shadow_test.cpp | 218 ++++++++++++++++++ 7 files changed, 428 insertions(+), 9 deletions(-) create mode 100644 mooncake-store/tests/replica_placement_master_shadow_test.cpp diff --git a/mooncake-store/include/master_config.h b/mooncake-store/include/master_config.h index fd893953f8..0cfd8a63be 100644 --- a/mooncake-store/include/master_config.h +++ b/mooncake-store/include/master_config.h @@ -7,6 +7,7 @@ #include #include "config_helper.h" +#include "replica_placement_shadow.h" #include "types.h" namespace mooncake { @@ -133,6 +134,11 @@ struct MasterConfig { // rich clusters may safely raise it. uint32_t promotion_max_per_heartbeat = 1; + // Dynamic replica placement observation. Empty keeps the feature fully + // disabled; targets have no built-in production defaults and must be + // supplied by the operator before the evaluator can be constructed. + std::optional replica_placement_shadow_config; + // KV Events publisher (RFC #1527) for cache-aware indexers. bool enable_kv_events = false; std::string kv_events_bind_endpoint; @@ -225,6 +231,7 @@ class MasterServiceSupervisorConfig { uint32_t promotion_admission_threshold = 2; uint32_t promotion_queue_limit = 50000; uint32_t promotion_max_per_heartbeat = 1; + std::optional replica_placement_shadow_config; bool enable_kv_events = false; std::string kv_events_bind_endpoint; std::string kv_events_model_name; @@ -279,6 +286,8 @@ class MasterServiceSupervisorConfig { promotion_admission_threshold = config.promotion_admission_threshold; promotion_queue_limit = config.promotion_queue_limit; promotion_max_per_heartbeat = config.promotion_max_per_heartbeat; + replica_placement_shadow_config = + config.replica_placement_shadow_config; enable_kv_events = config.enable_kv_events; kv_events_bind_endpoint = config.kv_events_bind_endpoint; kv_events_model_name = config.kv_events_model_name; @@ -441,6 +450,7 @@ class WrappedMasterServiceConfig { uint32_t promotion_admission_threshold = 2; uint32_t promotion_queue_limit = 50000; uint32_t promotion_max_per_heartbeat = 1; + std::optional replica_placement_shadow_config; bool enable_kv_events = false; std::string kv_events_bind_endpoint; std::string kv_events_model_name; @@ -526,6 +536,8 @@ class WrappedMasterServiceConfig { promotion_admission_threshold = config.promotion_admission_threshold; promotion_queue_limit = config.promotion_queue_limit; promotion_max_per_heartbeat = config.promotion_max_per_heartbeat; + replica_placement_shadow_config = + config.replica_placement_shadow_config; enable_kv_events = config.enable_kv_events; kv_events_bind_endpoint = config.kv_events_bind_endpoint; kv_events_model_name = config.kv_events_model_name; @@ -635,6 +647,8 @@ class WrappedMasterServiceConfig { promotion_admission_threshold = config.promotion_admission_threshold; promotion_queue_limit = config.promotion_queue_limit; promotion_max_per_heartbeat = config.promotion_max_per_heartbeat; + replica_placement_shadow_config = + config.replica_placement_shadow_config; enable_kv_events = config.enable_kv_events; kv_events_bind_endpoint = config.kv_events_bind_endpoint; kv_events_model_name = config.kv_events_model_name; @@ -1057,6 +1071,7 @@ class MasterServiceConfig { uint32_t promotion_admission_threshold = 2; uint32_t promotion_queue_limit = 50000; uint32_t promotion_max_per_heartbeat = 1; + std::optional replica_placement_shadow_config; bool enable_kv_events = false; std::string kv_events_bind_endpoint; std::string kv_events_model_name; @@ -1138,6 +1153,8 @@ class MasterServiceConfig { promotion_admission_threshold = config.promotion_admission_threshold; promotion_queue_limit = config.promotion_queue_limit; promotion_max_per_heartbeat = config.promotion_max_per_heartbeat; + replica_placement_shadow_config = + config.replica_placement_shadow_config; enable_kv_events = config.enable_kv_events; kv_events_bind_endpoint = config.kv_events_bind_endpoint; kv_events_model_name = config.kv_events_model_name; diff --git a/mooncake-store/include/master_service.h b/mooncake-store/include/master_service.h index e8f16c6e9b..eb2d880cc8 100644 --- a/mooncake-store/include/master_service.h +++ b/mooncake-store/include/master_service.h @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -115,6 +116,14 @@ class MasterService { const UUID& segment_id); std::optional GetTenantQuotaSnapshotForTesting( const std::string& tenant_id) const; + [[nodiscard]] bool ReplicaPlacementShadowEnabled() const noexcept; + [[nodiscard]] ReplicaPlacementSignalPublishStatus + PublishReplicaPlacementSignalSnapshot( + ReplicaPlacementSignalSnapshot snapshot); + [[nodiscard]] uint64_t ReplicaPlacementShadowSignalGeneration() + const noexcept; + [[nodiscard]] std::optional + GetReplicaPlacementShadowCounters() const noexcept; bool IsTenantQuotaEnabled() const; std::vector ListTenantQuotaSnapshots() const; std::optional GetTenantQuotaSnapshot( @@ -1938,6 +1947,15 @@ class MasterService { // from any GetReplicaList caller without additional locking. std::unique_ptr promotion_sketch_; + // Optional read-path observer. It owns no task queue and cannot mutate + // metadata; only explicit configuration constructs it. + std::unique_ptr + replica_placement_shadow_evaluator_; + + void ObserveReplicaPlacementShadow( + const ObjectIdentity& object_id, + std::span replicas); + const std::string ha_backend_type_; const std::string ha_backend_connstring_; diff --git a/mooncake-store/include/replica_placement_shadow.h b/mooncake-store/include/replica_placement_shadow.h index 28996c311c..73484c88ca 100644 --- a/mooncake-store/include/replica_placement_shadow.h +++ b/mooncake-store/include/replica_placement_shadow.h @@ -53,6 +53,7 @@ enum class ReplicaPlacementSignalPublishStatus : uint8_t { PUBLISHED = 0, INVALID_SNAPSHOT, GENERATION_NOT_INCREASING, + NOT_ENABLED, }; enum class ReplicaPlacementShadowSignalStatus : uint8_t { diff --git a/mooncake-store/src/master.cpp b/mooncake-store/src/master.cpp index 66daea1bbf..bb4a4ac8a4 100644 --- a/mooncake-store/src/master.cpp +++ b/mooncake-store/src/master.cpp @@ -2,12 +2,15 @@ #include #include // For std::atomic +#include #include // For std::chrono #include #include // For std::getenv #include // For std::ifstream #include // For std::unique_ptr +#include #include +#include #include // For std::thread #include #include @@ -225,6 +228,11 @@ DEFINE_uint32(promotion_max_per_heartbeat, 1, "SSD-read + RDMA-write on the client; serializing them avoids " "blocking past the client-liveness window. Default 1 is " "conservative."); +DEFINE_string( + replica_placement_shadow_config, "", + "Path to a YAML/JSON replica placement SHADOW profile. Empty disables " + "the observer. The profile must explicitly provide all 12 tier targets; " + "SHADOW never submits Copy/Move tasks."); DEFINE_bool(enable_kv_events, false, "Enable RFC #1527 KV cache event publisher over ZMQ"); DEFINE_string(kv_events_bind_endpoint, "", @@ -380,6 +388,77 @@ DEFINE_bool(enable_cxl, false, "Whether to enable CXL memory support"); namespace { +mooncake::ReplicaPlacementShadowConfig LoadReplicaPlacementShadowProfile( + const std::string& path) { + mooncake::DefaultConfig profile; + profile.SetPath(path); + profile.Load(); + + mooncake::ReplicaPlacementShadowConfig config; + uint32_t warm_threshold = config.warm_threshold; + uint32_t hot_threshold = config.hot_threshold; + uint64_t signal_ttl_ms = static_cast( + std::chrono::duration_cast(config.signal_ttl) + .count()); + uint64_t sketch_width = config.sketch_width; + uint64_t sketch_depth = config.sketch_depth; + profile.GetUInt32("warm_threshold", &warm_threshold, warm_threshold); + profile.GetUInt32("hot_threshold", &hot_threshold, hot_threshold); + profile.GetDurationMs("signal_ttl", &signal_ttl_ms, signal_ttl_ms); + profile.GetUInt64("sketch_width", &sketch_width, sketch_width); + profile.GetUInt64("sketch_depth", &sketch_depth, sketch_depth); + profile.GetUInt32("min_complete_replicas", + &config.policy.min_complete_replicas, + config.policy.min_complete_replicas); + profile.GetUInt32("max_total_replicas", &config.policy.max_total_replicas, + config.policy.max_total_replicas); + + if (warm_threshold > std::numeric_limits::max() || + hot_threshold > std::numeric_limits::max() || + sketch_width > std::numeric_limits::max() || + sketch_depth > std::numeric_limits::max()) { + throw std::runtime_error( + "replica placement SHADOW profile value exceeds type limit"); + } + config.warm_threshold = static_cast(warm_threshold); + config.hot_threshold = static_cast(hot_threshold); + config.signal_ttl = std::chrono::milliseconds(signal_ttl_ms); + config.sketch_width = static_cast(sketch_width); + config.sketch_depth = static_cast(sketch_depth); + + constexpr std::array kTemperatureNames = { + "cold", "warm", "hot"}; + constexpr std::array kTierNames = { + "memory", "local_disk", "nof_ssd", "remote_store"}; + constexpr uint32_t kMissingTarget = std::numeric_limits::max(); + for (size_t temperature = 0; temperature < kTemperatureNames.size(); + ++temperature) { + for (size_t tier = 0; tier < kTierNames.size(); ++tier) { + const std::string prefix = + "targets." + std::string(kTemperatureNames[temperature]) + "." + + std::string(kTierNames[tier]); + uint32_t desired = kMissingTarget; + bool required = false; + profile.GetUInt32(prefix + ".desired", &desired, kMissingTarget); + profile.GetBool(prefix + ".required", &required, false); + if (desired == kMissingTarget) { + throw std::runtime_error( + "replica placement SHADOW profile is missing " + prefix + + ".desired"); + } + config.policy.targets[temperature][tier] = + mooncake::ReplicaTierTarget{desired, required}; + } + } + return config; +} + +std::optional +LoadOptionalReplicaPlacementShadowProfile(const std::string& path) { + if (path.empty()) return std::nullopt; + return LoadReplicaPlacementShadowProfile(path); +} + std::string ResolveHABackendConnstring( const mooncake::MasterConfig& master_config) { return mooncake::ResolveConfiguredHABackendConnstring( @@ -506,6 +585,13 @@ void InitMasterConf(const mooncake::DefaultConfig& default_config, default_config.GetUInt32("promotion_max_per_heartbeat", &master_config.promotion_max_per_heartbeat, FLAGS_promotion_max_per_heartbeat); + std::string replica_placement_shadow_profile; + default_config.GetString("replica_placement_shadow_config", + &replica_placement_shadow_profile, + FLAGS_replica_placement_shadow_config); + master_config.replica_placement_shadow_config = + LoadOptionalReplicaPlacementShadowProfile( + replica_placement_shadow_profile); default_config.GetBool("enable_kv_events", &master_config.enable_kv_events, FLAGS_enable_kv_events); default_config.GetString("kv_events_bind_endpoint", @@ -854,6 +940,14 @@ void LoadConfigFromCmdline(mooncake::MasterConfig& master_config, master_config.promotion_max_per_heartbeat = FLAGS_promotion_max_per_heartbeat; } + if ((google::GetCommandLineFlagInfo("replica_placement_shadow_config", + &info) && + !info.is_default) || + !conf_set) { + master_config.replica_placement_shadow_config = + LoadOptionalReplicaPlacementShadowProfile( + FLAGS_replica_placement_shadow_config); + } if ((google::GetCommandLineFlagInfo("enable_kv_events", &info) && !info.is_default) || !conf_set) { @@ -1267,18 +1361,18 @@ int main(int argc, char* argv[]) { // Initialize the master configuration mooncake::MasterConfig master_config; std::string conf_path = FLAGS_config_path; - if (!conf_path.empty()) { - mooncake::DefaultConfig default_config; - default_config.SetPath(conf_path); - try { + try { + if (!conf_path.empty()) { + mooncake::DefaultConfig default_config; + default_config.SetPath(conf_path); default_config.Load(); - } catch (const std::exception& e) { - LOG(FATAL) << "Failed to initialize default config: " << e.what(); - return 1; + InitMasterConf(default_config, master_config); } - InitMasterConf(default_config, master_config); + LoadConfigFromCmdline(master_config, !conf_path.empty()); + } catch (const std::exception& e) { + LOG(ERROR) << "Failed to initialize master configuration: " << e.what(); + return 1; } - LoadConfigFromCmdline(master_config, !conf_path.empty()); ResolveRpcAddressFromInterfaceOrDie(master_config); // Fall back to environment variables for pod identity (K8s Downward API) diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index bf55656980..af6bc88133 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -411,6 +411,15 @@ MasterService::MasterService(const MasterServiceConfig& config) << ")"; } + if (config.replica_placement_shadow_config.has_value()) { + replica_placement_shadow_evaluator_ = + std::make_unique( + *config.replica_placement_shadow_config); + LOG(INFO) << "Replica placement SHADOW enabled: observations produce " + "fixed-cardinality intent counters only; Copy/Move " + "submission remains disabled"; + } + kv_event_publisher_ = std::make_unique(BuildKvEventConfig(config)); @@ -579,6 +588,41 @@ MasterService::~MasterService() { } } +bool MasterService::ReplicaPlacementShadowEnabled() const noexcept { + return replica_placement_shadow_evaluator_ != nullptr; +} + +ReplicaPlacementSignalPublishStatus +MasterService::PublishReplicaPlacementSignalSnapshot( + ReplicaPlacementSignalSnapshot snapshot) { + if (!replica_placement_shadow_evaluator_) { + return ReplicaPlacementSignalPublishStatus::NOT_ENABLED; + } + return replica_placement_shadow_evaluator_->PublishSignalSnapshot( + std::move(snapshot)); +} + +uint64_t MasterService::ReplicaPlacementShadowSignalGeneration() + const noexcept { + return replica_placement_shadow_evaluator_ + ? replica_placement_shadow_evaluator_->CurrentGeneration() + : 0; +} + +std::optional +MasterService::GetReplicaPlacementShadowCounters() const noexcept { + if (!replica_placement_shadow_evaluator_) return std::nullopt; + return replica_placement_shadow_evaluator_->Counters(); +} + +void MasterService::ObserveReplicaPlacementShadow( + const ObjectIdentity& object_id, + std::span replicas) { + if (!replica_placement_shadow_evaluator_) return; + (void)replica_placement_shadow_evaluator_->Observe( + MakeTenantScopedKey(object_id.tenant_id, object_id.user_key), replicas); +} + void MasterService::SetNoFProbeFnForTesting(NoFProbeFn fn) { #ifdef USE_NOF std::lock_guard lock(nof_probe_fn_mutex_); @@ -2532,6 +2576,7 @@ auto MasterService::GetReplicaList(const std::string& key, GetReplicaListResponse resp({}, default_kv_lease_ttl_); bool promotion_eligible = false; + std::vector shadow_replica_list; { MetadataAccessorRO accessor(this, object_id); @@ -2554,6 +2599,13 @@ auto MasterService::GetReplicaList(const std::string& key, return tl::make_unexpected(ErrorCode::REPLICA_IS_NOT_READY); } + if (replica_placement_shadow_evaluator_) { + shadow_replica_list.reserve(metadata.CountReplicas()); + for (const auto& replica : metadata.GetAllReplicas()) { + shadow_replica_list.emplace_back(replica.get_descriptor()); + } + } + // TODO: NoF SSD support (ranhaojia) if (replica_list[0].is_memory_replica()) { MasterMetricManager::instance().inc_mem_cache_hit_nums(); @@ -2595,6 +2647,7 @@ auto MasterService::GetReplicaList(const std::string& key, if (promotion_eligible) { TryPushPromotionQueue(object_id); } + ObserveReplicaPlacementShadow(object_id, shadow_replica_list); return resp; } @@ -2668,6 +2721,8 @@ MasterService::BatchGetReplicaList(const std::vector& keys, } std::vector promotion_candidates; + std::vector>> + shadow_candidates; std::shared_lock shared_lock(snapshot_mutex_); { MetadataShardAccessorRO shard(this, shard_idx); @@ -2724,6 +2779,17 @@ MasterService::BatchGetReplicaList(const std::vector& keys, MasterMetricManager::instance().inc_valid_get_nums(); GrantLeaseForGroup(tenant_state, key, metadata); + if (replica_placement_shadow_evaluator_) { + std::vector shadow_replicas; + shadow_replicas.reserve(metadata.CountReplicas()); + for (const auto& replica : metadata.GetAllReplicas()) { + shadow_replicas.emplace_back(replica.get_descriptor()); + } + shadow_candidates.emplace_back( + MakeObjectIdentity(key, normalized_tenant), + std::move(shadow_replicas)); + } + if (promotion_on_hit_) { const bool any_memory = metadata.HasReplica(&Replica::fn_is_memory_replica); @@ -2743,6 +2809,9 @@ MasterService::BatchGetReplicaList(const std::vector& keys, for (const auto& object_id : promotion_candidates) { TryPushPromotionQueue(object_id); } + for (const auto& [object_id, replicas] : shadow_candidates) { + ObserveReplicaPlacementShadow(object_id, replicas); + } } return results; diff --git a/mooncake-store/tests/CMakeLists.txt b/mooncake-store/tests/CMakeLists.txt index 3783200a2e..9196bf5b5f 100644 --- a/mooncake-store/tests/CMakeLists.txt +++ b/mooncake-store/tests/CMakeLists.txt @@ -46,6 +46,8 @@ set_tests_properties( PROPERTIES ENVIRONMENT "MC_STORE_REPLICA_SCORING=1") add_store_test(replica_placement_policy_test replica_placement_policy_test.cpp) add_store_test(replica_placement_shadow_test replica_placement_shadow_test.cpp) +add_store_test(replica_placement_master_shadow_test + replica_placement_master_shadow_test.cpp) add_store_test(eviction_strategy_test eviction_strategy_test.cpp) add_store_test(deadline_scheduler_test deadline_scheduler_test.cpp) add_store_test(kv_event_publisher_test kv_event_publisher_test.cpp) diff --git a/mooncake-store/tests/replica_placement_master_shadow_test.cpp b/mooncake-store/tests/replica_placement_master_shadow_test.cpp new file mode 100644 index 0000000000..7258d662a4 --- /dev/null +++ b/mooncake-store/tests/replica_placement_master_shadow_test.cpp @@ -0,0 +1,218 @@ +// Copyright 2026 Mooncake Authors + +#include "master_service.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace mooncake::test { +namespace { + +size_t Tier(ReplicaPlacementTier tier) { return static_cast(tier); } + +size_t Observation(ReplicaTemperature temperature, + ReplicaPlacementShadowSignalStatus status) { + return static_cast(temperature) * + kReplicaPlacementShadowSignalStatusCount + + static_cast(status); +} + +size_t Intent(ReplicaTemperature temperature, ReplicaPlacementTier tier) { + return static_cast(temperature) * kReplicaPlacementTierCount + + Tier(tier); +} + +uint64_t TotalObservations( + const ReplicaPlacementShadowCountersSnapshot& counters) { + uint64_t total = 0; + for (uint64_t count : counters.observations) total += count; + return total; +} + +ReplicaPlacementShadowConfig ShadowConfig() { + ReplicaPlacementShadowConfig config; + config.policy.targets[static_cast(ReplicaTemperature::COLD)] = { + ReplicaTierTarget{1, true}, ReplicaTierTarget{0, false}, + ReplicaTierTarget{0, false}, ReplicaTierTarget{0, false}}; + config.policy.targets[static_cast(ReplicaTemperature::WARM)] = { + ReplicaTierTarget{1, true}, ReplicaTierTarget{1, false}, + ReplicaTierTarget{0, false}, ReplicaTierTarget{0, false}}; + config.policy.targets[static_cast(ReplicaTemperature::HOT)] = { + ReplicaTierTarget{2, true}, ReplicaTierTarget{1, false}, + ReplicaTierTarget{0, false}, ReplicaTierTarget{0, false}}; + config.policy.min_complete_replicas = 1; + config.policy.max_total_replicas = 4; + config.warm_threshold = 2; + config.hot_threshold = 3; + config.signal_ttl = std::chrono::seconds(30); + config.sketch_width = 4096; + config.sketch_depth = 4; + return config; +} + +ReplicaPlacementSignalSnapshot AvailableSnapshot(uint64_t generation) { + ReplicaPlacementSignalSnapshot snapshot; + snapshot.generation = generation; + snapshot.ready = true; + snapshot.observed_at = std::chrono::steady_clock::now(); + for (auto& tier : snapshot.tiers) { + tier.allocation_state = ReplicaTierSignalState::AVAILABLE; + tier.health_state = ReplicaTierSignalState::AVAILABLE; + } + return snapshot; +} + +class ReplicaPlacementMasterShadowTest : public ::testing::Test { + protected: + static constexpr size_t kSegmentBase = 0x530000000; + static constexpr size_t kSegmentSize = 16 * 1024 * 1024; + + UUID PrepareMemorySegment(MasterService& service, const std::string& name) { + Segment segment; + segment.id = generate_uuid(); + segment.name = name; + segment.base = kSegmentBase; + segment.size = kSegmentSize; + segment.te_endpoint = name; + UUID client_id = generate_uuid(); + EXPECT_TRUE(service.MountSegment(segment, client_id).has_value()); + return client_id; + } + + void PutObject(MasterService& service, const UUID& client_id, + const std::string& key) { + ReplicateConfig config; + config.replica_num = 1; + ASSERT_TRUE(service.PutStart(client_id, key, "default", 1024, config) + .has_value()); + ASSERT_TRUE( + service.PutEnd(client_id, key, "default", ReplicaType::MEMORY) + .has_value()); + } +}; + +TEST_F(ReplicaPlacementMasterShadowTest, DefaultOffHasNoObserverState) { + MasterService service; + + EXPECT_FALSE(service.ReplicaPlacementShadowEnabled()); + EXPECT_EQ( + service.PublishReplicaPlacementSignalSnapshot(AvailableSnapshot(1)), + ReplicaPlacementSignalPublishStatus::NOT_ENABLED); + EXPECT_EQ(service.ReplicaPlacementShadowSignalGeneration(), 0); + EXPECT_FALSE(service.GetReplicaPlacementShadowCounters().has_value()); +} + +TEST_F(ReplicaPlacementMasterShadowTest, + MissingSnapshotRecordsDegradedObservationWithoutActuation) { + MasterServiceConfig config; + config.replica_placement_shadow_config = ShadowConfig(); + MasterService service(config); + const UUID client_id = PrepareMemorySegment(service, "shadow_missing"); + PutObject(service, client_id, "key"); + + ASSERT_TRUE(service.GetReplicaList("key", "default").has_value()); + + const auto counters = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(counters.has_value()); + EXPECT_EQ(TotalObservations(*counters), 1); + EXPECT_EQ(counters->observations[Observation( + ReplicaTemperature::COLD, + ReplicaPlacementShadowSignalStatus::MISSING_SNAPSHOT)], + 1); + service.RemoveAll(); +} + +TEST_F(ReplicaPlacementMasterShadowTest, + SingleAndBatchGetDriveShadowButAdminGetDoesNot) { + MasterServiceConfig config; + config.replica_placement_shadow_config = ShadowConfig(); + MasterService service(config); + ASSERT_EQ( + service.PublishReplicaPlacementSignalSnapshot(AvailableSnapshot(1)), + ReplicaPlacementSignalPublishStatus::PUBLISHED); + const UUID client_id = PrepareMemorySegment(service, "shadow_reads"); + PutObject(service, client_id, "hot"); + PutObject(service, client_id, "batch"); + + ASSERT_TRUE(service.GetReplicaList("hot", "default").has_value()); + ASSERT_TRUE(service.GetReplicaList("hot", "default").has_value()); + ASSERT_TRUE(service.GetReplicaList("hot", "default").has_value()); + const auto batch = service.BatchGetReplicaList({"batch"}, "default"); + ASSERT_EQ(batch.size(), 1); + ASSERT_TRUE(batch[0].has_value()); + + auto counters = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(counters.has_value()); + EXPECT_EQ(TotalObservations(*counters), 4); + EXPECT_EQ(counters->add_intents[Intent(ReplicaTemperature::WARM, + ReplicaPlacementTier::LOCAL_DISK)], + 1); + EXPECT_EQ(counters->add_intents[Intent(ReplicaTemperature::HOT, + ReplicaPlacementTier::MEMORY)], + 1); + + ASSERT_TRUE(service.GetReplicaListForAdmin("hot", "default").has_value()); + const auto after_admin = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(after_admin.has_value()); + EXPECT_EQ(TotalObservations(*after_admin), 4); + service.RemoveAll(); +} + +TEST_F(ReplicaPlacementMasterShadowTest, + ConcurrentGetsPreserveExactObservationCount) { + MasterServiceConfig config; + config.replica_placement_shadow_config = ShadowConfig(); + MasterService service(config); + ASSERT_EQ( + service.PublishReplicaPlacementSignalSnapshot(AvailableSnapshot(1)), + ReplicaPlacementSignalPublishStatus::PUBLISHED); + const UUID client_id = PrepareMemorySegment(service, "shadow_concurrent"); + PutObject(service, client_id, "key"); + + constexpr size_t kThreadCount = 8; + constexpr size_t kIterations = 500; + std::atomic failures{0}; + std::vector threads; + threads.reserve(kThreadCount); + for (size_t thread = 0; thread < kThreadCount; ++thread) { + threads.emplace_back([&]() { + for (size_t i = 0; i < kIterations; ++i) { + if (!service.GetReplicaList("key", "default").has_value()) { + failures.fetch_add(1, std::memory_order_relaxed); + } + } + }); + } + for (auto& thread : threads) thread.join(); + + const auto counters = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(counters.has_value()); + EXPECT_EQ(failures.load(std::memory_order_relaxed), 0); + EXPECT_EQ(TotalObservations(*counters), kThreadCount * kIterations); + service.RemoveAll(); +} + +TEST(ReplicaPlacementMasterShadowConfigTest, + WrappedConfigPropagatesExplicitOptInOnly) { + WrappedMasterServiceConfig wrapped; + wrapped.default_kv_lease_ttl = DEFAULT_DEFAULT_KV_LEASE_TTL; + MasterServiceConfig disabled(wrapped); + EXPECT_FALSE(disabled.replica_placement_shadow_config.has_value()); + + wrapped.replica_placement_shadow_config = ShadowConfig(); + MasterServiceConfig enabled(wrapped); + ASSERT_TRUE(enabled.replica_placement_shadow_config.has_value()); + EXPECT_EQ(enabled.replica_placement_shadow_config->warm_threshold, 2); + EXPECT_EQ(enabled.replica_placement_shadow_config->hot_threshold, 3); +} + +} // namespace +} // namespace mooncake::test From e29271f546c7c0ff14263f0de1d099fc64f7392a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Wed, 15 Jul 2026 10:39:30 +0800 Subject: [PATCH 4/6] feat(store): collect master replica tier signals --- mooncake-store/include/master_config.h | 19 +- mooncake-store/include/master_service.h | 11 ++ mooncake-store/src/master.cpp | 21 ++- mooncake-store/src/master_service.cpp | 176 +++++++++++++++++- .../replica_placement_master_shadow_test.cpp | 165 +++++++++++++++- 5 files changed, 374 insertions(+), 18 deletions(-) diff --git a/mooncake-store/include/master_config.h b/mooncake-store/include/master_config.h index 0cfd8a63be..18f832f2dc 100644 --- a/mooncake-store/include/master_config.h +++ b/mooncake-store/include/master_config.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include #include @@ -12,6 +13,12 @@ namespace mooncake { +struct MasterReplicaPlacementShadowConfig { + ReplicaPlacementShadowConfig evaluator; + bool auto_collect_master_signals{true}; + std::chrono::nanoseconds signal_refresh_interval{std::chrono::seconds(1)}; +}; + // Forwarded to the HA serve phase via MasterServiceSupervisorConfig. class HttpMetadataServer; @@ -137,7 +144,8 @@ struct MasterConfig { // Dynamic replica placement observation. Empty keeps the feature fully // disabled; targets have no built-in production defaults and must be // supplied by the operator before the evaluator can be constructed. - std::optional replica_placement_shadow_config; + std::optional + replica_placement_shadow_config; // KV Events publisher (RFC #1527) for cache-aware indexers. bool enable_kv_events = false; @@ -231,7 +239,8 @@ class MasterServiceSupervisorConfig { uint32_t promotion_admission_threshold = 2; uint32_t promotion_queue_limit = 50000; uint32_t promotion_max_per_heartbeat = 1; - std::optional replica_placement_shadow_config; + std::optional + replica_placement_shadow_config; bool enable_kv_events = false; std::string kv_events_bind_endpoint; std::string kv_events_model_name; @@ -450,7 +459,8 @@ class WrappedMasterServiceConfig { uint32_t promotion_admission_threshold = 2; uint32_t promotion_queue_limit = 50000; uint32_t promotion_max_per_heartbeat = 1; - std::optional replica_placement_shadow_config; + std::optional + replica_placement_shadow_config; bool enable_kv_events = false; std::string kv_events_bind_endpoint; std::string kv_events_model_name; @@ -1071,7 +1081,8 @@ class MasterServiceConfig { uint32_t promotion_admission_threshold = 2; uint32_t promotion_queue_limit = 50000; uint32_t promotion_max_per_heartbeat = 1; - std::optional replica_placement_shadow_config; + std::optional + replica_placement_shadow_config; bool enable_kv_events = false; std::string kv_events_bind_endpoint; std::string kv_events_model_name; diff --git a/mooncake-store/include/master_service.h b/mooncake-store/include/master_service.h index eb2d880cc8..219008c289 100644 --- a/mooncake-store/include/master_service.h +++ b/mooncake-store/include/master_service.h @@ -1951,10 +1951,21 @@ class MasterService { // metadata; only explicit configuration constructs it. std::unique_ptr replica_placement_shadow_evaluator_; + bool replica_placement_shadow_auto_collect_signals_{false}; + std::chrono::nanoseconds replica_placement_shadow_refresh_interval_{ + std::chrono::seconds(1)}; + std::atomic replica_placement_shadow_external_signal_source_{false}; + std::mutex replica_placement_shadow_refresh_mutex_; + std::optional + replica_placement_shadow_last_refresh_; void ObserveReplicaPlacementShadow( const ObjectIdentity& object_id, std::span replicas); + void RefreshReplicaPlacementShadowSignals(); + [[nodiscard]] ReplicaPlacementSignalSnapshot + CollectReplicaPlacementShadowSignals( + ReplicaPlacementSignalClock::TimePoint observed_at); const std::string ha_backend_type_; diff --git a/mooncake-store/src/master.cpp b/mooncake-store/src/master.cpp index bb4a4ac8a4..d3b9888da4 100644 --- a/mooncake-store/src/master.cpp +++ b/mooncake-store/src/master.cpp @@ -388,13 +388,14 @@ DEFINE_bool(enable_cxl, false, "Whether to enable CXL memory support"); namespace { -mooncake::ReplicaPlacementShadowConfig LoadReplicaPlacementShadowProfile( +mooncake::MasterReplicaPlacementShadowConfig LoadReplicaPlacementShadowProfile( const std::string& path) { mooncake::DefaultConfig profile; profile.SetPath(path); profile.Load(); - mooncake::ReplicaPlacementShadowConfig config; + mooncake::MasterReplicaPlacementShadowConfig master_config; + auto& config = master_config.evaluator; uint32_t warm_threshold = config.warm_threshold; uint32_t hot_threshold = config.hot_threshold; uint64_t signal_ttl_ms = static_cast( @@ -402,11 +403,21 @@ mooncake::ReplicaPlacementShadowConfig LoadReplicaPlacementShadowProfile( .count()); uint64_t sketch_width = config.sketch_width; uint64_t sketch_depth = config.sketch_depth; + uint64_t signal_refresh_interval_ms = static_cast( + std::chrono::duration_cast( + master_config.signal_refresh_interval) + .count()); profile.GetUInt32("warm_threshold", &warm_threshold, warm_threshold); profile.GetUInt32("hot_threshold", &hot_threshold, hot_threshold); profile.GetDurationMs("signal_ttl", &signal_ttl_ms, signal_ttl_ms); profile.GetUInt64("sketch_width", &sketch_width, sketch_width); profile.GetUInt64("sketch_depth", &sketch_depth, sketch_depth); + profile.GetBool("auto_collect_master_signals", + &master_config.auto_collect_master_signals, + master_config.auto_collect_master_signals); + profile.GetDurationMs("signal_refresh_interval", + &signal_refresh_interval_ms, + signal_refresh_interval_ms); profile.GetUInt32("min_complete_replicas", &config.policy.min_complete_replicas, config.policy.min_complete_replicas); @@ -425,6 +436,8 @@ mooncake::ReplicaPlacementShadowConfig LoadReplicaPlacementShadowProfile( config.signal_ttl = std::chrono::milliseconds(signal_ttl_ms); config.sketch_width = static_cast(sketch_width); config.sketch_depth = static_cast(sketch_depth); + master_config.signal_refresh_interval = + std::chrono::milliseconds(signal_refresh_interval_ms); constexpr std::array kTemperatureNames = { "cold", "warm", "hot"}; @@ -450,10 +463,10 @@ mooncake::ReplicaPlacementShadowConfig LoadReplicaPlacementShadowProfile( mooncake::ReplicaTierTarget{desired, required}; } } - return config; + return master_config; } -std::optional +std::optional LoadOptionalReplicaPlacementShadowProfile(const std::string& path) { if (path.empty()) return std::nullopt; return LoadReplicaPlacementShadowProfile(path); diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index af6bc88133..006ec5edf3 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -412,12 +413,28 @@ MasterService::MasterService(const MasterServiceConfig& config) } if (config.replica_placement_shadow_config.has_value()) { + const auto& shadow_config = *config.replica_placement_shadow_config; + if (shadow_config.auto_collect_master_signals && + (shadow_config.signal_refresh_interval <= + std::chrono::nanoseconds::zero() || + shadow_config.signal_refresh_interval > + shadow_config.evaluator.signal_ttl)) { + throw std::invalid_argument( + "replica placement SHADOW refresh interval must be positive " + "and no greater than signal TTL"); + } replica_placement_shadow_evaluator_ = std::make_unique( - *config.replica_placement_shadow_config); + shadow_config.evaluator); + replica_placement_shadow_auto_collect_signals_ = + shadow_config.auto_collect_master_signals; + replica_placement_shadow_refresh_interval_ = + shadow_config.signal_refresh_interval; LOG(INFO) << "Replica placement SHADOW enabled: observations produce " "fixed-cardinality intent counters only; Copy/Move " - "submission remains disabled"; + "submission remains disabled, auto_collect_master_" + "signals=" + << replica_placement_shadow_auto_collect_signals_; } kv_event_publisher_ = @@ -598,8 +615,14 @@ MasterService::PublishReplicaPlacementSignalSnapshot( if (!replica_placement_shadow_evaluator_) { return ReplicaPlacementSignalPublishStatus::NOT_ENABLED; } - return replica_placement_shadow_evaluator_->PublishSignalSnapshot( - std::move(snapshot)); + const auto status = + replica_placement_shadow_evaluator_->PublishSignalSnapshot( + std::move(snapshot)); + if (status == ReplicaPlacementSignalPublishStatus::PUBLISHED) { + replica_placement_shadow_external_signal_source_.store( + true, std::memory_order_release); + } + return status; } uint64_t MasterService::ReplicaPlacementShadowSignalGeneration() @@ -619,10 +642,155 @@ void MasterService::ObserveReplicaPlacementShadow( const ObjectIdentity& object_id, std::span replicas) { if (!replica_placement_shadow_evaluator_) return; + RefreshReplicaPlacementShadowSignals(); (void)replica_placement_shadow_evaluator_->Observe( MakeTenantScopedKey(object_id.tenant_id, object_id.user_key), replicas); } +void MasterService::RefreshReplicaPlacementShadowSignals() { + if (!replica_placement_shadow_auto_collect_signals_ || + replica_placement_shadow_external_signal_source_.load( + std::memory_order_acquire)) { + return; + } + const auto now = std::chrono::steady_clock::now(); + std::lock_guard lock(replica_placement_shadow_refresh_mutex_); + if (replica_placement_shadow_external_signal_source_.load( + std::memory_order_acquire) || + (replica_placement_shadow_last_refresh_.has_value() && + now - *replica_placement_shadow_last_refresh_ < + replica_placement_shadow_refresh_interval_)) { + return; + } + + auto snapshot = CollectReplicaPlacementShadowSignals(now); + const uint64_t current_generation = + replica_placement_shadow_evaluator_->CurrentGeneration(); + if (current_generation == std::numeric_limits::max()) return; + snapshot.generation = current_generation + 1; + if (replica_placement_shadow_evaluator_->PublishSignalSnapshot(std::move( + snapshot)) == ReplicaPlacementSignalPublishStatus::PUBLISHED) { + replica_placement_shadow_last_refresh_ = now; + } +} + +ReplicaPlacementSignalSnapshot +MasterService::CollectReplicaPlacementShadowSignals( + ReplicaPlacementSignalClock::TimePoint observed_at) { + ReplicaPlacementSignalSnapshot snapshot; + snapshot.ready = true; + snapshot.observed_at = observed_at; + auto set_tier = [&snapshot](ReplicaPlacementTier tier, + ReplicaTierSignalState allocation, + ReplicaTierSignalState health) { + auto& signals = snapshot.tiers[static_cast(tier)]; + signals.allocation_state = allocation; + signals.health_state = health; + }; + + { + auto access = segment_manager_.getAllocatorAccess(); + const auto& manager = access.getAllocatorManager(); + const bool mounted = !manager.getNames().empty(); + bool allocatable = false; + for (const auto& name : manager.getNames()) { + const auto* allocators = manager.getAllocators(name); + if (!allocators) continue; + for (const auto& allocator : *allocators) { + if (allocator && allocator->size() < allocator->capacity()) { + allocatable = true; + break; + } + } + if (allocatable) break; + } + set_tier(ReplicaPlacementTier::MEMORY, + allocatable ? ReplicaTierSignalState::AVAILABLE + : ReplicaTierSignalState::UNAVAILABLE, + mounted ? ReplicaTierSignalState::AVAILABLE + : ReplicaTierSignalState::UNAVAILABLE); + } + + { + auto access = segment_manager_.getLocalDiskSegmentAccess(); + auto& segments = access.getClientLocalDiskSegment(); + const bool mounted = !segments.empty(); + bool any_enabled = false; + bool all_enabled_capacity_known = true; + bool allocatable = false; + for (const auto& [client_id, segment] : segments) { + (void)client_id; + MutexLocker lock(&segment->offloading_mutex_); + if (!segment->enable_offloading) continue; + any_enabled = true; + const int64_t capacity = segment->ssd_total_capacity_bytes; + if (capacity <= 0) { + all_enabled_capacity_known = false; + continue; + } + if (segment->ssd_used_bytes.load(std::memory_order_relaxed) < + capacity) { + allocatable = true; + } + } + ReplicaTierSignalState allocation = ReplicaTierSignalState::UNKNOWN; + if (!mounted) { + allocation = ReplicaTierSignalState::UNAVAILABLE; + } else if (allocatable) { + allocation = ReplicaTierSignalState::AVAILABLE; + } else if (any_enabled && all_enabled_capacity_known) { + allocation = ReplicaTierSignalState::UNAVAILABLE; + } + set_tier(ReplicaPlacementTier::LOCAL_DISK, allocation, + mounted ? ReplicaTierSignalState::AVAILABLE + : ReplicaTierSignalState::UNAVAILABLE); + } + + { + auto access = nof_segment_manager_.getAllocatorAccess(); + const auto& manager = access.getAllocatorManager(); + const bool mounted = !manager.getNames().empty(); + bool allocatable = false; + for (const auto& name : manager.getNames()) { + const auto* allocators = manager.getAllocators(name); + if (!allocators) continue; + for (const auto& allocator : *allocators) { + if (allocator && allocator->size() < allocator->capacity()) { + allocatable = true; + break; + } + } + if (allocatable) break; + } + set_tier(ReplicaPlacementTier::NOF_SSD, + allocatable ? ReplicaTierSignalState::AVAILABLE + : ReplicaTierSignalState::UNAVAILABLE, + mounted ? ReplicaTierSignalState::AVAILABLE + : ReplicaTierSignalState::UNAVAILABLE); + } + + if (!use_disk_replica_) { + set_tier(ReplicaPlacementTier::REMOTE_STORE, + ReplicaTierSignalState::UNAVAILABLE, + ReplicaTierSignalState::UNAVAILABLE); + } else { + std::error_code error; + const auto space = std::filesystem::space( + std::filesystem::path(root_fs_dir_) / cluster_id_, error); + if (error) { + set_tier(ReplicaPlacementTier::REMOTE_STORE, + ReplicaTierSignalState::UNAVAILABLE, + ReplicaTierSignalState::UNAVAILABLE); + } else { + set_tier(ReplicaPlacementTier::REMOTE_STORE, + space.available > 0 ? ReplicaTierSignalState::AVAILABLE + : ReplicaTierSignalState::UNAVAILABLE, + ReplicaTierSignalState::AVAILABLE); + } + } + return snapshot; +} + void MasterService::SetNoFProbeFnForTesting(NoFProbeFn fn) { #ifdef USE_NOF std::lock_guard lock(nof_probe_fn_mutex_); diff --git a/mooncake-store/tests/replica_placement_master_shadow_test.cpp b/mooncake-store/tests/replica_placement_master_shadow_test.cpp index 7258d662a4..8058056701 100644 --- a/mooncake-store/tests/replica_placement_master_shadow_test.cpp +++ b/mooncake-store/tests/replica_placement_master_shadow_test.cpp @@ -6,11 +6,13 @@ #include #include #include +#include #include #include #include #include +#include #include namespace mooncake::test { @@ -58,6 +60,13 @@ ReplicaPlacementShadowConfig ShadowConfig() { return config; } +MasterReplicaPlacementShadowConfig ExplicitSignalShadowConfig() { + MasterReplicaPlacementShadowConfig config; + config.evaluator = ShadowConfig(); + config.auto_collect_master_signals = false; + return config; +} + ReplicaPlacementSignalSnapshot AvailableSnapshot(uint64_t generation) { ReplicaPlacementSignalSnapshot snapshot; snapshot.generation = generation; @@ -113,7 +122,7 @@ TEST_F(ReplicaPlacementMasterShadowTest, DefaultOffHasNoObserverState) { TEST_F(ReplicaPlacementMasterShadowTest, MissingSnapshotRecordsDegradedObservationWithoutActuation) { MasterServiceConfig config; - config.replica_placement_shadow_config = ShadowConfig(); + config.replica_placement_shadow_config = ExplicitSignalShadowConfig(); MasterService service(config); const UUID client_id = PrepareMemorySegment(service, "shadow_missing"); PutObject(service, client_id, "key"); @@ -133,7 +142,7 @@ TEST_F(ReplicaPlacementMasterShadowTest, TEST_F(ReplicaPlacementMasterShadowTest, SingleAndBatchGetDriveShadowButAdminGetDoesNot) { MasterServiceConfig config; - config.replica_placement_shadow_config = ShadowConfig(); + config.replica_placement_shadow_config = ExplicitSignalShadowConfig(); MasterService service(config); ASSERT_EQ( service.PublishReplicaPlacementSignalSnapshot(AvailableSnapshot(1)), @@ -169,7 +178,7 @@ TEST_F(ReplicaPlacementMasterShadowTest, TEST_F(ReplicaPlacementMasterShadowTest, ConcurrentGetsPreserveExactObservationCount) { MasterServiceConfig config; - config.replica_placement_shadow_config = ShadowConfig(); + config.replica_placement_shadow_config = ExplicitSignalShadowConfig(); MasterService service(config); ASSERT_EQ( service.PublishReplicaPlacementSignalSnapshot(AvailableSnapshot(1)), @@ -200,6 +209,148 @@ TEST_F(ReplicaPlacementMasterShadowTest, service.RemoveAll(); } +TEST_F(ReplicaPlacementMasterShadowTest, + AutoCollectorPublishesRealMountedMemorySignalsWithBoundedRefresh) { + MasterReplicaPlacementShadowConfig shadow; + shadow.evaluator = ShadowConfig(); + shadow.evaluator.policy + .targets[static_cast(ReplicaTemperature::COLD)] + [Tier(ReplicaPlacementTier::MEMORY)] + .desired = 2; + shadow.auto_collect_master_signals = true; + shadow.signal_refresh_interval = std::chrono::seconds(1); + MasterServiceConfig config; + config.replica_placement_shadow_config = shadow; + MasterService service(config); + const UUID client_id = PrepareMemorySegment(service, "shadow_collector"); + PutObject(service, client_id, "key"); + + ASSERT_TRUE(service.GetReplicaList("key", "default").has_value()); + EXPECT_EQ(service.ReplicaPlacementShadowSignalGeneration(), 1); + ASSERT_TRUE(service.GetReplicaList("key", "default").has_value()); + EXPECT_EQ(service.ReplicaPlacementShadowSignalGeneration(), 1) + << "refresh interval must prevent a signal scan on every Get"; + + const auto counters = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(counters.has_value()); + EXPECT_EQ(TotalObservations(*counters), 2); + EXPECT_EQ(counters->observations[Observation( + ReplicaTemperature::COLD, + ReplicaPlacementShadowSignalStatus::READY)], + 1); + EXPECT_EQ(counters->observations[Observation( + ReplicaTemperature::WARM, + ReplicaPlacementShadowSignalStatus::READY)], + 1); + EXPECT_EQ(counters->add_intents[Intent(ReplicaTemperature::COLD, + ReplicaPlacementTier::MEMORY)], + 1); + service.RemoveAll(); +} + +TEST_F(ReplicaPlacementMasterShadowTest, + AutoCollectorUsesReportedLocalDiskCapacity) { + MasterReplicaPlacementShadowConfig shadow; + shadow.evaluator = ShadowConfig(); + shadow.evaluator.policy.targets[static_cast( + ReplicaTemperature::COLD)][Tier(ReplicaPlacementTier::LOCAL_DISK)] = + ReplicaTierTarget{1, true}; + MasterServiceConfig config; + config.enable_offload = true; + config.replica_placement_shadow_config = shadow; + MasterService service(config); + const UUID client_id = PrepareMemorySegment(service, "shadow_local_disk"); + ASSERT_TRUE(service.MountLocalDiskSegment(client_id, true).has_value()); + ASSERT_TRUE(service.ReportSsdCapacity(client_id, 1024 * 1024).has_value()); + PutObject(service, client_id, "key"); + + ASSERT_TRUE(service.GetReplicaList("key", "default").has_value()); + const auto counters = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(counters.has_value()); + EXPECT_EQ(counters->add_intents[Intent(ReplicaTemperature::COLD, + ReplicaPlacementTier::LOCAL_DISK)], + 1); + service.RemoveAll(); +} + +TEST_F(ReplicaPlacementMasterShadowTest, + AutoCollectorKeepsUnreportedLocalDiskCapacityUnknown) { + MasterReplicaPlacementShadowConfig shadow; + shadow.evaluator = ShadowConfig(); + shadow.evaluator.policy.targets[static_cast( + ReplicaTemperature::COLD)][Tier(ReplicaPlacementTier::LOCAL_DISK)] = + ReplicaTierTarget{1, true}; + MasterServiceConfig config; + config.enable_offload = true; + config.replica_placement_shadow_config = shadow; + MasterService service(config); + const UUID client_id = + PrepareMemorySegment(service, "shadow_local_unknown"); + ASSERT_TRUE(service.MountLocalDiskSegment(client_id, true).has_value()); + PutObject(service, client_id, "key"); + + ASSERT_TRUE(service.GetReplicaList("key", "default").has_value()); + const auto counters = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(counters.has_value()); + EXPECT_EQ(counters->add_intents[Intent(ReplicaTemperature::COLD, + ReplicaPlacementTier::LOCAL_DISK)], + 0); + EXPECT_EQ( + counters->degraded[static_cast( + ReplicaPlacementDegradedReason::REQUIRED_TIER_SIGNAL_UNKNOWN)], + 1); + service.RemoveAll(); +} + +TEST_F(ReplicaPlacementMasterShadowTest, + AutoCollectorProbesConfiguredRemoteStoreDirectory) { + const auto root = std::filesystem::temp_directory_path() / + ("mooncake_shadow_remote_" + + std::to_string(static_cast(::getpid()))); + const std::string cluster_id = "cluster"; + ASSERT_TRUE(std::filesystem::create_directories(root / cluster_id)); + + MasterReplicaPlacementShadowConfig shadow; + shadow.evaluator = ShadowConfig(); + shadow.evaluator.policy.targets[static_cast( + ReplicaTemperature::COLD)][Tier(ReplicaPlacementTier::REMOTE_STORE)] = + ReplicaTierTarget{2, true}; + MasterServiceConfig config; + config.root_fs_dir = root.string(); + config.cluster_id = cluster_id; + config.replica_placement_shadow_config = shadow; + { + MasterService service(config); + const UUID client_id = + PrepareMemorySegment(service, "shadow_remote_store"); + PutObject(service, client_id, "key"); + + ASSERT_TRUE(service.GetReplicaList("key", "default").has_value()); + const auto counters = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(counters.has_value()); + EXPECT_EQ( + counters->add_intents[Intent(ReplicaTemperature::COLD, + ReplicaPlacementTier::REMOTE_STORE)], + 1); + service.RemoveAll(); + } + std::error_code error; + std::filesystem::remove_all(root, error); + EXPECT_FALSE(error); +} + +TEST(ReplicaPlacementMasterShadowConfigTest, + AutoCollectorRefreshCannotOutliveSignalTtl) { + MasterReplicaPlacementShadowConfig shadow; + shadow.evaluator = ShadowConfig(); + shadow.evaluator.signal_ttl = std::chrono::milliseconds(10); + shadow.signal_refresh_interval = std::chrono::milliseconds(11); + MasterServiceConfig config; + config.replica_placement_shadow_config = shadow; + + EXPECT_THROW((void)MasterService(config), std::invalid_argument); +} + TEST(ReplicaPlacementMasterShadowConfigTest, WrappedConfigPropagatesExplicitOptInOnly) { WrappedMasterServiceConfig wrapped; @@ -207,11 +358,13 @@ TEST(ReplicaPlacementMasterShadowConfigTest, MasterServiceConfig disabled(wrapped); EXPECT_FALSE(disabled.replica_placement_shadow_config.has_value()); - wrapped.replica_placement_shadow_config = ShadowConfig(); + wrapped.replica_placement_shadow_config = ExplicitSignalShadowConfig(); MasterServiceConfig enabled(wrapped); ASSERT_TRUE(enabled.replica_placement_shadow_config.has_value()); - EXPECT_EQ(enabled.replica_placement_shadow_config->warm_threshold, 2); - EXPECT_EQ(enabled.replica_placement_shadow_config->hot_threshold, 3); + EXPECT_EQ(enabled.replica_placement_shadow_config->evaluator.warm_threshold, + 2); + EXPECT_EQ(enabled.replica_placement_shadow_config->evaluator.hot_threshold, + 3); } } // namespace From 174e2240429f6e0a573e05c2d47c88394425ba1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Wed, 15 Jul 2026 10:46:26 +0800 Subject: [PATCH 5/6] feat(store): export replica placement shadow metrics --- .../include/master_metric_manager.h | 13 ++ mooncake-store/src/master_metric_manager.cpp | 67 ++++++++ mooncake-store/src/master_service.cpp | 3 +- .../replica_placement_master_shadow_test.cpp | 157 ++++++++++++++++++ 4 files changed, 239 insertions(+), 1 deletion(-) diff --git a/mooncake-store/include/master_metric_manager.h b/mooncake-store/include/master_metric_manager.h index fee705595b..4e31ddfdc6 100644 --- a/mooncake-store/include/master_metric_manager.h +++ b/mooncake-store/include/master_metric_manager.h @@ -12,6 +12,8 @@ namespace mooncake { +struct ReplicaPlacementShadowResult; + class MasterMetricManager { public: // --- Singleton Access --- @@ -389,6 +391,12 @@ class MasterMetricManager { int64_t get_update_task_requests(); int64_t get_update_task_failures(); + // Replica placement SHADOW metrics have fixed-cardinality labels derived + // only from enums. Object keys, tenants and endpoints are deliberately not + // accepted by this API. + void observe_replica_placement_shadow( + const ReplicaPlacementShadowResult& result); + // --- Serialization --- /** * @brief Serializes all managed metrics into Prometheus text format. @@ -686,6 +694,11 @@ class MasterMetricManager { ylt::metric::dynamic_counter_2t tenant_quota_reject_total_; ylt::metric::dynamic_counter_1t tenant_evict_bytes_total_; + ylt::metric::dynamic_counter_2t replica_placement_shadow_observations_; + ylt::metric::dynamic_counter_2t replica_placement_shadow_add_intents_; + ylt::metric::dynamic_counter_2t replica_placement_shadow_remove_intents_; + ylt::metric::dynamic_counter_1t replica_placement_shadow_degraded_; + // Snapshot Metrics ylt::metric::histogram_t snapshot_duration_ms_; ylt::metric::counter_t snapshot_success_; diff --git a/mooncake-store/src/master_metric_manager.cpp b/mooncake-store/src/master_metric_manager.cpp index ac4a38bfa4..50bb46a3df 100644 --- a/mooncake-store/src/master_metric_manager.cpp +++ b/mooncake-store/src/master_metric_manager.cpp @@ -7,6 +7,7 @@ #include // Required by histogram serialization #include +#include "replica_placement_shadow.h" #include "utils.h" namespace mooncake { @@ -381,6 +382,23 @@ MasterMetricManager::MasterMetricManager() tenant_evict_bytes_total_( "mooncake_tenant_evict_bytes_total", "Total bytes evicted by tenant-scoped quota eviction", {"tenant_id"}), + replica_placement_shadow_observations_( + "master_replica_placement_shadow_observations_total", + "Replica placement SHADOW observations by temperature and signal " + "status", + {"temperature", "status"}), + replica_placement_shadow_add_intents_( + "master_replica_placement_shadow_add_intents_total", + "Replica add intents proposed by replica placement SHADOW", + {"temperature", "tier"}), + replica_placement_shadow_remove_intents_( + "master_replica_placement_shadow_remove_intents_total", + "Replica remove intents proposed by replica placement SHADOW", + {"temperature", "tier"}), + replica_placement_shadow_degraded_( + "master_replica_placement_shadow_degraded_total", + "Replica placement SHADOW degraded observations by reason", + {"reason"}), // Snapshot Metrics snapshot_duration_ms_( @@ -1694,6 +1712,49 @@ int64_t MasterMetricManager::get_update_task_failures() { return mark_task_to_complete_failures_.value(); } +void MasterMetricManager::observe_replica_placement_shadow( + const ReplicaPlacementShadowResult& result) { + static constexpr std::array + kTemperatures = {"cold", "warm", "hot"}; + static constexpr std::array + kStatuses = {"ready", "missing_snapshot", "source_unavailable", + "stale_snapshot"}; + static constexpr std::array + kTiers = {"memory", "local_disk", "nof_ssd", "remote_store"}; + static constexpr std::array + kDegradedReasons = { + "required_tier_unavailable", "required_tier_unhealthy", + "required_tier_signal_unknown", "min_complete_unsatisfied"}; + + const size_t temperature = static_cast(result.temperature); + const size_t status = static_cast(result.signal_status); + if (temperature >= kTemperatures.size() || status >= kStatuses.size()) { + return; + } + replica_placement_shadow_observations_.inc( + {kTemperatures[temperature], kStatuses[status]}); + for (size_t tier = 0; tier < result.plan.adjustments.size(); ++tier) { + const auto& adjustment = result.plan.adjustments[tier]; + if (adjustment.add != 0) { + replica_placement_shadow_add_intents_.inc( + {kTemperatures[temperature], kTiers[tier]}, adjustment.add); + } + if (adjustment.remove != 0) { + replica_placement_shadow_remove_intents_.inc( + {kTemperatures[temperature], kTiers[tier]}, adjustment.remove); + } + } + for (size_t i = 0; i < result.plan.degraded_reason_count; ++i) { + const size_t reason = + static_cast(result.plan.degraded_reasons[i]); + if (reason < kDegradedReasons.size()) { + replica_placement_shadow_degraded_.inc({kDegradedReasons[reason]}); + } + } +} + // --- Serialization --- std::string MasterMetricManager::serialize_metrics() { // Note: Following Prometheus style, metrics with value 0 that haven't @@ -1837,6 +1898,12 @@ std::string MasterMetricManager::serialize_metrics() { serialize_metric(tenant_quota_reject_total_); serialize_metric(tenant_evict_bytes_total_); + // Serialize fixed-cardinality replica placement SHADOW metrics. + serialize_metric(replica_placement_shadow_observations_); + serialize_metric(replica_placement_shadow_add_intents_); + serialize_metric(replica_placement_shadow_remove_intents_); + serialize_metric(replica_placement_shadow_degraded_); + // Serialize Snapshot Metrics serialize_metric(snapshot_duration_ms_); serialize_metric(snapshot_success_); diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index 006ec5edf3..ba5148b2d5 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -643,8 +643,9 @@ void MasterService::ObserveReplicaPlacementShadow( std::span replicas) { if (!replica_placement_shadow_evaluator_) return; RefreshReplicaPlacementShadowSignals(); - (void)replica_placement_shadow_evaluator_->Observe( + const auto result = replica_placement_shadow_evaluator_->Observe( MakeTenantScopedKey(object_id.tenant_id, object_id.user_key), replicas); + MasterMetricManager::instance().observe_replica_placement_shadow(result); } void MasterService::RefreshReplicaPlacementShadowSignals() { diff --git a/mooncake-store/tests/replica_placement_master_shadow_test.cpp b/mooncake-store/tests/replica_placement_master_shadow_test.cpp index 8058056701..81bf5c23b0 100644 --- a/mooncake-store/tests/replica_placement_master_shadow_test.cpp +++ b/mooncake-store/tests/replica_placement_master_shadow_test.cpp @@ -2,6 +2,8 @@ #include "master_service.h" +#include "master_metric_manager.h" + #include #include #include @@ -139,6 +141,46 @@ TEST_F(ReplicaPlacementMasterShadowTest, service.RemoveAll(); } +TEST_F(ReplicaPlacementMasterShadowTest, + UnavailableAndStaleExternalSignalsRemainNonActuating) { + MasterServiceConfig config; + config.replica_placement_shadow_config = ExplicitSignalShadowConfig(); + MasterService service(config); + ReplicaPlacementSignalSnapshot unavailable; + unavailable.generation = 1; + unavailable.observed_at = std::chrono::steady_clock::now(); + ASSERT_EQ( + service.PublishReplicaPlacementSignalSnapshot(std::move(unavailable)), + ReplicaPlacementSignalPublishStatus::PUBLISHED); + const UUID client_id = + PrepareMemorySegment(service, "shadow_external_faults"); + PutObject(service, client_id, "key"); + + ASSERT_TRUE(service.GetReplicaList("key", "default").has_value()); + auto stale = AvailableSnapshot(2); + stale.observed_at = + std::chrono::steady_clock::now() - std::chrono::seconds(31); + ASSERT_EQ(service.PublishReplicaPlacementSignalSnapshot(std::move(stale)), + ReplicaPlacementSignalPublishStatus::PUBLISHED); + ASSERT_TRUE(service.GetReplicaList("key", "default").has_value()); + + const auto counters = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(counters.has_value()); + EXPECT_EQ(counters->observations[Observation( + ReplicaTemperature::COLD, + ReplicaPlacementShadowSignalStatus::SOURCE_UNAVAILABLE)], + 1); + EXPECT_EQ(counters->observations[Observation( + ReplicaTemperature::WARM, + ReplicaPlacementShadowSignalStatus::STALE_SNAPSHOT)], + 1); + uint64_t intents = 0; + for (uint64_t count : counters->add_intents) intents += count; + for (uint64_t count : counters->remove_intents) intents += count; + EXPECT_EQ(intents, 0); + service.RemoveAll(); +} + TEST_F(ReplicaPlacementMasterShadowTest, SingleAndBatchGetDriveShadowButAdminGetDoesNot) { MasterServiceConfig config; @@ -209,6 +251,51 @@ TEST_F(ReplicaPlacementMasterShadowTest, service.RemoveAll(); } +TEST_F(ReplicaPlacementMasterShadowTest, + PrometheusMetricsUseOnlyFixedCardinalityLabels) { + MasterServiceConfig config; + config.replica_placement_shadow_config = ExplicitSignalShadowConfig(); + MasterService service(config); + ASSERT_EQ( + service.PublishReplicaPlacementSignalSnapshot(AvailableSnapshot(1)), + ReplicaPlacementSignalPublishStatus::PUBLISHED); + const UUID client_id = PrepareMemorySegment(service, "shadow_metrics"); + const std::string secret_key = "must_not_appear_object_key_7f3a"; + PutObject(service, client_id, secret_key); + + ASSERT_TRUE(service.GetReplicaList(secret_key, "default").has_value()); + + ReplicaPlacementShadowResult synthetic; + synthetic.temperature = ReplicaTemperature::HOT; + synthetic.signal_status = ReplicaPlacementShadowSignalStatus::READY; + synthetic.plan.adjustments[Tier(ReplicaPlacementTier::LOCAL_DISK)].add = 2; + synthetic.plan.adjustments[Tier(ReplicaPlacementTier::MEMORY)].remove = 1; + synthetic.plan.degraded_reasons[0] = + ReplicaPlacementDegradedReason::REQUIRED_TIER_UNHEALTHY; + synthetic.plan.degraded_reason_count = 1; + MasterMetricManager::instance().observe_replica_placement_shadow(synthetic); + + const std::string metrics = + MasterMetricManager::instance().serialize_metrics(); + EXPECT_NE( + metrics.find("master_replica_placement_shadow_observations_total"), + std::string::npos); + EXPECT_NE(metrics.find("temperature=\"hot\",status=\"ready\""), + std::string::npos); + EXPECT_NE(metrics.find("temperature=\"hot\",tier=\"local_disk\""), + std::string::npos); + EXPECT_NE(metrics.find("temperature=\"hot\",tier=\"memory\""), + std::string::npos); + EXPECT_NE(metrics.find("reason=\"required_tier_unhealthy\""), + std::string::npos); + EXPECT_EQ(metrics.find(secret_key), std::string::npos); + EXPECT_EQ(metrics.find("tenant"), std::string::npos) + << "replica placement SHADOW metric labels must not contain tenants"; + EXPECT_EQ(metrics.find("endpoint"), std::string::npos) + << "replica placement SHADOW metric labels must not contain endpoints"; + service.RemoveAll(); +} + TEST_F(ReplicaPlacementMasterShadowTest, AutoCollectorPublishesRealMountedMemorySignalsWithBoundedRefresh) { MasterReplicaPlacementShadowConfig shadow; @@ -302,6 +389,42 @@ TEST_F(ReplicaPlacementMasterShadowTest, service.RemoveAll(); } +TEST_F(ReplicaPlacementMasterShadowTest, + AutoCollectorRejectsKnownFullLocalDisk) { + MasterReplicaPlacementShadowConfig shadow; + shadow.evaluator = ShadowConfig(); + shadow.evaluator.policy.targets[static_cast( + ReplicaTemperature::COLD)][Tier(ReplicaPlacementTier::LOCAL_DISK)] = + ReplicaTierTarget{2, true}; + MasterServiceConfig config; + config.enable_offload = true; + config.replica_placement_shadow_config = shadow; + MasterService service(config); + const UUID client_id = PrepareMemorySegment(service, "shadow_local_full"); + ASSERT_TRUE(service.MountLocalDiskSegment(client_id, true).has_value()); + constexpr int64_t kCapacity = 1024; + ASSERT_TRUE(service.ReportSsdCapacity(client_id, kCapacity).has_value()); + PutObject(service, client_id, "key"); + StorageObjectMetadata metadata; + metadata.data_size = kCapacity; + metadata.transport_endpoint = "shadow_local_full"; + OffloadTaskItem task{ + .tenant_id = "default", .key = "key", .size = kCapacity}; + ASSERT_TRUE(service.NotifyOffloadSuccess(client_id, {task}, {metadata}) + .has_value()); + + ASSERT_TRUE(service.GetReplicaList("key", "default").has_value()); + const auto counters = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(counters.has_value()); + EXPECT_EQ(counters->add_intents[Intent(ReplicaTemperature::COLD, + ReplicaPlacementTier::LOCAL_DISK)], + 0); + EXPECT_EQ(counters->degraded[static_cast( + ReplicaPlacementDegradedReason::REQUIRED_TIER_UNAVAILABLE)], + 1); + service.RemoveAll(); +} + TEST_F(ReplicaPlacementMasterShadowTest, AutoCollectorProbesConfiguredRemoteStoreDirectory) { const auto root = std::filesystem::temp_directory_path() / @@ -339,6 +462,40 @@ TEST_F(ReplicaPlacementMasterShadowTest, EXPECT_FALSE(error); } +TEST_F(ReplicaPlacementMasterShadowTest, + AutoCollectorRejectsMissingRemoteStoreDirectory) { + const auto root = std::filesystem::temp_directory_path() / + ("mooncake_shadow_missing_remote_" + + std::to_string(static_cast(::getpid()))); + std::error_code cleanup_error; + std::filesystem::remove_all(root, cleanup_error); + ASSERT_FALSE(std::filesystem::exists(root)); + + MasterReplicaPlacementShadowConfig shadow; + shadow.evaluator = ShadowConfig(); + shadow.evaluator.policy.targets[static_cast( + ReplicaTemperature::COLD)][Tier(ReplicaPlacementTier::REMOTE_STORE)] = + ReplicaTierTarget{2, true}; + MasterServiceConfig config; + config.root_fs_dir = root.string(); + config.cluster_id = "missing"; + config.replica_placement_shadow_config = shadow; + MasterService service(config); + const UUID client_id = PrepareMemorySegment(service, "shadow_remote_down"); + PutObject(service, client_id, "key"); + + ASSERT_TRUE(service.GetReplicaList("key", "default").has_value()); + const auto counters = service.GetReplicaPlacementShadowCounters(); + ASSERT_TRUE(counters.has_value()); + EXPECT_EQ(counters->add_intents[Intent(ReplicaTemperature::COLD, + ReplicaPlacementTier::REMOTE_STORE)], + 0); + EXPECT_EQ(counters->degraded[static_cast( + ReplicaPlacementDegradedReason::REQUIRED_TIER_UNHEALTHY)], + 1); + service.RemoveAll(); +} + TEST(ReplicaPlacementMasterShadowConfigTest, AutoCollectorRefreshCannotOutliveSignalTtl) { MasterReplicaPlacementShadowConfig shadow; From 3ec7650d9e71c59bbba027f57e1dc1c1a2b11910 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BD=A6=E7=BA=BE?= Date: Wed, 15 Jul 2026 15:47:25 +0800 Subject: [PATCH 6/6] [Store] Fix replica shadow snapshot storage portability --- .../include/replica_placement_shadow.h | 5 ++--- .../src/replica_placement_shadow.cpp | 18 +++++++++++------- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/mooncake-store/include/replica_placement_shadow.h b/mooncake-store/include/replica_placement_shadow.h index 73484c88ca..8521841fba 100644 --- a/mooncake-store/include/replica_placement_shadow.h +++ b/mooncake-store/include/replica_placement_shadow.h @@ -126,9 +126,8 @@ class ReplicaPlacementShadowEvaluator final { ReplicaPlacementPolicy policy_; CountMinSketch sketch_; std::shared_ptr clock_; - std::atomic> - snapshot_; - mutable std::mutex publish_mutex_; + std::shared_ptr snapshot_; + mutable std::mutex snapshot_mutex_; std::array, kReplicaTemperatureCount * kReplicaPlacementShadowSignalStatusCount> diff --git a/mooncake-store/src/replica_placement_shadow.cpp b/mooncake-store/src/replica_placement_shadow.cpp index 319631f91a..0172b11c0b 100644 --- a/mooncake-store/src/replica_placement_shadow.cpp +++ b/mooncake-store/src/replica_placement_shadow.cpp @@ -105,19 +105,19 @@ ReplicaPlacementShadowEvaluator::PublishSignalSnapshot( } } - std::lock_guard lock(publish_mutex_); - const auto current = snapshot_.load(std::memory_order_acquire); + std::lock_guard lock(snapshot_mutex_); + const auto current = snapshot_; if (current && snapshot.generation <= current->generation) { return ReplicaPlacementSignalPublishStatus::GENERATION_NOT_INCREASING; } - snapshot_.store(std::make_shared( - std::move(snapshot)), - std::memory_order_release); + snapshot_ = std::make_shared( + std::move(snapshot)); return ReplicaPlacementSignalPublishStatus::PUBLISHED; } uint64_t ReplicaPlacementShadowEvaluator::CurrentGeneration() const noexcept { - const auto snapshot = snapshot_.load(std::memory_order_acquire); + std::lock_guard lock(snapshot_mutex_); + const auto snapshot = snapshot_; return snapshot ? snapshot->generation : 0; } @@ -140,7 +140,11 @@ ReplicaPlacementShadowResult ReplicaPlacementShadowEvaluator::Observe( observations[i].pending = result.inventory[i].pending; } - const auto snapshot = snapshot_.load(std::memory_order_acquire); + std::shared_ptr snapshot; + { + std::lock_guard lock(snapshot_mutex_); + snapshot = snapshot_; + } if (!snapshot) { result.signal_status = ReplicaPlacementShadowSignalStatus::MISSING_SNAPSHOT;