Bucket entry counters and metrics (#4436)

# Description

Resolves #4426

* Adds `BucketEntryCounters` to `BucketIndexImpl` to count entry types
and duration.
* Adds functionality to `BucketList` to accumulate the counters for all
buckets
* Adds functionality to LedgerManager to emit these metrics to medida

Still TODO:
* Local testing of serialization of indexes with counters

Out of scope:
* XDR Changes for including counters in ledger close meta
* grafana updates

<!---

Describe what this pull request does, which issue it's resolving
(usually applicable for code changes).

--->

# Checklist
- [ ] Reviewed the
[contributing](https://github.com/stellar/stellar-core/blob/master/CONTRIBUTING.md#submitting-changes)
document
- [ ] Rebased on top of master (no merge commits)
- [ ] Ran `clang-format` v8.0.0 (via `make format` or the Visual Studio
extension)
- [ ] Compiles
- [ ] Ran all tests
- [ ] If change impacts performance, include supporting evidence per the
[performance
document](https://github.com/stellar/stellar-core/blob/master/performance-eval/performance-eval.md)

docs/metrics.md

@@ -32,6 +32,8 @@ bucket.memory.shared                      | counter   | number of buckets refere
 bucket.merge-time.level-<X>               | timer     | time to merge two buckets on level <X>
 bucket.snap.merge                         | timer     | time to merge two buckets
 bucketlist.size.bytes                     | counter   | total size of the BucketList in bytes
+bucketlist.entryCounts.-<X>               | counter   | number of entries of type <X> in the BucketList
+bucketlist.entrySizes.-<X>                | counter   | size of entries of type <X> in the BucketList
 bucketlistDB.bloom.lookups                | meter     | number of bloom filter lookups
 bucketlistDB.bloom.misses                 | meter     | number of bloom filter false positives
 bucketlistDB.bulk.loads                   | meter     | number of entries BucketListDB queried to prefetch

src/bucket/Bucket.cpp

@@ -863,4 +863,38 @@ Bucket::getBucketVersion(std::shared_ptr<Bucket const> const& bucket)
     BucketInputIterator it(bucket);
     return it.getMetadata().ledgerVersion;
 }
+
+BucketEntryCounters const&
+Bucket::getBucketEntryCounters() const
+{
+    releaseAssert(mIndex);
+    return mIndex->getBucketEntryCounters();
+}
+
+BucketEntryCounters&
+BucketEntryCounters::operator+=(BucketEntryCounters const& other)
+{
+    for (auto [type, count] : other.entryTypeCounts)
+    {
+        this->entryTypeCounts[type] += count;
+    }
+    for (auto [type, size] : other.entryTypeSizes)
+    {
+        this->entryTypeSizes[type] += size;
+    }
+    return *this;
+}
+
+bool
+BucketEntryCounters::operator==(BucketEntryCounters const& other) const
+{
+    return this->entryTypeCounts == other.entryTypeCounts &&
+           this->entryTypeSizes == other.entryTypeSizes;
+}
+
+bool
+BucketEntryCounters::operator!=(BucketEntryCounters const& other) const
+{
+    return !(*this == other);
 }
+}

src/bucket/Bucket.h

@@ -9,6 +9,7 @@
 #include "util/ProtocolVersion.h"
 #include "xdr/Stellar-ledger.h"
 #include <list>
+#include <map>
 #include <optional>
 #include <string>
 
@@ -43,6 +44,7 @@ class BucketManager;
 class SearchableBucketListSnapshot;
 struct EvictionResultEntry;
 class EvictionStatistics;
+struct BucketEntryCounters;
 
 class Bucket : public std::enable_shared_from_this<Bucket>,
                public NonMovableOrCopyable
@@ -179,7 +181,25 @@ class Bucket : public std::enable_shared_from_this<Bucket>,
     static uint32_t getBucketVersion(std::shared_ptr<Bucket> const& bucket);
     static uint32_t
     getBucketVersion(std::shared_ptr<Bucket const> const& bucket);
-
+    BucketEntryCounters const& getBucketEntryCounters() const;
     friend class BucketSnapshot;
 };
