MB-70483: Add expected duration for topkeys

And time out and clean up topkeys samples running too long

Change-Id: I3a9fa7aff2cde0ae6e85a14bb16c66140f085e14
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/239904
Reviewed-by: Jim Walker <jim@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: trondn

daemon/connection.cc

@@ -1034,6 +1034,11 @@ bool Connection::executeCommandsCallback() {
     // Make sure any core dumps from this code contain the bucket name.
     cb::DebugVariable bucketName(cb::toCharArrayN<32>(getBucket().name));
     const auto start = last_used_timestamp = std::chrono::steady_clock::now();
+
+    if (thread.keyTrace && thread.keyTrace->is_expired(start)) {
+        thread.keyTrace.reset();
+    }
+
     current_timeslice_end = start + Settings::instance().getCommandTimeSlice();
 
     processBlockedSendQueue(start);

daemon/ioctl.cc

@@ -354,6 +354,23 @@ static cb::engine_errc ioctlSetTopkeysStart(Cookie& cookie,
         }
     }
 
+    std::size_t expected_duration = 60;
+    if (args.contains("expected_duration")) {
+        try {
+            std::size_t val =
+                    std::stoul(args.find("expected_duration")->second) * 1.3;
+            expected_duration = std::min(val, expected_duration);
+            if (val == 0) {
+                cookie.setErrorContext("expected_duration cannot be zero");
+                return cb::engine_errc::invalid_arguments;
+            }
+        } catch (const std::exception&) {
+            cookie.setErrorContext(
+                    "Failed to parse expected_duration argument");
+            return cb::engine_errc::invalid_arguments;
+        }
+    }
+
     std::vector<std::size_t> bucket_filter;
     if (args.contains("bucket_filter")) {
         std::unordered_map<std::string, std::size_t> bucketnames;
@@ -385,8 +402,11 @@ static cb::engine_errc ioctlSetTopkeysStart(Cookie& cookie,
         }
     }
 
