diff --git a/mooncake-store/include/client_service.h b/mooncake-store/include/client_service.h index 8d18aa9e..e1418b4a 100644 --- a/mooncake-store/include/client_service.h +++ b/mooncake-store/include/client_service.h @@ -13,7 +13,7 @@ #include #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" diff --git a/mooncake-store/include/ha/backends/redis/redis_client_helper.h b/mooncake-store/include/ha/common/redis/redis_connection.h similarity index 96% rename from mooncake-store/include/ha/backends/redis/redis_client_helper.h rename to mooncake-store/include/ha/common/redis/redis_connection.h index 69a7b4b7..7cdb415d 100644 --- a/mooncake-store/include/ha/backends/redis/redis_client_helper.h +++ b/mooncake-store/include/ha/common/redis/redis_connection.h @@ -13,7 +13,7 @@ struct redisReply; namespace mooncake { namespace ha { -namespace backends { +namespace common { namespace redis { struct RedisEndpoint { @@ -50,6 +50,6 @@ tl::expected ConnectRedis( ErrorCode connection_error = ErrorCode::PERSISTENT_FAIL); } // namespace redis -} // namespace backends +} // namespace common } // namespace ha } // namespace mooncake diff --git a/mooncake-store/include/ha/backends/etcd/etcd_leader_coordinator.h b/mooncake-store/include/ha/leadership/backends/etcd/etcd_leader_coordinator.h similarity index 98% rename from mooncake-store/include/ha/backends/etcd/etcd_leader_coordinator.h rename to mooncake-store/include/ha/leadership/backends/etcd/etcd_leader_coordinator.h index 4dec172d..432d4da7 100644 --- a/mooncake-store/include/ha/backends/etcd/etcd_leader_coordinator.h +++ b/mooncake-store/include/ha/leadership/backends/etcd/etcd_leader_coordinator.h @@ -6,7 +6,7 @@ #include #include "etcd_helper.h" -#include "ha/leader_coordinator.h" +#include "ha/leadership/leader_coordinator.h" namespace mooncake { namespace ha { diff --git a/mooncake-store/include/ha/backends/redis/redis_leader_coordinator.h b/mooncake-store/include/ha/leadership/backends/redis/redis_leader_coordinator.h similarity index 98% rename from mooncake-store/include/ha/backends/redis/redis_leader_coordinator.h rename to mooncake-store/include/ha/leadership/backends/redis/redis_leader_coordinator.h index 11a33fac..1b4eb605 100644 --- a/mooncake-store/include/ha/backends/redis/redis_leader_coordinator.h +++ b/mooncake-store/include/ha/leadership/backends/redis/redis_leader_coordinator.h @@ -8,7 +8,7 @@ #include #include -#include "ha/leader_coordinator.h" +#include "ha/leadership/leader_coordinator.h" struct redisContext; diff --git a/mooncake-store/include/ha/leader_coordinator.h b/mooncake-store/include/ha/leadership/leader_coordinator.h similarity index 100% rename from mooncake-store/include/ha/leader_coordinator.h rename to mooncake-store/include/ha/leadership/leader_coordinator.h diff --git a/mooncake-store/include/ha/ha_backend_factory.h b/mooncake-store/include/ha/leadership/leader_coordinator_factory.h similarity index 86% rename from mooncake-store/include/ha/ha_backend_factory.h rename to mooncake-store/include/ha/leadership/leader_coordinator_factory.h index bbc31d90..bb4082be 100644 --- a/mooncake-store/include/ha/ha_backend_factory.h +++ b/mooncake-store/include/ha/leadership/leader_coordinator_factory.h @@ -5,7 +5,7 @@ #include #include "ha/ha_types.h" -#include "ha/leader_coordinator.h" +#include "ha/leadership/leader_coordinator.h" namespace mooncake { namespace ha { diff --git a/mooncake-store/include/ha/master_service_supervisor.h b/mooncake-store/include/ha/leadership/master_service_supervisor.h similarity index 100% rename from mooncake-store/include/ha/master_service_supervisor.h rename to mooncake-store/include/ha/leadership/master_service_supervisor.h diff --git a/mooncake-store/include/ha/serializer_snapshot_store.h b/mooncake-store/include/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.h similarity index 53% rename from mooncake-store/include/ha/serializer_snapshot_store.h rename to mooncake-store/include/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.h index 04401733..a73b79ff 100644 --- a/mooncake-store/include/ha/serializer_snapshot_store.h +++ b/mooncake-store/include/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.h @@ -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 diff --git a/mooncake-store/include/ha/backends/redis/redis_snapshot_store.h b/mooncake-store/include/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.h similarity index 70% rename from mooncake-store/include/ha/backends/redis/redis_snapshot_store.h rename to mooncake-store/include/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.h index 662116cb..c0855541 100644 --- a/mooncake-store/include/ha/backends/redis/redis_snapshot_store.h +++ b/mooncake-store/include/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.h @@ -2,19 +2,19 @@ #include -#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_; diff --git a/mooncake-store/include/ha/snapshot_store.h b/mooncake-store/include/ha/snapshot/catalog/snapshot_catalog_store.h similarity index 93% rename from mooncake-store/include/ha/snapshot_store.h rename to mooncake-store/include/ha/snapshot/catalog/snapshot_catalog_store.h index 99e79e88..d9c1ce8a 100644 --- a/mooncake-store/include/ha/snapshot_store.h +++ b/mooncake-store/include/ha/snapshot/catalog/snapshot_catalog_store.h @@ -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; diff --git a/mooncake-store/include/ha/snapshot/object/backends/local/local_file_snapshot_object_store.h b/mooncake-store/include/ha/snapshot/object/backends/local/local_file_snapshot_object_store.h new file mode 100644 index 00000000..1c51fbe0 --- /dev/null +++ b/mooncake-store/include/ha/snapshot/object/backends/local/local_file_snapshot_object_store.h @@ -0,0 +1,61 @@ +#pragma once + +#include +#include +#include + +#include + +#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 UploadBuffer( + const std::string& key, const std::vector& buffer) override; + + tl::expected DownloadBuffer( + const std::string& key, std::vector& buffer) override; + + tl::expected UploadString( + const std::string& key, const std::string& data) override; + + tl::expected DownloadString(const std::string& key, + std::string& data) override; + + tl::expected DeleteObjectsWithPrefix( + const std::string& prefix) override; + + tl::expected ListObjectsWithPrefix( + const std::string& prefix, + std::vector& 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 EnsureDirectoryExists( + const fs::path& dir_path) const; + bool IsPathWithinBase(const fs::path& path) const; + + fs::path base_path_; +}; + +} // namespace mooncake diff --git a/mooncake-store/include/ha/snapshot/object/backends/s3/s3_snapshot_object_store.h b/mooncake-store/include/ha/snapshot/object/backends/s3/s3_snapshot_object_store.h new file mode 100644 index 00000000..d2927b7c --- /dev/null +++ b/mooncake-store/include/ha/snapshot/object/backends/s3/s3_snapshot_object_store.h @@ -0,0 +1,54 @@ +#pragma once + +#include +#include +#include + +#include + +#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 UploadBuffer( + const std::string& key, const std::vector& buffer) override; + + tl::expected DownloadBuffer( + const std::string& key, std::vector& buffer) override; + + tl::expected UploadString( + const std::string& key, const std::string& data) override; + + tl::expected DownloadString(const std::string& key, + std::string& data) override; + + tl::expected DeleteObjectsWithPrefix( + const std::string& prefix) override; + + tl::expected ListObjectsWithPrefix( + const std::string& prefix, + std::vector& object_keys) override; + + bool IsNotFoundError(const std::string& error) const override; + std::string GetConnectionInfo() const override; + + private: + class Impl; + std::unique_ptr impl_; +}; + +#endif // HAVE_AWS_SDK + +} // namespace mooncake diff --git a/mooncake-store/include/ha/snapshot/object/snapshot_object_store.h b/mooncake-store/include/ha/snapshot/object/snapshot_object_store.h new file mode 100644 index 00000000..b64ae887 --- /dev/null +++ b/mooncake-store/include/ha/snapshot/object/snapshot_object_store.h @@ -0,0 +1,143 @@ +#pragma once + +#include +#include +#include +#include + +#include + +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(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 UploadBuffer( + const std::string& key, const std::vector& 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 DownloadBuffer( + const std::string& key, std::vector& 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 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 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 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 ListObjectsWithPrefix( + const std::string& prefix, std::vector& 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 Create( + SnapshotObjectStoreType type); +}; + +} // namespace mooncake diff --git a/mooncake-store/include/utils/snapshot_logger.h b/mooncake-store/include/ha/snapshot/snapshot_logger.h similarity index 100% rename from mooncake-store/include/utils/snapshot_logger.h rename to mooncake-store/include/ha/snapshot/snapshot_logger.h diff --git a/mooncake-store/include/snapshot_provider.h b/mooncake-store/include/ha/snapshot/snapshot_provider.h similarity index 100% rename from mooncake-store/include/snapshot_provider.h rename to mooncake-store/include/ha/snapshot/snapshot_provider.h diff --git a/mooncake-store/include/hot_standby_service.h b/mooncake-store/include/hot_standby_service.h index cb4cd163..091446fd 100644 --- a/mooncake-store/include/hot_standby_service.h +++ b/mooncake-store/include/hot_standby_service.h @@ -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" diff --git a/mooncake-store/include/master_config.h b/mooncake-store/include/master_config.h index f62276ac..df787000 100644 --- a/mooncake-store/include/master_config.h +++ b/mooncake-store/include/master_config.h @@ -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 = diff --git a/mooncake-store/include/master_service.h b/mooncake-store/include/master_service.h index 43c3afb3..fbe6737b 100644 --- a/mooncake-store/include/master_service.h +++ b/mooncake-store/include/master_service.h @@ -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 PersistState( const std::string& snapshot_id); - tl::expected UploadSnapshotFile( + tl::expected UploadSnapshotPayloadFile( const std::vector& data, const std::string& path, const std::string& local_filename, const std::string& snapshot_id); - std::unique_ptr CreateSnapshotStore(); + std::unique_ptr 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 snapshot_backend_; - std::unique_ptr snapshot_store_; + std::string snapshot_catalog_store_type_{}; + std::string snapshot_catalog_store_connstring_; + std::unique_ptr snapshot_object_store_; + std::unique_ptr snapshot_catalog_store_; mutable std::shared_mutex snapshot_mutex_; // Discarded replicas management diff --git a/mooncake-store/include/serialize/serializer_backend.h b/mooncake-store/include/serialize/serializer_backend.h deleted file mode 100644 index 313049d2..00000000 --- a/mooncake-store/include/serialize/serializer_backend.h +++ /dev/null @@ -1,246 +0,0 @@ -#pragma once - -#include -#include -#include -#include -#include - -#include - -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(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 UploadBuffer( - const std::string& key, const std::vector& 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 DownloadBuffer( - const std::string& key, std::vector& 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 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 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 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 ListObjectsWithPrefix( - const std::string& prefix, std::vector& 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 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 UploadBuffer( - const std::string& key, const std::vector& buffer) override; - - tl::expected DownloadBuffer( - const std::string& key, std::vector& buffer) override; - - tl::expected UploadString( - const std::string& key, const std::string& data) override; - - tl::expected DownloadString(const std::string& key, - std::string& data) override; - - tl::expected DeleteObjectsWithPrefix( - const std::string& prefix) override; - - tl::expected ListObjectsWithPrefix( - const std::string& prefix, - std::vector& object_keys) override; - - bool IsNotFoundError(const std::string& error) const override; - - std::string GetConnectionInfo() const override; - - private: - class Impl; - std::unique_ptr 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 UploadBuffer( - const std::string& key, const std::vector& buffer) override; - - tl::expected DownloadBuffer( - const std::string& key, std::vector& buffer) override; - - tl::expected UploadString( - const std::string& key, const std::string& data) override; - - tl::expected DownloadString(const std::string& key, - std::string& data) override; - - tl::expected DeleteObjectsWithPrefix( - const std::string& prefix) override; - - tl::expected ListObjectsWithPrefix( - const std::string& prefix, - std::vector& 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 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 diff --git a/mooncake-store/src/CMakeLists.txt b/mooncake-store/src/CMakeLists.txt index 11b528fc..d632cd51 100644 --- a/mooncake-store/src/CMakeLists.txt +++ b/mooncake-store/src/CMakeLists.txt @@ -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 diff --git a/mooncake-store/src/client_service.cpp b/mooncake-store/src/client_service.cpp index 47170a24..7f725834 100644 --- a/mooncake-store/src/client_service.cpp +++ b/mooncake-store/src/client_service.cpp @@ -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" diff --git a/mooncake-store/src/ha/redis_client_helper.cpp b/mooncake-store/src/ha/common/redis/redis_connection.cpp similarity index 91% rename from mooncake-store/src/ha/redis_client_helper.cpp rename to mooncake-store/src/ha/common/redis/redis_connection.cpp index 6eb35826..d694cb40 100644 --- a/mooncake-store/src/ha/redis_client_helper.cpp +++ b/mooncake-store/src/ha/common/redis/redis_connection.cpp @@ -1,4 +1,4 @@ -#include "ha/backends/redis/redis_client_helper.h" +#include "ha/common/redis/redis_connection.h" #include #include @@ -12,7 +12,7 @@ namespace mooncake { namespace ha { -namespace backends { +namespace common { namespace redis { namespace { @@ -195,10 +195,18 @@ tl::expected 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(redisCommand( - context.get(), "AUTH %b", password, std::strlen(password)))); + RedisReplyPtr reply; + if (username != nullptr && std::strlen(username) > 0) { + reply.reset(static_cast(redisCommand( + context.get(), "AUTH %b %b", username, std::strlen(username), + password, std::strlen(password)))); + } else { + reply.reset(static_cast(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 ConnectRedis( #endif } // namespace redis -} // namespace backends +} // namespace common } // namespace ha } // namespace mooncake diff --git a/mooncake-store/src/ha/etcd_leader_coordinator.cpp b/mooncake-store/src/ha/leadership/backends/etcd/etcd_leader_coordinator.cpp similarity index 99% rename from mooncake-store/src/ha/etcd_leader_coordinator.cpp rename to mooncake-store/src/ha/leadership/backends/etcd/etcd_leader_coordinator.cpp index 4b875f7e..58c1e238 100644 --- a/mooncake-store/src/ha/etcd_leader_coordinator.cpp +++ b/mooncake-store/src/ha/leadership/backends/etcd/etcd_leader_coordinator.cpp @@ -1,4 +1,4 @@ -#include "ha/backends/etcd/etcd_leader_coordinator.h" +#include "ha/leadership/backends/etcd/etcd_leader_coordinator.h" #include #include diff --git a/mooncake-store/src/ha/redis_leader_coordinator.cpp b/mooncake-store/src/ha/leadership/backends/redis/redis_leader_coordinator.cpp similarity index 99% rename from mooncake-store/src/ha/redis_leader_coordinator.cpp rename to mooncake-store/src/ha/leadership/backends/redis/redis_leader_coordinator.cpp index e832788e..57a5ce3a 100644 --- a/mooncake-store/src/ha/redis_leader_coordinator.cpp +++ b/mooncake-store/src/ha/leadership/backends/redis/redis_leader_coordinator.cpp @@ -1,4 +1,4 @@ -#include "ha/backends/redis/redis_leader_coordinator.h" +#include "ha/leadership/backends/redis/redis_leader_coordinator.h" #include #include @@ -16,7 +16,7 @@ #endif #include -#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); diff --git a/mooncake-store/src/ha/ha_backend_factory.cpp b/mooncake-store/src/ha/leadership/leader_coordinator_factory.cpp similarity index 87% rename from mooncake-store/src/ha/ha_backend_factory.cpp rename to mooncake-store/src/ha/leadership/leader_coordinator_factory.cpp index d23e0558..e6bc49af 100644 --- a/mooncake-store/src/ha/ha_backend_factory.cpp +++ b/mooncake-store/src/ha/leadership/leader_coordinator_factory.cpp @@ -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 { diff --git a/mooncake-store/src/ha/master_service_supervisor.cpp b/mooncake-store/src/ha/leadership/master_service_supervisor.cpp similarity index 98% rename from mooncake-store/src/ha/master_service_supervisor.cpp rename to mooncake-store/src/ha/leadership/master_service_supervisor.cpp index 4c7bc865..b76e6486 100644 --- a/mooncake-store/src/ha/master_service_supervisor.cpp +++ b/mooncake-store/src/ha/leadership/master_service_supervisor.cpp @@ -1,4 +1,4 @@ -#include "ha/master_service_supervisor.h" +#include "ha/leadership/master_service_supervisor.h" #include #include @@ -10,7 +10,7 @@ #include #include -#include "ha/ha_backend_factory.h" +#include "ha/leadership/leader_coordinator_factory.h" #include "rpc_service.h" namespace mooncake { diff --git a/mooncake-store/src/ha/serializer_snapshot_store.cpp b/mooncake-store/src/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.cpp similarity index 56% rename from mooncake-store/src/ha/serializer_snapshot_store.cpp rename to mooncake-store/src/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.cpp index ab650f03..6a6490c4 100644 --- a/mooncake-store/src/ha/serializer_snapshot_store.cpp +++ b/mooncake-store/src/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.cpp @@ -1,4 +1,4 @@ -#include "ha/serializer_snapshot_store.h" +#include "ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store.h" #include #include @@ -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, 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(); } 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(); } - 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( - snapshot_store_detail::MakeSnapshotDescriptor(latest_snapshot_id)); + snapshot_catalog_store_detail::MakeSnapshotDescriptor( + latest_snapshot_id)); } tl::expected, 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 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> 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 diff --git a/mooncake-store/src/ha/redis_snapshot_store.cpp b/mooncake-store/src/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.cpp similarity index 74% rename from mooncake-store/src/ha/redis_snapshot_store.cpp rename to mooncake-store/src/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.cpp index af6053ad..c872d5cd 100644 --- a/mooncake-store/src/ha/redis_snapshot_store.cpp +++ b/mooncake-store/src/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.cpp @@ -1,4 +1,4 @@ -#include "ha/backends/redis/redis_snapshot_store.h" +#include "ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store.h" #include #include @@ -10,7 +10,7 @@ #include #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 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, ErrorCode> -RedisSnapshotStore::GetLatest() { +RedisSnapshotCatalogStore::GetLatest() { return tl::make_unexpected(ErrorCode::UNAVAILABLE_IN_CURRENT_MODE); } tl::expected, 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, 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(); } - 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( - snapshot_store_detail::MakeSnapshotDescriptor(latest_snapshot_id)); + snapshot_catalog_store_detail::MakeSnapshotDescriptor( + latest_snapshot_id)); } tl::expected, 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"; diff --git a/mooncake-store/src/serialize/serializer_backend.cpp b/mooncake-store/src/ha/snapshot/object/backends/local/local_file_snapshot_object_store.cpp similarity index 56% rename from mooncake-store/src/serialize/serializer_backend.cpp rename to mooncake-store/src/ha/snapshot/object/backends/local/local_file_snapshot_object_store.cpp index d368cfd1..6a857857 100644 --- a/mooncake-store/src/serialize/serializer_backend.cpp +++ b/mooncake-store/src/ha/snapshot/object/backends/local/local_file_snapshot_object_store.cpp @@ -1,151 +1,31 @@ -#include "serialize/serializer_backend.h" +#include "ha/snapshot/object/backends/local/local_file_snapshot_object_store.h" #include -#include #include -#include #include #include -#include #include #include -#ifdef HAVE_AWS_SDK -#include "utils/s3_helper.h" -#endif - -namespace fs = std::filesystem; - namespace mooncake { -// ============================================================================ -// SerializerBackend factory method implementation -// ============================================================================ - -std::unique_ptr SerializerBackend::Create( - SnapshotBackendType type) { - switch (type) { -#ifdef HAVE_AWS_SDK - case SnapshotBackendType::S3: - return std::make_unique(); -#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(); - 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(lhs)) == - std::tolower(static_cast(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()) { - LOG(INFO) << "S3Backend initialized"; -} - -tl::expected S3Backend::UploadBuffer( - const std::string& key, const std::vector& buffer) { - return impl_->s3_helper_.UploadBufferMultipart(key, buffer); -} - -tl::expected S3Backend::DownloadBuffer( - const std::string& key, std::vector& buffer) { - return impl_->s3_helper_.DownloadBufferMultipart(key, buffer); -} - -tl::expected S3Backend::UploadString( - const std::string& key, const std::string& data) { - return impl_->s3_helper_.UploadString(key, data); -} - -tl::expected S3Backend::DownloadString( - const std::string& key, std::string& data) { - return impl_->s3_helper_.DownloadString(key, data); -} - -tl::expected S3Backend::DeleteObjectsWithPrefix( - const std::string& prefix) { - return impl_->s3_helper_.DeleteObjectsWithPrefix(prefix); -} - -tl::expected S3Backend::ListObjectsWithPrefix( - const std::string& prefix, std::vector& 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 LocalFileBackend::EnsureDirectoryExists( +tl::expected +LocalFileSnapshotObjectStore::EnsureDirectoryExists( const fs::path& dir_path) const { try { if (!fs::exists(dir_path)) { @@ -225,41 +101,39 @@ tl::expected 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 LocalFileBackend::UploadBuffer( +tl::expected LocalFileSnapshotObjectStore::UploadBuffer( const std::string& key, const std::vector& 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 LocalFileBackend::UploadBuffer( file.write(reinterpret_cast(buffer.data()), static_cast(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 LocalFileBackend::UploadBuffer( return {}; } -tl::expected LocalFileBackend::DownloadBuffer( +tl::expected LocalFileSnapshotObjectStore::DownloadBuffer( const std::string& key, std::vector& 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 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(buffer.data()), static_cast(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 LocalFileBackend::DownloadBuffer( return {}; } -tl::expected LocalFileBackend::UploadString( +tl::expected 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 LocalFileBackend::UploadString( return {}; } -tl::expected LocalFileBackend::DownloadString( +tl::expected 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 LocalFileBackend::DownloadString( return {}; } -tl::expected LocalFileBackend::DeleteObjectsWithPrefix( +tl::expected +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 LocalFileBackend::DeleteObjectsWithPrefix( return {}; } -tl::expected LocalFileBackend::ListObjectsWithPrefix( +tl::expected +LocalFileSnapshotObjectStore::ListObjectsWithPrefix( const std::string& prefix, std::vector& 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 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 diff --git a/mooncake-store/src/ha/snapshot/object/backends/s3/s3_snapshot_object_store.cpp b/mooncake-store/src/ha/snapshot/object/backends/s3/s3_snapshot_object_store.cpp new file mode 100644 index 00000000..4a0a63a8 --- /dev/null +++ b/mooncake-store/src/ha/snapshot/object/backends/s3/s3_snapshot_object_store.cpp @@ -0,0 +1,95 @@ +#include "ha/snapshot/object/backends/s3/s3_snapshot_object_store.h" + +#ifdef HAVE_AWS_SDK + +#include +#include +#include + +#include + +#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(lhs)) == + std::tolower(static_cast(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()) { + LOG(INFO) << "S3SnapshotObjectStore initialized"; +} + +tl::expected S3SnapshotObjectStore::UploadBuffer( + const std::string& key, const std::vector& buffer) { + return impl_->s3_helper_.UploadBufferMultipart(key, buffer); +} + +tl::expected S3SnapshotObjectStore::DownloadBuffer( + const std::string& key, std::vector& buffer) { + return impl_->s3_helper_.DownloadBufferMultipart(key, buffer); +} + +tl::expected S3SnapshotObjectStore::UploadString( + const std::string& key, const std::string& data) { + return impl_->s3_helper_.UploadString(key, data); +} + +tl::expected S3SnapshotObjectStore::DownloadString( + const std::string& key, std::string& data) { + return impl_->s3_helper_.DownloadString(key, data); +} + +tl::expected S3SnapshotObjectStore::DeleteObjectsWithPrefix( + const std::string& prefix) { + return impl_->s3_helper_.DeleteObjectsWithPrefix(prefix); +} + +tl::expected S3SnapshotObjectStore::ListObjectsWithPrefix( + const std::string& prefix, std::vector& 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 diff --git a/mooncake-store/src/ha/snapshot/object/snapshot_object_store.cpp b/mooncake-store/src/ha/snapshot/object/snapshot_object_store.cpp new file mode 100644 index 00000000..7ff678fc --- /dev/null +++ b/mooncake-store/src/ha/snapshot/object/snapshot_object_store.cpp @@ -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::Create( + SnapshotObjectStoreType type) { + switch (type) { +#ifdef HAVE_AWS_SDK + case SnapshotObjectStoreType::S3: + return std::make_unique(); +#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(); + default: + throw std::invalid_argument("Unknown snapshot object store type"); + } +} + +} // namespace mooncake diff --git a/mooncake-store/src/master.cpp b/mooncake-store/src/master.cpp index f29dc136..bc303f89 100644 --- a/mooncake-store/src/master.cpp +++ b/mooncake-store/src/master.cpp @@ -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 diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index 46573564..8b045503 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -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 -ParseSnapshotCatalogBackendKind(std::string_view backend_type) { - if (backend_type.empty() || backend_type == "serializer") { - return SnapshotCatalogBackendKind::kSerializer; +tl::expected 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 MasterService::CreateSnapshotStore() { - auto backend_kind = - ParseSnapshotCatalogBackendKind(snapshot_catalog_backend_type_); - if (!backend_kind) { - throw std::invalid_argument(backend_kind.error()); +std::unique_ptr +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( - 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( - 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 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 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 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 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 MasterService::PersistState( SNAPSHOT_SERIALIZER_VERSION, snapshot_id); std::vector 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 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 MasterService::PersistState( return {}; } -tl::expected MasterService::UploadSnapshotFile( +tl::expected MasterService::UploadSnapshotPayloadFile( const std::vector& 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 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 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 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 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, diff --git a/mooncake-store/tests/CMakeLists.txt b/mooncake-store/tests/CMakeLists.txt index 4bbe26d6..4cd80aa7 100644 --- a/mooncake-store/tests/CMakeLists.txt +++ b/mooncake-store/tests/CMakeLists.txt @@ -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) diff --git a/mooncake-store/tests/e2e/chaos_rand_test.cpp b/mooncake-store/tests/e2e/chaos_rand_test.cpp index 0f35d65b..cd714da2 100644 --- a/mooncake-store/tests/e2e/chaos_rand_test.cpp +++ b/mooncake-store/tests/e2e/chaos_rand_test.cpp @@ -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" diff --git a/mooncake-store/tests/e2e/chaos_test.cpp b/mooncake-store/tests/e2e/chaos_test.cpp index 24788b3a..619a89c5 100644 --- a/mooncake-store/tests/e2e/chaos_test.cpp +++ b/mooncake-store/tests/e2e/chaos_test.cpp @@ -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" diff --git a/mooncake-store/tests/redis_test_utils.h b/mooncake-store/tests/ha/common/redis/redis_test_utils.h similarity index 78% rename from mooncake-store/tests/redis_test_utils.h rename to mooncake-store/tests/ha/common/redis/redis_test_utils.h index 32958388..ace56e8d 100644 --- a/mooncake-store/tests/redis_test_utils.h +++ b/mooncake-store/tests/ha/common/redis/redis_test_utils.h @@ -6,24 +6,24 @@ #include #include -#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 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); } diff --git a/mooncake-store/tests/high_availability_redis_test.cpp b/mooncake-store/tests/ha/leadership/backends/redis/high_availability_redis_test.cpp similarity index 98% rename from mooncake-store/tests/high_availability_redis_test.cpp rename to mooncake-store/tests/ha/leadership/backends/redis/high_availability_redis_test.cpp index c43511d3..bc6df228 100644 --- a/mooncake-store/tests/high_availability_redis_test.cpp +++ b/mooncake-store/tests/ha/leadership/backends/redis/high_availability_redis_test.cpp @@ -12,9 +12,9 @@ #include -#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 { diff --git a/mooncake-store/tests/high_availability_test.cpp b/mooncake-store/tests/ha/leadership/high_availability_test.cpp similarity index 99% rename from mooncake-store/tests/high_availability_test.cpp rename to mooncake-store/tests/ha/leadership/high_availability_test.cpp index f356548f..4b91ecda 100644 --- a/mooncake-store/tests/high_availability_test.cpp +++ b/mooncake-store/tests/ha/leadership/high_availability_test.cpp @@ -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 { diff --git a/mooncake-store/tests/high_availability_test_fixture.h b/mooncake-store/tests/ha/leadership/high_availability_test_fixture.h similarity index 100% rename from mooncake-store/tests/high_availability_test_fixture.h rename to mooncake-store/tests/ha/leadership/high_availability_test_fixture.h diff --git a/mooncake-store/tests/serializer_snapshot_store_test.cpp b/mooncake-store/tests/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store_test.cpp similarity index 86% rename from mooncake-store/tests/serializer_snapshot_store_test.cpp rename to mooncake-store/tests/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store_test.cpp index 07d5f9d2..28f228ad 100644 --- a/mooncake-store/tests/serializer_snapshot_store_test.cpp +++ b/mooncake-store/tests/ha/snapshot/catalog/backends/embedded/embedded_snapshot_catalog_store_test.cpp @@ -7,11 +7,11 @@ #include #include -#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 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); diff --git a/mooncake-store/tests/redis_snapshot_store_test.cpp b/mooncake-store/tests/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store_test.cpp similarity index 84% rename from mooncake-store/tests/redis_snapshot_store_test.cpp rename to mooncake-store/tests/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store_test.cpp index 0256644e..10cda68c 100644 --- a/mooncake-store/tests/redis_snapshot_store_test.cpp +++ b/mooncake-store/tests/ha/snapshot/catalog/backends/redis/redis_snapshot_catalog_store_test.cpp @@ -7,9 +7,8 @@ #include -#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 UploadBuffer( const std::string& key, const std::vector& 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 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( - &payload_backend_, FLAGS_redis_endpoint, cluster_namespace_); + store_ = + std::make_unique( + &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 store_; + std::unique_ptr 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/"); } diff --git a/mooncake-store/tests/master_service_ssd_test_for_snapshot.cpp b/mooncake-store/tests/ha/snapshot/master_service_ssd_test_for_snapshot.cpp similarity index 100% rename from mooncake-store/tests/master_service_ssd_test_for_snapshot.cpp rename to mooncake-store/tests/ha/snapshot/master_service_ssd_test_for_snapshot.cpp diff --git a/mooncake-store/tests/master_service_test_for_snapshot.cpp b/mooncake-store/tests/ha/snapshot/master_service_test_for_snapshot.cpp similarity index 100% rename from mooncake-store/tests/master_service_test_for_snapshot.cpp rename to mooncake-store/tests/ha/snapshot/master_service_test_for_snapshot.cpp diff --git a/mooncake-store/tests/master_service_test_for_snapshot_base.h b/mooncake-store/tests/ha/snapshot/master_service_test_for_snapshot_base.h similarity index 97% rename from mooncake-store/tests/master_service_test_for_snapshot_base.h rename to mooncake-store/tests/ha/snapshot/master_service_test_for_snapshot_base.h index 51768fe5..893983ab 100644 --- a/mooncake-store/tests/master_service_test_for_snapshot_base.h +++ b/mooncake-store/tests/ha/snapshot/master_service_test_for_snapshot_base.h @@ -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 @@ -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 restored_service( @@ -784,16 +784,18 @@ class MasterServiceSnapshotTestBase : public ::testing::Test { << "Use 'service_.reset(new MasterService(...))' instead of " "'std::unique_ptr 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 diff --git a/mooncake-store/tests/local_file_backend_test.cpp b/mooncake-store/tests/ha/snapshot/object/backends/local/local_file_snapshot_object_store_test.cpp similarity index 73% rename from mooncake-store/tests/local_file_backend_test.cpp rename to mooncake-store/tests/ha/snapshot/object/backends/local/local_file_snapshot_object_store_test.cpp index 6fe3e0da..1104ab84 100644 --- a/mooncake-store/tests/local_file_backend_test.cpp +++ b/mooncake-store/tests/ha/snapshot/object/backends/local/local_file_snapshot_object_store_test.cpp @@ -6,31 +6,31 @@ #include #include -#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 backend_; + std::unique_ptr 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(tmp_dir()); + backend_ = std::make_unique(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 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 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 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 empty; auto result = backend_->UploadBuffer("test/empty", empty); EXPECT_FALSE(result.has_value()); diff --git a/mooncake-store/tests/snapshot_child_process_test.cpp b/mooncake-store/tests/ha/snapshot/snapshot_child_process_test.cpp similarity index 93% rename from mooncake-store/tests/snapshot_child_process_test.cpp rename to mooncake-store/tests/ha/snapshot/snapshot_child_process_test.cpp index 3f88f74b..445ec855 100644 --- a/mooncake-store/tests/snapshot_child_process_test.cpp +++ b/mooncake-store/tests/ha/snapshot/snapshot_child_process_test.cpp @@ -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 #include @@ -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(config); } @@ -96,15 +96,15 @@ class SnapshotChildProcessTest : public ::testing::Test { service_->CleanupOldSnapshot(keep_count, snapshot_id); } - tl::expected CallUploadSnapshotFile( + tl::expected CallUploadSnapshotPayloadFile( const std::vector& 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 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 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 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(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(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(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(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(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(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(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(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(config); diff --git a/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp b/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp index 968430d4..09098d0c 100644 --- a/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp +++ b/mooncake-transfer-engine/src/transfer_metadata_plugin.cpp @@ -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( - redisCommand(client_, "AUTH %s", password.c_str())); + redisReply *reply = nullptr; + if (!username.empty()) { + reply = static_cast(redisCommand( + client_, "AUTH %b %b", username.data(), username.size(), + password.data(), password.size())); + } else { + reply = static_cast(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::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::Create( } } - return std::make_shared(parsed_conn_string.second, - password_str, db_index); + return std::make_shared( + 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 \ No newline at end of file +} // namespace mooncake diff --git a/mooncake-transfer-engine/tent/include/tent/metastore/redis.h b/mooncake-transfer-engine/tent/include/tent/metastore/redis.h index e5fb9c4a..02ad204b 100644 --- a/mooncake-transfer-engine/tent/include/tent/metastore/redis.h +++ b/mooncake-transfer-engine/tent/include/tent/metastore/redis.h @@ -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 \ No newline at end of file +#endif // TENT_REDIS_H diff --git a/mooncake-transfer-engine/tent/include/tent/runtime/control_plane.h b/mooncake-transfer-engine/tent/include/tent/runtime/control_plane.h index 6e86fc81..34f69226 100644 --- a/mooncake-transfer-engine/tent/include/tent/runtime/control_plane.h +++ b/mooncake-transfer-engine/tent/include/tent/runtime/control_plane.h @@ -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; diff --git a/mooncake-transfer-engine/tent/include/tent/runtime/metastore.h b/mooncake-transfer-engine/tent/include/tent/runtime/metastore.h index c61a1c8d..a70fdd8f 100644 --- a/mooncake-transfer-engine/tent/include/tent/runtime/metastore.h +++ b/mooncake-transfer-engine/tent/include/tent/runtime/metastore.h @@ -31,6 +31,12 @@ struct MetaStore { const std::string &password, uint8_t db_index); + static std::shared_ptr 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 \ No newline at end of file +#endif // METASTORE_H diff --git a/mooncake-transfer-engine/tent/include/tent/runtime/segment_registry.h b/mooncake-transfer-engine/tent/include/tent/runtime/segment_registry.h index 6bad0cdf..97c586d0 100644 --- a/mooncake-transfer-engine/tent/include/tent/runtime/segment_registry.h +++ b/mooncake-transfer-engine/tent/include/tent/runtime/segment_registry.h @@ -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 \ No newline at end of file +#endif // SEGMENT_REGISTRY_H diff --git a/mooncake-transfer-engine/tent/src/metastore/redis.cpp b/mooncake-transfer-engine/tent/src/metastore/redis.cpp index 5b0caf83..0774acef 100644 --- a/mooncake-transfer-engine/tent/src/metastore/redis.cpp +++ b/mooncake-transfer-engine/tent/src/metastore/redis.cpp @@ -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) { diff --git a/mooncake-transfer-engine/tent/src/runtime/control_plane.cpp b/mooncake-transfer-engine/tent/src/runtime/control_plane.cpp index f1514122..c92a0bab 100644 --- a/mooncake-transfer-engine/tent/src/runtime/control_plane.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/control_plane.cpp @@ -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(); manager_ = std::make_unique(std::move(agent)); } else { auto agent = std::make_unique( - type, servers, password, db_index); + type, servers, username, password, db_index); manager_ = std::make_unique(std::move(agent)); } rpc_server_ = std::make_shared(); diff --git a/mooncake-transfer-engine/tent/src/runtime/metastore.cpp b/mooncake-transfer-engine/tent/src/runtime/metastore.cpp index 2df0ba8d..f797b4d8 100644 --- a/mooncake-transfer-engine/tent/src/runtime/metastore.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/metastore.cpp @@ -29,13 +29,21 @@ namespace mooncake { namespace tent { std::shared_ptr MetaStore::Create(const std::string &type, const std::string &servers) { - return Create(type, servers, "", 0); + return Create(type, servers, "", "", 0); } std::shared_ptr 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::Create(const std::string &type, + const std::string &servers, + const std::string &username, + const std::string &password, + uint8_t db_index) { std::shared_ptr plugin; #ifdef USE_ETCD if (type == "etcd") { @@ -45,7 +53,8 @@ std::shared_ptr MetaStore::Create(const std::string &type, #ifdef USE_REDIS if (type == "redis") { auto redis_plugin = std::make_shared(); - 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 { diff --git a/mooncake-transfer-engine/tent/src/runtime/segment_registry.cpp b/mooncake-transfer-engine/tent/src/runtime/segment_registry.cpp index 01397b94..f6f85971 100644 --- a/mooncake-transfer-engine/tent/src/runtime/segment_registry.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/segment_registry.cpp @@ -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, diff --git a/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp b/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp index e0876de9..0ab39a02 100644 --- a/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp +++ b/mooncake-transfer-engine/tent/src/runtime/transfer_engine_impl.cpp @@ -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( - 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_));