+
+enum class LedgerEntryTypeAndDurability : uint32_t;
+struct BucketEntryCounters
+{
+    std::map<LedgerEntryTypeAndDurability, size_t> entryTypeCounts;
+    std::map<LedgerEntryTypeAndDurability, size_t> entryTypeSizes;
+
+    BucketEntryCounters& operator+=(BucketEntryCounters const& other);
+    bool operator==(BucketEntryCounters const& other) const;
+    bool operator!=(BucketEntryCounters const& other) const;
+
+    template <class Archive>
+    void
+    serialize(Archive& ar)
+    {
+        ar(entryTypeCounts, entryTypeSizes);
+    }
+};
 }

src/bucket/BucketIndex.h

@@ -33,6 +33,7 @@ namespace stellar
  */
 
 class BucketManager;
+struct BucketEntryCounters;
 
 // BucketIndex abstract interface
 class BucketIndex : public NonMovableOrCopyable
@@ -74,7 +75,7 @@ class BucketIndex : public NonMovableOrCopyable
                                   IndividualIndex::const_iterator>;
 
     inline static const std::string DB_BACKEND_STATE = "bl";
-    inline static const uint32_t BUCKET_INDEX_VERSION = 3;
+    inline static const uint32_t BUCKET_INDEX_VERSION = 4;
 
     // Returns true if LedgerEntryType not supported by BucketListDB
     static bool typeNotSupported(LedgerEntryType t);
@@ -132,7 +133,7 @@ class BucketIndex : public NonMovableOrCopyable
 
     virtual void markBloomMiss() const = 0;
     virtual void markBloomLookup() const = 0;
-
+    virtual BucketEntryCounters const& getBucketEntryCounters() const = 0;
 #ifdef BUILD_TESTS
     virtual bool operator==(BucketIndex const& inRaw) const = 0;
 #endif

src/bucket/BucketIndexImpl.cpp

@@ -7,6 +7,7 @@
 #include "bucket/BucketManager.h"
 #include "crypto/Hex.h"
 #include "crypto/ShortHash.h"
+#include "ledger/LedgerTypeUtils.h"
 #include "main/Config.h"
 #include "util/BinaryFuseFilter.h"
 #include "util/Fs.h"
@@ -100,6 +101,17 @@ BucketIndexImpl<IndexT>::BucketIndexImpl(BucketManager& bm,
         std::vector<uint64_t> keyHashes;
         auto seed = shortHash::getShortHashInitKey();
 
+        auto countEntry = [&](BucketEntry const& be) {
+            if (be.type() == METAENTRY)
+            {
+                // Do not count meta entries.
+                return;
+            }
+            auto ledt = bucketEntryToLedgerEntryAndDurabilityType(be);
+            mData.counters.entryTypeCounts[ledt]++;
+            mData.counters.entryTypeSizes[ledt] += xdr::xdr_size(be);
+        };
+
         while (in && in.readOne(be))
         {
             // peridocially check if bucket manager is exiting to stop indexing
@@ -170,6 +182,7 @@ BucketIndexImpl<IndexT>::BucketIndexImpl(BucketManager& bm,
                 {
                     mData.keysToOffset.emplace_back(key, pos);
                 }
+                countEntry(be);
             }
 
             pos = in.pos();
@@ -551,6 +564,11 @@ BucketIndexImpl<IndexT>::operator==(BucketIndex const& inRaw) const
         }
     }
 
+    if (mData.counters != in.mData.counters)
+    {
+        return false;
+    }
+
     return true;
 }
 #endif
@@ -580,4 +598,11 @@ BucketIndexImpl<BucketIndex::RangeIndex>::markBloomLookup() const
 {
     mBloomLookupMeter.Mark();
 }
-}
+
+template <class IndexT>
+BucketEntryCounters const&
+BucketIndexImpl<IndexT>::getBucketEntryCounters() const
+{
+    return mData.counters;
+}
+}

src/bucket/BucketIndexImpl.h

@@ -4,6 +4,7 @@
 // under the Apache License, Version 2.0. See the COPYING file at the root
 // of this distribution or at http://www.apache.org/licenses/LICENSE-2.0
 
+#include "bucket/Bucket.h"
 #include "bucket/BucketIndex.h"
 #include "medida/meter.h"
 #include "util/BinaryFuseFilter.h"
