[BP] MB-57400: Use an estimated items_remaining count for Prometheus stats

[[Backport to 7.1.x for maintenance patch]]

Add to CheckpointManager an option to get an estimate of the remaining
items for a cursor - the estimate should provide near constant
performance as it just asks each the checkpoint from curstor to end
for an item count (read of a counter). The max checkpoints is 10, so
the estimated call will only iterate 10 at a maximum. The estimate
will be inaccurate if the cursor is partially through a checkpoint
and will over-count for items already processed.

For the Prometheus requests use the estimated statistic for.

1) The aggregated DCP connection stats (high-cardinality)
2) The aggregated DCP connection stats by type (low-cardinality) but
   only when the  type is !replication.

For cmd_stat, i.e. dcpagg use the accurate mode, so no change from
before this commit.

(cherry picked from commit c6397f2)

Change-Id: I712c2a7566735849130092d0f1db07d975cdb2c0
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/192799
Well-Formed: Restriction Checker
Reviewed-by: Paolo Cocchi <[email protected]>
Tested-by: Dave Rigby <[email protected]>

Author: jimwwalker

engines/ep/src/checkpoint_manager.cc

@@ -1257,18 +1257,21 @@ void CheckpointManager::checkOpenCheckpoint(
     }
 }
 
-size_t CheckpointManager::getNumItemsForCursor(
-        const CheckpointCursor* cursor) const {
+size_t CheckpointManager::getNumItemsForCursor(const CheckpointCursor* cursor,
+                                               bool accurate) const {
     std::lock_guard<std::mutex> lh(queueLock);
-    return getNumItemsForCursor(lh, cursor);
+    return getNumItemsForCursor(lh, cursor, accurate);
 }
 
 size_t CheckpointManager::getNumItemsForCursor(
         const std::lock_guard<std::mutex>& lh,
-        const CheckpointCursor* cursor) const {
+        const CheckpointCursor* cursor,
+        bool accurate) const {
     if (cursor && cursor->valid()) {
-        size_t items = cursor->getRemainingItemsCount();
         CheckpointList::const_iterator chkptIterator(cursor->getCheckpoint());
+        size_t items = accurate ? cursor->getRemainingItemsCount()
+                                : (*chkptIterator)->getNumItems();
+
         if (chkptIterator != checkpointList.end()) {
             ++chkptIterator;
         }
@@ -1595,7 +1598,7 @@ void CheckpointManager::addStats(const AddStatFn& add_stat,
                              "vb_%d:num_items_for_persistence",
                              vbucketId.get());
             add_casted_stat(buf.data(),
-                            getNumItemsForCursor(lh, persistenceCursor),
+                            getNumItemsForCursor(lh, persistenceCursor, true),
                             add_stat,
                             cookie);
         }
@@ -1635,10 +1638,21 @@ void CheckpointManager::addStats(const AddStatFn& add_stat,
                                  "vb_%d:%s:num_items_for_cursor",
                                  vbucketId.get(),
                                  cursor.second->getName().c_str());
-                add_casted_stat(buf.data(),
-                                getNumItemsForCursor(lh, cursor.second.get()),
-                                add_stat,
-                                cookie);
+                add_casted_stat(
+                        buf.data(),
+                        getNumItemsForCursor(lh, cursor.second.get(), true),
+                        add_stat,
+                        cookie);
+                checked_snprintf(buf.data(),
+                                 buf.size(),
+                                 "vb_%d:%s:estimated_num_items_for_cursor",
+                                 vbucketId.get(),
+                                 cursor.second->getName().c_str());
+                add_casted_stat(
+                        buf.data(),
+                        getNumItemsForCursor(lh, cursor.second.get(), false),
+                        add_stat,
+                        cookie);
             }
         }
 

engines/ep/src/checkpoint_manager.h

@@ -366,11 +366,19 @@ class CheckpointManager {
      * As such it is *not* safe to use when a precise count of remaining
      * items is needed.
      *
-     * Returns the count of Items (excluding meta items) that the given cursor
+     * @param cursor The cursor for which the caller want to know the item count
+     * @param accurate if true, then  the function will perform a count from the
+     *        the cursor to the end of the current checkpoint. If false the
+     *        total item count of the current checkpoint is used in the sum.
+     *        Warning that accurate=true is an O(n) cost, where n is the number
+     *        of items, it can be slow and trigger problems in other threads
+     *        that want to access the CheckpointManager (MB-57000)
+     * @return the count of Items (excluding meta items) that the given cursor
      * has yet to process (i.e. between the cursor's current position and the
-     * end of the last checkpoint).
+     * end of the last checkpoint). Note: see param accurate for more detail
      */
