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

[STORE] support Redis ACL username authentication and reorganize HA (#1757)

This commit is contained in:
EkiRui 2026-03-27 15:22:09 +08:00 committed by GitHub
parent af733e1209
commit e1220bb309
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
57 changed files with 1071 additions and 851 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -13,7 +13,7 @@
#include <unordered_set>
#include "client_metric.h"
#include "ha/leader_coordinator.h"
#include "ha/leadership/leader_coordinator.h"
#include "master_client.h"
#include "storage_backend.h"
#include "thread_pool.h"

4
mooncake-store/include/ha/backends/redis/redis_client_helper.h → mooncake-store/include/ha/common/redis/redis_connection.h
View File

@ -13,7 +13,7 @@ struct redisReply;
namespace mooncake {
namespace ha {
namespace backends {
namespace common {
namespace redis {
struct RedisEndpoint {
@ -50,6 +50,6 @@ tl::expected<RedisContextPtr, ErrorCode> ConnectRedis(
ErrorCode connection_error = ErrorCode::PERSISTENT_FAIL);
} // namespace redis
} // namespace backends
} // namespace common
} // namespace ha
} // namespace mooncake

2
mooncake-store/include/ha/backends/etcd/etcd_leader_coordinator.h → mooncake-store/include/ha/leadership/backends/etcd/etcd_leader_coordinator.h
View File

@ -6,7 +6,7 @@
#include <thread>
#include "etcd_helper.h"
#include "ha/leader_coordinator.h"
#include "ha/leadership/leader_coordinator.h"
namespace mooncake {
namespace ha {

2
mooncake-store/include/ha/backends/redis/redis_leader_coordinator.h → mooncake-store/include/ha/leadership/backends/redis/redis_leader_coordinator.h
View File

@ -8,7 +8,7 @@
#include <string>
#include <thread>
#include "ha/leader_coordinator.h"
#include "ha/leadership/leader_coordinator.h"
struct redisContext;

0
mooncake-store/include/ha/leader_coordinator.h → mooncake-store/include/ha/leadership/leader_coordinator.h
View File

2
mooncake-store/include/ha/ha_backend_factory.h → mooncake-store/include/ha/leadership/leader_coordinator_factory.h
View File

@ -5,7 +5,7 @@
#include <ylt/util/tl/expected.hpp>
#include "ha/ha_types.h"
#include "ha/leader_coordinator.h"
#include "ha/leadership/leader_coordinator.h"
namespace mooncake {
namespace ha {

0
mooncake-store/include/ha/master_service_supervisor.h → mooncake-store/include/ha/leadership/master_service_supervisor.h
View File

14
mooncake-store/include/ha/serializer_snapshot_store.h → mooncake-store/include/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.h
View File

@ -1,14 +1,16 @@
#pragma once
#include "ha/snapshot_store.h"
#include "serialize/serializer_backend.h"
#include "ha/snapshot/catalog/snapshot_catalog_store.h"
#include "ha/snapshot/object/snapshot_object_store.h"
namespace mooncake {
namespace ha {
namespace backends {
namespace embedded {
class SerializerSnapshotStore : public SnapshotStore {
class EmbeddedSnapshotCatalogStore : public SnapshotCatalogStore {
public:
explicit SerializerSnapshotStore(SerializerBackend* backend);
explicit EmbeddedSnapshotCatalogStore(SnapshotObjectStore* object_store);
ErrorCode Publish(const SnapshotDescriptor& snapshot) override;
@ -21,8 +23,10 @@ class SerializerSnapshotStore : public SnapshotStore {
ErrorCode Delete(const SnapshotId& snapshot_id) override;
private:
SerializerBackend* backend_;
SnapshotObjectStore* object_store_;
};
} // namespace embedded
} // namespace backends
} // namespace ha
} // namespace mooncake

14
mooncake-store/include/ha/backends/redis/redis_snapshot_store.h → mooncake-store/include/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.h
View File

@ -2,19 +2,19 @@
#include <string>
#include "ha/snapshot_store.h"
#include "serialize/serializer_backend.h"
#include "ha/snapshot/catalog/snapshot_catalog_store.h"
#include "ha/snapshot/object/snapshot_object_store.h"
namespace mooncake {
namespace ha {
namespace backends {
namespace redis {
class RedisSnapshotStore final : public SnapshotStore {
class RedisSnapshotCatalogStore final : public SnapshotCatalogStore {
public:
RedisSnapshotStore(SerializerBackend* payload_backend,
std::string connstring,
ClusterNamespace cluster_namespace);
RedisSnapshotCatalogStore(SnapshotObjectStore* object_store,
std::string connstring,
ClusterNamespace cluster_namespace);
ErrorCode Publish(const SnapshotDescriptor& snapshot) override;
@ -33,7 +33,7 @@ class RedisSnapshotStore final : public SnapshotStore {
const ClusterNamespace& cluster_namespace);
static std::string BuildIndexKey(const ClusterNamespace& cluster_namespace);
SerializerBackend* payload_backend_;
SnapshotObjectStore* object_store_;
std::string connstring_;
ClusterNamespace cluster_namespace_;
std::string latest_key_;

8
mooncake-store/include/ha/snapshot_store.h → mooncake-store/include/ha/snapshot/catalog/snapshot_catalog_store.h
View File

@ -14,7 +14,7 @@
namespace mooncake {
namespace ha {
namespace snapshot_store_detail {
namespace snapshot_catalog_store_detail {
constexpr std::string_view kSnapshotRoot = "mooncake_master_snapshot/";
constexpr std::string_view kSnapshotLatest = "latest.txt";
@ -77,11 +77,11 @@ inline SnapshotDescriptor MakeSnapshotDescriptor(
return descriptor;
}
} // namespace snapshot_store_detail
} // namespace snapshot_catalog_store_detail
class SnapshotStore {
class SnapshotCatalogStore {
public:
virtual ~SnapshotStore() = default;
virtual ~SnapshotCatalogStore() = default;
virtual ErrorCode Publish(const SnapshotDescriptor& snapshot) = 0;

61
mooncake-store/include/ha/snapshot/object/backends/local/local_file_snapshot_object_store.h Normal file
View File

@ -0,0 +1,61 @@
#pragma once
#include <filesystem>
#include <string>
#include <vector>
#include <ylt/util/tl/expected.hpp>
#include "ha/snapshot/object/snapshot_object_store.h"
namespace mooncake {
namespace fs = std::filesystem;
/**
* @brief Local-file snapshot object store
*
* Stores snapshot data to local file system.
* Storage path MUST be configured via MOONCAKE_SNAPSHOT_LOCAL_PATH
* environment variable. No default path is provided.
*/
class LocalFileSnapshotObjectStore final : public SnapshotObjectStore {
public:
LocalFileSnapshotObjectStore();
explicit LocalFileSnapshotObjectStore(const std::string& base_path);
~LocalFileSnapshotObjectStore() override = default;
tl::expected<void, std::string> UploadBuffer(
const std::string& key, const std::vector<uint8_t>& buffer) override;
tl::expected<void, std::string> DownloadBuffer(
const std::string& key, std::vector<uint8_t>& buffer) override;
tl::expected<void, std::string> UploadString(
const std::string& key, const std::string& data) override;
tl::expected<void, std::string> DownloadString(const std::string& key,
std::string& data) override;
tl::expected<void, std::string> DeleteObjectsWithPrefix(
const std::string& prefix) override;
tl::expected<void, std::string> ListObjectsWithPrefix(
const std::string& prefix,
std::vector<std::string>& object_keys) override;
bool IsNotFoundError(const std::string& error) const override;
std::string GetConnectionInfo() const override;
private:
fs::path KeyToPath(const std::string& key) const;
tl::expected<void, std::string> EnsureDirectoryExists(
const fs::path& dir_path) const;
bool IsPathWithinBase(const fs::path& path) const;
fs::path base_path_;
};
} // namespace mooncake

54
mooncake-store/include/ha/snapshot/object/backends/s3/s3_snapshot_object_store.h Normal file
View File

@ -0,0 +1,54 @@
#pragma once
#include <memory>
#include <string>
#include <vector>
#include <ylt/util/tl/expected.hpp>
#include "ha/snapshot/object/snapshot_object_store.h"
namespace mooncake {
#ifdef HAVE_AWS_SDK
/**
* @brief S3-backed snapshot object store
*
* Wraps S3Helper to provide snapshot object storage in S3.
*/
class S3SnapshotObjectStore final : public SnapshotObjectStore {
public:
S3SnapshotObjectStore();
~S3SnapshotObjectStore() override = default;
tl::expected<void, std::string> UploadBuffer(
const std::string& key, const std::vector<uint8_t>& buffer) override;
tl::expected<void, std::string> DownloadBuffer(
const std::string& key, std::vector<uint8_t>& buffer) override;
tl::expected<void, std::string> UploadString(
const std::string& key, const std::string& data) override;
tl::expected<void, std::string> DownloadString(const std::string& key,
std::string& data) override;
tl::expected<void, std::string> DeleteObjectsWithPrefix(
const std::string& prefix) override;
tl::expected<void, std::string> ListObjectsWithPrefix(
const std::string& prefix,
std::vector<std::string>& object_keys) override;
bool IsNotFoundError(const std::string& error) const override;
std::string GetConnectionInfo() const override;
private:
class Impl;
std::unique_ptr<Impl> impl_;
};
#endif // HAVE_AWS_SDK
} // namespace mooncake

143
mooncake-store/include/ha/snapshot/object/snapshot_object_store.h Normal file
View File

@ -0,0 +1,143 @@
#pragma once
#include <memory>
#include <stdexcept>
#include <string>
#include <vector>
#include <ylt/util/tl/expected.hpp>
namespace mooncake {
// Snapshot object store type enumeration
enum class SnapshotObjectStoreType {
LOCAL_FILE = 0, // Local file system
S3 = 1 // S3 storage
};
// Convert string to SnapshotObjectStoreType
inline SnapshotObjectStoreType ParseSnapshotObjectStoreType(
const std::string& type_str) {
#ifdef HAVE_AWS_SDK
if (type_str == "s3" || type_str == "S3") {
return SnapshotObjectStoreType::S3;
}
#else
if (type_str == "s3" || type_str == "S3") {
throw std::invalid_argument(
"S3 snapshot object store requested but AWS SDK is not "
"available. Please rebuild with HAVE_AWS_SDK or use the "
"'local' object store.");
}
#endif
if (type_str == "local" || type_str == "LOCAL") {
return SnapshotObjectStoreType::LOCAL_FILE;
}
// Unknown object store type - fail fast.
throw std::invalid_argument("Unknown snapshot object store type: '" +
type_str + "'");
}
// Convert SnapshotObjectStoreType to string
inline std::string SnapshotObjectStoreTypeToString(
SnapshotObjectStoreType type) {
switch (type) {
case SnapshotObjectStoreType::S3:
return "s3";
case SnapshotObjectStoreType::LOCAL_FILE:
return "local";
default:
throw std::invalid_argument("Unknown SnapshotObjectStoreType: " +
std::to_string(static_cast<int>(type)));
}
}
/**
* @brief Abstract interface for snapshot object storage
*
* Defines a unified storage interface for snapshot objects across
* different implementations (S3, local file system, etc.)
*/
class SnapshotObjectStore {
public:
virtual ~SnapshotObjectStore() = default;
/**
* @brief Upload binary data
* @param key Storage key (path)
* @param buffer Binary data
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> UploadBuffer(
const std::string& key, const std::vector<uint8_t>& buffer) = 0;
/**
* @brief Download binary data (supports chunked download for large files)
* @param key Storage key (path)
* @param buffer Output buffer
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> DownloadBuffer(
const std::string& key, std::vector<uint8_t>& buffer) = 0;
/**
* @brief Upload string data
* @param key Storage key (path)
* @param data String data
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> UploadString(
const std::string& key, const std::string& data) = 0;
/**
* @brief Download string data
* @param key Storage key (path)
* @param data Output string
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> DownloadString(
const std::string& key, std::string& data) = 0;
/**
* @brief Delete all objects with specified prefix
* @param prefix Prefix
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> DeleteObjectsWithPrefix(
const std::string& prefix) = 0;
/**
* @brief List all objects with specified prefix
* @param prefix Prefix
* @param object_keys Output list of object keys
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> ListObjectsWithPrefix(
const std::string& prefix, std::vector<std::string>& object_keys) = 0;
/**
* @brief Tell whether an error means the target object is missing
* @param error Store-specific error string
* @return true when the error represents a missing object
*/
virtual bool IsNotFoundError(const std::string& error) const {
return false;
}
/**
* @brief Get connection/configuration info (for logging)
* @return Connection info string
*/
virtual std::string GetConnectionInfo() const = 0;
/**
* @brief Factory method: create object store instance by type
* @param type Object store type
* @return Smart pointer to object store instance
*/
static std::unique_ptr<SnapshotObjectStore> Create(
SnapshotObjectStoreType type);
};
} // namespace mooncake

0
mooncake-store/include/utils/snapshot_logger.h → mooncake-store/include/ha/snapshot/snapshot_logger.h
View File

0
mooncake-store/include/snapshot_provider.h → mooncake-store/include/ha/snapshot/snapshot_provider.h
View File

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

@ -15,7 +15,7 @@
#include "oplog_applier.h"
#include "oplog_manager.h"
#include "oplog_watcher.h"
#include "snapshot_provider.h"
#include "ha/snapshot/snapshot_provider.h"
#include "standby_state_machine.h"
#include "types.h"

109
mooncake-store/include/master_config.h
View File

@ -58,17 +58,17 @@ struct MasterConfig {
uint64_t snapshot_child_timeout_seconds;
uint32_t snapshot_retention_count;
// Snapshot payload storage backend type: "local" or "s3", required when
// Snapshot object store type: "local" or "s3", required when
// snapshot or restore is enabled
std::string snapshot_backend_type;
std::string snapshot_object_store_type;
// Snapshot catalog backend type: ""/"serializer" or "redis". Empty keeps
// the existing serializer-backed catalog behavior.
std::string snapshot_catalog_backend_type;
// Snapshot catalog store type: ""/"embedded" or "redis". Empty keeps the
// embedded catalog behavior. "payload" remains a deprecated alias.
std::string snapshot_catalog_store_type;
// Optional connection string for snapshot catalog backend. When empty, the
// Optional connection string for snapshot catalog store. When empty, the
// implementation may fall back to a backend-specific default.
std::string snapshot_catalog_backend_connstring;
std::string snapshot_catalog_store_connstring;
// Task manager configuration
uint32_t max_total_finished_tasks;
@ -132,9 +132,9 @@ class MasterServiceSupervisorConfig {
uint64_t snapshot_child_timeout_seconds =
DEFAULT_SNAPSHOT_CHILD_TIMEOUT_SEC;
uint32_t snapshot_retention_count = DEFAULT_SNAPSHOT_RETENTION_COUNT;
std::string snapshot_backend_type;
std::string snapshot_catalog_backend_type;
std::string snapshot_catalog_backend_connstring;
std::string snapshot_object_store_type;
std::string snapshot_catalog_store_type;
std::string snapshot_catalog_store_connstring;
std::string cxl_path = DEFAULT_CXL_PATH;
size_t cxl_size = DEFAULT_CXL_SIZE;
@ -191,10 +191,10 @@ class MasterServiceSupervisorConfig {
snapshot_interval_seconds = config.snapshot_interval_seconds;
snapshot_child_timeout_seconds = config.snapshot_child_timeout_seconds;
snapshot_retention_count = config.snapshot_retention_count;
snapshot_backend_type = config.snapshot_backend_type;
snapshot_catalog_backend_type = config.snapshot_catalog_backend_type;
snapshot_catalog_backend_connstring =
config.snapshot_catalog_backend_connstring;
snapshot_object_store_type = config.snapshot_object_store_type;
snapshot_catalog_store_type = config.snapshot_catalog_store_type;
snapshot_catalog_store_connstring =
config.snapshot_catalog_store_connstring;
max_total_finished_tasks = config.max_total_finished_tasks;
max_total_pending_tasks = config.max_total_pending_tasks;
max_total_processing_tasks = config.max_total_processing_tasks;
@ -285,9 +285,9 @@ class WrappedMasterServiceConfig {
uint64_t snapshot_child_timeout_seconds =
DEFAULT_SNAPSHOT_CHILD_TIMEOUT_SEC;
uint32_t snapshot_retention_count = DEFAULT_SNAPSHOT_RETENTION_COUNT;
std::string snapshot_backend_type;
std::string snapshot_catalog_backend_type;
std::string snapshot_catalog_backend_connstring;
std::string snapshot_object_store_type;
std::string snapshot_catalog_store_type;
std::string snapshot_catalog_store_connstring;
uint32_t max_total_finished_tasks = DEFAULT_MAX_TOTAL_FINISHED_TASKS;
uint32_t max_total_pending_tasks = DEFAULT_MAX_TOTAL_PENDING_TASKS;
uint32_t max_total_processing_tasks = DEFAULT_MAX_TOTAL_PROCESSING_TASKS;
@ -359,10 +359,10 @@ class WrappedMasterServiceConfig {
snapshot_interval_seconds = config.snapshot_interval_seconds;
snapshot_child_timeout_seconds = config.snapshot_child_timeout_seconds;
snapshot_retention_count = config.snapshot_retention_count;
snapshot_backend_type = config.snapshot_backend_type;
snapshot_catalog_backend_type = config.snapshot_catalog_backend_type;
snapshot_catalog_backend_connstring =
config.snapshot_catalog_backend_connstring;
snapshot_object_store_type = config.snapshot_object_store_type;
snapshot_catalog_store_type = config.snapshot_catalog_store_type;
snapshot_catalog_store_connstring =
config.snapshot_catalog_store_connstring;
max_total_finished_tasks = config.max_total_finished_tasks;
max_total_pending_tasks = config.max_total_pending_tasks;
max_total_processing_tasks = config.max_total_processing_tasks;
@ -410,10 +410,10 @@ class WrappedMasterServiceConfig {
snapshot_interval_seconds = config.snapshot_interval_seconds;
snapshot_child_timeout_seconds = config.snapshot_child_timeout_seconds;
snapshot_retention_count = config.snapshot_retention_count;
snapshot_backend_type = config.snapshot_backend_type;
snapshot_catalog_backend_type = config.snapshot_catalog_backend_type;
snapshot_catalog_backend_connstring =
config.snapshot_catalog_backend_connstring;
snapshot_object_store_type = config.snapshot_object_store_type;
snapshot_catalog_store_type = config.snapshot_catalog_store_type;
snapshot_catalog_store_connstring =
config.snapshot_catalog_store_connstring;
max_total_finished_tasks = config.max_total_finished_tasks;
max_total_pending_tasks = config.max_total_pending_tasks;
max_total_processing_tasks = config.max_total_processing_tasks;
@ -462,9 +462,9 @@ class MasterServiceConfigBuilder {
uint64_t snapshot_child_timeout_seconds_ =
DEFAULT_SNAPSHOT_CHILD_TIMEOUT_SEC;
uint32_t snapshot_retention_count_ = DEFAULT_SNAPSHOT_RETENTION_COUNT;
std::string snapshot_backend_type_;
std::string snapshot_catalog_backend_type_;
std::string snapshot_catalog_backend_connstring_;
std::string snapshot_object_store_type_;
std::string snapshot_catalog_store_type_;
std::string snapshot_catalog_store_connstring_;
uint32_t max_total_finished_tasks_ = DEFAULT_MAX_TOTAL_FINISHED_TASKS;
uint32_t max_total_pending_tasks_ = DEFAULT_MAX_TOTAL_PENDING_TASKS;
uint32_t max_total_processing_tasks_ = DEFAULT_MAX_TOTAL_PROCESSING_TASKS;
@ -605,22 +605,43 @@ class MasterServiceConfigBuilder {
return *this;
}
MasterServiceConfigBuilder& set_snapshot_backend_type(
MasterServiceConfigBuilder& set_snapshot_object_store_type(
const std::string& type) {
snapshot_backend_type_ = type;
snapshot_object_store_type_ = type;
return *this;
}
// Deprecated compatibility shims for older tests and call sites.
MasterServiceConfigBuilder& set_snapshot_payload_store_type(
const std::string& type) {
return set_snapshot_object_store_type(type);
}
MasterServiceConfigBuilder& set_snapshot_payload_backend_type(
const std::string& type) {
return set_snapshot_object_store_type(type);
}
MasterServiceConfigBuilder& set_snapshot_catalog_store_type(
const std::string& type) {
snapshot_catalog_store_type_ = type;
return *this;
}
MasterServiceConfigBuilder& set_snapshot_catalog_backend_type(
const std::string& type) {
snapshot_catalog_backend_type_ = type;
return set_snapshot_catalog_store_type(type);
}
MasterServiceConfigBuilder& set_snapshot_catalog_store_connstring(
const std::string& connstring) {
snapshot_catalog_store_connstring_ = connstring;
return *this;
}
MasterServiceConfigBuilder& set_snapshot_catalog_backend_connstring(
const std::string& connstring) {
snapshot_catalog_backend_connstring_ = connstring;
return *this;
return set_snapshot_catalog_store_connstring(connstring);
}
MasterServiceConfigBuilder& set_max_total_finished_tasks(
@ -717,9 +738,9 @@ class MasterServiceConfig {
uint64_t snapshot_child_timeout_seconds =
DEFAULT_SNAPSHOT_CHILD_TIMEOUT_SEC;
uint32_t snapshot_retention_count = DEFAULT_SNAPSHOT_RETENTION_COUNT;
std::string snapshot_backend_type;
std::string snapshot_catalog_backend_type;
std::string snapshot_catalog_backend_connstring;
std::string snapshot_object_store_type;
std::string snapshot_catalog_store_type;
std::string snapshot_catalog_store_connstring;
TaskManagerConfig task_manager_config = {
.max_total_finished_tasks = DEFAULT_MAX_TOTAL_FINISHED_TASKS,
.max_total_pending_tasks = DEFAULT_MAX_TOTAL_PENDING_TASKS,
@ -766,10 +787,10 @@ class MasterServiceConfig {
snapshot_interval_seconds = config.snapshot_interval_seconds;
snapshot_child_timeout_seconds = config.snapshot_child_timeout_seconds;
snapshot_retention_count = config.snapshot_retention_count;
snapshot_backend_type = config.snapshot_backend_type;
snapshot_catalog_backend_type = config.snapshot_catalog_backend_type;
snapshot_catalog_backend_connstring =
config.snapshot_catalog_backend_connstring;
snapshot_object_store_type = config.snapshot_object_store_type;
snapshot_catalog_store_type = config.snapshot_catalog_store_type;
snapshot_catalog_store_connstring =
config.snapshot_catalog_store_connstring;
task_manager_config.max_total_finished_tasks =
config.max_total_finished_tasks;
@ -819,10 +840,10 @@ inline MasterServiceConfig MasterServiceConfigBuilder::build() const {
config.snapshot_interval_seconds = snapshot_interval_seconds_;
config.snapshot_child_timeout_seconds = snapshot_child_timeout_seconds_;
config.snapshot_retention_count = snapshot_retention_count_;
config.snapshot_backend_type = snapshot_backend_type_;
config.snapshot_catalog_backend_type = snapshot_catalog_backend_type_;
config.snapshot_catalog_backend_connstring =
snapshot_catalog_backend_connstring_;
config.snapshot_object_store_type = snapshot_object_store_type_;
config.snapshot_catalog_store_type = snapshot_catalog_store_type_;
config.snapshot_catalog_store_connstring =
snapshot_catalog_store_connstring_;
config.task_manager_config.max_total_finished_tasks =
max_total_finished_tasks_;
config.task_manager_config.max_total_pending_tasks =

18
mooncake-store/include/master_service.h
View File

@ -26,12 +26,12 @@
#include "master_config.h"
#include "rpc_types.h"
#include "replica.h"
#include "serialize/serializer_backend.h"
#include "ha/snapshot/object/snapshot_object_store.h"
#include "task_manager.h"
namespace mooncake {
namespace ha {
class SnapshotStore;
class SnapshotCatalogStore;
}
// Forward declarations
@ -438,13 +438,13 @@ class MasterService {
tl::expected<void, SerializationError> PersistState(
const std::string& snapshot_id);
tl::expected<void, SerializationError> UploadSnapshotFile(
tl::expected<void, SerializationError> UploadSnapshotPayloadFile(
const std::vector<uint8_t>& data, const std::string& path,
const std::string& local_filename, const std::string& snapshot_id);
std::unique_ptr<ha::SnapshotStore> CreateSnapshotStore();
std::unique_ptr<ha::SnapshotCatalogStore> CreateSnapshotCatalogStore();
void CleanupOldSnapshot(int keep_count, const std::string& snapshot_id);
ha::SnapshotStore* GetSnapshotStore();
ha::SnapshotCatalogStore* GetSnapshotCatalogStore();
// Restore master state
void RestoreState();
@ -1068,10 +1068,10 @@ class MasterService {
uint64_t snapshot_child_timeout_seconds_ =
DEFAULT_SNAPSHOT_CHILD_TIMEOUT_SEC;
uint32_t snapshot_retention_count_ = DEFAULT_SNAPSHOT_RETENTION_COUNT;
std::string snapshot_catalog_backend_type_{};
std::string snapshot_catalog_backend_connstring_;
std::unique_ptr<SerializerBackend> snapshot_backend_;
std::unique_ptr<ha::SnapshotStore> snapshot_store_;
std::string snapshot_catalog_store_type_{};
std::string snapshot_catalog_store_connstring_;
std::unique_ptr<SnapshotObjectStore> snapshot_object_store_;
std::unique_ptr<ha::SnapshotCatalogStore> snapshot_catalog_store_;
mutable std::shared_mutex snapshot_mutex_;
// Discarded replicas management

246
mooncake-store/include/serialize/serializer_backend.h
View File

@ -1,246 +0,0 @@
#pragma once
#include <filesystem>
#include <memory>
#include <stdexcept>
#include <string>
#include <vector>
#include <ylt/util/tl/expected.hpp>
namespace mooncake {
namespace fs = std::filesystem;
// Snapshot storage backend type enumeration
enum class SnapshotBackendType {
LOCAL_FILE = 0, // Local file system
S3 = 1 // S3 storage
};
// Convert string to SnapshotBackendType
inline SnapshotBackendType ParseSnapshotBackendType(
const std::string& type_str) {
#ifdef HAVE_AWS_SDK
if (type_str == "s3" || type_str == "S3") {
return SnapshotBackendType::S3;
}
#else
if (type_str == "s3" || type_str == "S3") {
throw std::invalid_argument(
"S3 backend requested but AWS SDK is not available. "
"Please rebuild with HAVE_AWS_SDK or use 'local' backend.");
}
#endif
if (type_str == "local" || type_str == "LOCAL") {
return SnapshotBackendType::LOCAL_FILE;
}
// Unknown backend type - fail fast
throw std::invalid_argument("Unknown snapshot backend type: '" + type_str +
"'");
}
// Convert SnapshotBackendType to string
inline std::string SnapshotBackendTypeToString(SnapshotBackendType type) {
switch (type) {
case SnapshotBackendType::S3:
return "s3";
case SnapshotBackendType::LOCAL_FILE:
return "local";
default:
throw std::invalid_argument("Unknown SnapshotBackendType: " +
std::to_string(static_cast<int>(type)));
}
}
/**
* @brief Abstract interface for serialization storage backend
*
* Defines a unified storage interface supporting different implementations (S3,
* local file system, etc.)
*/
class SerializerBackend {
public:
virtual ~SerializerBackend() = default;
/**
* @brief Upload binary data
* @param key Storage key (path)
* @param buffer Binary data
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> UploadBuffer(
const std::string& key, const std::vector<uint8_t>& buffer) = 0;
/**
* @brief Download binary data (supports chunked download for large files)
* @param key Storage key (path)
* @param buffer Output buffer
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> DownloadBuffer(
const std::string& key, std::vector<uint8_t>& buffer) = 0;
/**
* @brief Upload string data
* @param key Storage key (path)
* @param data String data
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> UploadString(
const std::string& key, const std::string& data) = 0;
/**
* @brief Download string data
* @param key Storage key (path)
* @param data Output string
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> DownloadString(
const std::string& key, std::string& data) = 0;
/**
* @brief Delete all objects with specified prefix
* @param prefix Prefix
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> DeleteObjectsWithPrefix(
const std::string& prefix) = 0;
/**
* @brief List all objects with specified prefix
* @param prefix Prefix
* @param object_keys Output list of object keys
* @return Empty on success, error message on failure
*/
virtual tl::expected<void, std::string> ListObjectsWithPrefix(
const std::string& prefix, std::vector<std::string>& object_keys) = 0;
/**
* @brief Tell whether an error means the target object is missing
* @param error Backend-specific error string
* @return true when the error represents a missing object
*/
virtual bool IsNotFoundError(const std::string& error) const {
return false;
}
/**
* @brief Get connection/configuration info (for logging)
* @return Connection info string
*/
virtual std::string GetConnectionInfo() const = 0;
/**
* @brief Factory method: create backend instance by type
* @param type Backend type
* @return Smart pointer to backend instance
*/
static std::unique_ptr<SerializerBackend> Create(SnapshotBackendType type);
};
#ifdef HAVE_AWS_SDK
/**
* @brief S3 storage backend implementation
*
* Wraps S3Helper to provide S3 storage functionality
* Note: Only available when HAVE_AWS_SDK macro is defined at compile time
*/
class S3Backend : public SerializerBackend {
public:
S3Backend();
~S3Backend() override = default;
tl::expected<void, std::string> UploadBuffer(
const std::string& key, const std::vector<uint8_t>& buffer) override;
tl::expected<void, std::string> DownloadBuffer(
const std::string& key, std::vector<uint8_t>& buffer) override;
tl::expected<void, std::string> UploadString(
const std::string& key, const std::string& data) override;
tl::expected<void, std::string> DownloadString(const std::string& key,
std::string& data) override;
tl::expected<void, std::string> DeleteObjectsWithPrefix(
const std::string& prefix) override;
tl::expected<void, std::string> ListObjectsWithPrefix(
const std::string& prefix,
std::vector<std::string>& object_keys) override;
bool IsNotFoundError(const std::string& error) const override;
std::string GetConnectionInfo() const override;
private:
class Impl;
std::unique_ptr<Impl> impl_;
};
#endif // HAVE_AWS_SDK
/**
* @brief Local file storage backend implementation
*
* Stores snapshot data to local file system
* Storage path MUST be configured via MOONCAKE_SNAPSHOT_LOCAL_PATH environment
* variable. No default path is provided — the environment variable is required.
*/
class LocalFileBackend : public SerializerBackend {
public:
/**
* @brief Default constructor
* Reads storage path from MOONCAKE_SNAPSHOT_LOCAL_PATH environment variable
* @throws std::runtime_error if environment variable is not set
*/
LocalFileBackend();
/**
* @brief Constructor with specified path
* @param base_path Base storage path (must not be empty)
* @throws std::runtime_error if base_path is empty
*/
explicit LocalFileBackend(const std::string& base_path);
~LocalFileBackend() override = default;
tl::expected<void, std::string> UploadBuffer(
const std::string& key, const std::vector<uint8_t>& buffer) override;
tl::expected<void, std::string> DownloadBuffer(
const std::string& key, std::vector<uint8_t>& buffer) override;
tl::expected<void, std::string> UploadString(
const std::string& key, const std::string& data) override;
tl::expected<void, std::string> DownloadString(const std::string& key,
std::string& data) override;
tl::expected<void, std::string> DeleteObjectsWithPrefix(
const std::string& prefix) override;
tl::expected<void, std::string> ListObjectsWithPrefix(
const std::string& prefix,
std::vector<std::string>& object_keys) override;
bool IsNotFoundError(const std::string& error) const override;
std::string GetConnectionInfo() const override;
private:
fs::path base_path_; // Base path for local file storage
// Convert key to full file path
fs::path KeyToPath(const std::string& key) const;
// Ensure directory exists
tl::expected<void, std::string> EnsureDirectoryExists(
const fs::path& dir_path) const;
// Check if path is within base_path_ (security check)
bool IsPathWithinBase(const fs::path& path) const;
};
} // namespace mooncake

18
mooncake-store/src/CMakeLists.txt
View File

@ -26,14 +26,16 @@ set(MOONCAKE_STORE_SOURCES
http_metadata_server.cpp
file_storage.cpp
serialize/serializer.cpp
serialize/serializer_backend.cpp
ha/ha_backend_factory.cpp
ha/serializer_snapshot_store.cpp
ha/etcd_leader_coordinator.cpp
ha/master_service_supervisor.cpp
ha/redis_client_helper.cpp
ha/redis_leader_coordinator.cpp
ha/redis_snapshot_store.cpp
ha/leadership/leader_coordinator_factory.cpp
ha/leadership/backends/etcd/etcd_leader_coordinator.cpp
ha/common/redis/redis_connection.cpp
ha/leadership/backends/redis/redis_leader_coordinator.cpp
ha/leadership/master_service_supervisor.cpp
ha/snapshot/object/snapshot_object_store.cpp
ha/snapshot/object/backends/local/local_file_snapshot_object_store.cpp
ha/snapshot/object/backends/s3/s3_snapshot_object_store.cpp
ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.cpp
ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.cpp
utils/type_util.cpp
utils/file_util.cpp
task_manager.cpp

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

@ -21,7 +21,7 @@
#include "transfer_task.h"
#include "transport/transport.h"
#include "config.h"
#include "ha/ha_backend_factory.h"
#include "ha/leadership/leader_coordinator_factory.h"
#include "types.h"
#include "client_buffer.hpp"
#include "utils.h"

18
mooncake-store/src/ha/redis_client_helper.cpp → mooncake-store/src/ha/common/redis/redis_connection.cpp
View File

@ -1,4 +1,4 @@
#include "ha/backends/redis/redis_client_helper.h"
#include "ha/common/redis/redis_connection.h"
#include <algorithm>
#include <chrono>
@ -12,7 +12,7 @@
namespace mooncake {
namespace ha {
namespace backends {
namespace common {
namespace redis {
namespace {
@ -195,10 +195,18 @@ tl::expected<RedisContextPtr, ErrorCode> ConnectRedis(
return tl::make_unexpected(connection_error);
}
const char* username = std::getenv("MC_REDIS_USERNAME");
const char* password = std::getenv("MC_REDIS_PASSWORD");
if (password != nullptr && std::strlen(password) > 0) {
RedisReplyPtr reply(static_cast<redisReply*>(redisCommand(
context.get(), "AUTH %b", password, std::strlen(password))));
RedisReplyPtr reply;
if (username != nullptr && std::strlen(username) > 0) {
reply.reset(static_cast<redisReply*>(redisCommand(
context.get(), "AUTH %b %b", username, std::strlen(username),
password, std::strlen(password))));
} else {
reply.reset(static_cast<redisReply*>(redisCommand(
context.get(), "AUTH %b", password, std::strlen(password))));
}
if (reply == nullptr || reply->type == REDIS_REPLY_ERROR) {
return tl::make_unexpected(connection_error);
}
@ -218,6 +226,6 @@ tl::expected<RedisContextPtr, ErrorCode> ConnectRedis(
#endif
} // namespace redis
} // namespace backends
} // namespace common
} // namespace ha
} // namespace mooncake

2
mooncake-store/src/ha/etcd_leader_coordinator.cpp → mooncake-store/src/ha/leadership/backends/etcd/etcd_leader_coordinator.cpp
View File

@ -1,4 +1,4 @@
#include "ha/backends/etcd/etcd_leader_coordinator.h"
#include "ha/leadership/backends/etcd/etcd_leader_coordinator.h"
#include <algorithm>
#include <atomic>

9
mooncake-store/src/ha/redis_leader_coordinator.cpp → mooncake-store/src/ha/leadership/backends/redis/redis_leader_coordinator.cpp
View File

@ -1,4 +1,4 @@
#include "ha/backends/redis/redis_leader_coordinator.h"
#include "ha/leadership/backends/redis/redis_leader_coordinator.h"
#include <chrono>
#include <exception>
@ -16,7 +16,7 @@
#endif
#include <ylt/util/tl/expected.hpp>
#include "ha/backends/redis/redis_client_helper.h"
#include "ha/common/redis/redis_connection.h"
namespace mooncake {
namespace ha {
@ -25,6 +25,11 @@ namespace redis {
namespace {
using common::redis::ConnectRedis;
using common::redis::IsStringReply;
using common::redis::RedisReplyPtr;
using common::redis::SanitizeHashTagComponent;
#ifdef STORE_USE_REDIS
constexpr auto kViewChangePollInterval = std::chrono::milliseconds(200);

6
mooncake-store/src/ha/ha_backend_factory.cpp → mooncake-store/src/ha/leadership/leader_coordinator_factory.cpp
View File

@ -1,7 +1,7 @@
#include "ha/ha_backend_factory.h"
#include "ha/leadership/leader_coordinator_factory.h"
#include "ha/backends/etcd/etcd_leader_coordinator.h"
#include "ha/backends/redis/redis_leader_coordinator.h"
#include "ha/leadership/backends/etcd/etcd_leader_coordinator.h"
#include "ha/leadership/backends/redis/redis_leader_coordinator.h"
namespace mooncake {
namespace ha {

4
mooncake-store/src/ha/master_service_supervisor.cpp → mooncake-store/src/ha/leadership/master_service_supervisor.cpp
View File

@ -1,4 +1,4 @@
#include "ha/master_service_supervisor.h"
#include "ha/leadership/master_service_supervisor.h"
#include <chrono>
#include <csignal>
@ -10,7 +10,7 @@
#include <glog/logging.h>
#include <ylt/coro_rpc/coro_rpc_server.hpp>
#include "ha/ha_backend_factory.h"
#include "ha/leadership/leader_coordinator_factory.h"
#include "rpc_service.h"
namespace mooncake {

81
mooncake-store/src/ha/serializer_snapshot_store.cpp → mooncake-store/src/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.cpp
View File

@ -1,4 +1,4 @@
#include "ha/serializer_snapshot_store.h"
#include "ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.h"
#include <algorithm>
#include <set>
@ -6,30 +6,35 @@
namespace mooncake {
namespace ha {
namespace backends {
namespace embedded {
namespace {
constexpr size_t kUnlimitedSnapshotList = 0;
ErrorCode ValidateBackend(SerializerBackend* backend) {
return backend == nullptr ? ErrorCode::INVALID_PARAMS : ErrorCode::OK;
ErrorCode ValidateObjectStore(SnapshotObjectStore* object_store) {
return object_store == nullptr ? ErrorCode::INVALID_PARAMS : ErrorCode::OK;
}
} // namespace
SerializerSnapshotStore::SerializerSnapshotStore(SerializerBackend* backend)
: backend_(backend) {}
EmbeddedSnapshotCatalogStore::EmbeddedSnapshotCatalogStore(
SnapshotObjectStore* object_store)
: object_store_(object_store) {}
ErrorCode SerializerSnapshotStore::Publish(const SnapshotDescriptor& snapshot) {
auto err = ValidateBackend(backend_);
ErrorCode EmbeddedSnapshotCatalogStore::Publish(
const SnapshotDescriptor& snapshot) {
auto err = ValidateObjectStore(object_store_);
if (err != ErrorCode::OK) {
return err;
}
if (!snapshot_store_detail::IsValidSnapshotId(snapshot.snapshot_id)) {
if (!snapshot_catalog_store_detail::IsValidSnapshotId(
snapshot.snapshot_id)) {
return ErrorCode::INVALID_PARAMS;
}
auto publish_result = backend_->UploadString(
snapshot_store_detail::BuildLatestKey(), snapshot.snapshot_id);
auto publish_result = object_store_->UploadString(
snapshot_catalog_store_detail::BuildLatestKey(), snapshot.snapshot_id);
if (!publish_result) {
return ErrorCode::PERSISTENT_FAIL;
}
@ -38,64 +43,67 @@ ErrorCode SerializerSnapshotStore::Publish(const SnapshotDescriptor& snapshot) {
}
tl::expected<std::optional<SnapshotDescriptor>, ErrorCode>
SerializerSnapshotStore::GetLatest() {
auto err = ValidateBackend(backend_);
EmbeddedSnapshotCatalogStore::GetLatest() {
auto err = ValidateObjectStore(object_store_);
if (err != ErrorCode::OK) {
return tl::make_unexpected(err);
}
std::string latest_snapshot_id;
auto get_result = backend_->DownloadString(
snapshot_store_detail::BuildLatestKey(), latest_snapshot_id);
auto get_result = object_store_->DownloadString(
snapshot_catalog_store_detail::BuildLatestKey(), latest_snapshot_id);
if (!get_result) {
if (backend_->IsNotFoundError(get_result.error())) {
if (object_store_->IsNotFoundError(get_result.error())) {
return std::optional<SnapshotDescriptor>();
}
return tl::make_unexpected(ErrorCode::PERSISTENT_FAIL);
}
latest_snapshot_id = snapshot_store_detail::TrimAsciiWhitespace(
latest_snapshot_id = snapshot_catalog_store_detail::TrimAsciiWhitespace(
std::move(latest_snapshot_id));
if (latest_snapshot_id.empty()) {
return std::optional<SnapshotDescriptor>();
}
if (!snapshot_store_detail::IsValidSnapshotId(latest_snapshot_id)) {
if (!snapshot_catalog_store_detail::IsValidSnapshotId(latest_snapshot_id)) {
return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
}
return std::optional<SnapshotDescriptor>(
snapshot_store_detail::MakeSnapshotDescriptor(latest_snapshot_id));
snapshot_catalog_store_detail::MakeSnapshotDescriptor(
latest_snapshot_id));
}
tl::expected<std::vector<SnapshotDescriptor>, ErrorCode>
SerializerSnapshotStore::List(size_t limit) {
auto err = ValidateBackend(backend_);
EmbeddedSnapshotCatalogStore::List(size_t limit) {
auto err = ValidateObjectStore(object_store_);
if (err != ErrorCode::OK) {
return tl::make_unexpected(err);
}
std::vector<std::string> object_keys;
auto list_result = backend_->ListObjectsWithPrefix(
std::string(snapshot_store_detail::kSnapshotRoot), object_keys);
auto list_result = object_store_->ListObjectsWithPrefix(
std::string(snapshot_catalog_store_detail::kSnapshotRoot), object_keys);
if (!list_result) {
return tl::make_unexpected(ErrorCode::PERSISTENT_FAIL);
}
std::set<SnapshotId, std::greater<>> snapshot_ids;
for (const auto& object_key : object_keys) {
if (object_key.size() <= snapshot_store_detail::kSnapshotRoot.size()) {
if (object_key.size() <=
snapshot_catalog_store_detail::kSnapshotRoot.size()) {
continue;
}
std::string_view suffix(object_key);
suffix.remove_prefix(snapshot_store_detail::kSnapshotRoot.size());
suffix.remove_prefix(
snapshot_catalog_store_detail::kSnapshotRoot.size());
const size_t slash_pos = suffix.find('/');
if (slash_pos == std::string_view::npos) {
continue;
}
const auto snapshot_id = suffix.substr(0, slash_pos);
if (!snapshot_store_detail::IsValidSnapshotId(snapshot_id)) {
if (!snapshot_catalog_store_detail::IsValidSnapshotId(snapshot_id)) {
continue;
}
@ -112,18 +120,18 @@ SerializerSnapshotStore::List(size_t limit) {
break;
}
snapshots.emplace_back(
snapshot_store_detail::MakeSnapshotDescriptor(snapshot_id));
snapshot_catalog_store_detail::MakeSnapshotDescriptor(snapshot_id));
}
return snapshots;
}
ErrorCode SerializerSnapshotStore::Delete(const SnapshotId& snapshot_id) {
auto err = ValidateBackend(backend_);
ErrorCode EmbeddedSnapshotCatalogStore::Delete(const SnapshotId& snapshot_id) {
auto err = ValidateObjectStore(object_store_);
if (err != ErrorCode::OK) {
return err;
}
if (!snapshot_store_detail::IsValidSnapshotId(snapshot_id)) {
if (!snapshot_catalog_store_detail::IsValidSnapshotId(snapshot_id)) {
return ErrorCode::INVALID_PARAMS;
}
@ -150,8 +158,8 @@ ErrorCode SerializerSnapshotStore::Delete(const SnapshotId& snapshot_id) {
}
}
auto delete_result = backend_->DeleteObjectsWithPrefix(
snapshot_store_detail::BuildSnapshotPrefix(snapshot_id));
auto delete_result = object_store_->DeleteObjectsWithPrefix(
snapshot_catalog_store_detail::BuildSnapshotPrefix(snapshot_id));
if (!delete_result) {
return ErrorCode::PERSISTENT_FAIL;
}
@ -161,16 +169,17 @@ ErrorCode SerializerSnapshotStore::Delete(const SnapshotId& snapshot_id) {
}
if (next_latest.has_value()) {
auto publish_result = backend_->UploadString(
snapshot_store_detail::BuildLatestKey(), next_latest->snapshot_id);
auto publish_result = object_store_->UploadString(
snapshot_catalog_store_detail::BuildLatestKey(),
next_latest->snapshot_id);
if (!publish_result) {
return ErrorCode::PERSISTENT_FAIL;
}
return ErrorCode::OK;
}
auto clear_result = backend_->DeleteObjectsWithPrefix(
snapshot_store_detail::BuildLatestKey());
auto clear_result = object_store_->DeleteObjectsWithPrefix(
snapshot_catalog_store_detail::BuildLatestKey());
if (!clear_result) {
return ErrorCode::PERSISTENT_FAIL;
}
@ -178,5 +187,7 @@ ErrorCode SerializerSnapshotStore::Delete(const SnapshotId& snapshot_id) {
return ErrorCode::OK;
}
} // namespace embedded
} // namespace backends
} // namespace ha
} // namespace mooncake

76
mooncake-store/src/ha/redis_snapshot_store.cpp → mooncake-store/src/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.cpp
View File

@ -1,4 +1,4 @@
#include "ha/backends/redis/redis_snapshot_store.h"
#include "ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.h"
#include <exception>
#include <memory>
@ -10,7 +10,7 @@
#include <hiredis/hiredis.h>
#endif
#include "ha/backends/redis/redis_client_helper.h"
#include "ha/common/redis/redis_connection.h"
namespace mooncake {
namespace ha {
@ -19,9 +19,14 @@ namespace redis {
namespace {
using common::redis::ConnectRedis;
using common::redis::IsStringReply;
using common::redis::RedisReplyPtr;
using common::redis::SanitizeHashTagComponent;
tl::expected<long long, ErrorCode> ParseSnapshotScore(
std::string_view snapshot_id) {
if (!snapshot_store_detail::IsValidSnapshotId(snapshot_id)) {
if (!snapshot_catalog_store_detail::IsValidSnapshotId(snapshot_id)) {
return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
}
@ -65,10 +70,10 @@ return 1
} // namespace
RedisSnapshotStore::RedisSnapshotStore(SerializerBackend* payload_backend,
std::string connstring,
ClusterNamespace cluster_namespace)
: payload_backend_(payload_backend),
RedisSnapshotCatalogStore::RedisSnapshotCatalogStore(
SnapshotObjectStore* object_store, std::string connstring,
ClusterNamespace cluster_namespace)
: object_store_(object_store),
connstring_(std::move(connstring)),
cluster_namespace_(ResolveClusterNamespace(cluster_namespace)),
latest_key_(BuildLatestKey(cluster_namespace_)),
@ -76,46 +81,49 @@ RedisSnapshotStore::RedisSnapshotStore(SerializerBackend* payload_backend,
#ifndef STORE_USE_REDIS
ErrorCode RedisSnapshotStore::Publish(const SnapshotDescriptor& snapshot) {
ErrorCode RedisSnapshotCatalogStore::Publish(
const SnapshotDescriptor& snapshot) {
(void)snapshot;
return ErrorCode::UNAVAILABLE_IN_CURRENT_MODE;
}
tl::expected<std::optional<SnapshotDescriptor>, ErrorCode>
RedisSnapshotStore::GetLatest() {
RedisSnapshotCatalogStore::GetLatest() {
return tl::make_unexpected(ErrorCode::UNAVAILABLE_IN_CURRENT_MODE);
}
tl::expected<std::vector<SnapshotDescriptor>, ErrorCode>
RedisSnapshotStore::List(size_t limit) {
RedisSnapshotCatalogStore::List(size_t limit) {
(void)limit;
return tl::make_unexpected(ErrorCode::UNAVAILABLE_IN_CURRENT_MODE);
}
ErrorCode RedisSnapshotStore::Delete(const SnapshotId& snapshot_id) {
ErrorCode RedisSnapshotCatalogStore::Delete(const SnapshotId& snapshot_id) {
(void)snapshot_id;
return ErrorCode::UNAVAILABLE_IN_CURRENT_MODE;
}
ClusterNamespace RedisSnapshotStore::ResolveClusterNamespace(
ClusterNamespace RedisSnapshotCatalogStore::ResolveClusterNamespace(
const ClusterNamespace& cluster_namespace) {
return cluster_namespace;
}
std::string RedisSnapshotStore::BuildLatestKey(
std::string RedisSnapshotCatalogStore::BuildLatestKey(
const ClusterNamespace& cluster_namespace) {
return cluster_namespace;
}
std::string RedisSnapshotStore::BuildIndexKey(
std::string RedisSnapshotCatalogStore::BuildIndexKey(
const ClusterNamespace& cluster_namespace) {
return cluster_namespace;
}
#else
ErrorCode RedisSnapshotStore::Publish(const SnapshotDescriptor& snapshot) {
if (!snapshot_store_detail::IsValidSnapshotId(snapshot.snapshot_id) ||
ErrorCode RedisSnapshotCatalogStore::Publish(
const SnapshotDescriptor& snapshot) {
if (!snapshot_catalog_store_detail::IsValidSnapshotId(
snapshot.snapshot_id) ||
connstring_.empty()) {
return ErrorCode::INVALID_PARAMS;
}
@ -143,7 +151,7 @@ ErrorCode RedisSnapshotStore::Publish(const SnapshotDescriptor& snapshot) {
}
tl::expected<std::optional<SnapshotDescriptor>, ErrorCode>
RedisSnapshotStore::GetLatest() {
RedisSnapshotCatalogStore::GetLatest() {
if (connstring_.empty()) {
return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
}
@ -165,21 +173,23 @@ RedisSnapshotStore::GetLatest() {
return tl::make_unexpected(ErrorCode::PERSISTENT_FAIL);
}
auto latest_snapshot_id = snapshot_store_detail::TrimAsciiWhitespace(
std::string(reply->str, reply->len));
auto latest_snapshot_id =
snapshot_catalog_store_detail::TrimAsciiWhitespace(
std::string(reply->str, reply->len));
if (latest_snapshot_id.empty()) {
return std::optional<SnapshotDescriptor>();
}
if (!snapshot_store_detail::IsValidSnapshotId(latest_snapshot_id)) {
if (!snapshot_catalog_store_detail::IsValidSnapshotId(latest_snapshot_id)) {
return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
}
return std::optional<SnapshotDescriptor>(
snapshot_store_detail::MakeSnapshotDescriptor(latest_snapshot_id));
snapshot_catalog_store_detail::MakeSnapshotDescriptor(
latest_snapshot_id));
}
tl::expected<std::vector<SnapshotDescriptor>, ErrorCode>
RedisSnapshotStore::List(size_t limit) {
RedisSnapshotCatalogStore::List(size_t limit) {
if (connstring_.empty()) {
return tl::make_unexpected(ErrorCode::INVALID_PARAMS);
}
@ -209,20 +219,20 @@ RedisSnapshotStore::List(size_t limit) {
}
const std::string snapshot_id(element->str, element->len);
if (!snapshot_store_detail::IsValidSnapshotId(snapshot_id)) {
if (!snapshot_catalog_store_detail::IsValidSnapshotId(snapshot_id)) {
continue;
}
snapshots.emplace_back(
snapshot_store_detail::MakeSnapshotDescriptor(snapshot_id));
snapshot_catalog_store_detail::MakeSnapshotDescriptor(snapshot_id));
}
return snapshots;
}
ErrorCode RedisSnapshotStore::Delete(const SnapshotId& snapshot_id) {
if (!snapshot_store_detail::IsValidSnapshotId(snapshot_id) ||
connstring_.empty() || payload_backend_ == nullptr) {
ErrorCode RedisSnapshotCatalogStore::Delete(const SnapshotId& snapshot_id) {
if (!snapshot_catalog_store_detail::IsValidSnapshotId(snapshot_id) ||
connstring_.empty() || object_store_ == nullptr) {
return ErrorCode::INVALID_PARAMS;
}
@ -239,10 +249,10 @@ ErrorCode RedisSnapshotStore::Delete(const SnapshotId& snapshot_id) {
return ErrorCode::PERSISTENT_FAIL;
}
auto delete_result = payload_backend_->DeleteObjectsWithPrefix(
snapshot_store_detail::BuildSnapshotPrefix(snapshot_id));
auto delete_result = object_store_->DeleteObjectsWithPrefix(
snapshot_catalog_store_detail::BuildSnapshotPrefix(snapshot_id));
if (!delete_result) {
LOG(ERROR) << "Failed to delete snapshot payload after Redis catalog "
LOG(ERROR) << "Failed to delete snapshot objects after Redis catalog "
"update, snapshot_id="
<< snapshot_id << ", error=" << delete_result.error();
return ErrorCode::PERSISTENT_FAIL;
@ -251,7 +261,7 @@ ErrorCode RedisSnapshotStore::Delete(const SnapshotId& snapshot_id) {
return ErrorCode::OK;
}
ClusterNamespace RedisSnapshotStore::ResolveClusterNamespace(
ClusterNamespace RedisSnapshotCatalogStore::ResolveClusterNamespace(
const ClusterNamespace& cluster_namespace) {
if (cluster_namespace.empty()) {
return "mooncake";
@ -259,13 +269,13 @@ ClusterNamespace RedisSnapshotStore::ResolveClusterNamespace(
return cluster_namespace;
}
std::string RedisSnapshotStore::BuildLatestKey(
std::string RedisSnapshotCatalogStore::BuildLatestKey(
const ClusterNamespace& cluster_namespace) {
const auto hash_tag = SanitizeHashTagComponent(cluster_namespace);
return "mooncake-store/{" + hash_tag + "}/snapshot/latest";
}
std::string RedisSnapshotStore::BuildIndexKey(
std::string RedisSnapshotCatalogStore::BuildIndexKey(
const ClusterNamespace& cluster_namespace) {
const auto hash_tag = SanitizeHashTagComponent(cluster_namespace);
return "mooncake-store/{" + hash_tag + "}/snapshot/index";

240
mooncake-store/src/serialize/serializer_backend.cpp → mooncake-store/src/ha/snapshot/object/backends/local/local_file_snapshot_object_store.cpp
View File

@ -1,151 +1,31 @@
#include "serialize/serializer_backend.h"
#include "ha/snapshot/object/backends/local/local_file_snapshot_object_store.h"
#include <algorithm>
#include <cctype>
#include <cstdlib>
#include <filesystem>
#include <fstream>
#include <sstream>
#include <string_view>
#include <fmt/format.h>
#include <glog/logging.h>
#ifdef HAVE_AWS_SDK
#include "utils/s3_helper.h"
#endif
namespace fs = std::filesystem;
namespace mooncake {
// ============================================================================
// SerializerBackend factory method implementation
// ============================================================================
std::unique_ptr<SerializerBackend> SerializerBackend::Create(
SnapshotBackendType type) {
switch (type) {
#ifdef HAVE_AWS_SDK
case SnapshotBackendType::S3:
return std::make_unique<S3Backend>();
#else
case SnapshotBackendType::S3:
throw std::runtime_error(
"S3 backend requested but AWS SDK not available. "
"Please rebuild with HAVE_AWS_SDK or use 'local' backend.");
#endif
case SnapshotBackendType::LOCAL_FILE:
return std::make_unique<LocalFileBackend>();
default:
throw std::invalid_argument("Unknown snapshot backend type");
}
}
// ============================================================================
// S3Backend implementation (compiled only when HAVE_AWS_SDK is defined)
// ============================================================================
#ifdef HAVE_AWS_SDK
namespace {
bool ContainsAsciiInsensitive(std::string_view haystack,
std::string_view needle) {
return std::search(
haystack.begin(), haystack.end(), needle.begin(), needle.end(),
[](char lhs, char rhs) {
return std::tolower(static_cast<unsigned char>(lhs)) ==
std::tolower(static_cast<unsigned char>(rhs));
}) != haystack.end();
}
} // namespace
class S3Backend::Impl {
public:
Impl() : initialized_(InitializeOnce()), s3_helper_("", "", "") {}
~Impl() { S3Helper::ShutdownAPI(); }
private:
static bool InitializeOnce() {
S3Helper::InitAPI();
return true;
}
bool initialized_;
public:
S3Helper s3_helper_;
};
S3Backend::S3Backend() : impl_(std::make_unique<Impl>()) {
LOG(INFO) << "S3Backend initialized";
}
tl::expected<void, std::string> S3Backend::UploadBuffer(
const std::string& key, const std::vector<uint8_t>& buffer) {
return impl_->s3_helper_.UploadBufferMultipart(key, buffer);
}
tl::expected<void, std::string> S3Backend::DownloadBuffer(
const std::string& key, std::vector<uint8_t>& buffer) {
return impl_->s3_helper_.DownloadBufferMultipart(key, buffer);
}
tl::expected<void, std::string> S3Backend::UploadString(
const std::string& key, const std::string& data) {
return impl_->s3_helper_.UploadString(key, data);
}
tl::expected<void, std::string> S3Backend::DownloadString(
const std::string& key, std::string& data) {
return impl_->s3_helper_.DownloadString(key, data);
}
tl::expected<void, std::string> S3Backend::DeleteObjectsWithPrefix(
const std::string& prefix) {
return impl_->s3_helper_.DeleteObjectsWithPrefix(prefix);
}
tl::expected<void, std::string> S3Backend::ListObjectsWithPrefix(
const std::string& prefix, std::vector<std::string>& object_keys) {
return impl_->s3_helper_.ListObjectsWithPrefix(prefix, object_keys);
}
bool S3Backend::IsNotFoundError(const std::string& error) const {
return ContainsAsciiInsensitive(error, "nosuchkey") ||
ContainsAsciiInsensitive(error, "not found") ||
ContainsAsciiInsensitive(error, "does not exist") ||
ContainsAsciiInsensitive(error, "404");
}
std::string S3Backend::GetConnectionInfo() const {
return impl_->s3_helper_.GetConnectionInfo();
}
#endif // HAVE_AWS_SDK
// ============================================================================
// LocalFileBackend implementation
// ============================================================================
namespace {
constexpr const char* kEnvLocalPath = "MOONCAKE_SNAPSHOT_LOCAL_PATH";
} // namespace
LocalFileBackend::LocalFileBackend() {
LocalFileSnapshotObjectStore::LocalFileSnapshotObjectStore() {
const char* env_path = std::getenv(kEnvLocalPath);
if (!env_path || !*env_path) {
throw std::runtime_error(
"MOONCAKE_SNAPSHOT_LOCAL_PATH environment variable is not set. "
"Please set it to a persistent directory path for snapshot "
"storage. "
"Example: export "
"storage. Example: export "
"MOONCAKE_SNAPSHOT_LOCAL_PATH=/data/mooncake_snapshots");
}
// Create directory if not exists, then canonicalize
std::error_code ec;
fs::create_directories(env_path, ec);
if (ec) {
@ -159,25 +39,24 @@ LocalFileBackend::LocalFileBackend() {
throw std::runtime_error(fmt::format(
"Failed to resolve base_path '{}': {}", env_path, ec.message()));
}
// Verify base_path_ is a directory
if (!fs::is_directory(base_path_)) {
throw std::runtime_error(fmt::format(
"base_path '{}' is not a directory", base_path_.string()));
}
LOG(INFO) << "LocalFileBackend initialized with path: " << base_path_;
LOG(INFO) << "LocalFileSnapshotObjectStore initialized with path: "
<< base_path_;
}
LocalFileBackend::LocalFileBackend(const std::string& base_path) {
LocalFileSnapshotObjectStore::LocalFileSnapshotObjectStore(
const std::string& base_path) {
if (base_path.empty()) {
throw std::runtime_error(
"LocalFileBackend base_path is empty. "
"LocalFileSnapshotObjectStore base_path is empty. "
"Please provide a valid persistent directory path for snapshot "
"storage.");
}
// Create directory if not exists, then canonicalize
std::error_code ec;
fs::create_directories(base_path, ec);
if (ec) {
@ -191,24 +70,21 @@ LocalFileBackend::LocalFileBackend(const std::string& base_path) {
throw std::runtime_error(fmt::format(
"Failed to resolve base_path '{}': {}", base_path, ec.message()));
}
// Verify base_path_ is a directory
if (!fs::is_directory(base_path_)) {
throw std::runtime_error(fmt::format(
"base_path '{}' is not a directory", base_path_.string()));
}
LOG(INFO) << "LocalFileBackend initialized with path: " << base_path_;
LOG(INFO) << "LocalFileSnapshotObjectStore initialized with path: "
<< base_path_;
}
fs::path LocalFileBackend::KeyToPath(const std::string& key) const {
// key format: "mooncake_master_snapshot/20231201_123456_000/metadata"
// converts to:
// "/base_path/mooncake_master_snapshot/20231201_123456_000/metadata"
fs::path LocalFileSnapshotObjectStore::KeyToPath(const std::string& key) const {
return base_path_ / key;
}
tl::expected<void, std::string> LocalFileBackend::EnsureDirectoryExists(
tl::expected<void, std::string>
LocalFileSnapshotObjectStore::EnsureDirectoryExists(
const fs::path& dir_path) const {
try {
if (!fs::exists(dir_path)) {
@ -225,41 +101,39 @@ tl::expected<void, std::string> LocalFileBackend::EnsureDirectoryExists(
}
}
bool LocalFileBackend::IsPathWithinBase(const fs::path& path) const {
bool LocalFileSnapshotObjectStore::IsPathWithinBase(
const fs::path& path) const {
std::error_code ec;
// Use weakly_canonical to handle paths that may not exist yet
fs::path canonical_path = fs::weakly_canonical(path, ec);
if (ec) return false;
if (ec) {
return false;
}
// Check if path starts with base_path_ using iterator comparison
auto [base_end, path_it] =
std::mismatch(base_path_.begin(), base_path_.end(),
canonical_path.begin(), canonical_path.end());
(void)path_it;
return base_end == base_path_.end();
}
tl::expected<void, std::string> LocalFileBackend::UploadBuffer(
tl::expected<void, std::string> LocalFileSnapshotObjectStore::UploadBuffer(
const std::string& key, const std::vector<uint8_t>& buffer) {
if (buffer.empty()) {
return tl::make_unexpected("Error: Buffer is empty");
}
fs::path full_path = KeyToPath(key);
// Security check: verify target path is within base_path_
if (!IsPathWithinBase(full_path)) {
return tl::make_unexpected(
fmt::format("Security error: Path {} is outside base directory {}",
full_path.string(), base_path_.string()));
}
// Ensure parent directory exists
auto dir_result = EnsureDirectoryExists(full_path.parent_path());
if (!dir_result) {
return dir_result;
}
// Write to file
std::ofstream file(full_path, std::ios::binary | std::ios::trunc);
if (!file) {
return tl::make_unexpected(fmt::format(
@ -268,7 +142,6 @@ tl::expected<void, std::string> LocalFileBackend::UploadBuffer(
file.write(reinterpret_cast<const char*>(buffer.data()),
static_cast<std::streamsize>(buffer.size()));
if (!file) {
return tl::make_unexpected(fmt::format(
"Failed to write data to file: {}", full_path.string()));
@ -280,24 +153,19 @@ tl::expected<void, std::string> LocalFileBackend::UploadBuffer(
return {};
}
tl::expected<void, std::string> LocalFileBackend::DownloadBuffer(
tl::expected<void, std::string> LocalFileSnapshotObjectStore::DownloadBuffer(
const std::string& key, std::vector<uint8_t>& buffer) {
fs::path full_path = KeyToPath(key);
// Security check: verify target path is within base_path_
if (!IsPathWithinBase(full_path)) {
return tl::make_unexpected(
fmt::format("Security error: Path {} is outside base directory {}",
full_path.string(), base_path_.string()));
}
// Check if file exists
if (!fs::exists(full_path)) {
return tl::make_unexpected(
fmt::format("File not found: {}", full_path.string()));
}
// Get file size
std::error_code ec;
auto file_size = fs::file_size(full_path, ec);
if (ec) {
@ -306,18 +174,15 @@ tl::expected<void, std::string> LocalFileBackend::DownloadBuffer(
full_path.string(), ec.message()));
}
// Open file
std::ifstream file(full_path, std::ios::binary);
if (!file) {
return tl::make_unexpected(fmt::format(
"Failed to open file for reading: {}", full_path.string()));
}
// Read file content
buffer.resize(file_size);
file.read(reinterpret_cast<char*>(buffer.data()),
static_cast<std::streamsize>(file_size));
if (!file) {
return tl::make_unexpected(fmt::format(
"Failed to read data from file: {}", full_path.string()));
@ -328,24 +193,20 @@ tl::expected<void, std::string> LocalFileBackend::DownloadBuffer(
return {};
}
tl::expected<void, std::string> LocalFileBackend::UploadString(
tl::expected<void, std::string> LocalFileSnapshotObjectStore::UploadString(
const std::string& key, const std::string& data) {
fs::path full_path = KeyToPath(key);
// Security check: verify target path is within base_path_
if (!IsPathWithinBase(full_path)) {
return tl::make_unexpected(
fmt::format("Security error: Path {} is outside base directory {}",
full_path.string(), base_path_.string()));
}
// Ensure parent directory exists
auto dir_result = EnsureDirectoryExists(full_path.parent_path());
if (!dir_result) {
return dir_result;
}
// Write to file
std::ofstream file(full_path, std::ios::trunc);
if (!file) {
return tl::make_unexpected(fmt::format(
@ -363,31 +224,25 @@ tl::expected<void, std::string> LocalFileBackend::UploadString(
return {};
}
tl::expected<void, std::string> LocalFileBackend::DownloadString(
tl::expected<void, std::string> LocalFileSnapshotObjectStore::DownloadString(
const std::string& key, std::string& data) {
fs::path full_path = KeyToPath(key);
// Security check: verify target path is within base_path_
if (!IsPathWithinBase(full_path)) {
return tl::make_unexpected(
fmt::format("Security error: Path {} is outside base directory {}",
full_path.string(), base_path_.string()));
}
// Check if file exists
if (!fs::exists(full_path)) {
return tl::make_unexpected(
fmt::format("File not found: {}", full_path.string()));
}
// Open file
std::ifstream file(full_path);
if (!file) {
return tl::make_unexpected(fmt::format(
"Failed to open file for reading: {}", full_path.string()));
}
// Read file content
std::stringstream buffer;
buffer << file.rdbuf();
data = buffer.str();
@ -396,48 +251,38 @@ tl::expected<void, std::string> LocalFileBackend::DownloadString(
return {};
}
tl::expected<void, std::string> LocalFileBackend::DeleteObjectsWithPrefix(
tl::expected<void, std::string>
LocalFileSnapshotObjectStore::DeleteObjectsWithPrefix(
const std::string& prefix) {
// In snapshot scenarios, the prefix is always a directory path
// (e.g., "mooncake_master_snapshot/20240101_123456_000/").
// This method deletes the entire directory corresponding to the prefix.
fs::path target_dir = KeyToPath(prefix);
// Security check: verify target path is within base_path_
if (!IsPathWithinBase(target_dir)) {
LOG(ERROR) << "Security violation: Attempted to delete path "
"outside base directory. "
<< "base_path=" << base_path_
<< ", target_path=" << target_dir;
LOG(ERROR) << "Security violation: Attempted to delete path outside "
"base directory. base_path="
<< base_path_ << ", target_path=" << target_dir;
return tl::make_unexpected(
fmt::format("Security error: Path {} is outside base directory {}",
target_dir.string(), base_path_.string()));
}
// Don't allow deleting base_path_ itself
std::error_code ec;
fs::path canonical_target = fs::weakly_canonical(target_dir, ec);
if (!ec && canonical_target == base_path_) {
LOG(ERROR) << "Security violation: Attempted to delete base "
"directory itself. "
<< "base_path=" << base_path_;
"directory itself. base_path="
<< base_path_;
return tl::make_unexpected(
fmt::format("Security error: Cannot delete base directory {}",
base_path_.string()));
}
// Directory doesn't exist, treat as successful deletion
if (!fs::exists(target_dir)) {
return {};
}
// Verify target is a directory (per design constraint)
if (!fs::is_directory(target_dir)) {
return tl::make_unexpected(fmt::format(
"Target path '{}' is not a directory", target_dir.string()));
}
// Delete the entire directory
auto removed_count = fs::remove_all(target_dir, ec);
if (ec) {
return tl::make_unexpected(
@ -449,37 +294,26 @@ tl::expected<void, std::string> LocalFileBackend::DeleteObjectsWithPrefix(
return {};
}
tl::expected<void, std::string> LocalFileBackend::ListObjectsWithPrefix(
tl::expected<void, std::string>
LocalFileSnapshotObjectStore::ListObjectsWithPrefix(
const std::string& prefix, std::vector<std::string>& object_keys) {
// In snapshot scenarios, the prefix is always a directory path
// (e.g., "mooncake_master_snapshot/" or
// "mooncake_master_snapshot/20240101_123456_000/"). This method lists all
// files recursively under that directory.
object_keys.clear();
fs::path target_dir = KeyToPath(prefix);
// Security check: verify target path is within base_path_
if (!IsPathWithinBase(target_dir)) {
return tl::make_unexpected(
fmt::format("Security error: Path {} is outside base directory {}",
target_dir.string(), base_path_.string()));
}
// Directory doesn't exist, return empty list
if (!fs::exists(target_dir)) {
return {};
}
// Verify target is a directory (per design constraint)
if (!fs::is_directory(target_dir)) {
return tl::make_unexpected(fmt::format(
"Target path '{}' is not a directory", target_dir.string()));
}
// Recursively traverse all files in the directory
for (const auto& entry : fs::recursive_directory_iterator(target_dir)) {
if (entry.is_regular_file()) {
// Use path API to compute relative path
fs::path relative = entry.path().lexically_relative(base_path_);
object_keys.push_back(relative.string());
}
@ -490,12 +324,14 @@ tl::expected<void, std::string> LocalFileBackend::ListObjectsWithPrefix(
return {};
}
bool LocalFileBackend::IsNotFoundError(const std::string& error) const {
bool LocalFileSnapshotObjectStore::IsNotFoundError(
const std::string& error) const {
return error.starts_with("File not found:");
}
std::string LocalFileBackend::GetConnectionInfo() const {
return fmt::format("LocalFileBackend: base_path={}", base_path_.string());
std::string LocalFileSnapshotObjectStore::GetConnectionInfo() const {
return fmt::format("LocalFileSnapshotObjectStore: base_path={}",
base_path_.string());
}
} // namespace mooncake

95
mooncake-store/src/ha/snapshot/object/backends/s3/s3_snapshot_object_store.cpp Normal file
View File

@ -0,0 +1,95 @@
#include "ha/snapshot/object/backends/s3/s3_snapshot_object_store.h"
#ifdef HAVE_AWS_SDK
#include <algorithm>
#include <cctype>
#include <string_view>
#include <glog/logging.h>
#include "utils/s3_helper.h"
namespace mooncake {
namespace {
bool ContainsAsciiInsensitive(std::string_view haystack,
std::string_view needle) {
return std::search(
haystack.begin(), haystack.end(), needle.begin(), needle.end(),
[](char lhs, char rhs) {
return std::tolower(static_cast<unsigned char>(lhs)) ==
std::tolower(static_cast<unsigned char>(rhs));
}) != haystack.end();
}
} // namespace
class S3SnapshotObjectStore::Impl {
public:
Impl() : initialized_(InitializeOnce()), s3_helper_("", "", "") {}
~Impl() { S3Helper::ShutdownAPI(); }
private:
static bool InitializeOnce() {
S3Helper::InitAPI();
return true;
}
bool initialized_;
public:
S3Helper s3_helper_;
};
S3SnapshotObjectStore::S3SnapshotObjectStore()
: impl_(std::make_unique<Impl>()) {
LOG(INFO) << "S3SnapshotObjectStore initialized";
}
tl::expected<void, std::string> S3SnapshotObjectStore::UploadBuffer(
const std::string& key, const std::vector<uint8_t>& buffer) {
return impl_->s3_helper_.UploadBufferMultipart(key, buffer);
}
tl::expected<void, std::string> S3SnapshotObjectStore::DownloadBuffer(
const std::string& key, std::vector<uint8_t>& buffer) {
return impl_->s3_helper_.DownloadBufferMultipart(key, buffer);
}
tl::expected<void, std::string> S3SnapshotObjectStore::UploadString(
const std::string& key, const std::string& data) {
return impl_->s3_helper_.UploadString(key, data);
}
tl::expected<void, std::string> S3SnapshotObjectStore::DownloadString(
const std::string& key, std::string& data) {
return impl_->s3_helper_.DownloadString(key, data);
}
tl::expected<void, std::string> S3SnapshotObjectStore::DeleteObjectsWithPrefix(
const std::string& prefix) {
return impl_->s3_helper_.DeleteObjectsWithPrefix(prefix);
}
tl::expected<void, std::string> S3SnapshotObjectStore::ListObjectsWithPrefix(
const std::string& prefix, std::vector<std::string>& object_keys) {
return impl_->s3_helper_.ListObjectsWithPrefix(prefix, object_keys);
}
bool S3SnapshotObjectStore::IsNotFoundError(const std::string& error) const {
return ContainsAsciiInsensitive(error, "nosuchkey") ||
ContainsAsciiInsensitive(error, "not found") ||
ContainsAsciiInsensitive(error, "does not exist") ||
ContainsAsciiInsensitive(error, "404");
}
std::string S3SnapshotObjectStore::GetConnectionInfo() const {
return impl_->s3_helper_.GetConnectionInfo();
}
} // namespace mooncake
#endif // HAVE_AWS_SDK

30
mooncake-store/src/ha/snapshot/object/snapshot_object_store.cpp Normal file
View File

@ -0,0 +1,30 @@
#include "ha/snapshot/object/snapshot_object_store.h"
#include "ha/snapshot/object/backends/local/local_file_snapshot_object_store.h"
#ifdef HAVE_AWS_SDK
#include "ha/snapshot/object/backends/s3/s3_snapshot_object_store.h"
#endif
namespace mooncake {
std::unique_ptr<SnapshotObjectStore> SnapshotObjectStore::Create(
SnapshotObjectStoreType type) {
switch (type) {
#ifdef HAVE_AWS_SDK
case SnapshotObjectStoreType::S3:
return std::make_unique<S3SnapshotObjectStore>();
#else
case SnapshotObjectStoreType::S3:
throw std::runtime_error(
"S3 snapshot object store requested but AWS SDK is not "
"available. Please rebuild with HAVE_AWS_SDK or use the "
"'local' object store.");
#endif
case SnapshotObjectStoreType::LOCAL_FILE:
return std::make_unique<LocalFileSnapshotObjectStore>();
default:
throw std::invalid_argument("Unknown snapshot object store type");
}
}
} // namespace mooncake

155
mooncake-store/src/master.cpp
View File

@ -10,7 +10,7 @@
#include "default_config.h"
#include "duration_utils.h"
#include "ha/master_service_supervisor.h"
#include "ha/leadership/master_service_supervisor.h"
#include "http_metadata_server.h"
#include "rpc_service.h"
#include "types.h"
@ -167,13 +167,22 @@ DEFINE_uint32(snapshot_retention_count,
mooncake::DEFAULT_SNAPSHOT_RETENTION_COUNT,
"Number of recent snapshots to keep (older snapshots will be "
"automatically deleted)");
DEFINE_string(snapshot_backend_type, "",
"Snapshot storage backend type: 'local' for local filesystem, "
DEFINE_string(snapshot_object_store_type, "",
"Snapshot object store type: 'local' for local filesystem, "
"'s3' for S3 storage");
DEFINE_string(snapshot_payload_store_type, "",
"Deprecated alias of --snapshot_object_store_type");
DEFINE_string(snapshot_payload_backend_type, "",
"Deprecated alias of --snapshot_object_store_type");
DEFINE_string(snapshot_catalog_store_type, "",
"Snapshot catalog store type: ''/'embedded' or 'redis' "
"('payload' is kept as a deprecated alias)");
DEFINE_string(snapshot_catalog_backend_type, "",
"Snapshot catalog backend type: ''/'serializer' or 'redis'");
"Deprecated alias of --snapshot_catalog_store_type");
DEFINE_string(snapshot_catalog_store_connstring, "",
"Optional connection string for snapshot catalog store");
DEFINE_string(snapshot_catalog_backend_connstring, "",
"Optional connection string for snapshot catalog backend");
"Deprecated alias of --snapshot_catalog_store_connstring");
// Task manager configuration
DEFINE_uint32(max_total_finished_tasks, 10000,
"Maximum number of finished tasks to keep in memory");
@ -311,15 +320,36 @@ void InitMasterConf(const mooncake::DefaultConfig& default_config,
default_config.GetUInt32("snapshot_retention_count",
&master_config.snapshot_retention_count,
FLAGS_snapshot_retention_count);
default_config.GetString("snapshot_backend_type",
&master_config.snapshot_backend_type,
FLAGS_snapshot_backend_type);
default_config.GetString("snapshot_catalog_backend_type",
&master_config.snapshot_catalog_backend_type,
FLAGS_snapshot_catalog_backend_type);
default_config.GetString("snapshot_catalog_backend_connstring",
&master_config.snapshot_catalog_backend_connstring,
FLAGS_snapshot_catalog_backend_connstring);
default_config.GetString("snapshot_object_store_type",
&master_config.snapshot_object_store_type,
FLAGS_snapshot_object_store_type);
if (master_config.snapshot_object_store_type.empty()) {
default_config.GetString("snapshot_payload_store_type",
&master_config.snapshot_object_store_type,
master_config.snapshot_object_store_type);
}
if (master_config.snapshot_object_store_type.empty()) {
default_config.GetString("snapshot_payload_backend_type",
&master_config.snapshot_object_store_type,
master_config.snapshot_object_store_type);
}
default_config.GetString("snapshot_catalog_store_type",
&master_config.snapshot_catalog_store_type,
FLAGS_snapshot_catalog_store_type);
if (master_config.snapshot_catalog_store_type.empty()) {
default_config.GetString("snapshot_catalog_backend_type",
&master_config.snapshot_catalog_store_type,
master_config.snapshot_catalog_store_type);
}
default_config.GetString("snapshot_catalog_store_connstring",
&master_config.snapshot_catalog_store_connstring,
FLAGS_snapshot_catalog_store_connstring);
if (master_config.snapshot_catalog_store_connstring.empty()) {
default_config.GetString(
"snapshot_catalog_backend_connstring",
&master_config.snapshot_catalog_store_connstring,
master_config.snapshot_catalog_store_connstring);
}
default_config.GetUInt32("max_total_finished_tasks",
&master_config.max_total_finished_tasks,
FLAGS_max_total_finished_tasks);
@ -620,24 +650,86 @@ void LoadConfigFromCmdline(mooncake::MasterConfig& master_config,
!conf_set) {
master_config.snapshot_backup_dir = FLAGS_snapshot_backup_dir;
}
if ((google::GetCommandLineFlagInfo("snapshot_backend_type", &info) &&
!info.is_default) ||
!conf_set) {
master_config.snapshot_backend_type = FLAGS_snapshot_backend_type;
bool use_snapshot_object_store_flag = false;
bool use_snapshot_payload_store_flag = false;
bool use_snapshot_payload_backend_flag = false;
if (google::GetCommandLineFlagInfo("snapshot_object_store_type", &info) &&
!info.is_default) {
use_snapshot_object_store_flag = true;
}
if ((google::GetCommandLineFlagInfo("snapshot_catalog_backend_type",
&info) &&
!info.is_default) ||
!conf_set) {
master_config.snapshot_catalog_backend_type =
if (google::GetCommandLineFlagInfo("snapshot_payload_store_type", &info) &&
!info.is_default) {
use_snapshot_payload_store_flag = true;
}
if (google::GetCommandLineFlagInfo("snapshot_payload_backend_type",
&info) &&
!info.is_default) {
use_snapshot_payload_backend_flag = true;
}
if (use_snapshot_object_store_flag) {
master_config.snapshot_object_store_type =
FLAGS_snapshot_object_store_type;
} else if (use_snapshot_payload_store_flag) {
LOG(WARNING) << "--snapshot_payload_store_type is deprecated; use "
<< "--snapshot_object_store_type instead";
master_config.snapshot_object_store_type =
FLAGS_snapshot_payload_store_type;
} else if (use_snapshot_payload_backend_flag) {
LOG(WARNING) << "--snapshot_payload_backend_type is deprecated; use "
<< "--snapshot_object_store_type instead";
master_config.snapshot_object_store_type =
FLAGS_snapshot_payload_backend_type;
} else if (!conf_set) {
master_config.snapshot_object_store_type =
FLAGS_snapshot_object_store_type;
}
bool use_snapshot_catalog_store_flag = false;
bool use_snapshot_catalog_backend_flag = false;
if (google::GetCommandLineFlagInfo("snapshot_catalog_store_type", &info) &&
!info.is_default) {
use_snapshot_catalog_store_flag = true;
}
if (google::GetCommandLineFlagInfo("snapshot_catalog_backend_type",
&info) &&
!info.is_default) {
use_snapshot_catalog_backend_flag = true;
}
if (use_snapshot_catalog_store_flag) {
master_config.snapshot_catalog_store_type =
FLAGS_snapshot_catalog_store_type;
} else if (use_snapshot_catalog_backend_flag) {
LOG(WARNING) << "--snapshot_catalog_backend_type is deprecated; use "
<< "--snapshot_catalog_store_type instead";
master_config.snapshot_catalog_store_type =
FLAGS_snapshot_catalog_backend_type;
} else if (!conf_set) {
master_config.snapshot_catalog_store_type =
FLAGS_snapshot_catalog_store_type;
}
if ((google::GetCommandLineFlagInfo("snapshot_catalog_backend_connstring",
&info) &&
!info.is_default) ||
!conf_set) {
master_config.snapshot_catalog_backend_connstring =
bool use_snapshot_catalog_store_connstring_flag = false;
bool use_snapshot_catalog_backend_connstring_flag = false;
if (google::GetCommandLineFlagInfo("snapshot_catalog_store_connstring",
&info) &&
!info.is_default) {
use_snapshot_catalog_store_connstring_flag = true;
}
if (google::GetCommandLineFlagInfo("snapshot_catalog_backend_connstring",
&info) &&
!info.is_default) {
use_snapshot_catalog_backend_connstring_flag = true;
}
if (use_snapshot_catalog_store_connstring_flag) {
master_config.snapshot_catalog_store_connstring =
FLAGS_snapshot_catalog_store_connstring;
} else if (use_snapshot_catalog_backend_connstring_flag) {
LOG(WARNING)
<< "--snapshot_catalog_backend_connstring is deprecated; use "
<< "--snapshot_catalog_store_connstring instead";
master_config.snapshot_catalog_store_connstring =
FLAGS_snapshot_catalog_backend_connstring;
} else if (!conf_set) {
master_config.snapshot_catalog_store_connstring =
FLAGS_snapshot_catalog_store_connstring;
}
}
@ -778,9 +870,10 @@ int main(int argc, char* argv[]) {
<< ", snapshot_interval_seconds="
<< master_config.snapshot_interval_seconds
<< ", snapshot_backup_dir=" << master_config.snapshot_backup_dir
<< ", snapshot_backend_type=" << master_config.snapshot_backend_type
<< ", snapshot_catalog_backend_type="
<< master_config.snapshot_catalog_backend_type
<< ", snapshot_object_store_type="
<< master_config.snapshot_object_store_type
<< ", snapshot_catalog_store_type="
<< master_config.snapshot_catalog_store_type
<< ", snapshot_retention_count="
<< master_config.snapshot_retention_count
<< ", max_retry_attempts=" << master_config.max_retry_attempts

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

@ -13,14 +13,14 @@
#include "master_metric_manager.h"
#include "segment.h"
#include "ha/backends/redis/redis_snapshot_store.h"
#include "ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.h"
#include "ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.h"
#include "ha/snapshot/object/snapshot_object_store.h"
#include "types.h"
#include "ha/serializer_snapshot_store.h"
#include "serialize/serializer.hpp"
#include "serialize/serializer_backend.h"
#include "ha/snapshot/snapshot_logger.h"
#include "utils/zstd_util.h"
#include "utils/file_util.h"
#include "utils/snapshot_logger.h"
#include "utils.h"
namespace mooncake {
@ -44,20 +44,21 @@ namespace {
constexpr size_t kUnlimitedSnapshotList = 0;
enum class SnapshotCatalogBackendKind {
kSerializer,
kEmbedded,
kRedis,
};
tl::expected<SnapshotCatalogBackendKind, std::string>
ParseSnapshotCatalogBackendKind(std::string_view backend_type) {
if (backend_type.empty() || backend_type == "serializer") {
return SnapshotCatalogBackendKind::kSerializer;
tl::expected<SnapshotCatalogBackendKind, std::string> ParseSnapshotCatalogKind(
std::string_view store_type) {
if (store_type.empty() || store_type == "embedded" ||
store_type == "payload") {
return SnapshotCatalogBackendKind::kEmbedded;
}
if (backend_type == "redis") {
if (store_type == "redis") {
return SnapshotCatalogBackendKind::kRedis;
}
return tl::make_unexpected("unknown snapshot catalog backend type: " +
std::string(backend_type));
return tl::make_unexpected("unknown snapshot catalog store type: " +
std::string(store_type));
}
} // namespace
@ -89,9 +90,9 @@ MasterService::MasterService(const MasterServiceConfig& config)
snapshot_interval_seconds_(config.snapshot_interval_seconds),
snapshot_child_timeout_seconds_(config.snapshot_child_timeout_seconds),
snapshot_retention_count_(config.snapshot_retention_count),
snapshot_catalog_backend_type_(config.snapshot_catalog_backend_type),
snapshot_catalog_backend_connstring_(
config.snapshot_catalog_backend_connstring),
snapshot_catalog_store_type_(config.snapshot_catalog_store_type),
snapshot_catalog_store_connstring_(
config.snapshot_catalog_store_connstring),
put_start_discard_timeout_sec_(config.put_start_discard_timeout_sec),
put_start_release_timeout_sec_(config.put_start_release_timeout_sec),
task_manager_(config.task_manager_config),
@ -100,14 +101,15 @@ MasterService::MasterService(const MasterServiceConfig& config)
enable_cxl_(config.enable_cxl) {
if (enable_snapshot_ || enable_snapshot_restore_) {
try {
auto backend_type =
ParseSnapshotBackendType(config.snapshot_backend_type);
snapshot_backend_ = SerializerBackend::Create(backend_type);
snapshot_store_ = CreateSnapshotStore();
auto object_store_type =
ParseSnapshotObjectStoreType(config.snapshot_object_store_type);
snapshot_object_store_ =
SnapshotObjectStore::Create(object_store_type);
snapshot_catalog_store_ = CreateSnapshotCatalogStore();
} catch (const std::exception& e) {
LOG(ERROR) << "Failed to create snapshot backend: " << e.what();
LOG(ERROR) << "Failed to create snapshot stores: " << e.what();
throw std::runtime_error(
fmt::format("Failed to create snapshot backend: {}", e.what()));
fmt::format("Failed to create snapshot stores: {}", e.what()));
}
if (!snapshot_backup_dir_.empty()) {
use_snapshot_backup_dir_ = true;
@ -181,39 +183,40 @@ MasterService::MasterService(const MasterServiceConfig& config)
}
}
std::unique_ptr<ha::SnapshotStore> MasterService::CreateSnapshotStore() {
auto backend_kind =
ParseSnapshotCatalogBackendKind(snapshot_catalog_backend_type_);
if (!backend_kind) {
throw std::invalid_argument(backend_kind.error());
std::unique_ptr<ha::SnapshotCatalogStore>
MasterService::CreateSnapshotCatalogStore() {
auto catalog_kind = ParseSnapshotCatalogKind(snapshot_catalog_store_type_);
if (!catalog_kind) {
throw std::invalid_argument(catalog_kind.error());
}
switch (backend_kind.value()) {
case SnapshotCatalogBackendKind::kSerializer:
return std::make_unique<ha::SerializerSnapshotStore>(
snapshot_backend_.get());
switch (catalog_kind.value()) {
case SnapshotCatalogBackendKind::kEmbedded:
return std::make_unique<
ha::backends::embedded::EmbeddedSnapshotCatalogStore>(
snapshot_object_store_.get());
case SnapshotCatalogBackendKind::kRedis: {
#ifndef STORE_USE_REDIS
throw std::invalid_argument(
"redis snapshot catalog backend is unavailable in the current "
"redis snapshot catalog store is unavailable in the current "
"build");
#else
const auto connstring =
!snapshot_catalog_backend_connstring_.empty()
? snapshot_catalog_backend_connstring_
: ha_backend_connstring_;
const auto connstring = !snapshot_catalog_store_connstring_.empty()
? snapshot_catalog_store_connstring_
: ha_backend_connstring_;
if (connstring.empty()) {
throw std::invalid_argument(
"redis snapshot catalog backend requires a connection "
"redis snapshot catalog store requires a connection "
"string");
}
return std::make_unique<ha::backends::redis::RedisSnapshotStore>(
snapshot_backend_.get(), connstring, cluster_id_);
return std::make_unique<
ha::backends::redis::RedisSnapshotCatalogStore>(
snapshot_object_store_.get(), connstring, cluster_id_);
#endif
}
}
throw std::invalid_argument("unknown snapshot catalog backend type");
throw std::invalid_argument("unknown snapshot catalog store type");
}
MasterService::~MasterService() {
@ -2237,11 +2240,11 @@ void MasterService::HandleChildExit(pid_t pid, int status,
tl::expected<void, SerializationError> MasterService::PersistState(
const std::string& snapshot_id) {
try {
auto* snapshot_store = GetSnapshotStore();
if (!snapshot_store) {
return tl::make_unexpected(
SerializationError(ErrorCode::PERSISTENT_FAIL,
"snapshot backend is not initialized"));
auto* snapshot_catalog_store = GetSnapshotCatalogStore();
if (!snapshot_catalog_store) {
return tl::make_unexpected(SerializationError(
ErrorCode::PERSISTENT_FAIL,
"snapshot catalog store is not initialized"));
}
SNAP_LOG_INFO(
@ -2303,13 +2306,13 @@ tl::expected<void, SerializationError> MasterService::PersistState(
bool upload_success = true;
std::string error_msg;
SNAP_LOG_INFO("[Snapshot] Backend info: {}",
snapshot_backend_->GetConnectionInfo());
snapshot_object_store_->GetConnectionInfo());
// Upload metadata
std::string metadata_path = path_prefix + SNAPSHOT_METADATA_FILE;
auto upload_result =
UploadSnapshotFile(serialized_metadata, metadata_path,
SNAPSHOT_METADATA_FILE, snapshot_id);
UploadSnapshotPayloadFile(serialized_metadata, metadata_path,
SNAPSHOT_METADATA_FILE, snapshot_id);
if (!upload_result) {
SNAP_LOG_ERROR(
"[Snapshot] metadata upload failed, snapshot_id={}, "
@ -2326,8 +2329,9 @@ tl::expected<void, SerializationError> MasterService::PersistState(
// Upload segment
std::string segment_path = path_prefix + SNAPSHOT_SEGMENTS_FILE;
upload_result = UploadSnapshotFile(serialized_segment, segment_path,
SNAPSHOT_SEGMENTS_FILE, snapshot_id);
upload_result =
UploadSnapshotPayloadFile(serialized_segment, segment_path,
SNAPSHOT_SEGMENTS_FILE, snapshot_id);
if (!upload_result) {
SNAP_LOG_ERROR(
"[Snapshot] segment upload failed, snapshot_id={}, "
@ -2344,9 +2348,9 @@ tl::expected<void, SerializationError> MasterService::PersistState(
// Upload task manager
std::string task_manager_path =
path_prefix + SNAPSHOT_TASK_MANAGER_FILE;
upload_result =
UploadSnapshotFile(serialized_task_manager, task_manager_path,
SNAPSHOT_TASK_MANAGER_FILE, snapshot_id);
upload_result = UploadSnapshotPayloadFile(
serialized_task_manager, task_manager_path,
SNAPSHOT_TASK_MANAGER_FILE, snapshot_id);
if (!upload_result) {
SNAP_LOG_ERROR(
"[Snapshot] task_manager upload failed, snapshot_id={}, "
@ -2368,8 +2372,8 @@ tl::expected<void, SerializationError> MasterService::PersistState(
SNAPSHOT_SERIALIZER_VERSION, snapshot_id);
std::vector<uint8_t> manifest_bytes(manifest_content.begin(),
manifest_content.end());
upload_result = UploadSnapshotFile(manifest_bytes, manifest_path,
SNAPSHOT_MANIFEST_FILE, snapshot_id);
upload_result = UploadSnapshotPayloadFile(
manifest_bytes, manifest_path, SNAPSHOT_MANIFEST_FILE, snapshot_id);
if (!upload_result) {
SNAP_LOG_ERROR(
"[Snapshot] manifest upload failed, snapshot_id={}, "
@ -2397,7 +2401,7 @@ tl::expected<void, SerializationError> MasterService::PersistState(
descriptor.snapshot_id = snapshot_id;
descriptor.manifest_key = manifest_path;
descriptor.object_prefix = path_prefix;
auto publish_result = snapshot_store->Publish(descriptor);
auto publish_result = snapshot_catalog_store->Publish(descriptor);
if (publish_result != ErrorCode::OK) {
SNAP_LOG_ERROR(
"[Snapshot] latest update failed, snapshot_id={}, file={}, "
@ -2450,14 +2454,14 @@ tl::expected<void, SerializationError> MasterService::PersistState(
return {};
}
tl::expected<void, SerializationError> MasterService::UploadSnapshotFile(
tl::expected<void, SerializationError> MasterService::UploadSnapshotPayloadFile(
const std::vector<uint8_t>& data, const std::string& path,
const std::string& local_filename, const std::string& snapshot_id) {
SNAP_LOG_INFO("[Snapshot] Uploading {} to: {}, snapshot_id={}",
local_filename, path, snapshot_id);
std::string error_msg;
auto upload_result = snapshot_backend_->UploadBuffer(path, data);
auto upload_result = snapshot_object_store_->UploadBuffer(path, data);
if (!upload_result) {
SNAP_LOG_ERROR(
"[Snapshot] {} upload failed, snapshot_id={}, file={}, error={}",
@ -2493,16 +2497,16 @@ tl::expected<void, SerializationError> MasterService::UploadSnapshotFile(
void MasterService::CleanupOldSnapshot(int keep_count,
const std::string& snapshot_id) {
auto* snapshot_store = GetSnapshotStore();
if (!snapshot_store) {
auto* snapshot_catalog_store = GetSnapshotCatalogStore();
if (!snapshot_catalog_store) {
SNAP_LOG_ERROR(
"[Snapshot] snapshot store is not initialized, "
"[Snapshot] snapshot catalog store is not initialized, "
"snapshot_id={}",
snapshot_id);
return;
}
auto list_result = snapshot_store->List(kUnlimitedSnapshotList);
auto list_result = snapshot_catalog_store->List(kUnlimitedSnapshotList);
if (!list_result) {
SNAP_LOG_ERROR("[Snapshot] error=list failed, snapshot_id={}, code={}",
snapshot_id, toString(list_result.error()));
@ -2524,15 +2528,15 @@ void MasterService::CleanupOldSnapshot(int keep_count,
continue;
}
auto delete_result = snapshot_store->Delete(old_state_dir);
auto delete_result = snapshot_catalog_store->Delete(old_state_dir);
if (delete_result != ErrorCode::OK) {
SNAP_LOG_ERROR(
"[Snapshot] Failed to delete old state directory {}, "
"[Snapshot] Failed to delete old snapshot {}, "
"snapshot_id={}, code={}",
old_state_dir, snapshot_id, toString(delete_result));
} else {
SNAP_LOG_INFO(
"[Snapshot] Successfully deleted old state directory {}, "
"[Snapshot] Successfully deleted old snapshot {}, "
"snapshot_id={}",
old_state_dir, snapshot_id);
}
@ -2542,19 +2546,19 @@ void MasterService::CleanupOldSnapshot(int keep_count,
void MasterService::RestoreState() {
try {
auto* snapshot_store = GetSnapshotStore();
if (!snapshot_store) {
LOG(ERROR) << "[Restore] Snapshot backend is not initialized, "
<< "starting fresh";
auto* snapshot_catalog_store = GetSnapshotCatalogStore();
if (!snapshot_catalog_store) {
LOG(ERROR) << "[Restore] Snapshot catalog store is not "
"initialized, starting fresh";
return;
}
auto now = std::chrono::system_clock::now();
LOG(INFO) << "[Restore] Backend info: "
<< snapshot_backend_->GetConnectionInfo();
<< snapshot_object_store_->GetConnectionInfo();
// 1. Resolve the latest snapshot from the snapshot catalog.
auto latest_result = snapshot_store->GetLatest();
auto latest_result = snapshot_catalog_store->GetLatest();
if (!latest_result) {
LOG(ERROR) << "[Restore] Failed to load latest snapshot: "
<< toString(latest_result.error()) << ", starting fresh";
@ -2579,8 +2583,8 @@ void MasterService::RestoreState() {
manifest_path = path_prefix + SNAPSHOT_MANIFEST_FILE;
}
std::string manifest_content;
if (!snapshot_backend_->DownloadString(manifest_path,
manifest_content)) {
if (!snapshot_object_store_->DownloadString(manifest_path,
manifest_content)) {
LOG(ERROR) << "[Restore] Failed to download manifest file: "
<< manifest_path << " , starting fresh";
return;
@ -2635,8 +2639,8 @@ void MasterService::RestoreState() {
// 3. Download metadata
std::string metadata_path = path_prefix + SNAPSHOT_METADATA_FILE;
std::vector<uint8_t> metadata_content;
auto download_result =
snapshot_backend_->DownloadBuffer(metadata_path, metadata_content);
auto download_result = snapshot_object_store_->DownloadBuffer(
metadata_path, metadata_content);
if (!download_result) {
LOG(ERROR) << "[Restore] Failed to download metadata file: "
<< metadata_path << "error=" << download_result.error();
@ -2658,8 +2662,8 @@ void MasterService::RestoreState() {
// 4. Download segments
std::string segments_path = path_prefix + SNAPSHOT_SEGMENTS_FILE;
std::vector<uint8_t> segments_content;
download_result =
snapshot_backend_->DownloadBuffer(segments_path, segments_content);
download_result = snapshot_object_store_->DownloadBuffer(
segments_path, segments_content);
if (!download_result) {
LOG(ERROR) << "Failed to download segments file: " << segments_path
<< " error=" << download_result.error();
@ -2681,7 +2685,7 @@ void MasterService::RestoreState() {
std::string task_manager_path =
path_prefix + SNAPSHOT_TASK_MANAGER_FILE;
std::vector<uint8_t> task_manager_content;
download_result = snapshot_backend_->DownloadBuffer(
download_result = snapshot_object_store_->DownloadBuffer(
task_manager_path, task_manager_content);
if (!download_result) {
LOG(ERROR) << "Failed to download task manager file: "
@ -2881,8 +2885,8 @@ void MasterService::RestoreState() {
}
}
ha::SnapshotStore* MasterService::GetSnapshotStore() {
return snapshot_store_.get();
ha::SnapshotCatalogStore* MasterService::GetSnapshotCatalogStore() {
return snapshot_catalog_store_.get();
}
void MasterService::BatchEvict(double evict_ratio_target,

33
mooncake-store/tests/CMakeLists.txt
View File

@ -1,5 +1,6 @@
function(add_store_test name)
add_executable(${name} ${ARGN})
target_include_directories(${name} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
target_link_libraries(
${name}
PUBLIC mooncake_store
@ -16,6 +17,7 @@ endfunction()
function(add_hot_standby_ut_test name)
add_executable(${name} ${ARGN})
target_include_directories(${name} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
target_link_libraries(
${name}
PUBLIC mooncake_store
@ -37,7 +39,7 @@ add_store_test(eviction_strategy_test eviction_strategy_test.cpp)
add_store_test(master_service_test master_service_test.cpp)
add_store_test(master_service_ssd_test master_service_ssd_test.cpp)
add_store_test(master_service_ssd_test_for_snapshot
master_service_ssd_test_for_snapshot.cpp)
ha/snapshot/master_service_ssd_test_for_snapshot.cpp)
add_store_test(client_integration_test client_integration_test.cpp)
if(USE_CXL)
add_store_test(cxl_client_integration_test cxl_client_integration_test.cpp)
@ -55,14 +57,16 @@ add_store_test(pybind_client_test pybind_client_test.cpp)
add_store_test(ipv6_client_test ipv6_client_test.cpp)
add_store_test(client_metrics_test client_metrics_test.cpp)
add_store_test(serializer_test serializer_test.cpp)
add_store_test(serializer_snapshot_store_test
serializer_snapshot_store_test.cpp)
add_store_test(embedded_snapshot_catalog_store_test
ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store_test.cpp)
add_store_test(zstd_util_test zstd_util_test.cpp)
add_store_test(local_file_backend_test local_file_backend_test.cpp)
add_store_test(local_file_snapshot_object_store_test
ha/snapshot/object/backends/local/local_file_snapshot_object_store_test.cpp)
add_store_test(file_util_test file_util_test.cpp)
add_store_test(snapshot_child_process_test snapshot_child_process_test.cpp)
add_store_test(snapshot_child_process_test
ha/snapshot/snapshot_child_process_test.cpp)
add_store_test(master_service_test_for_snapshot
master_service_test_for_snapshot.cpp)
ha/snapshot/master_service_test_for_snapshot.cpp)
add_store_test(non_ha_reconnect_test non_ha_reconnect_test.cpp)
add_store_test(storage_backend_test storage_backend_test.cpp)
add_store_test(mutex_test mutex_test.cpp)
@ -74,23 +78,28 @@ add_store_test(dummy_client_get_buffer_test dummy_client_get_buffer_test.cpp)
add_store_test(health_check_test health_check_test.cpp)
add_subdirectory(e2e)
add_executable(high_availability_test high_availability_test.cpp)
add_executable(high_availability_test ha/leadership/high_availability_test.cpp)
target_include_directories(high_availability_test PRIVATE
${CMAKE_CURRENT_SOURCE_DIR})
if(STORE_USE_REDIS)
target_sources(high_availability_test PRIVATE
high_availability_redis_test.cpp)
ha/leadership/backends/redis/high_availability_redis_test.cpp)
target_include_directories(high_availability_test PRIVATE
${MOONCAKE_STORE_HIREDIS_INCLUDE_DIR})
target_link_libraries(high_availability_test PRIVATE
${MOONCAKE_STORE_HIREDIS_LIBRARY})
add_executable(redis_snapshot_store_test redis_snapshot_store_test.cpp)
target_include_directories(redis_snapshot_store_test PRIVATE
add_executable(redis_snapshot_catalog_store_test
ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store_test.cpp)
target_include_directories(redis_snapshot_catalog_store_test PRIVATE
${CMAKE_CURRENT_SOURCE_DIR}
${MOONCAKE_STORE_HIREDIS_INCLUDE_DIR})
target_link_libraries(redis_snapshot_store_test
target_link_libraries(redis_snapshot_catalog_store_test
PUBLIC mooncake_store transfer_engine cachelib_memory_allocator
${ETCD_WRAPPER_LIB} glog gflags ibverbs gtest pthread
PRIVATE ${MOONCAKE_STORE_HIREDIS_LIBRARY})
add_test(NAME redis_snapshot_store_test COMMAND redis_snapshot_store_test)
add_test(NAME redis_snapshot_catalog_store_test
COMMAND redis_snapshot_catalog_store_test)
endif()
target_link_libraries(high_availability_test PUBLIC mooncake_store transfer_engine cachelib_memory_allocator ${ETCD_WRAPPER_LIB} glog gtest gtest_main pthread)
if (STORE_USE_ETCD OR STORE_USE_REDIS)

2
mooncake-store/tests/e2e/chaos_rand_test.cpp
View File

@ -8,7 +8,7 @@
#include "client_wrapper.h"
#include "e2e_utils.h"
#include "ha/ha_backend_factory.h"
#include "ha/leadership/leader_coordinator_factory.h"
#include "process_handler.h"
#include "types.h"
#include "utils.h"

2
mooncake-store/tests/e2e/chaos_test.cpp
View File

@ -8,7 +8,7 @@
#include "client_wrapper.h"
#include "e2e_utils.h"
#include "ha/ha_backend_factory.h"
#include "ha/leadership/leader_coordinator_factory.h"
#include "process_handler.h"
#include "types.h"
#include "utils.h"

10
mooncake-store/tests/redis_test_utils.h → mooncake-store/tests/ha/common/redis/redis_test_utils.h
View File

@ -6,24 +6,24 @@
#include <hiredis/hiredis.h>
#include <ylt/util/tl/expected.hpp>
#include "ha/backends/redis/redis_client_helper.h"
#include "ha/common/redis/redis_connection.h"
#include "ha/ha_types.h"
#include "types.h"
namespace mooncake {
namespace testing {
using RedisContextPtr = ha::backends::redis::RedisContextPtr;
using RedisReplyPtr = ha::backends::redis::RedisReplyPtr;
using RedisContextPtr = ha::common::redis::RedisContextPtr;
using RedisReplyPtr = ha::common::redis::RedisReplyPtr;
inline tl::expected<RedisContextPtr, ErrorCode> ConnectRedisForTest(
std::string_view endpoint) {
return ha::backends::redis::ConnectRedis(endpoint);
return ha::common::redis::ConnectRedis(endpoint);
}
inline std::string BuildRedisScopedKey(
const ha::ClusterNamespace& cluster_namespace, std::string_view suffix) {
const auto hash_tag = ha::backends::redis::SanitizeHashTagComponent(
const auto hash_tag = ha::common::redis::SanitizeHashTagComponent(
std::string(cluster_namespace));
return "mooncake-store/{" + hash_tag + "}/" + std::string(suffix);
}

6
mooncake-store/tests/high_availability_redis_test.cpp → mooncake-store/tests/ha/leadership/backends/redis/high_availability_redis_test.cpp
View File

@ -12,9 +12,9 @@
#include <hiredis/hiredis.h>
#include "ha/ha_backend_factory.h"
#include "high_availability_test_fixture.h"
#include "redis_test_utils.h"
#include "ha/leadership/leader_coordinator_factory.h"
#include "ha/common/redis/redis_test_utils.h"
#include "ha/leadership/high_availability_test_fixture.h"
#include "types.h"
namespace mooncake {

4
mooncake-store/tests/high_availability_test.cpp → mooncake-store/tests/ha/leadership/high_availability_test.cpp
View File

@ -11,8 +11,8 @@
#ifdef STORE_USE_ETCD
#include "etcd_helper.h"
#endif
#include "ha/ha_backend_factory.h"
#include "high_availability_test_fixture.h"
#include "ha/leadership/leader_coordinator_factory.h"
#include "ha/leadership/high_availability_test_fixture.h"
#include "types.h"
namespace mooncake {

0
mooncake-store/tests/high_availability_test_fixture.h → mooncake-store/tests/ha/leadership/high_availability_test_fixture.h
View File

32
mooncake-store/tests/serializer_snapshot_store_test.cpp → mooncake-store/tests/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store_test.cpp
View File

@ -7,11 +7,11 @@
#include <utility>
#include <vector>
#include "ha/serializer_snapshot_store.h"
#include "ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.h"
namespace mooncake::test {
class FakeSerializerBackend final : public SerializerBackend {
class FakeSnapshotObjectStore final : public SnapshotObjectStore {
public:
void SetDownloadStringError(std::string error) {
download_string_error_ = std::move(error);
@ -87,7 +87,7 @@ class FakeSerializerBackend final : public SerializerBackend {
std::unordered_map<std::string, std::string> objects_;
};
class SerializerSnapshotStoreTest : public ::testing::Test {
class EmbeddedSnapshotCatalogStoreTest : public ::testing::Test {
protected:
static ha::SnapshotDescriptor MakeDescriptor(
const std::string& snapshot_id) {
@ -105,11 +105,11 @@ class SerializerSnapshotStoreTest : public ::testing::Test {
ASSERT_TRUE(result.has_value()) << result.error();
}
FakeSerializerBackend backend_;
ha::SerializerSnapshotStore store_{&backend_};
FakeSnapshotObjectStore backend_;
ha::backends::embedded::EmbeddedSnapshotCatalogStore store_{&backend_};
};
TEST_F(SerializerSnapshotStoreTest, PublishAndGetLatestRoundTrip) {
TEST_F(EmbeddedSnapshotCatalogStoreTest, PublishAndGetLatestRoundTrip) {
auto publish_result = store_.Publish(MakeDescriptor("20240301_120000_001"));
ASSERT_EQ(publish_result, ErrorCode::OK);
@ -123,13 +123,15 @@ TEST_F(SerializerSnapshotStoreTest, PublishAndGetLatestRoundTrip) {
"mooncake_master_snapshot/20240301_120000_001/");
}
TEST_F(SerializerSnapshotStoreTest, GetLatestReturnsEmptyWhenMarkerMissing) {
TEST_F(EmbeddedSnapshotCatalogStoreTest,
GetLatestReturnsEmptyWhenMarkerMissing) {
auto latest = store_.GetLatest();
ASSERT_TRUE(latest.has_value());
EXPECT_FALSE(latest->has_value());
}
TEST_F(SerializerSnapshotStoreTest, GetLatestReturnsErrorOnBackendReadFailure) {
TEST_F(EmbeddedSnapshotCatalogStoreTest,
GetLatestReturnsErrorOnBackendReadFailure) {
backend_.SetDownloadStringError("permission denied");
auto latest = store_.GetLatest();
@ -137,7 +139,7 @@ TEST_F(SerializerSnapshotStoreTest, GetLatestReturnsErrorOnBackendReadFailure) {
EXPECT_EQ(latest.error(), ErrorCode::PERSISTENT_FAIL);
}
TEST_F(SerializerSnapshotStoreTest, GetLatestTrimsWhitespaceMarker) {
TEST_F(EmbeddedSnapshotCatalogStoreTest, GetLatestTrimsWhitespaceMarker) {
PutObject("mooncake_master_snapshot/latest.txt",
" \n20240301_120000_002\t\r\n");
@ -147,7 +149,8 @@ TEST_F(SerializerSnapshotStoreTest, GetLatestTrimsWhitespaceMarker) {
EXPECT_EQ(latest->value().snapshot_id, "20240301_120000_002");
}
TEST_F(SerializerSnapshotStoreTest, ListReturnsSnapshotsInDescendingOrder) {
TEST_F(EmbeddedSnapshotCatalogStoreTest,
ListReturnsSnapshotsInDescendingOrder) {
PutObject("mooncake_master_snapshot/20240301_120000_001/manifest.txt",
"m1");
PutObject("mooncake_master_snapshot/20240303_120000_001/metadata", "d3");
@ -162,7 +165,7 @@ TEST_F(SerializerSnapshotStoreTest, ListReturnsSnapshotsInDescendingOrder) {
EXPECT_EQ(snapshots->at(1).snapshot_id, "20240302_120000_001");
}
TEST_F(SerializerSnapshotStoreTest, DeleteRemovesSnapshotObjectsByPrefix) {
TEST_F(EmbeddedSnapshotCatalogStoreTest, DeleteRemovesSnapshotObjectsByPrefix) {
PutObject("mooncake_master_snapshot/20240301_120000_001/manifest.txt",
"m1");
PutObject("mooncake_master_snapshot/20240301_120000_001/metadata", "d1");
@ -178,7 +181,8 @@ TEST_F(SerializerSnapshotStoreTest, DeleteRemovesSnapshotObjectsByPrefix) {
EXPECT_EQ(snapshots->at(0).snapshot_id, "20240302_120000_001");
}
TEST_F(SerializerSnapshotStoreTest, DeleteLatestFallsBackToPreviousSnapshot) {
TEST_F(EmbeddedSnapshotCatalogStoreTest,
DeleteLatestFallsBackToPreviousSnapshot) {
PutObject("mooncake_master_snapshot/20240301_120000_001/manifest.txt",
"m1");
PutObject("mooncake_master_snapshot/20240302_120000_001/manifest.txt",
@ -196,7 +200,7 @@ TEST_F(SerializerSnapshotStoreTest, DeleteLatestFallsBackToPreviousSnapshot) {
EXPECT_EQ(latest->value().snapshot_id, "20240301_120000_001");
}
TEST_F(SerializerSnapshotStoreTest, DeleteLastSnapshotClearsLatestMarker) {
TEST_F(EmbeddedSnapshotCatalogStoreTest, DeleteLastSnapshotClearsLatestMarker) {
PutObject("mooncake_master_snapshot/20240301_120000_001/manifest.txt",
"m1");
ASSERT_EQ(store_.Publish(MakeDescriptor("20240301_120000_001")),
@ -209,7 +213,7 @@ TEST_F(SerializerSnapshotStoreTest, DeleteLastSnapshotClearsLatestMarker) {
EXPECT_FALSE(latest->has_value());
}
TEST_F(SerializerSnapshotStoreTest, RejectsInvalidSnapshotIds) {
TEST_F(EmbeddedSnapshotCatalogStoreTest, RejectsInvalidSnapshotIds) {
EXPECT_EQ(store_.Publish(MakeDescriptor("invalid-id")),
ErrorCode::INVALID_PARAMS);
EXPECT_EQ(store_.Delete("invalid-id"), ErrorCode::INVALID_PARAMS);

31
mooncake-store/tests/redis_snapshot_store_test.cpp → mooncake-store/tests/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store_test.cpp
View File

@ -7,9 +7,8 @@
#include <hiredis/hiredis.h>
#include "ha/backends/redis/redis_snapshot_store.h"
#include "redis_test_utils.h"
#include "serialize/serializer_backend.h"
#include "ha/common/redis/redis_test_utils.h"
#include "ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.h"
#include "types.h"
namespace mooncake::test {
@ -19,7 +18,7 @@ DEFINE_string(redis_endpoint, "",
namespace {
class FakePayloadBackend final : public SerializerBackend {
class FakeObjectStore final : public SnapshotObjectStore {
public:
tl::expected<void, std::string> UploadBuffer(
const std::string& key, const std::vector<uint8_t>& buffer) override {
@ -63,7 +62,7 @@ class FakePayloadBackend final : public SerializerBackend {
return {};
}
std::string GetConnectionInfo() const override { return "fake://payload"; }
std::string GetConnectionInfo() const override { return "fake://object"; }
std::vector<std::string> deleted_prefixes;
};
@ -77,7 +76,7 @@ ha::SnapshotDescriptor MakeDescriptor(const std::string& snapshot_id) {
return descriptor;
}
class RedisSnapshotStoreTest : public ::testing::Test {
class RedisSnapshotCatalogStoreTest : public ::testing::Test {
protected:
void SetUp() override {
if (FLAGS_redis_endpoint.empty()) {
@ -85,8 +84,9 @@ class RedisSnapshotStoreTest : public ::testing::Test {
}
cluster_namespace_ =
"snapshot-redis-test-" + UuidToString(generate_uuid());
store_ = std::make_unique<ha::backends::redis::RedisSnapshotStore>(
&payload_backend_, FLAGS_redis_endpoint, cluster_namespace_);
store_ =
std::make_unique<ha::backends::redis::RedisSnapshotCatalogStore>(
&object_store_, FLAGS_redis_endpoint, cluster_namespace_);
}
void TearDown() override {
@ -108,18 +108,18 @@ class RedisSnapshotStoreTest : public ::testing::Test {
ASSERT_NE(reply->type, REDIS_REPLY_ERROR);
}
FakePayloadBackend payload_backend_;
FakeObjectStore object_store_;
std::string cluster_namespace_;
std::unique_ptr<ha::backends::redis::RedisSnapshotStore> store_;
std::unique_ptr<ha::backends::redis::RedisSnapshotCatalogStore> store_;
};
TEST_F(RedisSnapshotStoreTest, GetLatestReturnsEmptyWhenCatalogMissing) {
TEST_F(RedisSnapshotCatalogStoreTest, GetLatestReturnsEmptyWhenCatalogMissing) {
auto latest = store_->GetLatest();
ASSERT_TRUE(latest.has_value());
EXPECT_FALSE(latest->has_value());
}
TEST_F(RedisSnapshotStoreTest, PublishListAndGetLatestRoundTrip) {
TEST_F(RedisSnapshotCatalogStoreTest, PublishListAndGetLatestRoundTrip) {
ASSERT_EQ(store_->Publish(MakeDescriptor("20240301_120000_001")),
ErrorCode::OK);
ASSERT_EQ(store_->Publish(MakeDescriptor("20240302_120000_001")),
@ -139,7 +139,8 @@ TEST_F(RedisSnapshotStoreTest, PublishListAndGetLatestRoundTrip) {
EXPECT_EQ(snapshots->at(1).snapshot_id, "20240301_120000_001");
}
TEST_F(RedisSnapshotStoreTest, DeleteUpdatesLatestAndDeletesPayloadPrefix) {
TEST_F(RedisSnapshotCatalogStoreTest,
DeleteUpdatesLatestAndDeletesPayloadPrefix) {
ASSERT_EQ(store_->Publish(MakeDescriptor("20240301_120000_001")),
ErrorCode::OK);
ASSERT_EQ(store_->Publish(MakeDescriptor("20240302_120000_001")),
@ -157,8 +158,8 @@ TEST_F(RedisSnapshotStoreTest, DeleteUpdatesLatestAndDeletesPayloadPrefix) {
ASSERT_EQ(snapshots->size(), 1u);
EXPECT_EQ(snapshots->at(0).snapshot_id, "20240301_120000_001");
ASSERT_EQ(payload_backend_.deleted_prefixes.size(), 1u);
EXPECT_EQ(payload_backend_.deleted_prefixes.front(),
ASSERT_EQ(object_store_.deleted_prefixes.size(), 1u);
EXPECT_EQ(object_store_.deleted_prefixes.front(),
"mooncake_master_snapshot/20240302_120000_001/");
}

0
mooncake-store/tests/master_service_ssd_test_for_snapshot.cpp → mooncake-store/tests/ha/snapshot/master_service_ssd_test_for_snapshot.cpp
View File

0
mooncake-store/tests/master_service_test_for_snapshot.cpp → mooncake-store/tests/ha/snapshot/master_service_test_for_snapshot.cpp
View File

22
mooncake-store/tests/master_service_test_for_snapshot_base.h → mooncake-store/tests/ha/snapshot/master_service_test_for_snapshot_base.h
View File

@ -3,8 +3,8 @@
#include "master_service.h"
#include "master_metric_manager.h"
#include "segment.h"
#include "ha/snapshot_store.h"
#include "serialize/serializer_backend.h"
#include "ha/snapshot/catalog/snapshot_catalog_store.h"
#include "ha/snapshot/object/snapshot_object_store.h"
#include "task_manager.h"
#include <glog/logging.h>
@ -61,7 +61,7 @@ class MasterServiceSnapshotTestBase : public ::testing::Test {
ASSERT_NE(dir, nullptr) << "Failed to create temp directory";
tmp_dir_ = dir;
// Set MOONCAKE_SNAPSHOT_LOCAL_PATH for LocalFileBackend
// Set MOONCAKE_SNAPSHOT_LOCAL_PATH for LocalFileSnapshotObjectStore
::setenv(kEnvSnapshotLocalPath, tmp_dir().c_str(), 1);
}
@ -723,7 +723,7 @@ class MasterServiceSnapshotTestBase : public ::testing::Test {
MasterServiceConfig::builder()
.set_memory_allocator(BufferAllocatorType::OFFSET)
.set_enable_snapshot_restore(true)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.set_root_fs_dir(service->root_fs_dir_)
.build();
std::unique_ptr<MasterService> restored_service(
@ -784,16 +784,18 @@ class MasterServiceSnapshotTestBase : public ::testing::Test {
<< "Use 'service_.reset(new MasterService(...))' instead of "
"'std::unique_ptr<MasterService> service_(...)'";
// Ensure snapshot_backend_ is initialized for PersistState
// Ensure snapshot_object_store_ is initialized for PersistState
// Some test configs may not enable snapshot/restore, so the backend
// is not created in the constructor. We create it here for TearDown
// validation.
if (!service_->snapshot_backend_) {
service_->snapshot_backend_ =
SerializerBackend::Create(SnapshotBackendType::LOCAL_FILE);
if (!service_->snapshot_object_store_) {
service_->snapshot_object_store_ = SnapshotObjectStore::Create(
SnapshotObjectStoreType::LOCAL_FILE);
}
if (!service_->snapshot_store_ && service_->snapshot_backend_) {
service_->snapshot_store_ = service_->CreateSnapshotStore();
if (!service_->snapshot_catalog_store_ &&
service_->snapshot_object_store_) {
service_->snapshot_catalog_store_ =
service_->CreateSnapshotCatalogStore();
}
// Test snapshot and restore functionality for all test cases

38
mooncake-store/tests/local_file_backend_test.cpp → mooncake-store/tests/ha/snapshot/object/backends/local/local_file_snapshot_object_store_test.cpp
View File

@ -6,31 +6,31 @@
#include <string>
#include <vector>
#include "serialize/serializer_backend.h"
#include "ha/snapshot/object/backends/local/local_file_snapshot_object_store.h"
namespace mooncake::test {
namespace fs = std::filesystem;
class LocalFileBackendTest : public ::testing::Test {
class LocalFileSnapshotObjectStoreTest : public ::testing::Test {
protected:
const std::string& tmp_dir() const { return tmp_dir_; }
std::unique_ptr<LocalFileBackend> backend_;
std::unique_ptr<LocalFileSnapshotObjectStore> backend_;
void SetUp() override {
google::InitGoogleLogging("LocalFileBackendTest");
google::InitGoogleLogging("LocalFileSnapshotObjectStoreTest");
FLAGS_logtostderr = true;
// Create a unique temporary directory
std::string tmpl =
(fs::temp_directory_path() / "local_file_backend_test_XXXXXX")
.string();
std::string tmpl = (fs::temp_directory_path() /
"local_file_snapshot_object_store_test_XXXXXX")
.string();
char* dir = mkdtemp(tmpl.data());
ASSERT_NE(dir, nullptr) << "Failed to create temp directory";
tmp_dir_ = dir;
backend_ = std::make_unique<LocalFileBackend>(tmp_dir());
backend_ = std::make_unique<LocalFileSnapshotObjectStore>(tmp_dir());
}
void TearDown() override {
@ -47,7 +47,7 @@ class LocalFileBackendTest : public ::testing::Test {
// ========== Normal Functionality ==========
TEST_F(LocalFileBackendTest, UploadDownloadBuffer_Roundtrip) {
TEST_F(LocalFileSnapshotObjectStoreTest, UploadDownloadBuffer_Roundtrip) {
std::vector<uint8_t> data = {0, 1, 2, 128, 254, 255};
auto upload_result = backend_->UploadBuffer("test/buf", data);
ASSERT_TRUE(upload_result.has_value()) << upload_result.error();
@ -58,7 +58,7 @@ TEST_F(LocalFileBackendTest, UploadDownloadBuffer_Roundtrip) {
EXPECT_EQ(downloaded, data);
}
TEST_F(LocalFileBackendTest, UploadDownloadString_Roundtrip) {
TEST_F(LocalFileSnapshotObjectStoreTest, UploadDownloadString_Roundtrip) {
std::string data = "hello mooncake snapshot";
auto upload_result = backend_->UploadString("test/str", data);
ASSERT_TRUE(upload_result.has_value()) << upload_result.error();
@ -69,7 +69,7 @@ TEST_F(LocalFileBackendTest, UploadDownloadString_Roundtrip) {
EXPECT_EQ(downloaded, data);
}
TEST_F(LocalFileBackendTest, ListObjectsWithPrefix) {
TEST_F(LocalFileSnapshotObjectStoreTest, ListObjectsWithPrefix) {
// Upload several files under the same prefix
backend_->UploadString("snap/20240101/metadata", "m");
backend_->UploadString("snap/20240101/segments", "s");
@ -87,7 +87,7 @@ TEST_F(LocalFileBackendTest, ListObjectsWithPrefix) {
EXPECT_EQ(keys.size(), 3u);
}
TEST_F(LocalFileBackendTest, DeleteObjectsWithPrefix) {
TEST_F(LocalFileSnapshotObjectStoreTest, DeleteObjectsWithPrefix) {
backend_->UploadString("snap/20240101/metadata", "m");
backend_->UploadString("snap/20240101/segments", "s");
@ -100,12 +100,12 @@ TEST_F(LocalFileBackendTest, DeleteObjectsWithPrefix) {
EXPECT_FALSE(dl.has_value());
}
TEST_F(LocalFileBackendTest, GetConnectionInfo) {
TEST_F(LocalFileSnapshotObjectStoreTest, GetConnectionInfo) {
auto info = backend_->GetConnectionInfo();
EXPECT_NE(info.find(tmp_dir()), std::string::npos);
}
TEST_F(LocalFileBackendTest, UploadBuffer_CreatesSubdirectories) {
TEST_F(LocalFileSnapshotObjectStoreTest, UploadBuffer_CreatesSubdirectories) {
std::vector<uint8_t> data = {42};
auto result = backend_->UploadBuffer("a/b/c/deep_file", data);
ASSERT_TRUE(result.has_value()) << result.error();
@ -118,23 +118,23 @@ TEST_F(LocalFileBackendTest, UploadBuffer_CreatesSubdirectories) {
// ========== Error Handling ==========
TEST_F(LocalFileBackendTest, Constructor_EmptyPath_Throws) {
EXPECT_THROW(LocalFileBackend(""), std::runtime_error);
TEST_F(LocalFileSnapshotObjectStoreTest, Constructor_EmptyPath_Throws) {
EXPECT_THROW(LocalFileSnapshotObjectStore(""), std::runtime_error);
}
TEST_F(LocalFileBackendTest, DownloadBuffer_NonExistentKey) {
TEST_F(LocalFileSnapshotObjectStoreTest, DownloadBuffer_NonExistentKey) {
std::vector<uint8_t> buf;
auto result = backend_->DownloadBuffer("no/such/key", buf);
EXPECT_FALSE(result.has_value());
}
TEST_F(LocalFileBackendTest, DownloadString_NonExistentKey) {
TEST_F(LocalFileSnapshotObjectStoreTest, DownloadString_NonExistentKey) {
std::string data;
auto result = backend_->DownloadString("no/such/key", data);
EXPECT_FALSE(result.has_value());
}
TEST_F(LocalFileBackendTest, UploadBuffer_EmptyBuffer) {
TEST_F(LocalFileSnapshotObjectStoreTest, UploadBuffer_EmptyBuffer) {
std::vector<uint8_t> empty;
auto result = backend_->UploadBuffer("test/empty", empty);
EXPECT_FALSE(result.has_value());

54
mooncake-store/tests/snapshot_child_process_test.cpp → mooncake-store/tests/ha/snapshot/snapshot_child_process_test.cpp
View File

@ -1,6 +1,6 @@
#include "master_service.h"
#include "master_metric_manager.h"
#include "serialize/serializer_backend.h"
#include "ha/snapshot/object/snapshot_object_store.h"
#include <glog/logging.h>
#include <gtest/gtest.h>
@ -48,7 +48,7 @@ class SnapshotChildProcessTest : public ::testing::Test {
ASSERT_NE(dir, nullptr) << "Failed to create temp directory";
tmp_dir_ = dir;
// Set env for LocalFileBackend
// Set env for LocalFileSnapshotObjectStore
::setenv(kEnvSnapshotLocalPath, tmp_dir().c_str(), 1);
}
@ -71,7 +71,7 @@ class SnapshotChildProcessTest : public ::testing::Test {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
service_ = std::make_unique<MasterService>(config);
}
@ -96,15 +96,15 @@ class SnapshotChildProcessTest : public ::testing::Test {
service_->CleanupOldSnapshot(keep_count, snapshot_id);
}
tl::expected<void, SerializationError> CallUploadSnapshotFile(
tl::expected<void, SerializationError> CallUploadSnapshotPayloadFile(
const std::vector<uint8_t>& data, const std::string& path,
const std::string& local_filename, const std::string& snapshot_id) {
return service_->UploadSnapshotFile(data, path, local_filename,
snapshot_id);
return service_->UploadSnapshotPayloadFile(data, path, local_filename,
snapshot_id);
}
SerializerBackend* GetSnapshotBackend() {
return service_->snapshot_backend_.get();
SnapshotObjectStore* GetSnapshotObjectStore() {
return service_->snapshot_object_store_.get();
}
tl::expected<void, SerializationError> CallPersistState(
@ -214,7 +214,7 @@ TEST_F(SnapshotChildProcessTest, HandleChildTimeout_KillsSleepingChild) {
TEST_F(SnapshotChildProcessTest, CleanupOldSnapshot_KeepsRecentDeletesOld) {
CreateDefaultService();
auto* backend = GetSnapshotBackend();
auto* backend = GetSnapshotObjectStore();
ASSERT_NE(backend, nullptr);
// Create 5 fake snapshot directories with timestamp-like names
@ -246,19 +246,20 @@ TEST_F(SnapshotChildProcessTest, CleanupOldSnapshot_KeepsRecentDeletesOld) {
EXPECT_GE(remaining.size(), 2u);
}
// ========== UploadSnapshotFile ==========
// ========== UploadSnapshotPayloadFile ==========
TEST_F(SnapshotChildProcessTest, UploadSnapshotFile_Success) {
TEST_F(SnapshotChildProcessTest, UploadSnapshotPayloadFile_Success) {
CreateDefaultService();
std::vector<uint8_t> data = {10, 20, 30, 40, 50};
std::string path = "mooncake_master_snapshot/test_upload/metadata";
auto result = CallUploadSnapshotFile(data, path, "metadata", "test_upload");
auto result =
CallUploadSnapshotPayloadFile(data, path, "metadata", "test_upload");
ASSERT_TRUE(result.has_value())
<< "UploadSnapshotFile failed: " << result.error().message;
<< "UploadSnapshotPayloadFile failed: " << result.error().message;
// Verify data can be downloaded back
std::vector<uint8_t> downloaded;
auto dl = GetSnapshotBackend()->DownloadBuffer(path, downloaded);
auto dl = GetSnapshotObjectStore()->DownloadBuffer(path, downloaded);
ASSERT_TRUE(dl.has_value()) << dl.error();
EXPECT_EQ(downloaded, data);
}
@ -274,7 +275,7 @@ TEST_F(SnapshotChildProcessTest, AutoSnapshot_GeneratesFiles) {
.set_snapshot_interval_seconds(2)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
auto auto_service = std::make_unique<MasterService>(config);
@ -315,7 +316,7 @@ TEST_F(SnapshotChildProcessTest, RestoreWithBackupDir_CreatesBackupFiles) {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
auto restore_service = std::make_unique<MasterService>(config);
@ -352,7 +353,7 @@ TEST_F(SnapshotChildProcessTest, RestoreWithoutBackupDir_NoBackupFiles) {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
auto restore_service = std::make_unique<MasterService>(config);
@ -385,10 +386,11 @@ TEST_F(SnapshotChildProcessTest, EnableSnapshotWithoutEnvVar_Throws) {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
// LocalFileBackend default constructor throws when env var is missing
// LocalFileSnapshotObjectStore default constructor throws when env var is
// missing
EXPECT_THROW(MasterService service(config), std::runtime_error);
}
@ -402,7 +404,7 @@ TEST_F(SnapshotChildProcessTest, DisableSnapshotWithoutEnvVar_NoThrow) {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
// With snapshot disabled, backend is never created, so no throw
@ -420,7 +422,7 @@ TEST_F(SnapshotChildProcessTest, RestoreCleansNonCompleteReplica) {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
service_ = std::make_unique<MasterService>(config);
@ -470,7 +472,7 @@ TEST_F(SnapshotChildProcessTest, RestoreCleansNonCompleteReplica) {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
auto restored_service = std::make_unique<MasterService>(restore_config);
@ -492,7 +494,7 @@ TEST_F(SnapshotChildProcessTest, RestoreCleansExpiredLease) {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.set_default_kv_lease_ttl(600000) // 10 min lease
.build();
service_ = std::make_unique<MasterService>(config);
@ -544,7 +546,7 @@ TEST_F(SnapshotChildProcessTest, RestoreCleansExpiredLease) {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
auto restored_service = std::make_unique<MasterService>(restore_config);
@ -569,7 +571,7 @@ TEST_F(SnapshotChildProcessTest, PersistState_FailFast_StopsOnFirstError) {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
service_ = std::make_unique<MasterService>(config);
@ -637,7 +639,7 @@ TEST_F(SnapshotChildProcessTest, UploadFail_WithBackupDir_SavesAllFiles) {
.set_snapshot_interval_seconds(100)
.set_snapshot_child_timeout_seconds(60)
.set_snapshot_retention_count(3)
.set_snapshot_backend_type("local")
.set_snapshot_object_store_type("local")
.build();
service_ = std::make_unique<MasterService>(config);

23
mooncake-transfer-engine/src/transfer_metadata_plugin.cpp
View File

@ -87,15 +87,23 @@ struct RedisStoragePlugin : public MetadataStoragePlugin {
}
RedisStoragePlugin(const std::string &metadata_uri,
const std::string &password, const uint8_t &db_index)
const std::string &username, const std::string &password,
const uint8_t &db_index)
: RedisStoragePlugin(metadata_uri) {
if (!client_) {
return;
}
if (!password.empty()) {
auto *reply = static_cast<redisReply *>(
redisCommand(client_, "AUTH %s", password.c_str()));
redisReply *reply = nullptr;
if (!username.empty()) {
reply = static_cast<redisReply *>(redisCommand(
client_, "AUTH %b %b", username.data(), username.size(),
password.data(), password.size()));
} else {
reply = static_cast<redisReply *>(redisCommand(
client_, "AUTH %b", password.data(), password.size()));
}
if (!reply || reply->type == REDIS_REPLY_ERROR) {
LOG(ERROR) << "RedisStoragePlugin: authentication failed for "
<< metadata_uri_;
@ -542,6 +550,9 @@ std::shared_ptr<MetadataStoragePlugin> MetadataStoragePlugin::Create(
#ifdef USE_REDIS
if (parsed_conn_string.first == "redis") {
const char *username = std::getenv("MC_REDIS_USERNAME");
std::string username_str = username ? username : "";
const char *password = std::getenv("MC_REDIS_PASSWORD");
std::string password_str = password ? password : "";
@ -563,8 +574,8 @@ std::shared_ptr<MetadataStoragePlugin> MetadataStoragePlugin::Create(
}
}
return std::make_shared<RedisStoragePlugin>(parsed_conn_string.second,
password_str, db_index);
return std::make_shared<RedisStoragePlugin>(
parsed_conn_string.second, username_str, password_str, db_index);
}
#endif // USE_REDIS
@ -1211,4 +1222,4 @@ uint16_t findAvailableTcpPort(int &sockfd, bool set_range) {
return 0;
}
} // namespace mooncake
} // namespace mooncake

5
mooncake-transfer-engine/tent/include/tent/metastore/redis.h
View File

@ -39,6 +39,9 @@ class RedisMetaStore : public MetaStore {
virtual Status connect(const std::string &endpoint);
Status connect(const std::string &endpoint, const std::string &username,
const std::string &password, uint8_t db_index);
Status connect(const std::string &endpoint, const std::string &password,
uint8_t db_index);
@ -64,4 +67,4 @@ class RedisMetaStore : public MetaStore {
} // namespace tent
} // namespace mooncake
#endif // TENT_REDIS_H
#endif // TENT_REDIS_H

4
mooncake-transfer-engine/tent/include/tent/runtime/control_plane.h
View File

@ -108,6 +108,10 @@ class ControlService {
const std::string& password, uint8_t db_index,
TransferEngineImpl* impl);
ControlService(const std::string& type, const std::string& servers,
const std::string& username, const std::string& password,
uint8_t db_index, TransferEngineImpl* impl);
~ControlService();
ControlService(const ControlService&) = delete;

8
mooncake-transfer-engine/tent/include/tent/runtime/metastore.h
View File

@ -31,6 +31,12 @@ struct MetaStore {
const std::string &password,
uint8_t db_index);
static std::shared_ptr<MetaStore> Create(const std::string &type,
const std::string &servers,
const std::string &username,
const std::string &password,
uint8_t db_index);
MetaStore() {}
virtual ~MetaStore() {}
@ -46,4 +52,4 @@ struct MetaStore {
} // namespace tent
} // namespace mooncake
#endif // METASTORE_H
#endif // METASTORE_H

6
mooncake-transfer-engine/tent/include/tent/runtime/segment_registry.h
View File

@ -56,6 +56,10 @@ class CentralSegmentRegistry : public SegmentRegistry {
CentralSegmentRegistry(const std::string &type, const std::string &servers,
const std::string &password, uint8_t db_index);
CentralSegmentRegistry(const std::string &type, const std::string &servers,
const std::string &username,
const std::string &password, uint8_t db_index);
virtual ~CentralSegmentRegistry() {}
public:
@ -92,4 +96,4 @@ class PeerSegmentRegistry : public SegmentRegistry {
} // namespace tent
} // namespace mooncake
#endif // SEGMENT_REGISTRY_H
#endif // SEGMENT_REGISTRY_H

20
mooncake-transfer-engine/tent/src/metastore/redis.cpp
View File

@ -45,11 +45,17 @@ RedisMetaStore::RedisMetaStore() {}
RedisMetaStore::~RedisMetaStore() { disconnect(); }
Status RedisMetaStore::connect(const std::string &endpoint) {
return connect(endpoint, "", 0);
return connect(endpoint, "", "", 0);
}
Status RedisMetaStore::connect(const std::string &endpoint,
const std::string &password, uint8_t db_index) {
return connect(endpoint, "", password, db_index);
}
Status RedisMetaStore::connect(const std::string &endpoint,
const std::string &username,
const std::string &password, uint8_t db_index) {
if (connected_) {
return Status::MetadataError(
"Redis connection already established" LOC_MARK);
@ -85,9 +91,15 @@ Status RedisMetaStore::connect(const std::string &endpoint,
// Authenticate if password is provided
if (!password.empty()) {
// Use binary-safe authentication to prevent password leakage
redisReply *reply = (redisReply *)redisCommand(
client_, "AUTH %b", password.data(), password.size());
redisReply *reply = nullptr;
if (!username.empty()) {
reply = (redisReply *)redisCommand(
client_, "AUTH %b %b", username.data(), username.size(),
password.data(), password.size());
} else {
reply = (redisReply *)redisCommand(
client_, "AUTH %b", password.data(), password.size());
}
RedisReplyGuard reply_guard(reply);
if (!reply || reply->type == REDIS_REPLY_ERROR) {

11
mooncake-transfer-engine/tent/src/runtime/control_plane.cpp
View File

@ -151,19 +151,26 @@ Status ControlClient::unpinStageBuffer(const std::string& server_addr,
ControlService::ControlService(const std::string& type,
const std::string& servers,
TransferEngineImpl* impl)
: ControlService(type, servers, "", 0, impl) {}
: ControlService(type, servers, "", "", 0, impl) {}
ControlService::ControlService(const std::string& type,
const std::string& servers,
const std::string& password, uint8_t db_index,
TransferEngineImpl* impl)
: ControlService(type, servers, "", password, db_index, impl) {}
ControlService::ControlService(const std::string& type,
const std::string& servers,
const std::string& username,
const std::string& password, uint8_t db_index,
TransferEngineImpl* impl)
: bootstrap_callback_(nullptr), notify_callback_(nullptr), impl_(impl) {
if (type == "p2p") {
auto agent = std::make_unique<PeerSegmentRegistry>();
manager_ = std::make_unique<SegmentManager>(std::move(agent));
} else {
auto agent = std::make_unique<CentralSegmentRegistry>(
type, servers, password, db_index);
type, servers, username, password, db_index);
manager_ = std::make_unique<SegmentManager>(std::move(agent));
}
rpc_server_ = std::make_shared<CoroRpcAgent>();

13
mooncake-transfer-engine/tent/src/runtime/metastore.cpp
View File

@ -29,13 +29,21 @@ namespace mooncake {
namespace tent {
std::shared_ptr<MetaStore> MetaStore::Create(const std::string &type,
const std::string &servers) {
return Create(type, servers, "", 0);
return Create(type, servers, "", "", 0);
}
std::shared_ptr<MetaStore> MetaStore::Create(const std::string &type,
const std::string &servers,
const std::string &password,
uint8_t db_index) {
return Create(type, servers, "", password, db_index);
}
std::shared_ptr<MetaStore> MetaStore::Create(const std::string &type,
const std::string &servers,
const std::string &username,
const std::string &password,
uint8_t db_index) {
std::shared_ptr<MetaStore> plugin;
#ifdef USE_ETCD
if (type == "etcd") {
@ -45,7 +53,8 @@ std::shared_ptr<MetaStore> MetaStore::Create(const std::string &type,
#ifdef USE_REDIS
if (type == "redis") {
auto redis_plugin = std::make_shared<RedisMetaStore>();
auto status = redis_plugin->connect(servers, password, db_index);
auto status =
redis_plugin->connect(servers, username, password, db_index);
if (status.ok())
return redis_plugin;
else {

10
mooncake-transfer-engine/tent/src/runtime/segment_registry.cpp
View File

@ -38,7 +38,15 @@ CentralSegmentRegistry::CentralSegmentRegistry(const std::string &type,
const std::string &servers,
const std::string &password,
uint8_t db_index) {
plugin_ = MetaStore::Create(type, servers, password, db_index);
plugin_ = MetaStore::Create(type, servers, "", password, db_index);
}
CentralSegmentRegistry::CentralSegmentRegistry(const std::string &type,
const std::string &servers,
const std::string &username,
const std::string &password,
uint8_t db_index) {
plugin_ = MetaStore::Create(type, servers, username, password, db_index);
}
Status CentralSegmentRegistry::getSegmentDesc(SegmentDescRef &desc,

9
mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp
View File

@ -266,6 +266,12 @@ Status TransferEngineImpl::construct() {
redis_password = env_password;
}
std::string redis_username;
const char* env_username = std::getenv("MC_REDIS_USERNAME");
if (env_username && *env_username) {
redis_username = env_username;
}
// Get Redis DB index from environment variable or config
int redis_db_index_config = conf_->get("redis_db_index", 0);
const char* env_db_index = std::getenv("MC_REDIS_DB_INDEX");
@ -305,7 +311,8 @@ Status TransferEngineImpl::construct() {
}
metadata_ = std::make_shared<ControlService>(
metadata_type, metadata_servers, redis_password, db_index, this);
metadata_type, metadata_servers, redis_username, redis_password,
db_index, this);
CHECK_STATUS(metadata_->start(port_, ipv6_));