Store naming

cpp/arcticdb/async/async_store.hpp

@@ -371,6 +371,10 @@ std::vector<folly::Future<bool>> batch_key_exists(
         library_->set_failure_sim(cfg);
     }
 
+    std::string name() const override {
+        return library_->name();
+    }
+
 private:
     std::shared_ptr<storage::Library> library_;
     std::shared_ptr<arcticdb::proto::encoding::VariantCodec> codec_;

cpp/arcticdb/entity/key.cpp

@@ -64,6 +64,7 @@ KeyData get_key_data(KeyType key_type) {
     STRING_KEY(KeyType::APPEND_DATA, app, 'b')
     // Unused
     STRING_KEY(KeyType::PARTITION, pref, 'p')
+    STRING_KEY(KeyType::REPLICATION_FAIL_INFO, rfail, 'F')
     STRING_REF(KeyType::STORAGE_INFO, sref, 'h')
     NUMERIC_KEY(KeyType::STREAM_GROUP, sg, 'g')
     NUMERIC_KEY(KeyType::GENERATION, gen, 'G')

cpp/arcticdb/entity/key.hpp

@@ -178,6 +178,10 @@ enum class KeyType : int {
      * Contains column stats about the index key with the same stream ID and version number
      */
     COLUMN_STATS = 25,
+    /*
+     * Used for storing the ids of storages that failed to sync
+     */
+    REPLICATION_FAIL_INFO = 26,
     UNDEFINED
 };
 

cpp/arcticdb/storage/azure/azure_storage.cpp

@@ -300,6 +300,10 @@ bool do_key_exists_impl(
 }
 } //namespace detail
 
+std::string AzureStorage::name() const {
+    return fmt::format("azure_storage-{}/{}", container_name_, root_folder_);
+}
+
 void AzureStorage::do_write(Composite<KeySegmentPair>&& kvs) {
     detail::do_write_impl(std::move(kvs), root_folder_, *azure_client_, FlatBucketizer{}, upload_option_, request_timeout_);
 }
@@ -338,6 +342,7 @@ using namespace Azure::Storage::Blobs;
 AzureStorage::AzureStorage(const LibraryPath &library_path, OpenMode mode, const Config &conf) :
     Storage(library_path, mode),
     root_folder_(object_store_utils::get_root_folder(library_path)),
