MB-58531: Allow the config to define collections per OSO snapshot

Add a new configuration value (default to 1)

* dcp_oso_max_collections_per_backfill

Read the new config parameter when ActiveStream decides if an OSO
snapshot can be used (replacing the singleCollection() test).

Add a unit test which configures > 1 collection per snapshot and checks
that two collections are delivered.

Change-Id: Ic8d56e08a033b448cc7e494ba0b334b48a204b64
Reviewed-on: https://review.couchbase.org/c/kv_engine/+/197004
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: jimwwalker

engines/ep/configuration.json

@@ -1114,6 +1114,12 @@
             "dynamic": true,
             "type": "size_t"
         },
+        "dcp_oso_max_collections_per_backfill": {
+            "default": "1",
+            "desr": "This is the maximum number of collections that a DCP stream can be filtering to be eligible for OSO",
+            "dynamic": true,
+            "type": "size_t"
+        },
         "dcp_scan_byte_limit": {
             "default": {
                 "on-prem": "4 * 1024 * 1024",

engines/ep/src/collections/vbucket_filter.cc

@@ -503,6 +503,15 @@ cb::engine_errc Filter::checkPrivileges(
     return cb::engine_errc::success;
 }
 
+bool Filter::isOsoSuitable(size_t limit) const {
+    // Note: passthrough means all collections match, we could probably do OSO
+    // for passthrough, setting the key range to match every collection, i.e.
+    // namespace \00 to \01 and \08 to \ff, but for now just consider the
+    // explicitly filter case and when the filter size is acceptable for the
+    // given limit.
+    return !passthrough && filter.size() <= limit;
+}
+
 void Filter::dump() const {
     std::cerr << *this << std::endl;
 }

engines/ep/src/collections/vbucket_filter.h

@@ -194,11 +194,6 @@ class Filter {
         return filter.end();
     }
 
-    /// @return true if this filter represents a single collection
-    bool singleCollection() const {
-        return !passthrough && filter.size() == 1;
-    }
-
     /**
      * Method to check if the filter dose not filter collections
      * @return true if the filter is a pass-through filter
@@ -218,6 +213,15 @@ class Filter {
         return !isPassThroughFilter() && !isLegacyFilter();
     }
 
+    /**
+     * Check if the filter is suitable for an OSO snapshot given the limit.
+     * Currently a !passthrough filter with a size that does not exceed the
+     * limit would result in true.
+     *
+     * @return true if an OSO snapshot is ok for this filter
+     */
+    bool isOsoSuitable(size_t limit) const;
+
     /**
      * Dump this to std::cerr
      */

engines/ep/src/dcp/active_stream.cc

@@ -2030,18 +2030,19 @@ void ActiveStream::scheduleBackfill_UNLOCKED(DcpProducer& producer,
 bool ActiveStream::tryAndScheduleOSOBackfill(DcpProducer& producer,
                                              VBucket& vb) {
     // OSO only _allowed_ (but may not be chosen):
-    // if the filter is set to a single collection.
-    // if this is the initial backfill request
     // if the client has enabled OSO
-    if (producer.isOutOfOrderSnapshotsEnabled() && filter.singleCollection() &&
+    // if the size of the collection filter fits the current configured max
+    // if this is the initial backfill request (diskonly or not)
+    const auto& config = engine->getConfiguration();
+
+    if (producer.isOutOfOrderSnapshotsEnabled() &&
+        filter.isOsoSuitable(config.getDcpOsoMaxCollectionsPerBackfill()) &&
         lastReadSeqno.load() == 0 &&
         ((curChkSeqno.load() > lastReadSeqno.load() + 1) || (isDiskOnly()))) {
-
         // however OSO is only _used_ if:
         // - dcp_oso_backfill is set to enabled,
         // - dcp_oso_backfill is set to "auto", and OSO is predicted to be
         //   faster for this backfill.
-        const auto& config = engine->getConfiguration();
         const auto osoBackfill = config.getDcpOsoBackfill();
         if (osoBackfill == "disabled") {
             return false;

engines/ep/src/ep_engine.cc

@@ -957,6 +957,9 @@ cb::engine_errc EventuallyPersistentEngine::setDcpParam(const std::string& key,
             getConfiguration().setDcpTakeoverMaxTime(std::stoull(val));
         } else if (key == "dcp_backfill_byte_limit") {
             getConfiguration().setDcpBackfillByteLimit(std::stoull(val));
+        } else if (key == "dcp_oso_max_collections_per_backfill") {
+            getConfiguration().setDcpOsoMaxCollectionsPerBackfill(
+                    std::stoull(val));
         } else {
             msg = "Unknown config param";
             rv = cb::engine_errc::no_such_key;

engines/ep/tests/ep_testsuite.cc

@@ -6547,6 +6547,7 @@ static enum test_result test_mb19687_fixed(EngineIface* h) {
               "ep_dcp_oso_backfill_large_value_ratio",
               "ep_dcp_oso_backfill_small_value_ratio",
               "ep_dcp_oso_backfill_small_item_size_threshold",
+              "ep_dcp_oso_max_collections_per_backfill",
               "ep_dcp_scan_byte_limit",
               "ep_dcp_scan_item_limit",
               "ep_dcp_takeover_max_time",
@@ -6808,6 +6809,7 @@ static enum test_result test_mb19687_fixed(EngineIface* h) {
               "ep_dcp_oso_backfill_large_value_ratio",
               "ep_dcp_oso_backfill_small_value_ratio",
               "ep_dcp_oso_backfill_small_item_size_threshold",
+              "ep_dcp_oso_max_collections_per_backfill",
               "ep_dcp_producer_snapshot_marker_yield_limit",
               "ep_dcp_scan_byte_limit",
               "ep_dcp_scan_item_limit",

engines/ep/tests/module_tests/collections/collections_oso_dcp_test.cc

@@ -21,6 +21,8 @@
 #include "tests/module_tests/test_helpers.h"
 #include "vbucket.h"
 
+#include <folly/portability/GMock.h>
+#include <folly/portability/GTest.h>
 #include <spdlog/fmt/fmt.h>
 
 class CollectionsOSODcpTest : public CollectionsDcpParameterizedTest {
@@ -29,10 +31,12 @@ class CollectionsOSODcpTest : public CollectionsDcpParameterizedTest {
     }
 
     void SetUp() override {
-        config_string += "collections_enabled=true";
         // Disable OSO backfill auto-selection to simplify most of the
         // functional tests - set to always.
-        config_string += ";dcp_oso_backfill=enabled";
+        if (!config_string.empty()) {
+            config_string += ";";
+        }
+        config_string += "dcp_oso_backfill=enabled";
 
         CollectionsDcpParameterizedTest::SetUp();
         producers = std::make_unique<CollectionsDcpTestProducers>();
@@ -997,6 +1001,90 @@ TEST_P(CollectionsOSODcpAutoSelectTest, SmallThresholdDynamic) {
     testDcpOsoBackfillAutomaticMode(0.004, 0.006, 300);
 }
 
+class CollectionsOSOMultiTest : public CollectionsOSODcpTest {
+public:
+    void SetUp() override {
+        CollectionsOSODcpTest::SetUp();
+        std::string msg;
+        ASSERT_EQ(cb::engine_errc::success,
+                  engine->setDcpParam(
+                          "dcp_oso_max_collections_per_backfill", "100", msg));
+    }
+};
+
+// The test uses three collections and filters for two and expects the two
+// filtered collections to be sent in the OSO snapshot
+TEST_P(CollectionsOSOMultiTest, multi) {
+    using namespace cb::mcbp;
+
+    CollectionsManifest cm;
+    setCollections(cookie,
+                   cm.add(CollectionEntry::vegetable)
+                           .add(CollectionEntry::fruit)
+                           .add(CollectionEntry::dairy));
+    flush_vbucket_to_disk(vbid, 3);
+
+    // 3 collections
+    std::array<CollectionID, 3> collections = {CollectionUid::fruit,
+                                               CollectionUid::dairy,
+                                               CollectionUid::vegetable};
+    // 4 keys
+    std::array<std::string, 4> keys = {{"a", "b", "c", "d"}};
+
+    // combine!
+    for (auto cid : collections) {
+        for (const auto& key : keys) {
+            store_item(vbid, makeStoredDocKey(key, cid), "value" + key);
+        }
+        flush_vbucket_to_disk(vbid, keys.size());
+    }
+
+    ensureDcpWillBackfill();
+
+    // filter on collections 1 and 2 of the 3 that have been written to.
+    nlohmann::json filter = {{"collections",
+                              {collections.at(1).to_string(false),
+                               collections.at(2).to_string(false)}}};
+    createDcpObjects(filter.dump(), OutOfOrderSnapshots::Yes, 0);
+
+    runBackfill();
+
+    stepAndExpect(ClientOpcode::DcpOsoSnapshot);
+    EXPECT_EQ(uint32_t(request::DcpOsoSnapshotFlags::Start),
+              producers->last_oso_snapshot_flags);
+
+    std::unordered_set<std::string> keys1, keys2;
+
+    // This test is written to not assume any ordering of the snapshot.
+    // This loop will collect all keys and check they are for the correct
+    // collections and the correct keys
+    while (producer->stepWithBorderGuard(*producers) ==
+                   cb::engine_errc::success &&
+           producers->last_op != ClientOpcode::DcpOsoSnapshot) {
+        EXPECT_THAT(producers->last_op,
+                    testing::AnyOf(ClientOpcode::DcpSystemEvent,
+                                   ClientOpcode::DcpMutation));
+        EXPECT_THAT(producers->last_collection_id,
+                    testing::AnyOf(collections.at(1), collections.at(2)));
+
+        if (producers->last_op == ClientOpcode::DcpMutation) {
+            EXPECT_THAT(producers->last_key, testing::AnyOfArray(keys));
+
+            if (producers->last_collection_id == collections.at(1)) {
+                keys1.emplace(producers->last_key);
+            } else {
+                keys2.emplace(producers->last_key);
+            }
+        }
+    }
+
+    EXPECT_EQ(uint32_t(request::DcpOsoSnapshotFlags::End),
+              producers->last_oso_snapshot_flags);
+    // All Keys from each collection must of been observed
+    EXPECT_EQ(keys.size(), keys1.size());
+    EXPECT_EQ(keys.size(), keys2.size());
+}
+
 INSTANTIATE_TEST_SUITE_P(CollectionsOSOEphemeralTests,
                          CollectionsOSOEphemeralTest,
                          STParameterizedBucketTest::ephConfigValues(),
@@ -1011,3 +1099,8 @@ INSTANTIATE_TEST_SUITE_P(CollectionsOSODcpAutoSelectTests,
                          CollectionsOSODcpAutoSelectTest,
                          STParameterizedBucketTest::persistentConfigValues(),
                          STParameterizedBucketTest::PrintToStringParamName);
+
+INSTANTIATE_TEST_SUITE_P(CollectionsOSOMultiTests,
+                         CollectionsOSOMultiTest,
+                         STParameterizedBucketTest::persistentConfigValues(),
+                         STParameterizedBucketTest::PrintToStringParamName);

include/memcached/dockey.h

@@ -128,7 +128,8 @@ class CollectionID {
     /// Get network byte order of the value
     CollectionIDNetworkOrder to_network() const;
 
-    std::string to_string() const;
+    /// @param xPrefix true if the result should be 0x prefixed
+    std::string to_string(bool xPrefix = true) const;
 
     /**
      * A number of interfaces pass collection-ID as a string, e.g. the Manifest

utilities/dockey.cc

@@ -80,8 +80,8 @@ std::ostream& operator<<(std::ostream& os, const CollectionID& cid) {
     return os << cid.to_string();
 }
 
-std::string CollectionID::to_string() const {
-    return fmt::format("{:#x}", value);
+std::string CollectionID::to_string(bool xPrefix) const {
+    return fmt::format(xPrefix ? "{:#x}" : "{:x}", value);
 }
 
 bool operator==(ScopeIDType lhs, const ScopeID& rhs) {