kv/Rocksdbstore: change output format for short txc dump.

Plus refactors txc dumping  code to permit Unit Testing for it.

Here is the new output sample:
Short form:
 DF:D*3,DF:O,MF:A,PF:O*2,PF:P,...*2
Verbose form:
PutCF(prefix = O, key = 'key', val len = 5)
PutCF(prefix = P, key = 'key2', val len = 6)
PutCF(prefix = O, key = 'key3', val len = 6)
DeleteCF(prefix = O, key = 'A1')
DeleteCF(prefix = D, key = 'A1')
DeleteCF(prefix = D, key = 'A2')
DeleteCF(prefix = D, key = 'A3')
MergeCF(prefix = A, key = 'A5', val len = 6)
<...>*2

Signed-off-by: Igor Fedotov <igor.fedotov@croit.io>

diff --git a/src/kv/RocksDBStore.cc b/src/kv/RocksDBStore.cc
index 3671ee775f49a5c212eb61ca08fd7695ebbd7a33..945d467a0dbb5e59e2fbe8d515ca56525888187c 100644 (file)
--- a/src/kv/RocksDBStore.cc
+++ b/src/kv/RocksDBStore.cc
@@ -8,6 +8,7 @@
 #include <string>
 #include <string_view>
 #include <tuple>
+#include <unordered_map>
 #include <errno.h>
 #include <unistd.h>
 #include <sys/types.h>
@@ -1548,88 +1549,6 @@ void RocksDBStore::get_statistics(Formatter *f)
   }
 }
 