-    size_t getNumItemsForCursor(const CheckpointCursor* cursor) const;
+    size_t getNumItemsForCursor(const CheckpointCursor* cursor,
+                                bool accurate = true) const;
 
     /* WARNING! This method can return inaccurate counts - see MB-28431. It
      * at *least* can suffer from overcounting by at least 1 (in scenarios as
@@ -741,7 +749,8 @@ class CheckpointManager {
      * @return number of items to be processed
      */
     size_t getNumItemsForCursor(const std::lock_guard<std::mutex>& lh,
-                                const CheckpointCursor* cursor) const;
+                                const CheckpointCursor* cursor,
+                                bool accurate) const;
 
     /**
      * Clears this CM, effectively removing all checkpoints in the list and

engines/ep/src/connhandler.h

@@ -280,7 +280,8 @@ class ConnHandler : public DcpConnHandlerIface {
 
     virtual void addStats(const AddStatFn& add_stat, const CookieIface* c);
 
-    virtual void aggregateQueueStats(ConnCounter& stats_aggregator) const {
+    virtual void aggregateQueueStats(ConnCounter& stats_aggregator,
+                                     bool accurateItemsRemaining) const {
         // Empty
     }
 

engines/ep/src/dcp/active_stream.cc

@@ -2103,7 +2103,7 @@ void ActiveStream::transitionState(StreamState newState) {
     }
 }
 
-size_t ActiveStream::getItemsRemaining() {
+size_t ActiveStream::getItemsRemaining(bool accurateItemsRemaining) {
     VBucketPtr vbucket = engine->getVBucket(vb_);
 
     if (!vbucket || !isActive()) {
@@ -2113,9 +2113,20 @@ size_t ActiveStream::getItemsRemaining() {
     // Items remaining is the sum of:
     // (a) Items outstanding in checkpoints
     // (b) Items pending in our readyQ, excluding any meta items.
+    //
+    // Note: if accurateItemsRemaining=true this call could be slow, the
+    // returned value will count from the cursor to the end of the checkpoint.
+    // That's an O(n) n=number-items.
+    // if accurateItemsRemaining=false the returned value represents the sum of
+    // items for current and all outstanding checkpoints, so is inaccurate in
+    // worst case by chk_max_items - 1 (if the cursor happens to be at the end
+    // of the checkpoint). The cost of the estimated call is O(n) where is
+    // number of checkpoints (and that is capped by max_checkpoints, 10 as
+    // default).
     size_t ckptItems = 0;
     if (auto sp = cursor.lock()) {
-        ckptItems = vbucket->checkpointManager->getNumItemsForCursor(sp.get());
+        ckptItems = vbucket->checkpointManager->getNumItemsForCursor(
+                sp.get(), accurateItemsRemaining);
     }
     return ckptItems + readyQ_non_meta_items;
 }

engines/ep/src/dcp/active_stream.h

@@ -273,7 +273,7 @@ class ActiveStream : public Stream,
     /* Returns a count of how many items are outstanding to be sent for this
      * stream's vBucket.
      */
-    size_t getItemsRemaining();
+    size_t getItemsRemaining(bool accurateItemsRemaining = true);
 
     uint64_t getLastReadSeqno() const;
 

engines/ep/src/dcp/consumer.cc

@@ -1204,7 +1204,8 @@ void DcpConsumer::addStats(const AddStatFn& add_stat, const CookieIface* c) {
     addStat("synchronous_replication", isSyncReplicationEnabled(), add_stat, c);
 }
 
