Revert "MB-57400: Use an estimated items_remaining count for Prometheus stats"

paolococchi · paolococchi · commit 87dd4209e7b2 · 2023-08-03T13:34:16.000Z
This reverts commit c6397f2. - Reason for revert - After the backport of MB-52276 to Neo "estimate" is no longer needed. This is the second of 2 reverts for MB-57400. Change-Id: I7f6493706a30db76bcf131aea14f3a6ff5cbe95e Reviewed-on: https://review.couchbase.org/c/kv_engine/+/194197 Well-Formed: Restriction Checker Tested-by: Build Bot <build@couchbase.com> Reviewed-by: Dave Rigby <daver@couchbase.com>
diff --git a/engines/ep/src/checkpoint_manager.cc b/engines/ep/src/checkpoint_manager.cc
@@ -1355,16 +1355,15 @@ void CheckpointManager::checkOpenCheckpoint(
     }
 }
 
-size_t CheckpointManager::getNumItemsForCursor(const CheckpointCursor* cursor,
-                                               bool accurate) const {
+size_t CheckpointManager::getNumItemsForCursor(
+        const CheckpointCursor* cursor) const {
     std::lock_guard<std::mutex> lh(queueLock);
-    return getNumItemsForCursor(lh, cursor, accurate);
+    return getNumItemsForCursor(lh, cursor);
 }
 
 size_t CheckpointManager::getNumItemsForCursor(
         const std::lock_guard<std::mutex>& lh,
-        const CheckpointCursor* cursor,
-        bool accurate) const {
+        const CheckpointCursor* cursor) const {
     if (cursor && cursor->valid()) {
         size_t items = cursor->getRemainingItemsInCurrentCheckpoint();
         CheckpointList::const_iterator chkptIterator(cursor->getCheckpoint());
@@ -1690,7 +1689,7 @@ void CheckpointManager::addStats(const AddStatFn& add_stat,
                              "vb_%d:num_items_for_persistence",
                              vbucketId.get());
             add_casted_stat(buf.data(),
-                            getNumItemsForCursor(lh, persistenceCursor, true),
+                            getNumItemsForCursor(lh, persistenceCursor),
                             add_stat,
                             cookie);
         }
@@ -1730,21 +1729,10 @@ 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(), 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);
+                add_casted_stat(buf.data(),
+                                getNumItemsForCursor(lh, cursor.second.get()),
+                                add_stat,
+                                cookie);
             }
         }
 
