From 93a7389442ab68959d1ea8e4df8211e119d63577 Mon Sep 17 00:00:00 2001 From: Jay Zhuang Date: Tue, 29 Jun 2021 11:46:44 -0700 Subject: [PATCH] Add statistics support on CompactionService remote side (#8368) Summary: Add statistics option on CompactionService remote side. Pull Request resolved: https://github.com/facebook/rocksdb/pull/8368 Test Plan: `make check` Reviewed By: ajkr Differential Revision: D28944427 Pulled By: jay-zhuang fbshipit-source-id: 2a19217f4a69b6e511af87eed12391860ef00c5e --- HISTORY.md | 2 + db/compaction/compaction_service_test.cc | 47 ++++++++++++++---------- db/db_impl/db_impl_secondary.cc | 1 + include/rocksdb/options.h | 5 +++ 4 files changed, 36 insertions(+), 19 deletions(-) diff --git a/HISTORY.md b/HISTORY.md index b080090ff..7893629dc 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -11,6 +11,8 @@ * ldb has a new feature, `list_live_files_metadata`, that shows the live SST files, as well as their LSM storage level and the column family they belong to. * The new BlobDB implementation now tracks the amount of garbage in each blob file in the MANIFEST. * Integrated BlobDB now supports Merge with base values (Put/Delete etc.). +* RemoteCompaction supports sub-compaction, the job_id in the user interface is changed from `int` to `uint64_t` to support sub-compaction id. +* Expose statistics option in RemoteCompaction worker. ### Public API change * Added APIs to the Customizable class to allow developers to create their own Customizable classes. Created the utilities/customizable_util.h file to contain helper methods for developing new Customizable classes. diff --git a/db/compaction/compaction_service_test.cc b/db/compaction/compaction_service_test.cc index 03e626181..9128bba98 100644 --- a/db/compaction/compaction_service_test.cc +++ b/db/compaction/compaction_service_test.cc @@ -12,9 +12,9 @@ namespace ROCKSDB_NAMESPACE { class MyTestCompactionService : public CompactionService { public: - MyTestCompactionService(const std::string& db_path, - std::shared_ptr fs, Options& options) - : db_path_(db_path), fs_(fs), options_(options) {} + MyTestCompactionService(const std::string& db_path, Options& options, + std::shared_ptr statistics = nullptr) + : db_path_(db_path), options_(options), statistics_(statistics) {} static const char* kClassName() { return "MyTestCompactionService"; } @@ -54,6 +54,7 @@ class MyTestCompactionService : public CompactionService { options_override.prefix_extractor = options_.prefix_extractor; options_override.table_factory = options_.table_factory; options_override.sst_partitioner_factory = options_.sst_partitioner_factory; + options_override.statistics = statistics_; Status s = DB::OpenAndCompact( db_path_, db_path_ + "/" + ROCKSDB_NAMESPACE::ToString(job_id), @@ -75,8 +76,8 @@ class MyTestCompactionService : public CompactionService { std::atomic_int compaction_num_{0}; std::map jobs_; const std::string db_path_; - std::shared_ptr fs_; Options options_; + std::shared_ptr statistics_; }; class CompactionServiceTest : public DBTestBase { @@ -123,8 +124,10 @@ class CompactionServiceTest : public DBTestBase { TEST_F(CompactionServiceTest, BasicCompactions) { Options options = CurrentOptions(); options.env = env_; + options.statistics = CreateDBStatistics(); + std::shared_ptr compactor_statistics = CreateDBStatistics(); options.compaction_service = std::make_shared( - dbname_, env_->GetFileSystem(), options); + dbname_, options, compactor_statistics); DestroyAndReopen(options); @@ -158,6 +161,12 @@ TEST_F(CompactionServiceTest, BasicCompactions) { dynamic_cast(options.compaction_service.get()); ASSERT_GE(my_cs->GetCompactionNum(), 1); + // make sure the compaction statistics is only recorded on remote side + ASSERT_GE( + compactor_statistics->getTickerCount(COMPACTION_KEY_DROP_NEWER_ENTRY), 1); + ASSERT_EQ(options.statistics->getTickerCount(COMPACTION_KEY_DROP_NEWER_ENTRY), + 0); + // Test failed compaction SyncPoint::GetInstance()->SetCallBack( "DBImplSecondary::CompactWithoutInstallation::End", [&](void* status) { @@ -195,8 +204,8 @@ TEST_F(CompactionServiceTest, ManualCompaction) { Options options = CurrentOptions(); options.env = env_; options.disable_auto_compactions = true; - options.compaction_service = std::make_shared( - dbname_, env_->GetFileSystem(), options); + options.compaction_service = + std::make_shared(dbname_, options); DestroyAndReopen(options); GenerateTestData(); @@ -236,8 +245,8 @@ TEST_F(CompactionServiceTest, FailedToStart) { Options options = CurrentOptions(); options.env = env_; options.disable_auto_compactions = true; - options.compaction_service = std::make_shared( - dbname_, env_->GetFileSystem(), options); + options.compaction_service = + std::make_shared(dbname_, options); DestroyAndReopen(options); GenerateTestData(); @@ -261,8 +270,8 @@ TEST_F(CompactionServiceTest, InvalidResult) { Options options = CurrentOptions(); options.env = env_; options.disable_auto_compactions = true; - options.compaction_service = std::make_shared( - dbname_, env_->GetFileSystem(), options); + options.compaction_service = + std::make_shared(dbname_, options); DestroyAndReopen(options); GenerateTestData(); @@ -288,8 +297,8 @@ TEST_F(CompactionServiceTest, SubCompaction) { options.max_subcompactions = 10; options.target_file_size_base = 1 << 10; // 1KB options.disable_auto_compactions = true; - options.compaction_service = std::make_shared( - dbname_, env_->GetFileSystem(), options); + options.compaction_service = + std::make_shared(dbname_, options); DestroyAndReopen(options); GenerateTestData(); @@ -330,8 +339,8 @@ TEST_F(CompactionServiceTest, CompactionFilter) { options.env = env_; auto delete_comp_filter = PartialDeleteCompactionFilter(); options.compaction_filter = &delete_comp_filter; - options.compaction_service = std::make_shared( - dbname_, env_->GetFileSystem(), options); + options.compaction_service = + std::make_shared(dbname_, options); DestroyAndReopen(options); @@ -373,8 +382,8 @@ TEST_F(CompactionServiceTest, CompactionFilter) { TEST_F(CompactionServiceTest, Snapshot) { Options options = CurrentOptions(); options.env = env_; - options.compaction_service = std::make_shared( - dbname_, env_->GetFileSystem(), options); + options.compaction_service = + std::make_shared(dbname_, options); DestroyAndReopen(options); @@ -400,8 +409,8 @@ TEST_F(CompactionServiceTest, ConcurrentCompaction) { Options options = CurrentOptions(); options.level0_file_num_compaction_trigger = 100; options.env = env_; - options.compaction_service = std::make_shared( - dbname_, env_->GetFileSystem(), options); + options.compaction_service = + std::make_shared(dbname_, options); options.max_background_jobs = 20; DestroyAndReopen(options); diff --git a/db/db_impl/db_impl_secondary.cc b/db/db_impl/db_impl_secondary.cc index 10c04b3e1..dae004cdd 100644 --- a/db/db_impl/db_impl_secondary.cc +++ b/db/db_impl/db_impl_secondary.cc @@ -767,6 +767,7 @@ Status DB::OpenAndCompact( compaction_input.db_options.env = override_options.env; compaction_input.db_options.file_checksum_gen_factory = override_options.file_checksum_gen_factory; + compaction_input.db_options.statistics = override_options.statistics; compaction_input.column_family.options.comparator = override_options.comparator; compaction_input.column_family.options.merge_operator = diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h index 7b8d238f7..affbb7aa0 100644 --- a/include/rocksdb/options.h +++ b/include/rocksdb/options.h @@ -1797,6 +1797,11 @@ struct CompactionServiceOptionsOverride { std::shared_ptr prefix_extractor = nullptr; std::shared_ptr table_factory; std::shared_ptr sst_partitioner_factory = nullptr; + + // statistics is used to collect DB operation metrics, the metrics won't be + // returned to CompactionService primary host, to collect that, the user needs + // to set it here. + std::shared_ptr statistics = nullptr; }; } // namespace ROCKSDB_NAMESPACE