-    auto collector =
-            cb::trace::topkeys::Collector::create(limit, shards, bucket_filter);
+    auto expiry_time = cb::time::steady_clock::now() +
+                       std::chrono::seconds(expected_duration);
+
+    auto collector = cb::trace::topkeys::Collector::create(
+            limit, shards, expiry_time, bucket_filter);
     if (!collector) {
         cookie.setErrorContext("Failed to create topkeys collector");
         LOG_WARNING_CTX("Failed to create topkeys collector",

daemon/top_keys.cc

@@ -75,8 +75,10 @@ void to_json(nlohmann::json& json, const Result& result) {
 
 class CountingCollector : public Collector {
 public:
-    CountingCollector(std::size_t max, std::size_t shards)
-        : limit(max), shardmaps(shards) {
+    CountingCollector(std::size_t max,
+                      std::size_t shards,
+                      cb::time::steady_clock::time_point expiry_time)
+        : Collector(expiry_time), limit(max), shardmaps(shards) {
     }
 
     void access(const size_t bucket,
@@ -209,8 +211,10 @@ class FilteredCountingCollector : public CountingCollector {
 public:
     FilteredCountingCollector(std::size_t max,
                               std::size_t shards,
+                              cb::time::steady_clock::time_point expiry_time,
                               std::vector<std::size_t> buckets)
-        : CountingCollector(max, shards), bucketfilter(std::move(buckets)) {
+        : CountingCollector(max, shards, expiry_time),
+          bucketfilter(std::move(buckets)) {
     }
 
     void access(const size_t bucket,
@@ -225,15 +229,18 @@ class FilteredCountingCollector : public CountingCollector {
     std::vector<std::size_t> bucketfilter;
 };
 
-std::shared_ptr<Collector> Collector::create(std::size_t num_keys,
-                                             std::size_t shards,
-                                             std::vector<std::size_t> buckets) {
+std::shared_ptr<Collector> Collector::create(
+        std::size_t num_keys,
+        std::size_t shards,
+        cb::time::steady_clock::time_point expiry_time,
+        std::vector<std::size_t> buckets) {
     if (num_keys) {
         if (buckets.empty()) {
-            return std::make_unique<CountingCollector>(num_keys, shards);
+            return std::make_unique<CountingCollector>(
+                    num_keys, shards, expiry_time);
         }
         return std::make_unique<FilteredCountingCollector>(
-                num_keys, shards, std::move(buckets));
+                num_keys, shards, expiry_time, std::move(buckets));
     }
     return {};
 }

daemon/top_keys.h

@@ -13,6 +13,7 @@
 #include "memcached/dockey_view.h"
 
 #include <nlohmann/json_fwd.hpp>
+#include <platform/cb_time.h>
 #include <memory>
 #include <unordered_map>
 #include <vector>
@@ -51,6 +52,9 @@ class Collector {
      *
      * @param num_keys The maximum number of keys to track in a shard
      * @param shards The number of internal maps to shard the keyspace across
+     * @param expiry_time The time after which the collected data should be
+     *                    considered expired and can be discarded (and the
+     *                    collector can be reset)
      * @param buckets If non-empty only track keys accessed in the specified
      *                bucket. For space efficiencty we use the bucket *id*
      *                and not the bucket name (as buckets typically don't
@@ -59,8 +63,18 @@ class Collector {
     static std::shared_ptr<Collector> create(
             std::size_t num_keys,
             std::size_t shards,
+            cb::time::steady_clock::time_point expiry_time =
+                    cb::time::steady_clock::now() + std::chrono::minutes(1),
             std::vector<std::size_t> buckets = {});
 
+    /// Is this collector expired or not (e.g. should we discard the collected
+    /// data). The time is passed in as an argument to avoid having to fetch the
+    /// clock multiple times when we want to check multiple collectors (e.g. for
+    /// each front end thread).
+    bool is_expired(cb::time::steady_clock::time_point now) {
+        return now >= expiry_time;
+    }
+
     /**
      * Register access for a key in a given bucket
      *
@@ -80,7 +94,8 @@ class Collector {
     virtual Result getResults(size_t limit) const = 0;
 
 protected:
-    Collector() = default;
+    Collector(cb::time::steady_clock::time_point exp) : expiry_time(exp) {};
+    const cb::time::steady_clock::time_point expiry_time;
 };
 
 } // namespace cb::trace::topkeys

daemon/top_keys_test.cc

@@ -77,7 +77,11 @@ TEST(TopKeyCollector, TestAccessNoFilter) {
 TEST(TopKeyCollector, testFiltered) {
     constexpr std::size_t num_shards = 4;
     constexpr std::size_t num_keys = 10;
-    auto collector = Collector::create(num_keys, num_shards, {{0, 1}});
+    auto collector = Collector::create(
+            num_keys,
+            num_shards,
+            cb::time::steady_clock::now() + std::chrono::minutes(1),
+            {{0, 1}});
     for (int ii = 0; ii < 10; ++ii) {
         collector->access(ii, false, fmt::format("key-{}", ii));
     }
@@ -154,3 +158,13 @@ TEST(TopKeyCollector, TestSingleShard) {
     auto& keys = bucket["cid:0x0"];
     ASSERT_EQ(100, keys.size()) << "json: " << keys.dump(2);
 }
+
+TEST(TopKeyCollector, testExpiration) {
+    cb::time::steady_clock::use_chrono = false;
+    auto collector = Collector::create(
+            1, 1, cb::time::steady_clock::now() + std::chrono::minutes(1));
+    EXPECT_FALSE(collector->is_expired(cb::time::steady_clock::now()));
+    cb::time::steady_clock::advance(std::chrono::minutes(2));
+    EXPECT_TRUE(collector->is_expired(cb::time::steady_clock::now()));
+    cb::time::steady_clock::use_chrono = true;
+}

programs/mctopkeys/mctopkeys.cc

@@ -191,8 +191,9 @@ int main(const int argc, char** argv) {
         }
 
         const auto start_command = fmt::format(
-                "topkeys.start?limit={}{}{}",
+                "topkeys.start?limit={}&expected_duration={}{}{}",
                 collect_limit,
+                duration.count(),
                 bucket_filter.empty()
                         ? ""
                         : fmt::format("&bucket_filter={}", bucket_filter),