From: Igor Fedotov Date: Wed, 1 Jul 2026 22:13:32 +0000 (+0300) Subject: kv/Rocksdbstore: change output format for short txc dump. X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=df23e844bc6f71d196f17aef666368047ecd3c0e;p=ceph.git 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 --- diff --git a/src/kv/RocksDBStore.cc b/src/kv/RocksDBStore.cc index 3671ee775f49..945d467a0dbb 100644 --- a/src/kv/RocksDBStore.cc +++ b/src/kv/RocksDBStore.cc @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -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 << " "; - } - 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 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 0dc36485f1d6..840b7ed77a1a 100644 --- a/src/kv/RocksDBStore.h +++ b/src/kv/RocksDBStore.h @@ -82,6 +82,7 @@ inline rocksdb::Slice make_slice(const std::optional& 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& 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 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 f53366455943..7b0eade176a3 100644 --- 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(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(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)); {