diff --git a/engines/ep/src/checkpoint_manager.h b/engines/ep/src/checkpoint_manager.h
@@ -371,12 +371,8 @@ class CheckpointManager {
      * Returns the count of all items (empty item excluded) that the given
      * cursor has yet to process (i.e. between the cursor's current position and
      * the end of the last checkpoint).
-     *
-     * @param cursor The cursor for which the caller want to know the item count
-     * @param accurate [unused] Added in MB-57400, @todo: remove
      */
-    size_t getNumItemsForCursor(const CheckpointCursor* cursor,
-                                bool accurate = true) const;
+    size_t getNumItemsForCursor(const CheckpointCursor* cursor) 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
@@ -746,8 +742,7 @@ class CheckpointManager {
      * @return number of items to be processed
      */
     size_t getNumItemsForCursor(const std::lock_guard<std::mutex>& lh,
-                                const CheckpointCursor* cursor,
-                                bool accurate) const;
+                                const CheckpointCursor* cursor) const;
 
     /**
      * Clears this CM, effectively removing all checkpoints in the list and
diff --git a/engines/ep/src/connhandler.h b/engines/ep/src/connhandler.h
@@ -289,8 +289,7 @@ class ConnHandler : public DcpConnHandlerIface {
 
     virtual void addStats(const AddStatFn& add_stat, const CookieIface* c);
 
-    virtual void aggregateQueueStats(ConnCounter& stats_aggregator,
-                                     bool accurateItemsRemaining) const {
+    virtual void aggregateQueueStats(ConnCounter& stats_aggregator) const {
         // Empty
     }
 
diff --git a/engines/ep/src/dcp/active_stream.cc b/engines/ep/src/dcp/active_stream.cc
@@ -2263,7 +2263,7 @@ void ActiveStream::transitionState(StreamState newState) {
     }
 }
 
-size_t ActiveStream::getItemsRemaining(bool accurateItemsRemaining) {
+size_t ActiveStream::getItemsRemaining() {
     VBucketPtr vbucket = engine->getVBucket(vb_);
 
     if (!vbucket || !isActive()) {
@@ -2275,8 +2275,7 @@ size_t ActiveStream::getItemsRemaining(bool accurateItemsRemaining) {
     // (b) Items pending in our readyQ
     size_t ckptItems = 0;
     if (auto sp = cursor.lock()) {
-        ckptItems = vbucket->checkpointManager->getNumItemsForCursor(
-                sp.get(), accurateItemsRemaining);
+        ckptItems = vbucket->checkpointManager->getNumItemsForCursor(sp.get());
     }
 
     // Note: concurrent access to readyQ guarded by streamMutex
diff --git a/engines/ep/src/dcp/active_stream.h b/engines/ep/src/dcp/active_stream.h
@@ -286,10 +286,8 @@ class ActiveStream : public Stream,
     /**
      * Returns a count of how many items are outstanding to be sent for this
      * stream's vBucket.
-     *
-     * @param accurate [unused] Added in MB-57400, @todo: remove
      */
-    size_t getItemsRemaining(bool accurateItemsRemaining = true);
+    size_t getItemsRemaining();
 
     /// @returns the count of items backfilled from disk.
     size_t getBackfillItemsDisk() const;
diff --git a/engines/ep/src/dcp/consumer.cc b/engines/ep/src/dcp/consumer.cc
@@ -1246,8 +1246,7 @@ void DcpConsumer::addStats(const AddStatFn& add_stat, const CookieIface* c) {
     addStat("synchronous_replication", isSyncReplicationEnabled(), add_stat, c);
 }
 
-void DcpConsumer::aggregateQueueStats(ConnCounter& aggregator,
-                                      bool accurateItemsRemaining) const {
+void DcpConsumer::aggregateQueueStats(ConnCounter& aggregator) const {
     aggregator.conn_queueBackoff += backoffs;
 }
 
diff --git a/engines/ep/src/dcp/consumer.h b/engines/ep/src/dcp/consumer.h
@@ -258,8 +258,7 @@ class DcpConsumer : public ConnHandler,
 
     void addStats(const AddStatFn& add_stat, const CookieIface* c) override;
 
-    void aggregateQueueStats(ConnCounter& aggregator,
-                             bool accurateItemsRemaining) const override;
+    void aggregateQueueStats(ConnCounter& aggregator) const override;
 
     void notifyStreamReady(Vbid vbucket);
 
diff --git a/engines/ep/src/dcp/producer.cc b/engines/ep/src/dcp/producer.cc
@@ -1610,13 +1610,12 @@ void DcpProducer::addTakeoverStats(const AddStatFn& add_stat,
     add_casted_stat("backfillRemaining", 0, add_stat, c);
 }
 
-void DcpProducer::aggregateQueueStats(ConnCounter& aggregator,
-                                      bool accurateItemsRemaining) const {
+void DcpProducer::aggregateQueueStats(ConnCounter& aggregator) const {
     ++aggregator.totalProducers;
     aggregator.conn_queueDrain += itemsSent;
     aggregator.conn_totalBytes += totalBytesSent;
     aggregator.conn_totalUncompressedDataSize += totalUncompressedDataSize;
-    auto streamAggStats = getStreamAggStats(accurateItemsRemaining);
+    auto streamAggStats = getStreamAggStats();
 
     aggregator.conn_queueRemaining += streamAggStats.itemsRemaining;
     aggregator.conn_queueMemory += streamAggStats.readyQueueMemory;
@@ -1978,19 +1977,17 @@ size_t DcpProducer::getItemsRemaining() const {
     return remainingSize;
 }
 
-DcpProducer::StreamAggStats DcpProducer::getStreamAggStats(
-        bool accurateItemsRemaining) const {
+DcpProducer::StreamAggStats DcpProducer::getStreamAggStats() const {
     DcpProducer::StreamAggStats stats;
 
     std::for_each(
             streams->begin(),
             streams->end(),
-            [&stats, accurateItemsRemaining](const StreamsMap::value_type& vt) {
+            [&stats](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(accurateItemsRemaining);
+                        stats.itemsRemaining += as->getItemsRemaining();
                         stats.readyQueueMemory += as->getReadyQueueMemory();
                         stats.backfillItemsDisk += as->getBackfillItemsDisk();
                         stats.backfillItemsMemory +=
diff --git a/engines/ep/src/dcp/producer.h b/engines/ep/src/dcp/producer.h
@@ -123,8 +123,7 @@ class DcpProducer : public ConnHandler,
                           const CookieIface* c,
                           const VBucket& vb);
 
-    void aggregateQueueStats(ConnCounter& aggregator,
-                             bool accurateItemsRemaining) const override;
+    void aggregateQueueStats(ConnCounter& aggregator) const override;
 
     /**
      * ALERT: Do NOT call this function while holding ConnMap::connLock.
@@ -481,7 +480,7 @@ class DcpProducer : public ConnHandler,
         size_t backfillItemsMemory{};
     };
 
-    StreamAggStats getStreamAggStats(bool accurateItemsRemaining) const;
+    StreamAggStats getStreamAggStats() const;
 
     /**
      * Map the cb::mcbp::DcpStreamEndStatus to one the client can understand.
diff --git a/engines/ep/src/ep_engine.cc b/engines/ep/src/ep_engine.cc
@@ -389,9 +389,7 @@ 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) {
-                    // Do not use the potentially slower "accurate" items
-                    // remaining. MB-57400
-                    tp->aggregateQueueStats(aggregator, false);
+                    tp->aggregateQueueStats(aggregator);
                 }
             });
             addAggregatedProducerStats(collector, aggregator);
@@ -409,7 +407,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, ":", false);
+            if (status = doConnAggStatsInner(collector, ":");
                 status != cb::engine_errc::success) {
                 return status;
             }
@@ -4043,7 +4041,7 @@ struct ConnStatBuilder {
             tc->addStats(add_stat, cookie);
             auto tp = std::dynamic_pointer_cast<DcpProducer>(tc);
             if (tp) {
-                tp->aggregateQueueStats(aggregator, true);
+                tp->aggregateQueueStats(aggregator);
             }
         }
     }
@@ -4060,17 +4058,11 @@ struct ConnStatBuilder {
 
 struct ConnAggStatBuilder {
     /**
-     * Construct with the separator and a configuration bool.
+     * Construct with the separator.
      * @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) {
+    ConnAggStatBuilder(std::string_view sep) : sep(sep) {
     }
 
     /**
@@ -4088,23 +4080,22 @@ struct ConnAggStatBuilder {
      * returns nullptr.
      *
      * @param tc connection
-     * @return pair of counter for the given connection, or nullptr and true
-     *         if the connection is considered a replication stream
+     * @return counter for the given connection, or nullptr
      */
-    std::pair<ConnCounter*, bool> getCounterForConnType(std::string_view name) {
+    ConnCounter* 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, false};
+            return nullptr;
         }
 
         name.remove_prefix(pos1 + 1);
 
         // find the given separator
         size_t pos2 = name.find(sep);
         if (pos2 == std::string_view::npos) {
-            return {nullptr, false};
+            return nullptr;
         }
 
         // extract upto given separator e.g.,
@@ -4115,18 +4106,14 @@ struct ConnAggStatBuilder {
         //  prefix is "replication"
         std::string prefix(name.substr(0, pos2));
 
-        // 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"};
+        return &counters[prefix];
     }
 
-    void aggregate(ConnHandler& conn, ConnCounter* tc, bool isReplication) {
+    void aggregate(ConnHandler& conn, ConnCounter* tc) {
         ConnCounter counter;
         ++counter.totalConns;
 
-        conn.aggregateQueueStats(counter, isReplication);
+        conn.aggregateQueueStats(counter);
 
         ConnCounter& total = getTotalCounter();
         total += counter;
@@ -4142,8 +4129,8 @@ struct ConnAggStatBuilder {
 
     void operator()(std::shared_ptr<ConnHandler> tc) {
         if (tc) {
-            auto aggregator = getCounterForConnType(tc->getName());
-            aggregate(*tc, aggregator.first, aggregator.second);
+            ConnCounter* aggregator = getCounterForConnType(tc->getName());
+            aggregate(*tc, aggregator);
         }
     }
 
@@ -4153,7 +4140,6 @@ struct ConnAggStatBuilder {
 
     std::map<std::string, ConnCounter> counters;
     std::string_view sep;
-    const bool alwaysUseAccurateItemsRemaining{false};
 };
 
 /// @endcond
@@ -4250,8 +4236,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, true);
+                    ep->doConnAggStatsInner(col.forBucket(ep->getName()),
+                                            separator);
                     return cb::engine_errc::success;
                 });
         ExecutorPool::get()->schedule(task);
@@ -4263,15 +4249,13 @@ cb::engine_errc EventuallyPersistentEngine::doConnAggStats(
 }
 
 cb::engine_errc EventuallyPersistentEngine::doConnAggStatsInner(
-        const BucketStatCollector& collector,
-        std::string_view sep,
-        bool cmdStat) {
+        const BucketStatCollector& collector, std::string_view sep) {
     // 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, cmdStat);
+    ConnAggStatBuilder visitor(sep);
     dcpConnMap_->each(visitor);
 
     for (const auto& [connType, counter] : visitor.getCounters()) {
diff --git a/engines/ep/src/ep_engine.h b/engines/ep/src/ep_engine.h
@@ -1043,8 +1043,7 @@ class EventuallyPersistentEngine : public EngineIface, public DcpIface {
                                    const AddStatFn& add_stat,
                                    std::string_view sep);
     cb::engine_errc doConnAggStatsInner(const BucketStatCollector& collector,
-                                        std::string_view sep,
-                                        bool cmdStat);
+                                        std::string_view sep);
     void doTimingStats(const BucketStatCollector& collector);
     cb::engine_errc doSchedulerStats(const CookieIface* cookie,
                                      const AddStatFn& add_stat);
diff --git a/engines/ep/tests/module_tests/checkpoint_test.cc b/engines/ep/tests/module_tests/checkpoint_test.cc
diff --git a/engines/ep/tests/module_tests/dcp_test.cc b/engines/ep/tests/module_tests/dcp_test.cc

Original file line number	Diff line number	Diff line change
`@@ -289,8 +289,7 @@ class ConnHandler : public DcpConnHandlerIface {`
`289`	`289`
`290`	`290`	`virtual void addStats(const AddStatFn& add_stat, const CookieIface* c);`
`291`	`291`
`292`		`- virtual void aggregateQueueStats(ConnCounter& stats_aggregator,`
`293`		`- bool accurateItemsRemaining) const {`
	`292`	`+ virtual void aggregateQueueStats(ConnCounter& stats_aggregator) const {`
`294`	`293`	`// Empty`
`295`	`294`	`}`
`296`	`295`
Original file line number	Diff line number	Diff line change
`@@ -2263,7 +2263,7 @@ void ActiveStream::transitionState(StreamState newState) {`
`2263`	`2263`	`}`
`2264`	`2264`	`}`
`2265`	`2265`
`2266`		`-size_t ActiveStream::getItemsRemaining(bool accurateItemsRemaining) {`
	`2266`	`+size_t ActiveStream::getItemsRemaining() {`
`2267`	`2267`	`VBucketPtr vbucket = engine->getVBucket(vb_);`
`2268`	`2268`
`2269`	`2269`	`if (!vbucket \|\| !isActive()) {`
`@@ -2275,8 +2275,7 @@ size_t ActiveStream::getItemsRemaining(bool accurateItemsRemaining) {`
`2275`	`2275`	`// (b) Items pending in our readyQ`
`2276`	`2276`	`size_t ckptItems = 0;`
`2277`	`2277`	`if (auto sp = cursor.lock()) {`
`2278`		`- ckptItems = vbucket->checkpointManager->getNumItemsForCursor(`
`2279`		`- sp.get(), accurateItemsRemaining);`
	`2278`	`+ ckptItems = vbucket->checkpointManager->getNumItemsForCursor(sp.get());`
`2280`	`2279`	`}`
`2281`	`2280`
`2282`	`2281`	`// Note: concurrent access to readyQ guarded by streamMutex`
Original file line number	Diff line number	Diff line change
`@@ -1246,8 +1246,7 @@ void DcpConsumer::addStats(const AddStatFn& add_stat, const CookieIface* c) {`
`1246`	`1246`	`addStat("synchronous_replication", isSyncReplicationEnabled(), add_stat, c);`
`1247`	`1247`	`}`
`1248`	`1248`
`1249`		`-void DcpConsumer::aggregateQueueStats(ConnCounter& aggregator,`
`1250`		`- bool accurateItemsRemaining) const {`
	`1249`	`+void DcpConsumer::aggregateQueueStats(ConnCounter& aggregator) const {`
`1251`	`1250`	`aggregator.conn_queueBackoff += backoffs;`
`1252`	`1251`	`}`
`1253`	`1252`