-void DcpConsumer::aggregateQueueStats(ConnCounter& aggregator) const {
+void DcpConsumer::aggregateQueueStats(ConnCounter& aggregator,
+                                      bool accurateItemsRemaining) const {
     aggregator.conn_queueBackoff += backoffs;
 }
 

engines/ep/src/dcp/consumer.h

@@ -258,7 +258,8 @@ class DcpConsumer : public ConnHandler,
 
     void addStats(const AddStatFn& add_stat, const CookieIface* c) override;
 
-    void aggregateQueueStats(ConnCounter& aggregator) const override;
+    void aggregateQueueStats(ConnCounter& aggregator,
+                             bool accurateItemsRemaining) const override;
 
     void notifyStreamReady(Vbid vbucket);
 

engines/ep/src/dcp/producer.cc

@@ -1560,12 +1560,13 @@ void DcpProducer::addTakeoverStats(const AddStatFn& add_stat,
     add_casted_stat("backfillRemaining", 0, add_stat, c);
 }
 
-void DcpProducer::aggregateQueueStats(ConnCounter& aggregator) const {
+void DcpProducer::aggregateQueueStats(ConnCounter& aggregator,
+                                      bool accurateItemsRemaining) const {
     ++aggregator.totalProducers;
     aggregator.conn_queueDrain += itemsSent;
     aggregator.conn_totalBytes += totalBytesSent;
     aggregator.conn_totalUncompressedDataSize += totalUncompressedDataSize;
-    auto streamAggStats = getStreamAggStats();
+    auto streamAggStats = getStreamAggStats(accurateItemsRemaining);
 
     aggregator.conn_queueRemaining += streamAggStats.itemsRemaining;
     aggregator.conn_queueMemory += streamAggStats.readyQueueMemory;
@@ -1908,17 +1909,19 @@ size_t DcpProducer::getItemsRemaining() const {
     return remainingSize;
 }
 
-DcpProducer::StreamAggStats DcpProducer::getStreamAggStats() const {
+DcpProducer::StreamAggStats DcpProducer::getStreamAggStats(
+        bool accurateItemsRemaining) const {
     DcpProducer::StreamAggStats stats{};
 
     std::for_each(
             streams->begin(),
             streams->end(),
-            [&stats](const StreamsMap::value_type& vt) {
+            [&stats, accurateItemsRemaining](const StreamsMap::value_type& vt) {
                 for (auto itr = vt.second->rlock(); !itr.end(); itr.next()) {
                     auto* as = itr.get().get();
                     if (as) {
-                        stats.itemsRemaining += as->getItemsRemaining();
+                        stats.itemsRemaining +=
+                                as->getItemsRemaining(accurateItemsRemaining);
                         stats.readyQueueMemory += as->getReadyQueueMemory();
                     }
                 }

engines/ep/src/dcp/producer.h

@@ -121,7 +121,8 @@ class DcpProducer : public ConnHandler,
                           const CookieIface* c,
                           const VBucket& vb);
 
-    void aggregateQueueStats(ConnCounter& aggregator) const override;
+    void aggregateQueueStats(ConnCounter& aggregator,
+                             bool accurateItemsRemaining) const override;
 
     /**
      * ALERT: Do NOT call this function while holding ConnMap::connLock.
@@ -452,7 +453,7 @@ class DcpProducer : public ConnHandler,
         size_t readyQueueMemory;
     };
 
-    StreamAggStats getStreamAggStats() const;
+    StreamAggStats getStreamAggStats(bool accurateItemsRemaining) const;
 
     /**
      * Map the cb::mcbp::DcpStreamEndStatus to one the client can understand.

engines/ep/src/ep_engine.cc

@@ -389,7 +389,9 @@ cb::engine_errc EventuallyPersistentEngine::get_prometheus_stats(
                                       const std::shared_ptr<ConnHandler>& tc) {
                 ++aggregator.totalConns;
                 if (auto tp = std::dynamic_pointer_cast<DcpProducer>(tc); tp) {
-                    tp->aggregateQueueStats(aggregator);
+                    // Do not use the potentially slower "accurate" items
+                    // remaining. MB-57400
+                    tp->aggregateQueueStats(aggregator, false);
                 }
             });
             addAggregatedProducerStats(collector, aggregator);
@@ -402,7 +404,7 @@ cb::engine_errc EventuallyPersistentEngine::get_prometheus_stats(
             }
             // do dcp aggregated stats, using ":" as the separator to split
             // connection names to find the connection type.
-            if (status = doConnAggStatsInner(collector, ":");
+            if (status = doConnAggStatsInner(collector, ":", false);
                 status != cb::engine_errc::success) {
                 return status;
             }
@@ -3916,7 +3918,7 @@ struct ConnStatBuilder {
             tc->addStats(add_stat, cookie);
             auto tp = std::dynamic_pointer_cast<DcpProducer>(tc);
             if (tp) {
-                tp->aggregateQueueStats(aggregator);
+                tp->aggregateQueueStats(aggregator, true);
             }
         }
     }
@@ -3932,7 +3934,18 @@ struct ConnStatBuilder {
 };
 
 struct ConnAggStatBuilder {
-    ConnAggStatBuilder(std::string_view sep) : sep(sep) {
+    /**
+     * Construct with the separator and a configuration bool.
+     * @param sep The separator used for determining "type" of DCP connection
+     *            by splitting the connection name with sep.
+     * @param alwaysUseAccurateItemsRemaining if true the aggregated stats will
+     *        include an "accurate" items-remaining. If false only connections
+     *        labelled as "replication" will use an accurate value. See MB-57400
+     */
+    ConnAggStatBuilder(std::string_view sep,
+                       bool alwaysUseAccurateItemsRemaining)
+        : sep(sep),
+          alwaysUseAccurateItemsRemaining(alwaysUseAccurateItemsRemaining) {
     }
 
     /**
@@ -3950,22 +3963,23 @@ struct ConnAggStatBuilder {
      * returns nullptr.
      *
      * @param tc connection
-     * @return counter for the given connection, or nullptr
+     * @return pair of counter for the given connection, or nullptr and true
+     *         if the connection is considered a replication stream
      */
-    ConnCounter* getCounterForConnType(std::string_view name) {
+    std::pair<ConnCounter*, bool> getCounterForConnType(std::string_view name) {
         // strip everything upto and including the first colon,
         // e.g., "eq_dcpq:"
         size_t pos1 = name.find(':');
         if (pos1 == std::string_view::npos) {
-            return nullptr;
+            return {nullptr, false};
         }
 
         name.remove_prefix(pos1 + 1);
 
         // find the given separator
         size_t pos2 = name.find(sep);
         if (pos2 == std::string_view::npos) {
-            return nullptr;
+            return {nullptr, false};
         }
 
         // extract upto given separator e.g.,
@@ -3976,14 +3990,18 @@ struct ConnAggStatBuilder {
         //  prefix is "replication"
         std::string prefix(name.substr(0, pos2));
 
-        return &counters[prefix];
+        // class is created with the itemsRemaining "policy" - which if not true
+        // replication will always use the accurate items remaining - all
+        // other connection types use a faster estimate.
+        return {&counters[prefix],
+                alwaysUseAccurateItemsRemaining || prefix == "replication"};
     }
 
-    void aggregate(ConnHandler& conn, ConnCounter* tc) {
+    void aggregate(ConnHandler& conn, ConnCounter* tc, bool isReplication) {
         ConnCounter counter;
         ++counter.totalConns;
 
-        conn.aggregateQueueStats(counter);
+        conn.aggregateQueueStats(counter, isReplication);
 
         ConnCounter& total = getTotalCounter();
         total += counter;
@@ -3999,8 +4017,8 @@ struct ConnAggStatBuilder {
 
     void operator()(std::shared_ptr<ConnHandler> tc) {
         if (tc) {
-            ConnCounter* aggregator = getCounterForConnType(tc->getName());
-            aggregate(*tc, aggregator);
+            auto aggregator = getCounterForConnType(tc->getName());
+            aggregate(*tc, aggregator.first, aggregator.second);
         }
     }
 
@@ -4010,6 +4028,7 @@ struct ConnAggStatBuilder {
 
     std::map<std::string, ConnCounter> counters;
     std::string_view sep;
+    const bool alwaysUseAccurateItemsRemaining{false};
 };
 
 /// @endcond
@@ -4102,8 +4121,8 @@ cb::engine_errc EventuallyPersistentEngine::doConnAggStats(
                     // write them as responses (which would be racy).
                     // a later call to maybeWriteResponse will do that.
                     CBStatCollector col(deferredAddStat, cookie);
-                    ep->doConnAggStatsInner(col.forBucket(ep->getName()),
-                                            separator);
+                    ep->doConnAggStatsInner(
+                            col.forBucket(ep->getName()), separator, true);
                     return cb::engine_errc::success;
                 });
         ExecutorPool::get()->schedule(task);
@@ -4115,13 +4134,15 @@ cb::engine_errc EventuallyPersistentEngine::doConnAggStats(
 }
 
 cb::engine_errc EventuallyPersistentEngine::doConnAggStatsInner(
-        const BucketStatCollector& collector, std::string_view sep) {
+        const BucketStatCollector& collector,
+        std::string_view sep,
+        bool cmdStat) {
     // The separator is, in all current usage, ":" so the length will
     // normally be 1
     const size_t max_sep_len(8);
     sep = sep.substr(0, max_sep_len);
 
-    ConnAggStatBuilder visitor(sep);
+    ConnAggStatBuilder visitor(sep, cmdStat);
     dcpConnMap_->each(visitor);
 
     for (const auto& [connType, counter] : visitor.getCounters()) {