@@ -15,7 +16,6 @@
 
 namespace stellar
 {
-
 // Index maps either individual keys or a key range of BucketEntry's to the
 // associated offset within the bucket file. Index stored as vector of pairs:
 // First: LedgerKey/Key ranges sorted in the same scheme as LedgerEntryCmp
@@ -33,13 +33,15 @@ template <class IndexT> class BucketIndexImpl : public BucketIndex
         std::streamoff pageSize{};
         std::unique_ptr<BinaryFuseFilter16> filter{};
         std::map<Asset, std::vector<PoolID>> assetToPoolID{};
+        BucketEntryCounters counters{};
 
         template <class Archive>
         void
         save(Archive& ar) const
         {
             auto version = BUCKET_INDEX_VERSION;
-            ar(version, pageSize, assetToPoolID, keysToOffset, filter);
+            ar(version, pageSize, assetToPoolID, keysToOffset, filter,
+               counters);
         }
 
         // Note: version and pageSize must be loaded before this function is
@@ -50,7 +52,7 @@ template <class IndexT> class BucketIndexImpl : public BucketIndex
         void
         load(Archive& ar)
         {
-            ar(assetToPoolID, keysToOffset, filter);
+            ar(assetToPoolID, keysToOffset, filter, counters);
         }
     } mData;
 
@@ -111,6 +113,7 @@ template <class IndexT> class BucketIndexImpl : public BucketIndex
 
     virtual void markBloomMiss() const override;
     virtual void markBloomLookup() const override;
+    virtual BucketEntryCounters const& getBucketEntryCounters() const override;
 
 #ifdef BUILD_TESTS
     virtual bool operator==(BucketIndex const& inRaw) const override;

src/bucket/BucketList.cpp

@@ -4,6 +4,7 @@
 
 #include "BucketList.h"
 #include "bucket/Bucket.h"
+#include "bucket/BucketIndexImpl.h"
 #include "bucket/BucketInputIterator.h"
 #include "bucket/BucketManager.h"
 #include "bucket/BucketSnapshot.h"
@@ -631,6 +632,24 @@ BucketList::addBatch(Application& app, uint32_t currLedger,
     }
 }
 
+BucketEntryCounters
+BucketList::sumBucketEntryCounters() const
+{
+    BucketEntryCounters counters;
+    for (auto const& lev : mLevels)
+    {
+        for (auto const& b : {lev.getCurr(), lev.getSnap()})
+        {
+            if (b->isIndexed())
+            {
+                auto c = b->getBucketEntryCounters();
+                counters += c;
+            }
+        }
+    }
+    return counters;
+}
+
 void
 BucketList::updateStartingEvictionIterator(EvictionIterator& iter,
                                            uint32_t firstScanLevel,

src/bucket/BucketList.h

@@ -345,6 +345,7 @@ namespace stellar
 class AbstractLedgerTxn;
 class Application;
 class Bucket;
+struct BucketEntryCounters;
 class Config;
 struct EvictionCounters;
 struct InflationWinner;
@@ -523,5 +524,6 @@ class BucketList
                   std::vector<LedgerEntry> const& initEntries,
                   std::vector<LedgerEntry> const& liveEntries,
                   std::vector<LedgerKey> const& deadEntries);
+    BucketEntryCounters sumBucketEntryCounters() const;
 };
 }

src/bucket/BucketManager.h

@@ -385,5 +385,6 @@ class BucketManager : NonMovableOrCopyable
     // Get bucketlist snapshot
     virtual std::shared_ptr<SearchableBucketListSnapshot>
     getSearchableBucketListSnapshot() = 0;
+    virtual void reportBucketEntryCountMetrics() = 0;
 };
 }

src/bucket/BucketManagerImpl.cpp

@@ -4,6 +4,7 @@
 
 #include "bucket/BucketManagerImpl.h"
 #include "bucket/Bucket.h"
+#include "bucket/BucketIndexImpl.h"
 #include "bucket/BucketInputIterator.h"
 #include "bucket/BucketList.h"
 #include "bucket/BucketListSnapshot.h"
