MB-38283: Stop PassiveStream repeatedly logging ENOMEM

PassiveStream would previously log each time it tried to
processMessage, quickly filling logs. Make it instead log when first
backing off, and when it resumes processing.

Change-Id: Icc73ca9429a307140882a083b2c435bc489379ed
Reviewed-on: http://review.couchbase.org/c/kv_engine/+/142428
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Paolo Cocchi <[email protected]>
Tested-by: Paolo Cocchi <[email protected]>

Author: jameseh96

engines/ep/src/dcp/stream.cc

@@ -2506,17 +2506,24 @@ ENGINE_ERROR_CODE PassiveStream::processMutation(MutationResponse* mutation) {
     }
 
     if (ret != ENGINE_SUCCESS) {
-        log(EXTENSION_LOG_WARNING,
-            "vb:%" PRIu16
-            " Got error '%s' while trying to process "
-            "mutation with seqno:%" PRId64,
-            vb_,
-            cb::to_string(cb::to_engine_errc(ret)).c_str(),
-            mutation->getItem()->getBySeqno());
+        // ENOMEM logging is handled by maybeLogMemoryState
+        if (ret != ENGINE_ENOMEM) {
+            log(EXTENSION_LOG_WARNING,
+                "vb:%" PRIu16
+                " Got error '%s' while trying to process "
+                "mutation with seqno:%" PRId64,
+                vb_,
+                cb::to_string(cb::to_engine_errc(ret)).c_str(),
+                mutation->getItem()->getBySeqno());
+        }
     } else {
         handleSnapshotEnd(vb, *mutation->getBySeqno());
     }
 
+    maybeLogMemoryState(cb::to_engine_errc(ret),
+                        "mutation",
+                        mutation->getItem()->getBySeqno());
+
     return ret;
 }
 
@@ -2588,17 +2595,23 @@ ENGINE_ERROR_CODE PassiveStream::processDeletion(MutationResponse* deletion) {
     }
 
     if (ret != ENGINE_SUCCESS) {
-        log(EXTENSION_LOG_WARNING,
-            "vb:%" PRIu16
-            " Got error '%s' while trying to process "
-            "deletion with seqno:%" PRId64,
-            vb_,
-            cb::to_string(cb::to_engine_errc(ret)).c_str(),
-            *deletion->getBySeqno());
+        // ENOMEM logging is handled by maybeLogMemoryState
+        if (ret != ENGINE_ENOMEM) {
+            log(EXTENSION_LOG_WARNING,
+                "vb:%" PRIu16
+                " Got error '%s' while trying to process "
+                "deletion with seqno:%" PRId64,
+                vb_,
+                cb::to_string(cb::to_engine_errc(ret)).c_str(),
+                *deletion->getBySeqno());
+        }
     } else {
         handleSnapshotEnd(vb, *deletion->getBySeqno());
     }
 
+    maybeLogMemoryState(
+            cb::to_engine_errc(ret), "deletion", *deletion->getBySeqno());
+
     return ret;
 }
 
@@ -2916,6 +2929,27 @@ void PassiveStream::log(EXTENSION_LOG_LEVEL severity,
     va_end(va);
 }
 
+void PassiveStream::maybeLogMemoryState(cb::engine_errc status,
+                                        const std::string& msgType,
+                                        int64_t seqno) {
+    bool previousNoMem = isNoMemory.load();
+    if (status == cb::engine_errc::no_memory && !previousNoMem) {
+        log(EXTENSION_LOG_WARNING,
+            "vb:%" PRIu16
+            " Got error '%s' while trying to process %s with seqno:%" PRId64,
+            vb_,
+            cb::to_string(status).c_str(),
+            msgType.c_str(),
+            seqno);
+        isNoMemory.store(true);
+    } else if (status == cb::engine_errc::success && previousNoMem) {
+        log(EXTENSION_LOG_INFO,
+            "vb:%" PRIu16 " PassiveStream resuming after no-memory backoff",
+            vb_);
+        isNoMemory.store(false);
+    }
+}
+
 void PassiveStream::notifyStreamReady() {
     auto consumer = consumerPtr.lock();
     if (!consumer) {

engines/ep/src/dcp/stream.h

@@ -813,6 +813,24 @@ class PassiveStream : public Stream {
 
     void log(EXTENSION_LOG_LEVEL severity, const char* fmt, ...) const override;
 
+    /**
+     * Log when the stream backs off or resumes processing items.
+     *
+     * Determines if the stream has backed off or resumed based on
+     * the passed status. Assumes no_memory indicates the stream will
+     * backoff, and that success indicates the stream has resumed
+     * processing.
+     *
+     * @param status last error code returned by the engine when processing
+     *               a message
+     * @param msgType string indicating the type of the last message processed
+     *                ("mutation", "deletion", "expiration", "prepare")
+     * @param seqno seqno of last message processed
+     */
+    void maybeLogMemoryState(cb::engine_errc status,
+                             const std::string& msgType,
+                             int64_t seqno);
+
     /**
      * Notifies the consumer connection that the stream has items ready to be
      * pick up.
@@ -887,6 +905,10 @@ class PassiveStream : public Stream {
      * messages in the buffer.
      */
     std::function<void()> processBufferedMessages_postFront_Hook;
+
+    // Flag indicating if the most recent call to processMessage
+    // backed off due to ENOMEM. Only used for limiting logging
+    std::atomic<bool> isNoMemory{false};
 };
 
 #endif  // SRC_DCP_STREAM_H_