MB-43680: Record Snappy decompress times

daverigby · trondn · commit 71804e2ddbed · 2021-01-19T06:37:02.000Z
Time the duration of Snappy decompress called made as part of executing client requests, recoring the timings to: * A new trace span. * A new per-bucket histogram. The trace span will be reported as part of Slow operation logging - for example: WARNING 57: Slow operation. {...,"trace":"... snappy.decompress=6446937532739980:18555 ... The histograms will be reported via mctimings using the stat key 'snappy_decompress', this will be added to cbcollect_info via a future ns_server patch. Change-Id: I72d44cd79aaef3e20be8fad543d6e950e7bbfd75 Reviewed-on: http://review.couchbase.org/c/kv_engine/+/143527 Tested-by: Build Bot <build@couchbase.com> Reviewed-by: Trond Norbye <trond.norbye@couchbase.com>
diff --git a/daemon/buckets.h b/daemon/buckets.h
@@ -134,6 +134,9 @@ class Bucket {
      */
     Hdr1sfMicroSecHistogram subjson_operation_times;
 
+    /// Snappy decompression time histogram.
+    Hdr1sfMicroSecHistogram snappyDecompressionTimes;
+
     using ResponseCounter = cb::RelaxedAtomic<uint64_t>;
 
     /**
diff --git a/daemon/cookie.cc b/daemon/cookie.cc
@@ -36,6 +36,8 @@
 #include <phosphor/phosphor.h>
 #include <platform/checked_snprintf.h>
 #include <platform/compress.h>
+#include <platform/histogram.h>
+#include <platform/scope_timer.h>
 #include <platform/string_hex.h>
 #include <platform/timeutils.h>
 #include <platform/uuid.h>
@@ -601,8 +603,7 @@ bool Cookie::inflateInputPayload(const cb::mcbp::Header& header) {
 
     try {
         auto val = header.getValue();
-        if (!cb::compression::inflate(
-                    cb::compression::Algorithm::Snappy,
+        if (!inflateSnappy(
                     {reinterpret_cast<const char*>(val.data()), val.size()},
                     inflated_input_payload)) {
             setErrorContext("Failed to inflate payload");
@@ -616,6 +617,20 @@ bool Cookie::inflateInputPayload(const cb::mcbp::Header& header) {
     return true;
 }
 
+bool Cookie::inflateSnappy(std::string_view input,
+                           cb::compression::Buffer& output) {
+    // Record how long Snappy decompression takes to both Tracer and
+    // bucket-level histogram.
+    using namespace cb::tracing;
+    ScopeTimer2<HdrMicroSecStopwatch, SpanStopwatch> timer(
+            std::forward_as_tuple(
+                    getConnection().getBucket().snappyDecompressionTimes),
+            std::forward_as_tuple(*this, Code::SnappyDecompress));
+
+    return cb::compression::inflate(
+            cb::compression::Algorithm::Snappy, input, output);
+}
+
 cb::rbac::PrivilegeAccess Cookie::checkPrivilege(
         cb::rbac::Privilege privilege,
         std::optional<ScopeID> sid,
diff --git a/daemon/cookie.h b/daemon/cookie.h
@@ -36,6 +36,9 @@ class Header;
 class Request;
 class Response;
 } // namespace cb::mcbp
+namespace cb::compression {
+class Buffer;
+} // namespace cb::compression
 
 /**
  * The Cookie class represents the cookie passed from the memcached core
@@ -476,14 +479,26 @@ class Cookie : public cb::tracing::Traceable {
     std::string_view getInflatedInputPayload() const;
 
     /**
-     * Inflate the value (if deflated)
+     * Inflate the value (if deflated); caching the inflated value inside the
+     * cookie.
      *
      * @param header The packet header
      * @return true if success, false if an error occurs (the error context
      *         contains the reason why)
      */
     bool inflateInputPayload(const cb::mcbp::Header& header);
 
+    /**
+     * Helper function to inflate the specified Snappy-compressed buffer.
+     *
+     * Records the time taken to decompress as SnappyDecompress tracing span
+     * against this cookie.
+     * @param input Snappy-compressed input buffer.
+     * @param output Buffer to write uncompressed data to.
+     * @returns true if successfully inflated, else false.
+     */
+    bool inflateSnappy(std::string_view input, cb::compression::Buffer& output);
+
     /// Set the current collection meta information. The packet validator
     /// is responsible for checking that the requested collection identifier
     /// is a legal scope (and return an error if it isn't) and to avoid having
diff --git a/daemon/mcbp.cc b/daemon/mcbp.cc
@@ -49,8 +49,7 @@ static bool mcbp_response_handler(std::string_view key,
          mcbp::datatype::is_xattr(datatype))) {
         // The client is not snappy-aware, and the content contains
         // snappy encoded data. Or it's xattr compressed. We need to inflate it!
-        if (!cb::compression::inflate(cb::compression::Algorithm::Snappy,
-                                      payload, buffer)) {
+        if (!cookie->inflateSnappy(payload, buffer)) {
             std::string mykey(key.data(), key.size());
             LOG_WARNING(
                     "<{} ERROR: Failed to inflate body, "
diff --git a/daemon/protocol/mcbp/appendprepend_context.cc b/daemon/protocol/mcbp/appendprepend_context.cc
@@ -95,9 +95,7 @@ ENGINE_ERROR_CODE AppendPrependCommandContext::getItem() {
                 std::string_view payload(
                         static_cast<const char*>(oldItemInfo.value[0].iov_base),
                         oldItemInfo.value[0].iov_len);
-                if (!cb::compression::inflate(cb::compression::Algorithm::Snappy,
-                                              payload,
-                                              buffer)) {
+                if (!cookie.inflateSnappy(payload, buffer)) {
                     return ENGINE_FAILED;
                 }
             } catch (const std::bad_alloc&) {
diff --git a/daemon/protocol/mcbp/arithmetic_context.cc b/daemon/protocol/mcbp/arithmetic_context.cc
@@ -57,10 +57,7 @@ ENGINE_ERROR_CODE ArithmeticCommandContext::getItem() {
                 std::string_view payload(
                         static_cast<const char*>(oldItemInfo.value[0].iov_base),
                         oldItemInfo.value[0].iov_len);
-                if (!cb::compression::inflate(
-                    cb::compression::Algorithm::Snappy,
-                    payload,
-                    buffer)) {
+                if (!cookie.inflateSnappy(payload, buffer)) {
                     return ENGINE_FAILED;
                 }
             } catch (const std::bad_alloc&) {
diff --git a/daemon/protocol/mcbp/gat_context.cc b/daemon/protocol/mcbp/gat_context.cc
@@ -86,8 +86,7 @@ ENGINE_ERROR_CODE GatCommandContext::getAndTouchItem() {
 
 ENGINE_ERROR_CODE GatCommandContext::inflateItem() {
     try {
-        if (!cb::compression::inflate(cb::compression::Algorithm::Snappy,
-                                      payload, buffer)) {
+        if (!cookie.inflateSnappy(payload, buffer)) {
             LOG_WARNING("{}: Failed to inflate item", connection.getId());
             return ENGINE_FAILED;
         }
diff --git a/daemon/protocol/mcbp/get_context.cc b/daemon/protocol/mcbp/get_context.cc
@@ -61,8 +61,7 @@ ENGINE_ERROR_CODE GetCommandContext::getItem() {
 
 ENGINE_ERROR_CODE GetCommandContext::inflateItem() {
     try {
-        if (!cb::compression::inflate(cb::compression::Algorithm::Snappy,
-                                      payload, buffer)) {
+        if (!cookie.inflateSnappy(payload, buffer)) {
             LOG_WARNING("{}: Failed to inflate item", connection.getId());
             return ENGINE_FAILED;
         }
diff --git a/daemon/protocol/mcbp/get_locked_context.cc b/daemon/protocol/mcbp/get_locked_context.cc
@@ -64,8 +64,7 @@ ENGINE_ERROR_CODE GetLockedCommandContext::getAndLockItem() {
 
 ENGINE_ERROR_CODE GetLockedCommandContext::inflateItem() {
     try {
-        if (!cb::compression::inflate(cb::compression::Algorithm::Snappy,
-                                      payload, buffer)) {
+        if (!cookie.inflateSnappy(payload, buffer)) {
             LOG_WARNING(
                     "{}: GetLockedCommandContext::inflateItem:"
                     " Failed to inflate item",
diff --git a/daemon/protocol/mcbp/stats_context.cc b/daemon/protocol/mcbp/stats_context.cc
@@ -303,6 +303,36 @@ static ENGINE_ERROR_CODE stat_connections_executor(const std::string& arg,
     return ENGINE_EWOULDBLOCK;
 }
 
+/**
+ * Helper function to append JSON-formatted histogram statistics for the
+ * specified Bucket histogram data member.
+ */
+static ENGINE_ERROR_CODE stat_histogram_executor(
+        const std::string& arg,
+        Cookie& cookie,
+        Hdr1sfMicroSecHistogram Bucket::*histogram) {
+    if (arg.empty()) {
+        const auto index = cookie.getConnection().getBucketIndex();
+        std::string json_str;
+        if (index == 0) {
+            // Aggregrated timings for all buckets.
+            Hdr1sfMicroSecHistogram aggregated{};
+            for (const auto& bucket : all_buckets) {
+                aggregated += bucket.*histogram;
+            }
+            json_str = aggregated.to_string();
+        } else {
+            // Timings for a specific bucket.
+            auto& bucket = all_buckets[cookie.getConnection().getBucketIndex()];
+            json_str = (bucket.*histogram).to_string();
+        }
+        append_stats({}, json_str, &cookie);
+        return ENGINE_SUCCESS;
+    } else {
+        return ENGINE_EINVAL;
+    }
+}
+
 /**
  * Handler for the <code>stats topkeys</code> command used to retrieve
  * the most popular keys in the attached bucket.
@@ -354,6 +384,19 @@ static ENGINE_ERROR_CODE stat_topkeys_json_executor(const std::string& arg,
     }
 }
 
+/**
+ * Handler for the <code>stats snappy_decompress</code> command used to retrieve
+ * histograms of how long it took to decompress Snappy compressed values.
+ *
+ * @param arg - should be empty
+ * @param cookie the command context
+ */
+static ENGINE_ERROR_CODE stat_snappy_decompress_executor(const std::string& arg,
+                                                         Cookie& cookie) {
+    return stat_histogram_executor(
+            arg, cookie, &Bucket::snappyDecompressionTimes);
+}
+
 /**
  * Handler for the <code>stats subdoc_execute</code> command used to retrieve
  * information from the subdoc subsystem.
@@ -363,26 +406,8 @@ static ENGINE_ERROR_CODE stat_topkeys_json_executor(const std::string& arg,
  */
 static ENGINE_ERROR_CODE stat_subdoc_execute_executor(const std::string& arg,
                                                       Cookie& cookie) {
-    if (arg.empty()) {
-        const auto index = cookie.getConnection().getBucketIndex();
-        std::string json_str;
-        if (index == 0) {
-            // Aggregrated timings for all buckets.
-            Hdr1sfMicroSecHistogram aggregated{};
-            for (const auto& bucket : all_buckets) {
-                aggregated += bucket.subjson_operation_times;
-            }
-            json_str = aggregated.to_string();
-        } else {
-            // Timings for a specific bucket.
-            auto& bucket = all_buckets[cookie.getConnection().getBucketIndex()];
-            json_str = bucket.subjson_operation_times.to_string();
-        }
-        append_stats({}, json_str, &cookie);
-        return ENGINE_SUCCESS;
-    } else {
-        return ENGINE_EINVAL;
-    }
+    return stat_histogram_executor(
+            arg, cookie, &Bucket::subjson_operation_times);
 }
 
 static ENGINE_ERROR_CODE stat_responses_json_executor(const std::string& arg,
@@ -561,6 +586,8 @@ static std::unordered_map<std::string, struct command_stat_handler>
                 {"topkeys", {false, true, true, stat_topkeys_executor}},
                 {"topkeys_json",
                  {false, true, true, stat_topkeys_json_executor}},
+                {"snappy_decompress",
+                 {false, true, true, stat_snappy_decompress_executor}},
                 {"subdoc_execute",
                  {false, true, true, stat_subdoc_execute_executor}},
                 {"responses",
diff --git a/daemon/subdocument_context.cc b/daemon/subdocument_context.cc
@@ -711,8 +711,7 @@ cb::mcbp::Status SubdocCmdContext::get_document_for_searching(
     if (mcbp::datatype::is_snappy(info.datatype)) {
         // Need to expand before attempting to extract from it.
         try {
-            using namespace cb::compression;
-            if (!inflate(Algorithm::Snappy, in_doc.view, inflated_doc_buffer)) {
+            if (!cookie.inflateSnappy(in_doc.view, inflated_doc_buffer)) {
                 char clean_key[KEY_MAX_LENGTH + 32];
                 if (buf_to_printable_buffer(
                             clean_key,
diff --git a/include/memcached/tracer.h b/include/memcached/tracer.h
@@ -27,6 +27,8 @@ namespace cb::tracing {
 enum class Code : uint8_t {
     /// Time spent in the entire request
     Request,
+    /// Time spent decompressing Snappy data.
+    SnappyDecompress,
     /// Time spent waiting for a background fetch operation to be scheduled.
     BackgroundWait,
     /// Time spent performing the actual background load from disk.
@@ -81,6 +83,9 @@ class MEMCACHED_PUBLIC_CLASS Tracer {
     /// End a Span, stopping at the specified time point (defaults to now).
     bool end(SpanId spanId, Clock::time_point endTime = Clock::now());
 
+    // Record a complete Span (when both start and end are already known).
+    void record(Code code, Clock::time_point start, Clock::time_point end);
+
     // Extract the trace vector (and clears the internal trace vector)
     std::vector<Span> extractDurations();
 
@@ -125,6 +130,55 @@ class MEMCACHED_PUBLIC_CLASS Traceable {
     Tracer tracer;
 };
 
+/**
+ * Helper class to assist in recording a Span into Tracer objects via
+ * ScopeTimerN<> classes.
+ *
+ * The start and stop methods record the duration of the span and it is
+ * injected into the provided traceable object as part of object destruction (as
+ * long as start() was at least called).
+ */
+class MEMCACHED_PUBLIC_CLASS SpanStopwatch {
+public:
+    SpanStopwatch(Traceable& traceable, Code code)
+        : traceable(traceable), code(code) {
+    }
+
+    // Disable copy and move: doesn't make sense for this class (could
+    // result in recording spurious spans).
+    SpanStopwatch(const SpanStopwatch&) = delete;
+    SpanStopwatch& operator=(const SpanStopwatch&) = delete;
+    SpanStopwatch& operator=(SpanStopwatch&&) = delete;
+#if defined(_MSC_VER) && (_MSC_VER < 1920)
+    // Prior to MSVC 2019, the move constructor must be defined, if it is
+    // deleted then the code fails to compile (even though MSVC doesn't
+    // actually call the move constructor :S)
+    SpanStopwatch(SpanStopwatch&&) = default;
+#else
+    SpanStopwatch(SpanStopwatch&&) = delete;
+#endif
+
+    ~SpanStopwatch() {
+        if (traceable.isTracingEnabled()) {
+            traceable.getTracer().record(code, startTime, stopTime);
+        }
+    }
+
+    void start(Clock::time_point tp) {
+        startTime = tp;
+    }
+
+    void stop(Clock::time_point tp) {
+        stopTime = tp;
+    }
+
+private:
+    Traceable& traceable;
+    Clock::time_point startTime;
+    Clock::time_point stopTime;
+    const Code code;
+};
+
 } // namespace cb::tracing
 
 MEMCACHED_PUBLIC_API std::ostream& operator<<(
diff --git a/tests/sizes/CMakeLists.txt b/tests/sizes/CMakeLists.txt
@@ -1,6 +1,6 @@
 ADD_EXECUTABLE(memcached_sizes
                sizes.cc)
-TARGET_LINK_LIBRARIES(memcached_sizes mcbp memcached_logger)
+TARGET_LINK_LIBRARIES(memcached_sizes mcbp mcd_tracing memcached_logger)
 
 ADD_TEST(NAME memcached-sizes
          WORKING_DIRECTORY ${CMAKE_RUNTIME_OUTPUT_DIRECTORY}
diff --git a/tracing/tracer.cc b/tracing/tracer.cc
@@ -44,6 +44,13 @@ bool Tracer::end(SpanId spanId, Clock::time_point endTime) {
     });
 }
 
+void Tracer::record(Code code, Clock::time_point start, Clock::time_point end) {
+    auto duration = std::chrono::duration_cast<Span::Duration>(end - start);
+    vecSpans.withLock([code, start, duration](auto& spans) {
+        spans.emplace_back(code, start, duration);
+    });
+}
+
 std::vector<Span> Tracer::extractDurations() {
     std::vector<Span> ret;
     vecSpans.swap(ret);
@@ -123,6 +130,8 @@ MEMCACHED_PUBLIC_API std::string to_string(const cb::tracing::Code tracecode) {
     switch (tracecode) {
     case Code::Request:
         return "request";
+    case Code::SnappyDecompress:
+        return "snappy.decompress";
     case Code::BackgroundWait:
         return "bg.wait";
     case Code::BackgroundLoad:

Original file line number	Diff line number	Diff line change
`@@ -86,8 +86,7 @@ ENGINE_ERROR_CODE GatCommandContext::getAndTouchItem() {`
`86`	`86`
`87`	`87`	`ENGINE_ERROR_CODE GatCommandContext::inflateItem() {`
`88`	`88`	`try {`
`89`		`- if (!cb::compression::inflate(cb::compression::Algorithm::Snappy,`
`90`		`- payload, buffer)) {`
	`89`	`+ if (!cookie.inflateSnappy(payload, buffer)) {`
`91`	`90`	`LOG_WARNING("{}: Failed to inflate item", connection.getId());`
`92`	`91`	`return ENGINE_FAILED;`
`93`	`92`	`}`
Original file line number	Diff line number	Diff line change
`@@ -61,8 +61,7 @@ ENGINE_ERROR_CODE GetCommandContext::getItem() {`
`61`	`61`
`62`	`62`	`ENGINE_ERROR_CODE GetCommandContext::inflateItem() {`
`63`	`63`	`try {`
`64`		`- if (!cb::compression::inflate(cb::compression::Algorithm::Snappy,`
`65`		`- payload, buffer)) {`
	`64`	`+ if (!cookie.inflateSnappy(payload, buffer)) {`
`66`	`65`	`LOG_WARNING("{}: Failed to inflate item", connection.getId());`
`67`	`66`	`return ENGINE_FAILED;`
`68`	`67`	`}`