Merge pull request ceph#57107 from ifed01/wip-ifed-async-compact

os/bluestore: enable async manual compactions

Reviewed-by: Adam Kupczyk <[email protected]>
Reviewed-by: Pere Diaz Bou <[email protected]>

Author: yuriw

src/common/options/global.yaml.in

@@ -5070,6 +5070,13 @@ options:
     This setting is used only when OSD is doing ``--mkfs``.
     Next runs of OSD retrieve sharding from disk.
   default: m(3) p(3,0-12) O(3,0-13)=block_cache={type=binned_lru} L=min_write_buffer_number_to_merge=32 P=min_write_buffer_number_to_merge=32
+- name: bluestore_async_db_compaction
+  type: bool
+  level: dev
+  desc: Perform DB compaction requests asynchronously
+  long_desc: 'How to perform DB compactions triggered either through async socket or
+    by OSD initialization procedure on start.'
+  default: true
 - name: bluestore_qfsck_on_mount
   type: bool
   level: dev

src/kv/RocksDBStore.cc

@@ -1243,7 +1243,9 @@ int RocksDBStore::do_open(ostream &out,
   plb.add_time_avg(l_rocksdb_submit_latency, "submit_latency", "Submit Latency");
   plb.add_time_avg(l_rocksdb_submit_sync_latency, "submit_sync_latency", "Submit Sync Latency");
   plb.add_u64_counter(l_rocksdb_compact, "compact", "Compactions");
-  plb.add_u64_counter(l_rocksdb_compact_range, "compact_range", "Compactions by range");
+  plb.add_u64_counter(l_rocksdb_compact_running, "compact_running", "Running compactions");
+  plb.add_u64_counter(l_rocksdb_compact_completed, "compact_completed", "Completed compactions");
+  plb.add_time(l_rocksdb_compact_lasted, "compact_lasted", "Last completed compaction duration");
   plb.add_u64_counter(l_rocksdb_compact_queue_merge, "compact_queue_merge", "Mergings of ranges in compaction queue");
   plb.add_u64(l_rocksdb_compact_queue_len, "compact_queue_len", "Length of compaction queue");
   plb.add_time_avg(l_rocksdb_write_wal_time, "rocksdb_write_wal_time", "Rocksdb write wal time");
@@ -1988,6 +1990,7 @@ int RocksDBStore::split_key(rocksdb::Slice in, string *prefix, string *key)
 
 void RocksDBStore::compact()
 {
+  dout(2) << __func__ << " starting" << dendl;
   logger->inc(l_rocksdb_compact);
   rocksdb::CompactRangeOptions options;
   db->CompactRange(options, default_cf, nullptr, nullptr);
@@ -1999,6 +2002,7 @@ void RocksDBStore::compact()
 	nullptr, nullptr);
     }
   }
+  dout(2) << __func__ << " completed" << dendl;
 }
 
 void RocksDBStore::compact_thread_entry()
@@ -2011,12 +2015,17 @@ void RocksDBStore::compact_thread_entry()
       compact_queue.pop_front();
       logger->set(l_rocksdb_compact_queue_len, compact_queue.size());
       l.unlock();
-      logger->inc(l_rocksdb_compact_range);
+      logger->inc(l_rocksdb_compact_running);
+      auto start = ceph_clock_now();
       if (range.first.empty() && range.second.empty()) {
         compact();
       } else {
         compact_range(range.first, range.second);
       }
+      auto lat = ceph_clock_now() - start;
+      logger->dec(l_rocksdb_compact_running);
+      logger->inc(l_rocksdb_compact_completed);
+      logger->tset(l_rocksdb_compact_lasted, lat);
       l.lock();
       continue;
     }

src/kv/RocksDBStore.h