-struct RocksDBStore::RocksWBHandler: public rocksdb::WriteBatch::Handler {
-  RocksWBHandler(const RocksDBStore& db, bool verbose)
-   : db(db), verbose(verbose) {}
-  const RocksDBStore& db;
-  std::stringstream seen;
-  int num_seen = 0;
-  bool verbose = true;
-
-  void dump(const char* op_name, const char* op_short,
-           uint32_t column_family_id,
-           const rocksdb::Slice& key_in,
-           const rocksdb::Slice* value = nullptr) {
-    string prefix;
-    string key;
-    if (column_family_id == 0) {
-      db.split_key(key_in, &prefix, &key);
-    } else {
-      auto it = db.cf_ids_to_prefix.find(column_family_id);
-      ceph_assert(it != db.cf_ids_to_prefix.end());
-      prefix = it->second;
-      key = key_in.ToString();
-    }
-    if (verbose) {
-      ssize_t size = value ? value->size() : -1;
-      seen << std::endl << op_name << "(";
-
-      seen << " prefix = " << prefix;
-      seen << " key = " << pretty_binary_string(key);
-      if (size != -1)
-        seen << " value size = " << std::to_string(size);
-      seen << ")";
-    } else {
-      if (num_seen)
-        seen << ",";
-      seen << op_short << ":" << prefix;
-    }
-    num_seen++;
-  }
-  void Put(const rocksdb::Slice& key,
-          const rocksdb::Slice& value) override {
-    dump("Put", " P", 0, key, &value);
-  }
-  rocksdb::Status PutCF(uint32_t column_family_id, const rocksdb::Slice& key,
-                       const rocksdb::Slice& value) override {
-    dump("PutCF", " p", column_family_id, key, &value);
-    return rocksdb::Status::OK();
-  }
-  void SingleDelete(const rocksdb::Slice& key) override {
-    dump("SingleDelete", "ds", 0, key);
-  }
-  rocksdb::Status SingleDeleteCF(uint32_t column_family_id, const rocksdb::Slice& key) override {
-    dump("SingleDeleteCF", "DS", column_family_id, key);
-    return rocksdb::Status::OK();
-  }
-  void Delete(const rocksdb::Slice& key) override {
-    dump("Delete", " D", 0, key);
-  }
-  rocksdb::Status DeleteCF(uint32_t column_family_id, const rocksdb::Slice& key) override {
-    dump("DeleteCF", " d", column_family_id, key);
-    return rocksdb::Status::OK();
-  }
-  void Merge(const rocksdb::Slice& key,
-            const rocksdb::Slice& value) override {
-    dump("Merge", " M", 0, key, &value);
-  }
-  rocksdb::Status MergeCF(uint32_t column_family_id, const rocksdb::Slice& key,
-                         const rocksdb::Slice& value) override {
-    dump("MergeCF", " m", column_family_id, key, &value);
-    return rocksdb::Status::OK();
-  }
-  bool Continue() override {
-    bool r = verbose ? num_seen < 128 : num_seen < 50;
-    if (!r) {
-      if (verbose) {
-        seen << std::endl;
-      }
-      seen << " <skipped>";
-    }
-    return r;
-  }
-};
-
 int RocksDBStore::submit_common(rocksdb::WriteOptions& woptions, KeyValueDB::Transaction t) 
 {
   // enable rocksdb breakdown
@@ -1645,14 +1564,14 @@ int RocksDBStore::submit_common(rocksdb::WriteOptions& woptions, KeyValueDB::Tra
   lgeneric_subdout(cct, rocksdb, 30) << __func__;
   RocksWBHandler bat_txc(*this, true);
   _t->bat.Iterate(&bat_txc);
-  *_dout << " Rocksdb transaction: " << bat_txc.seen.str() << dendl;
+  *_dout << " Rocksdb transaction: " << bat_txc.get_seen() << dendl;
   
   rocksdb::Status s = db->Write(woptions, &_t->bat);
   if (!s.ok()) {
     RocksWBHandler rocks_txc(*this, true);
     _t->bat.Iterate(&rocks_txc);
     derr << __func__ << " error: " << s.ToString() << " code = " << s.code()
-         << " Rocksdb transaction: " << rocks_txc.seen.str() << dendl;
+         << " Rocksdb transaction: " << rocks_txc.get_seen() << dendl;
   }
 
   if (cct->_conf->rocksdb_perf) {
@@ -1738,7 +1657,7 @@ string RocksDBStore::RocksDBTransactionImpl::get_summary_string(
   ceph_assert(db);
   RocksWBHandler bat_txc(*db, verbose);
   bat.Iterate(&bat_txc);
-  return bat_txc.seen.str();
+  return bat_txc.get_seen();
 }
 
 void RocksDBStore::RocksDBTransactionImpl::set(
@@ -4144,3 +4063,83 @@ void RocksDBStore::util_divide_key_range(
   dout(10) << "produced chunk size=" << full_size - base->size << " "
     << pretty_binary_string(base->key) << " " << pretty_binary_string(key_to) << dendl;
 }
+
+void RocksWBHandler::_finalize_seen(bool sorted)
+{
+  if (verbose) {
+    if (num_skipped) {
+      seen << std::endl << "<...>*" << num_skipped;
+    }
+  } else {
+    seen.clear();
+    bool first = true;
+    auto _add = [&](const std::string& k, size_t c) {
+      if (!first) {
+       seen << ",";
+      } else {
+       first = false;
+      }
+      seen << k;
+      if (c > 1) {
+       seen << "*" << c;
+      }
+    };
+    if (sorted) {
+      std::map<std::string, size_t> sorted_seen_counts;
+      for (auto& [k, c] : seen_counts) {
+       sorted_seen_counts.emplace(k, c);
+      }
+      for (auto& [k, c] : sorted_seen_counts) {
+       _add(k, c);
+      }
+    } else {
+      for(auto& [k, c] : seen_counts) {
+        _add(k,c);
+      }
+    }
+    if (num_skipped) {
+      if (!first) {
+       seen << ",";
+      }
+      seen << "...*" << num_skipped;
+    }
+  }
+}
+
+void RocksWBHandler::_dump(const char* op_name, const char* op_short,
+           uint32_t column_family_id,
+           const rocksdb::Slice& key_in,
+           const rocksdb::Slice* value)
+{
+  if (num_seen >= max) {
+    num_skipped++;
+    return;
+  }
+  string prefix;
+  string key;
+  if (column_family_id == 0) {
+    db.split_key(key_in, &prefix, &key);
+  } else {
+    auto it = db.cf_ids_to_prefix.find(column_family_id);
+    ceph_assert(it != db.cf_ids_to_prefix.end());
+    prefix = it->second;
+    key = key_in.ToString();
+  }
+  if (verbose) {
+    ssize_t size = value ? value->size() : -1;
+    seen << std::endl << op_name << "(";
+
+    seen << "prefix = " << prefix;
+    seen << ", key = " << pretty_binary_string(key);
+    if (size != -1)
+      seen << ", val len = " << std::to_string(size);
+    seen << ")";
+  } else {
+    string k = op_short;
+    k += ":";
+    k += prefix;
+    auto [it, found] = seen_counts.emplace(k, 0);
+    it->second++;
+  }
+  num_seen++;
+}

diff --git a/src/kv/RocksDBStore.h b/src/kv/RocksDBStore.h
index 0dc36485f1d6aa552e34dc0dbfb3827d47d984aa..840b7ed77a1a88104857a7e7f5ad80bad0f2a1f9 100644 (file)
--- a/src/kv/RocksDBStore.h
+++ b/src/kv/RocksDBStore.h
@@ -82,6 +82,7 @@ inline rocksdb::Slice make_slice(const std::optional<std::string>& bound) {
 /**
  * Uses RocksDB to implement the KeyValueDB interface
  */
+struct RocksWBHandler;
 class RocksDBStore : public KeyValueDB {
   CephContext *cct;
   PerfCounters *logger;
@@ -100,6 +101,7 @@ class RocksDBStore : public KeyValueDB {
   friend class ShardMergeIteratorImpl;
   friend class CFIteratorImpl;
   friend class WholeMergeIteratorImpl;
+  friend struct RocksWBHandler;
   /*
    *  See RocksDB's definition of a column family(CF) and how to use it.
    *  The interfaces of KeyValueDB is extended, when a column family is created.
@@ -315,7 +317,6 @@ public:
   int64_t estimate_range_size(const std::string& prefix,
                               const std::string& key_from,
                               const std::string& key_to) override;
-  struct RocksWBHandler;
   class RocksDBTransactionImpl : public KeyValueDB::TransactionImpl {
   public:
     rocksdb::WriteBatch bat;
@@ -600,4 +601,68 @@ public:
     std::vector<keyrange_t>& chunks) override;
 };
 
+class RocksWBHandler : public rocksdb::WriteBatch::Handler
+{
+  const RocksDBStore& db;
+  std::stringstream seen;
+  size_t num_seen = 0;
+  size_t num_skipped = 0;
+  bool verbose = true;
+  size_t max = 0;
+  std::unordered_map<std::string, size_t> seen_counts;
+
+  void _finalize_seen(bool sorted);
+  void _dump(const char* op_name, const char* op_short,
+            uint32_t column_family_id,
+            const rocksdb::Slice& key_in,
+            const rocksdb::Slice* value = nullptr);
+public:
+  RocksWBHandler(const RocksDBStore& db, bool verbose, size_t _max = 0)
+   : db(db), verbose(verbose), max(_max) {
+     if (max == 0) {
+       max = verbose ? 128 : 64;
+     }
+   }
+
+  std::string get_seen(bool sorted = false) {
+    _finalize_seen(sorted);
+    return seen.str();
+  }
+  void Put(const rocksdb::Slice& key,
+          const rocksdb::Slice& value) override {
+    _dump("Put", "P", 0, key, &value);
+  }
+  rocksdb::Status PutCF(uint32_t column_family_id, const rocksdb::Slice& key,
+                       const rocksdb::Slice& value) override {
+    _dump("PutCF", "PF", column_family_id, key, &value);
+    return rocksdb::Status::OK();
+  }
+  void SingleDelete(const rocksdb::Slice& key) override {
+    _dump("SingleDelete", "d", 0, key);
+  }
+  rocksdb::Status SingleDeleteCF(uint32_t column_family_id, const rocksdb::Slice& key) override {
+    _dump("SingleDeleteCF", "df", column_family_id, key);
+    return rocksdb::Status::OK();
+  }
+  void Delete(const rocksdb::Slice& key) override {
+    _dump("Delete", "D", 0, key);
+  }
+  rocksdb::Status DeleteCF(uint32_t column_family_id, const rocksdb::Slice& key) override {
+    _dump("DeleteCF", "DF", column_family_id, key);
+    return rocksdb::Status::OK();
+  }
+  void Merge(const rocksdb::Slice& key,
+            const rocksdb::Slice& value) override {
+    _dump("Merge", "M", 0, key, &value);
+  }
+  rocksdb::Status MergeCF(uint32_t column_family_id, const rocksdb::Slice& key,
+                         const rocksdb::Slice& value) override {
+    _dump("MergeCF", "MF", column_family_id, key, &value);
+    return rocksdb::Status::OK();
+  }
+  bool Continue() override {
+    return true;
+  }
+};
+
 #endif

diff --git a/src/test/objectstore/test_kv.cc b/src/test/objectstore/test_kv.cc
index f533664559437d5652c013a3942fbfd27dd9ec14..7b0eade176a339e35f5d3e0dbe175b9294bc3e2b 100644 (file)
--- a/src/test/objectstore/test_kv.cc
+++ b/src/test/objectstore/test_kv.cc
@@ -135,6 +135,61 @@ TEST_P(KVTest, OpenWriteRead) {
   fini();
 }
 
+TEST_P(KVTest, RocksDBDumpTransaction) {
+  RocksDBStore* rdb = dynamic_cast<RocksDBStore*>(db.get());
+  if (!rdb) {
+    return;
+  }
+
+  ASSERT_EQ(0, db->create_and_open(cout));
+  {
+    KeyValueDB::Transaction t = db->get_transaction();
+    bufferlist value;
+    value.append("value");
+    t->set("O", "key", value);
+    value.clear();
+    value.append("value2");
+    t->set("P", "key2", value);
+    value.clear();
+    value.append("value3");
+    t->set("O", "key3", value);
+
+    t->rmkey("O", "A1");
+    t->rmkey("D", "A1");
+    t->rmkey("D", "A2");
+    t->rmkey("D", "A3");
+    t->merge("A", "A5", value);
+    // following txcs to be skipped
+    t->merge("B", "A5", value);
+    t->merge("B", "A5", value);
+
+    auto* _t = dynamic_cast<RocksDBStore::RocksDBTransactionImpl*>(t.get());
+    ASSERT_TRUE(_t != nullptr);
+    RocksWBHandler bat_txc_short(*rdb, false, 8);
+    _t->bat.Iterate(&bat_txc_short);
+    auto seen = bat_txc_short.get_seen(true); // using 'sorted' result to ensure
+                                              // fixed ordering in the result
+    std::cout << "Seen short = " << seen << std::endl;
+    ASSERT_EQ(seen, "DF:D*3,DF:O,MF:A,PF:O*2,PF:P,...*2");
+
+    RocksWBHandler bat_txc_verbose(*rdb, true, 8);
+    _t->bat.Iterate(&bat_txc_verbose);
+    seen = bat_txc_verbose.get_seen();
+    std::cout << "Seen verbose:" << seen << std::endl;
+    ASSERT_EQ(seen,
+      "\nPutCF(prefix = O, key = 'key', val len = 5)"
+      "\nPutCF(prefix = P, key = 'key2', val len = 6)"
+      "\nPutCF(prefix = O, key = 'key3', val len = 6)"
+      "\nDeleteCF(prefix = O, key = 'A1')"
+      "\nDeleteCF(prefix = D, key = 'A1')"
+      "\nDeleteCF(prefix = D, key = 'A2')"
+      "\nDeleteCF(prefix = D, key = 'A3')"
+      "\nMergeCF(prefix = A, key = 'A5', val len = 6)"
+      "\n<...>*2");
+  }
+  fini();
+}
+
 TEST_P(KVTest, PutReopen) {
   ASSERT_EQ(0, db->create_and_open(cout));
   {