[STORE] abstract snapshot catalog and add Redis snapshot backend

mooncake-store/include/ha/backends/redis/redis_client_helper.h

@@ -0,0 +1,55 @@
+#pragma once
+
+#include <memory>
+#include <string>
+#include <string_view>
+
+#include <ylt/util/tl/expected.hpp>
+
+#include "types.h"
+
+struct redisContext;
+struct redisReply;
+
+namespace mooncake {
+namespace ha {
+namespace backends {
+namespace redis {
+
+struct RedisEndpoint {
+    std::string host;
+    int port = 6379;
+};
+
+struct RedisContextDeleter {
+    void operator()(redisContext* context) const;
+};
+
+using RedisContextPtr = std::unique_ptr<redisContext, RedisContextDeleter>;
+
+struct RedisReplyDeleter {
+    void operator()(redisReply* reply) const;
+};
+
+using RedisReplyPtr = std::unique_ptr<redisReply, RedisReplyDeleter>;
+
+bool IsStringReply(const redisReply* reply);
+
+tl::expected<int, ErrorCode> ParsePositiveInt(std::string_view text,
+                                              int min_value, int max_value);
+
+tl::expected<int, ErrorCode> ResolveRedisDbIndex();
+
+tl::expected<RedisEndpoint, ErrorCode> ParseRedisEndpoint(
+    std::string_view connstring);
+
+std::string SanitizeHashTagComponent(std::string component);
+
+tl::expected<RedisContextPtr, ErrorCode> ConnectRedis(
+    std::string_view connstring,
+    ErrorCode connection_error = ErrorCode::PERSISTENT_FAIL);
+
+}  // namespace redis
+}  // namespace backends
+}  // namespace ha
+}  // namespace mooncake

mooncake-store/include/ha/backends/redis/redis_snapshot_store.h

@@ -0,0 +1,46 @@
+#pragma once
+
+#include <string>
+
+#include "ha/snapshot_store.h"
+#include "serialize/serializer_backend.h"
+
+namespace mooncake {
+namespace ha {
+namespace backends {
+namespace redis {
+
+class RedisSnapshotStore final : public SnapshotStore {
+   public:
+    RedisSnapshotStore(SerializerBackend* payload_backend,
+                       std::string connstring,
+                       ClusterNamespace cluster_namespace);
+
+    ErrorCode Publish(const SnapshotDescriptor& snapshot) override;
+
+    tl::expected<std::optional<SnapshotDescriptor>, ErrorCode> GetLatest()
+        override;
+
+    tl::expected<std::vector<SnapshotDescriptor>, ErrorCode> List(
+        size_t limit) override;
+
+    ErrorCode Delete(const SnapshotId& snapshot_id) override;
+
+   private:
+    static ClusterNamespace ResolveClusterNamespace(
+        const ClusterNamespace& cluster_namespace);
+    static std::string BuildLatestKey(
+        const ClusterNamespace& cluster_namespace);
+    static std::string BuildIndexKey(const ClusterNamespace& cluster_namespace);
+
+    SerializerBackend* payload_backend_;
+    std::string connstring_;
+    ClusterNamespace cluster_namespace_;
+    std::string latest_key_;
+    std::string index_key_;
+};
+
+}  // namespace redis
+}  // namespace backends
+}  // namespace ha
+}  // namespace mooncake

mooncake-store/include/ha/serializer_snapshot_store.h

@@ -0,0 +1,28 @@
+#pragma once
+
+#include "ha/snapshot_store.h"
+#include "serialize/serializer_backend.h"
+
+namespace mooncake {
+namespace ha {
+
+class SerializerSnapshotStore : public SnapshotStore {
+   public:
+    explicit SerializerSnapshotStore(SerializerBackend* backend);
+
+    ErrorCode Publish(const SnapshotDescriptor& snapshot) override;
+
+    tl::expected<std::optional<SnapshotDescriptor>, ErrorCode> GetLatest()
+        override;
+
+    tl::expected<std::vector<SnapshotDescriptor>, ErrorCode> List(
+        size_t limit) override;
+
+    ErrorCode Delete(const SnapshotId& snapshot_id) override;
+
+   private:
+    SerializerBackend* backend_;
+};
+
+}  // namespace ha
+}  // namespace mooncake

mooncake-store/include/ha/snapshot_store.h

@@ -1,7 +1,10 @@
 #pragma once
 
+#include <cctype>
 #include <cstddef>
 #include <optional>
+#include <string>
+#include <string_view>
 #include <vector>
 
 #include <ylt/util/tl/expected.hpp>
@@ -11,6 +14,71 @@
 namespace mooncake {
 namespace ha {
 
+namespace snapshot_store_detail {
+
+constexpr std::string_view kSnapshotRoot = "mooncake_master_snapshot/";
+constexpr std::string_view kSnapshotLatest = "latest.txt";
+constexpr std::string_view kSnapshotManifest = "manifest.txt";
+
+inline bool IsAsciiDigit(char ch) {
+    return std::isdigit(static_cast<unsigned char>(ch)) != 0;
+}
+
+inline bool IsValidSnapshotId(std::string_view snapshot_id) {
+    if (snapshot_id.size() != 19) {
+        return false;
+    }
+
+    for (size_t i = 0; i < snapshot_id.size(); ++i) {
+        if (i == 8 || i == 15) {
+            if (snapshot_id[i] != '_') {
+                return false;
+            }
+            continue;
+        }
+
+        if (!IsAsciiDigit(snapshot_id[i])) {
+            return false;
+        }
+    }
+
+    return true;
+}
+
+inline std::string TrimAsciiWhitespace(std::string value) {
+    constexpr std::string_view kAsciiWhitespace = " \t\n\r\f\v";
+    const auto first = value.find_first_not_of(kAsciiWhitespace);
+    if (first == std::string::npos) {
+        return "";
+    }
+
+    const auto last = value.find_last_not_of(kAsciiWhitespace);
+    return value.substr(first, last - first + 1);
+}
+
+inline std::string BuildSnapshotPrefix(const SnapshotId& snapshot_id) {
+    return std::string(kSnapshotRoot) + snapshot_id + "/";
+}
+
+inline std::string BuildManifestKey(const SnapshotId& snapshot_id) {
+    return BuildSnapshotPrefix(snapshot_id) + std::string(kSnapshotManifest);
+}
+
+inline std::string BuildLatestKey() {
+    return std::string(kSnapshotRoot) + std::string(kSnapshotLatest);
+}
+
+inline SnapshotDescriptor MakeSnapshotDescriptor(
+    const SnapshotId& snapshot_id) {
+    SnapshotDescriptor descriptor;
+    descriptor.snapshot_id = snapshot_id;
+    descriptor.manifest_key = BuildManifestKey(snapshot_id);
+    descriptor.object_prefix = BuildSnapshotPrefix(snapshot_id);
+    return descriptor;
+}
+
+}  // namespace snapshot_store_detail
+
 class SnapshotStore {
    public:
     virtual ~SnapshotStore() = default;