@@ -189,6 +190,20 @@ BucketManagerImpl::BucketManagerImpl(Application& app)
     , mDeleteEntireBucketDirInDtor(
           app.getConfig().isInMemoryModeWithoutMinimalDB())
 {
+    for (uint32_t t =
+             static_cast<uint32_t>(LedgerEntryTypeAndDurability::ACCOUNT);
+         t < static_cast<uint32_t>(LedgerEntryTypeAndDurability::NUM_TYPES);
+         ++t)
+    {
+        auto type = static_cast<LedgerEntryTypeAndDurability>(t);
+        auto typeString = toString(type);
+        mBucketListEntryCountCounters.emplace(
+            type, app.getMetrics().NewCounter(
+                      {"bucketlist", "entryCounts", typeString}));
+        mBucketListEntrySizeCounters.emplace(
+            type, app.getMetrics().NewCounter(
+                      {"bucketlist", "entrySizes", typeString}));
+    }
 }
 
 const std::string BucketManagerImpl::kLockFilename = "stellar-core.lock";
@@ -905,6 +920,11 @@ BucketManagerImpl::addBatch(Application& app, LedgerHeader header,
     mBucketList->addBatch(app, header.ledgerSeq, header.ledgerVersion,
                           initEntries, liveEntries, deadEntries);
     mBucketListSizeCounter.set_count(mBucketList->getSize());
+
+    if (app.getConfig().isUsingBucketListDB())
+    {
+        reportBucketEntryCountMetrics();
+    }
 }
 
 #ifdef BUILD_TESTS
@@ -1524,4 +1544,43 @@ BucketManagerImpl::getSearchableBucketListSnapshot()
 
     return mSearchableBucketListSnapshot;
 }
+
+void
+BucketManagerImpl::reportBucketEntryCountMetrics()
+{
+    if (!mApp.getConfig().isUsingBucketListDB())
+    {
+        return;
+    }
+    auto bucketEntryCounters = mBucketList->sumBucketEntryCounters();
+    for (auto [type, count] : bucketEntryCounters.entryTypeCounts)
+    {
+        auto countCounter = mBucketListEntryCountCounters.find(type);
+        if (countCounter == mBucketListEntryCountCounters.end())
+        {
+            auto typeString = toString(type);
+            countCounter =
+                mBucketListEntryCountCounters
+                    .emplace(type,
+                             mApp.getMetrics().NewCounter(
+                                 {"bucketlist", "entryCounts", typeString}))
+                    .first;
+        }
+        countCounter->second.set_count(count);
+
+        auto sizeCounter = mBucketListEntrySizeCounters.find(type);
+        if (sizeCounter == mBucketListEntrySizeCounters.end())
+        {
+            auto typeString = toString(type);
+            sizeCounter =
+                mBucketListEntrySizeCounters
+                    .emplace(type,
+                             mApp.getMetrics().NewCounter(
+                                 {"bucketlist", "entrySizes", typeString}))
+                    .first;
+        }
+        sizeCounter->second.set_count(
+            bucketEntryCounters.entryTypeSizes.at(type));
+    }
+}
 }

src/bucket/BucketManagerImpl.h

@@ -31,6 +31,9 @@ class Application;
 class Bucket;
 class BucketList;
 class BucketSnapshotManager;
+struct BucketEntryCounters;
+enum class LedgerEntryTypeAndDurability : uint32_t;
+
 struct HistoryArchiveState;
 
 class BucketManagerImpl : public BucketManager
@@ -61,6 +64,10 @@ class BucketManagerImpl : public BucketManager
     EvictionCounters mBucketListEvictionCounters;
     MergeCounters mMergeCounters;
     std::shared_ptr<EvictionStatistics> mEvictionStatistics{};
+    std::map<LedgerEntryTypeAndDurability, medida::Counter&>
+        mBucketListEntryCountCounters;
+    std::map<LedgerEntryTypeAndDurability, medida::Counter&>
+        mBucketListEntrySizeCounters;
 
     std::future<EvictionResult> mEvictionFuture{};
 
@@ -192,6 +199,7 @@ class BucketManagerImpl : public BucketManager
 
     std::shared_ptr<SearchableBucketListSnapshot>
     getSearchableBucketListSnapshot() override;
+    void reportBucketEntryCountMetrics() override;
 };
 
 #define SKIP_1 50