+    container_name_(conf.container_name()),
     request_timeout_(conf.request_timeout() == 0 ? 200000 : conf.request_timeout()) {
         if(conf.use_mock_storage_for_testing()) {
             ARCTICDB_RUNTIME_DEBUG(log::storage(), "Using Mock Azure storage");

cpp/arcticdb/storage/azure/azure_storage.hpp

@@ -31,6 +31,8 @@ class AzureStorage final : public Storage {
 
     AzureStorage(const LibraryPath &lib, OpenMode mode, const Config &conf);
 
+    std::string name() const final;
+
   protected:
     void do_write(Composite<KeySegmentPair>&& kvs) final;
 
@@ -58,6 +60,7 @@ class AzureStorage final : public Storage {
     std::unique_ptr<AzureClientWrapper> azure_client_;
 
     std::string root_folder_;
+    std::string container_name_;
     unsigned int request_timeout_;
     Azure::Storage::Blobs::UploadBlockBlobFromOptions upload_option_;
     Azure::Storage::Blobs::DownloadBlobToOptions download_option_;

cpp/arcticdb/storage/file/mapped_file_storage.cpp

@@ -25,6 +25,10 @@ MappedFileStorage::MappedFileStorage(const LibraryPath &lib, OpenMode mode, Conf
     init();
 }
 
+std::string MappedFileStorage::name() const {
+    return fmt::format("mapped_file_storage-{}", config_.path());
+}
+
 void MappedFileStorage::do_write_raw(const uint8_t* data, size_t bytes) {
     ARCTICDB_DEBUG(log::storage(), "Writing {} bytes to mapped file storage at offset {}", bytes, offset_);
     memcpy(file_.data() + offset_, data, bytes);

cpp/arcticdb/storage/file/mapped_file_storage.hpp

@@ -26,6 +26,9 @@ class MappedFileStorage final : public SingleFileStorage {
 
     ~MappedFileStorage() override = default;
 
+    std::string name() const final;
+
+
   private:
     void do_write_raw(const uint8_t* data, size_t bytes) override;
 

cpp/arcticdb/storage/library.hpp

@@ -168,7 +168,8 @@ class Library {
     }
 
     std::string name() {
-        return library_path_.to_delim_path();
+        auto lib_name = storages_->name();
+        return lib_name;
     }
 
   private:

cpp/arcticdb/storage/lmdb/lmdb_storage.cpp

@@ -83,6 +83,10 @@ void LmdbStorage::do_write_internal(Composite<KeySegmentPair>&& kvs, ::lmdb::txn
     });
 }
 
+std::string LmdbStorage::name() const {
+    return fmt::format("lmdb_storage-{}", lib_dir_.string());
+}
+
 void LmdbStorage::do_write(Composite<KeySegmentPair>&& kvs) {
     ARCTICDB_SAMPLE(LmdbStorageWrite, 0)
     std::lock_guard<std::mutex> lock{*write_mutex_};

cpp/arcticdb/storage/lmdb/lmdb_storage.hpp

@@ -36,6 +36,8 @@ class LmdbStorage final : public Storage {
     LmdbStorage(LmdbStorage&& other) noexcept;
     ~LmdbStorage() override;
 
+    std::string name() const final;
+
   private:
     void do_write(Composite<KeySegmentPair>&& kvs) final;
 

cpp/arcticdb/storage/memory/memory_storage.cpp

@@ -18,6 +18,10 @@ namespace arcticdb::storage::memory {
 
     namespace fg = folly::gen;
 
+    std::string MemoryStorage::name() const {
+        return "memory_storage-0";
+    }
+
     void MemoryStorage::do_write(Composite<KeySegmentPair>&& kvs) {
         ARCTICDB_SAMPLE(MemoryStorageWrite, 0)
 

cpp/arcticdb/storage/memory/memory_storage.hpp

@@ -23,6 +23,8 @@ namespace arcticdb::storage::memory {
 
         MemoryStorage(const LibraryPath &lib, OpenMode mode, const Config &conf);
 
+        std::string name() const final;
+
     private:
         void do_write(Composite<KeySegmentPair>&& kvs) final;
 

cpp/arcticdb/storage/mongo/mongo_storage.cpp

@@ -63,6 +63,10 @@ void raise_if_unexpected_error(const mongocxx::operation_exception& e) {
     }
 }
 
+std::string MongoStorage::name() const {
+    return fmt::format("mongo_storage-{}", db_);
+}
+
 void MongoStorage::do_write(Composite<KeySegmentPair>&& kvs) {
     namespace fg = folly::gen;
     auto fmt_db = [](auto &&kv) { return kv.key_type(); };

cpp/arcticdb/storage/mongo/mongo_storage.hpp

@@ -25,6 +25,8 @@ class MongoStorage final : public Storage {
 
     MongoStorage(const LibraryPath &lib, OpenMode mode, const Config &conf);
 
+    std::string name() const final;
+
   private:
     void do_write(Composite<KeySegmentPair>&& kvs) final;
 

cpp/arcticdb/storage/rocksdb/rocksdb_storage.cpp

@@ -33,7 +33,7 @@ RocksDBStorage::RocksDBStorage(const LibraryPath &library_path, OpenMode mode, c
     auto lib_path_str = library_path.to_delim_path(fs::path::preferred_separator);
 
     auto lib_dir = root_path / lib_path_str;
-    auto db_name = lib_dir.generic_string();
+    db_name_ = lib_dir.generic_string();
 
     std::set<std::string> key_names;
     arcticdb::entity::foreach_key_type([&](KeyType&& key_type) {
@@ -45,7 +45,7 @@ RocksDBStorage::RocksDBStorage(const LibraryPath &library_path, OpenMode mode, c
     ::rocksdb::DBOptions db_options;
 
     ::rocksdb::ConfigOptions cfg_opts;
-    auto s = ::rocksdb::LoadLatestOptions(cfg_opts, db_name, &db_options, &column_families);
+    auto s = ::rocksdb::LoadLatestOptions(cfg_opts, db_name_, &db_options, &column_families);
     if (s.ok()) {
         std::set<std::string> existing_key_names{};
         for (const auto& desc : column_families) {
@@ -56,7 +56,7 @@ RocksDBStorage::RocksDBStorage(const LibraryPath &library_path, OpenMode mode, c
         util::check(existing_key_names == key_names, "Existing database has incorrect key columns.");
     } else if (s.IsNotFound()) {
         util::check_arg(mode > OpenMode::READ, "Missing dir {} for lib={}. mode={}",
-                            db_name, lib_path_str, mode);
+                            db_name_, lib_path_str, mode);
         // Default column family required, error if not provided.
         column_families.emplace_back(::rocksdb::kDefaultColumnFamilyName, ::rocksdb::ColumnFamilyOptions());
         for (const auto& key_name: key_names) {
@@ -75,7 +75,7 @@ RocksDBStorage::RocksDBStorage(const LibraryPath &library_path, OpenMode mode, c
     std::vector<::rocksdb::ColumnFamilyHandle*> handles;
     // Note: the "default" handle will be returned as well. It is necessary to delete this, but not
     // rocksdb's internal handle to the default column family as returned by DefaultColumnFamily().
-    s = ::rocksdb::DB::Open(db_options, db_name, column_families, &handles, &db_);
+    s = ::rocksdb::DB::Open(db_options, db_name_, column_families, &handles, &db_);
     storage::check<ErrorCode::E_UNEXPECTED_ROCKSDB_ERROR>(s.ok(), DEFAULT_ROCKSDB_NOT_OK_ERROR + s.ToString());
     util::check(handles.size() == column_families.size(), "Open returned wrong number of handles.");
     for (std::size_t i = 0; i < handles.size(); i++) {
@@ -95,6 +95,10 @@ RocksDBStorage::~RocksDBStorage() {
     delete db_;
 }
 
+std::string RocksDBStorage::name() const {
+    return fmt::format("rocksdb_storage-{}", db_name_);
+}
+
 void RocksDBStorage::do_write(Composite<KeySegmentPair>&& kvs) {
     ARCTICDB_SAMPLE(RocksDBStorageWrite, 0)
     do_write_internal(std::move(kvs));

cpp/arcticdb/storage/rocksdb/rocksdb_storage.hpp

@@ -25,6 +25,8 @@ namespace arcticdb::storage::rocksdb {
         RocksDBStorage(const LibraryPath &lib, OpenMode mode, const Config &conf);
         ~RocksDBStorage() override;
 
+        std::string name() const final;
+
     private:
         void do_write(Composite<KeySegmentPair>&& kvs) final override;
 
@@ -55,6 +57,7 @@ namespace arcticdb::storage::rocksdb {
         using MapKeyType = std::string;
         using HandleType = ::rocksdb::ColumnFamilyHandle*;
         std::unordered_map<MapKeyType, HandleType> handles_by_key_type_;
+        std::string db_name_;
 
         inline static const std::string DEFAULT_ROCKSDB_NOT_OK_ERROR = "RocksDB Unexpected Error, RocksDB status not OK: ";
     };

cpp/arcticdb/storage/s3/nfs_backed_storage.cpp

@@ -118,6 +118,10 @@ VariantKey unencode_object_id(const VariantKey& key) {
                         });
 }
 
+std::string NfsBackedStorage::name() const {
+    return fmt::format("nfs_backed_storage-{}/{}/{}", region_, bucket_name_, root_folder_);
+}
+
 void NfsBackedStorage::do_write(Composite<KeySegmentPair>&& kvs) {
     auto enc = kvs.transform([] (auto&& key_seg) {
         return KeySegmentPair{encode_object_id(key_seg.variant_key()), std::move(key_seg.segment())};
@@ -181,7 +185,8 @@ NfsBackedStorage::NfsBackedStorage(const LibraryPath &library_path, OpenMode mod
     Storage(library_path, mode),
     s3_api_(s3::S3ApiInstance::instance()),
     root_folder_(object_store_utils::get_root_folder(library_path)),
-    bucket_name_(conf.bucket_name()) {
+    bucket_name_(conf.bucket_name()),
+    region_(conf.region()) {
     s3_client_ = std::make_unique<s3::RealS3Client>(s3::get_aws_credentials(conf), s3::get_s3_config(conf), Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, false);
     if (!conf.prefix().empty()) {
         ARCTICDB_DEBUG(log::version(), "prefix found, using: {}", conf.prefix());

cpp/arcticdb/storage/s3/nfs_backed_storage.hpp

@@ -30,6 +30,8 @@ class NfsBackedStorage final : public Storage {
 
     NfsBackedStorage(const LibraryPath &lib, OpenMode mode, const Config &conf);
 
+    std::string name() const final;
+
 private:
     void do_write(Composite<KeySegmentPair>&& kvs) final;
 
@@ -56,11 +58,13 @@ class NfsBackedStorage final : public Storage {
     auto& client() { return s3_client_; }
     const std::string& bucket_name() const { return bucket_name_; }
     const std::string& root_folder() const { return root_folder_; }
+    const std::string& region() const { return region_; }
 
     std::shared_ptr<storage::s3::S3ApiInstance> s3_api_;
     std::unique_ptr<storage::s3::S3ClientWrapper> s3_client_;
     std::string root_folder_;
     std::string bucket_name_;
+    std::string region_;
 };
 
 inline arcticdb::proto::storage::VariantStorage pack_config(const std::string &bucket_name) {

cpp/arcticdb/storage/s3/s3_storage.cpp

@@ -53,6 +53,9 @@ namespace s3 {
 
 namespace fg = folly::gen;
 
+std::string S3Storage::name() const {
+    return fmt::format("s3_storage-{}/{}/{}", region_, bucket_name_, root_folder_);
+}
 
 std::string S3Storage::get_key_path(const VariantKey& key) const {
     auto b = FlatBucketizer{};
@@ -101,7 +104,8 @@ S3Storage::S3Storage(const LibraryPath &library_path, OpenMode mode, const Confi
     Storage(library_path, mode),
     s3_api_(S3ApiInstance::instance()),
     root_folder_(object_store_utils::get_root_folder(library_path)),
-    bucket_name_(conf.bucket_name()) {
+    bucket_name_(conf.bucket_name()),
+    region_(conf.region()) {
 
     auto creds = get_aws_credentials(conf);
 

cpp/arcticdb/storage/s3/s3_storage.hpp

@@ -39,6 +39,8 @@ class S3Storage final : public Storage {
      */
     std::string get_key_path(const VariantKey& key) const;
 
+    std::string name() const final;
+
   private:
     void do_write(Composite<KeySegmentPair>&& kvs) final;
 
@@ -70,6 +72,7 @@ class S3Storage final : public Storage {
     std::unique_ptr<S3ClientWrapper> s3_client_;
     std::string root_folder_;
     std::string bucket_name_;
+    std::string region_;
 };
 
 inline arcticdb::proto::storage::VariantStorage pack_config(const std::string &bucket_name) {

cpp/arcticdb/storage/single_file_storage.hpp

@@ -15,6 +15,8 @@ class SingleFileStorage : public Storage {
     SingleFileStorage(const LibraryPath &lib, OpenMode mode) :
         Storage(lib, mode) {}
 
+    std::string name() const = 0;
+
     void write_raw(const uint8_t* data, size_t bytes) {
         do_write_raw(data, bytes);
     }

cpp/arcticdb/storage/storage.hpp

@@ -173,6 +173,8 @@ class Storage {
     [[nodiscard]] const LibraryPath &library_path() const { return lib_path_; }
     [[nodiscard]] OpenMode open_mode() const { return mode_; }
 
+    virtual std::string name() const = 0;
+
 private:
     virtual void do_write(Composite<KeySegmentPair>&& kvs) = 0;
 

cpp/arcticdb/storage/storages.hpp

@@ -144,6 +144,10 @@ class Storages {
             return std::nullopt;
         }
     }
+    std::string name() const {
+        return primary().name();
+    }
+
   private:
     Storage& primary() {
         util::check(!storages_.empty(), "No storages configured");

cpp/arcticdb/storage/store.hpp

@@ -34,6 +34,8 @@ class Store : public stream::StreamSink, public stream::StreamSource, public std
     virtual folly::Future<VariantKey> copy(KeyType key_type, const StreamId& stream_id, VersionId version_id, const VariantKey& source_key) = 0;
 
     virtual VariantKey copy_sync(KeyType key_type, const StreamId& stream_id, VersionId version_id, const VariantKey& source_key) = 0;
+
+    virtual std::string name() const = 0;
 };
 
 } // namespace arcticdb

cpp/arcticdb/storage/test/in_memory_store.hpp

@@ -431,6 +431,10 @@ namespace arcticdb {
 
         void set_failure_sim(const arcticdb::proto::storage::VersionStoreConfig::StorageFailureSimulator &) override {}
 
+        std::string name() const override {
+            return "InMemoryStore";
+        }
+
         void add_segment(const AtomKey &key, SegmentInMemory &&seg) {
             StorageFailureSimulator::instance()->go(FailureType::WRITE);
             std::lock_guard lock{mutex_};

cpp/arcticdb/stream/test/mock_stores.hpp

@@ -142,6 +142,10 @@ class NullStore :
 
     void set_failure_sim(const arcticdb::proto::storage::VersionStoreConfig::StorageFailureSimulator &) override {}
 
+    std::string name() const override {
+        return "NullStore";
+    }
+
 };
 
 } //namespace arcticdb

cpp/arcticdb/version/version_constants.hpp

@@ -15,5 +15,8 @@ namespace arcticdb {
     static const char* const DeleteSnapshotId = "__delete_snapshot__";
     static const char* const LastSyncId = "__last_sync__";
     static const char* const LastBackupId = "__last_backup__";
+    static const char* const FailedTargetId = "__failed_target__";
     static const char* const StorageLogId = "__storage_log__";
+    static const char* const FailedStorageLogId = "__failed_storage_log__";
+
 }