@@ -35,7 +35,9 @@ enum {
   l_rocksdb_submit_latency,
   l_rocksdb_submit_sync_latency,
   l_rocksdb_compact,
-  l_rocksdb_compact_range,
+  l_rocksdb_compact_running,
+  l_rocksdb_compact_completed,
+  l_rocksdb_compact_lasted,
   l_rocksdb_compact_queue_merge,
   l_rocksdb_compact_queue_len,
   l_rocksdb_write_wal_time,

src/os/ObjectStore.h

@@ -791,7 +791,7 @@ class ObjectStore {
   virtual void inject_data_error(const ghobject_t &oid) {}
   virtual void inject_mdata_error(const ghobject_t &oid) {}
 
-  virtual void compact() {}
+  virtual int compact() { return -ENOTSUP; }
   virtual bool has_builtin_csum() const {
     return false;
   }

src/os/bluestore/BlueStore.cc

@@ -11305,6 +11305,22 @@ void BlueStore::inject_bluefs_file(std::string_view dir, std::string_view name,
   bluefs->close_writer(p_handle);
 }
 
+int BlueStore::compact()
+{
+  int r = 0;
+  ceph_assert(db);
+  if (cct->_conf.get_val<bool>("bluestore_async_db_compaction")) {
+    dout(1) << __func__ << " starting async.." << dendl;
+    db->compact_async();
+    r = -EINPROGRESS;
+  } else {
+    dout(1) << __func__ << " starting sync.." << dendl;
+    db->compact();
+    dout(1) << __func__ << " finished." << dendl;
+  }
+  return r;
+}
+
 void BlueStore::collect_metadata(map<string,string> *pm)
 {
   dout(10) << __func__ << dendl;

src/os/bluestore/BlueStore.h

@@ -3403,10 +3403,7 @@ class BlueStore : public ObjectStore,
 			  std::string_view name,
 			  size_t new_size);
 
-  void compact() override {
-    ceph_assert(db);
-    db->compact();
-  }
+  int compact() override;
   bool has_builtin_csum() const override {
     return true;
   }

src/os/kstore/KStore.h

@@ -580,9 +580,10 @@ class KStore : public ObjectStore {
     TrackedOpRef op = TrackedOpRef(),
     ThreadPool::TPHandle *handle = NULL) override;
 
-  void compact () override {
+  int compact () override {
     ceph_assert(db);
     db->compact();
+    return 0;
   }
 
 private:

src/osd/OSD.cc

@@ -3000,15 +3000,22 @@ will start to track new ops received afterwards.";
   } else if (prefix == "compact") {
     dout(1) << "triggering manual compaction" << dendl;
     auto start = ceph::coarse_mono_clock::now();
-    store->compact();
-    auto end = ceph::coarse_mono_clock::now();
-    double duration = std::chrono::duration<double>(end-start).count();
-    dout(1) << "finished manual compaction in "
-            << duration
-            << " seconds" << dendl;
-    f->open_object_section("compact_result");
-    f->dump_float("elapsed_time", duration);
-    f->close_section();
+    int r = store->compact();
+    if (r == 0) {
+      auto end = ceph::coarse_mono_clock::now();
+      double duration = std::chrono::duration<double>(end-start).count();
+
+      dout(1) << "finished manual compaction in "
+              << duration
+              << " seconds" << dendl;
+      f->open_object_section("compact_result");
+      f->dump_float("elapsed_time", duration);
+      f->close_section();
+    } else  if ( r == -EINPROGRESS) {
+      dout(1) << "manual compaction is being executed asynchronously" << dendl;
+    } else {
+      derr << "error starting manual compaction:" << cpp_strerror(r) << dendl;
+    }
   } else if (prefix == "get_mapped_pools") {
     f->open_array_section("mapped_pools");
     set<int64_t> poollist = get_mapped_pools();
@@ -3924,7 +3931,7 @@ int OSD::init()
   dout(2) << "superblock: I am osd." << superblock.whoami << dendl;
 
   if (cct->_conf.get_val<bool>("osd_compact_on_start")) {
-    dout(2) << "compacting object store's omap" << dendl;
+    dout(2) << "compacting object store's DB" << dendl;
     store->compact();
   }