add ServiceMapHistory class

10 files changed, 561 insertions, 5 deletions

diff --git a/slobrok/CMakeLists.txt b/slobrok/CMakeLists.txt
index 6601dda9304..00ddf7296ca 100644
--- a/slobrok/CMakeLists.txt
+++ b/slobrok/CMakeLists.txt
@@ -21,6 +21,7 @@ vespa_define_module(
     src/tests/configure
     src/tests/mirrorapi
     src/tests/registerapi
+    src/tests/service_map_history
     src/tests/standalone
     src/tests/startsome
     src/tests/startup
diff --git a/slobrok/src/tests/service_map_history/CMakeLists.txt b/slobrok/src/tests/service_map_history/CMakeLists.txt
new file mode 100644
index 00000000000..84aace8629c
--- /dev/null
+++ b/slobrok/src/tests/service_map_history/CMakeLists.txt
@@ -0,0 +1,9 @@
+# Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(slobrok_service_map_history_test_app TEST
+    SOURCES
+    service_map_history_test.cpp
+    DEPENDS
+    slobrok_slobrokserver
+    GTest::GTest
+)
+vespa_add_test(NAME slobrok_service_map_history_test_app COMMAND slobrok_service_map_history_test_app)
diff --git a/slobrok/src/tests/service_map_history/service_map_history_test.cpp b/slobrok/src/tests/service_map_history/service_map_history_test.cpp
new file mode 100644
index 00000000000..486a5d7b9cb
--- /dev/null
+++ b/slobrok/src/tests/service_map_history/service_map_history_test.cpp
@@ -0,0 +1,225 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/slobrok/server/service_map_history.h>
+#include <vespa/vespalib/gtest/gtest.h>
+#include <vespa/vespalib/util/stringfmt.h>
+#include <map>
+
+using namespace vespalib;
+using namespace slobrok;
+using vespalib::make_string_short::fmt;
+
+using Map = std::map<vespalib::string, vespalib::string>;
+
+struct Dumper : ServiceMapHistory::DiffCompletionHandler {
+    std::unique_ptr<MapDiff> got = {};
+    void handle(MapDiff diff) override {
+        got = std::make_unique<MapDiff>(diff);
+    }
+};
+
+MapDiff diffGen(ServiceMapHistory &history, uint32_t gen) {
+    Dumper dumper;
+    history.asyncGenerationDiff(&dumper, GenCnt(gen));
+    EXPECT_TRUE(dumper.got);
+    return *dumper.got;
+}
+
+Map dump(ServiceMapHistory &history) {
+    MapDiff full = diffGen(history, 0);
+    EXPECT_TRUE(full.is_full_dump());
+    Map result;
+    for (const auto & [ k, v ] : full.updated) {
+        result[k] = v;
+    }
+    return result;
+}
+
+
+vespalib::string lookup(ServiceMapHistory &history, const vespalib::string &name) {
+    auto map = dump(history);
+    auto iter = map.find(name);
+    if (iter == map.end()) {
+        return {};
+    } else {
+        return iter->second;
+    }
+}
+
+TEST(ServiceMapHistoryTest, empty_inspection) {
+    ServiceMapHistory p;
+    auto bar = dump(p);
+    EXPECT_TRUE(bar.empty());
+
+    auto gen = p.currentGen();
+    EXPECT_EQ(gen, GenCnt(1));
+
+    Dumper dumper;
+    {
+        ServiceMapHistory empty2;
+        empty2.asyncGenerationDiff(&dumper, gen);
+    }
+    EXPECT_TRUE(dumper.got);
+    auto diff1 = *dumper.got;
+    EXPECT_FALSE(diff1.is_full_dump());
+    EXPECT_EQ(diff1.fromGen, gen);
+    EXPECT_TRUE(diff1.removed.empty());
+    EXPECT_TRUE(diff1.updated.empty());
+    EXPECT_EQ(diff1.toGen, gen);
+
+    auto diff2 = diffGen(p, 42);
+    EXPECT_TRUE(diff2.is_full_dump());
+    EXPECT_EQ(diff2.fromGen, GenCnt(0));
+    EXPECT_TRUE(diff2.removed.empty());
+    EXPECT_TRUE(diff2.updated.empty());
+    EXPECT_EQ(diff2.toGen, gen);
+
+    auto diff3 = diffGen(p, 0);
+    EXPECT_TRUE(diff3.is_full_dump());
+    EXPECT_EQ(diff3.fromGen, GenCnt(0));
+    EXPECT_TRUE(diff3.removed.empty());
+    EXPECT_TRUE(diff3.updated.empty());
+    EXPECT_EQ(diff3.toGen, gen);
+}
+
+TEST(ServiceMapHistoryTest, full_inspection) {
+    Dumper dumper;
+    {
+        ServiceMapHistory p;
+        for (int i = 0; i < 1984; ++i) {
+            auto name = fmt("key/%d/name", i);
+            auto spec = fmt("tcp/host%d.domain.tld:19099", 10000+i);
+            p.update(ServiceMapping{name, spec});
+        }
+        EXPECT_EQ(p.currentGen(), GenCnt(1985));
+        p.remove("key/666/name");
+        EXPECT_EQ(p.currentGen(), GenCnt(1986));
+        p.update(ServiceMapping{"key/1969/name", "tcp/woodstock:19069"});
+        EXPECT_EQ(p.currentGen(), GenCnt(1987));
+
+        auto map = dump(p);
+        
+        EXPECT_FALSE(map.contains("foo"));
+        EXPECT_TRUE(map.contains("key/0/name"));
+        EXPECT_FALSE(map.contains("key/666/name"));
+        EXPECT_TRUE(map.contains("key/1983/name"));
+        EXPECT_FALSE(map.contains("key/1984/name"));
+        EXPECT_TRUE(map.contains("key/1969/name"));
+
+        auto foo = map["key/0/name"];
+        EXPECT_EQ(foo, "tcp/host10000.domain.tld:19099");
+
+        foo = map["key/123/name"];
+        EXPECT_EQ(foo, "tcp/host10123.domain.tld:19099");
+
+        foo = map["key/1983/name"];
+        EXPECT_EQ(foo, "tcp/host11983.domain.tld:19099");
+
+        foo = map["key/1969/name"];
+        EXPECT_EQ(foo, "tcp/woodstock:19069");
+        
+        EXPECT_EQ(map.size(), 1983ul);
+
+        auto gen = p.currentGen();
+        
+        auto diff2 = diffGen(p, 42);
+        EXPECT_TRUE(diff2.is_full_dump());
+        EXPECT_EQ(diff2.fromGen, GenCnt(0));
+        EXPECT_TRUE(diff2.removed.empty());
+        EXPECT_EQ(diff2.updated.size(), 1983ul);
+        EXPECT_EQ(diff2.toGen, gen);
+        
+        auto diff3 = diffGen(p, 1984);
+        EXPECT_FALSE(diff3.is_full_dump());
+        EXPECT_EQ(diff3.fromGen, GenCnt(1984));
+        EXPECT_EQ(diff3.removed.size(), 1ul);
+        EXPECT_EQ(diff3.updated.size(), 2ul);
+        EXPECT_EQ(diff3.toGen, gen);
+        
+        p.asyncGenerationDiff(&dumper, gen);
+        EXPECT_FALSE(dumper.got);
+    }
+    EXPECT_TRUE(dumper.got);
+    auto diff1 = *dumper.got;
+    EXPECT_EQ(diff1.fromGen, GenCnt(1987));
+    EXPECT_TRUE(diff1.removed.empty());
+    EXPECT_TRUE(diff1.updated.empty());
+    EXPECT_EQ(diff1.toGen, GenCnt(1987));
+    EXPECT_FALSE(diff1.is_full_dump());
+}
+
+class MockListener : public ServiceMapHistory::DiffCompletionHandler {
+public:
+    bool got_update = false;
+    GenCnt got_gen = GenCnt(0);
+    size_t got_removes = 0;
+    size_t got_updates = 0;
+
+    void handle(MapDiff diff) override {
+        got_update = true;
+        got_gen = diff.toGen;
+        got_removes = diff.removed.size();
+        got_updates = diff.updated.size();
+    }
+
+    ~MockListener();
+};
+
+MockListener::~MockListener() = default;
+
+TEST(ServiceMapHistoryTest, handlers_test) {
+    MockListener handler1;
+    MockListener handler2;
+    MockListener handler3;
+    MockListener handler4;
+    MockListener handler5;
+    {
+        ServiceMapHistory p;
+        p.asyncGenerationDiff(&handler1, GenCnt(0));
+        p.asyncGenerationDiff(&handler2, GenCnt(1));
+        EXPECT_TRUE(handler1.got_update);
+        EXPECT_FALSE(handler2.got_update);
+        EXPECT_FALSE(handler3.got_update);
+        EXPECT_EQ(handler1.got_gen, GenCnt(1));
+        EXPECT_EQ(handler1.got_removes, 0ul);
+        EXPECT_EQ(handler1.got_updates, 0ul);
+
+        handler1.got_update = false;
+        p.update(ServiceMapping{"foo", "bar"});
+        EXPECT_FALSE(handler1.got_update);
+        EXPECT_TRUE(handler2.got_update);
+        EXPECT_FALSE(handler3.got_update);
+        EXPECT_EQ(handler2.got_removes, 0ul);
+        EXPECT_EQ(handler2.got_updates, 1ul);
+
+        handler2.got_update = false;
+        p.asyncGenerationDiff(&handler3, GenCnt(2));
+        EXPECT_FALSE(handler3.got_update);
+        p.remove("foo");
+        EXPECT_FALSE(handler1.got_update);
+        EXPECT_FALSE(handler2.got_update);
+        EXPECT_TRUE(handler3.got_update);
+        EXPECT_EQ(handler3.got_removes, 1ul);
+        EXPECT_EQ(handler3.got_updates, 0ul);
+
+        p.asyncGenerationDiff(&handler4, GenCnt(3));
+        EXPECT_FALSE(handler4.got_update);
+        p.asyncGenerationDiff(&handler5, GenCnt(3));
+        EXPECT_FALSE(handler5.got_update);
+        p.cancel(&handler4);
+
+        handler1.got_update = false;
+        handler2.got_update = false;
+        handler3.got_update = false;
+    }
+    EXPECT_FALSE(handler1.got_update);
+    EXPECT_FALSE(handler2.got_update);
+    EXPECT_FALSE(handler3.got_update);
+    EXPECT_FALSE(handler4.got_update);
+    EXPECT_TRUE(handler5.got_update);
+    EXPECT_EQ(handler5.got_removes, 0ul);
+    EXPECT_EQ(handler5.got_updates, 0ul);
+}
+
+GTEST_MAIN_RUN_ALL_TESTS()
+
diff --git a/slobrok/src/vespa/slobrok/server/CMakeLists.txt b/slobrok/src/vespa/slobrok/server/CMakeLists.txt
index 3221e268e8d..89007f626f3 100644
--- a/slobrok/src/vespa/slobrok/server/CMakeLists.txt
+++ b/slobrok/src/vespa/slobrok/server/CMakeLists.txt
@@ -5,23 +5,26 @@ vespa_add_library(slobrok_slobrokserver
     configshim.cpp
     exchange_manager.cpp
     history.cpp
-    i_rpc_server_manager.cpp
     i_monitored_server.cpp
+    i_rpc_server_manager.cpp
     managed_rpc_server.cpp
+    map_diff.cpp
+    metrics_producer.cpp
     monitor.cpp
     named_service.cpp
+    reconfigurable_stateserver.cpp
     remote_check.cpp
     remote_slobrok.cpp
     reserved_name.cpp
-    rpc_server_manager.cpp
-    rpc_server_map.cpp
     rpchooks.cpp
     rpcmirror.cpp
+    rpc_server_manager.cpp
+    rpc_server_map.cpp
     sbenv.cpp
+    service_map_history.cpp
+    service_mapping.cpp
     slobrokserver.cpp
     visible_map.cpp
-    metrics_producer.cpp
-    reconfigurable_stateserver.cpp
     INSTALL lib64
     DEPENDS
     slobrok
diff --git a/slobrok/src/vespa/slobrok/server/map_diff.cpp b/slobrok/src/vespa/slobrok/server/map_diff.cpp
new file mode 100644
index 00000000000..d276a2fbf05
--- /dev/null
+++ b/slobrok/src/vespa/slobrok/server/map_diff.cpp
@@ -0,0 +1,9 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "map_diff.h"
+
+namespace slobrok {
+
+MapDiff::~MapDiff() = default;
+
+} // namespace slobrok
diff --git a/slobrok/src/vespa/slobrok/server/map_diff.h b/slobrok/src/vespa/slobrok/server/map_diff.h
new file mode 100644
index 00000000000..80886886519
--- /dev/null
+++ b/slobrok/src/vespa/slobrok/server/map_diff.h
@@ -0,0 +1,51 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <vespa/vespalib/util/gencnt.h>
+#include "service_mapping.h"
+
+namespace slobrok {
+
+/**
+ * represents an incremental update to a name->spec map,
+ * or optionally a full dump of it.
+ **/
+struct MapDiff {
+    /** construct incremental diff */
+    MapDiff(const vespalib::GenCnt &from,
+            std::vector<vespalib::string> remove,
+            ServiceMappingList update,            
+            const vespalib::GenCnt &to)
+      : fromGen(from),
+        removed(std::move(remove)),
+        updated(std::move(update)),
+        toGen(to)
+    {}
+
+    /** construct full map dump */
+    MapDiff(ServiceMappingList mappings,
+            const vespalib::GenCnt &to)
+      : MapDiff(0, {}, std::move(mappings), to)
+    {}
+
+    ~MapDiff();
+    
+    // is this a diff from the empty map:
+    bool is_full_dump() const { return fromGen == vespalib::GenCnt(0); }
+
+    // which generation this diff goes from:
+    vespalib::GenCnt fromGen;
+
+    // names to remove (empty if is_full_dump):
+    std::vector<vespalib::string> removed;
+
+    // name->spec pairs to add or update:
+    ServiceMappingList updated;
+
+    // which generation this diff brings you to:
+    vespalib::GenCnt toGen;
+};
+
+} // namespace slobrok
+
diff --git a/slobrok/src/vespa/slobrok/server/service_map_history.cpp b/slobrok/src/vespa/slobrok/server/service_map_history.cpp
new file mode 100644
index 00000000000..f9de736c093
--- /dev/null
+++ b/slobrok/src/vespa/slobrok/server/service_map_history.cpp
@@ -0,0 +1,134 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "service_map_history.h"
+
+#include <vespa/log/log.h>
+LOG_SETUP(".slobrok.publisher");
+
+namespace slobrok {
+
+ServiceMapHistory::UpdateLog::UpdateLog()
+  : startGeneration(1),
+    currentGeneration(1),
+    updates(keep_items + 1)
+{}
+        
+ServiceMapHistory::UpdateLog::~UpdateLog() = default;
+
+void ServiceMapHistory::UpdateLog::add(const vespalib::string &name) {
+    currentGeneration.add();
+    updates.push(name);
+    while (updates.size() > keep_items) {
+        startGeneration.add();
+        updates.pop();
+    }
+}
+        
+bool ServiceMapHistory::UpdateLog::isInRange(const Generation &gen) const {
+    return gen.inRangeInclusive(startGeneration, currentGeneration);
+}
+
+std::vector<vespalib::string>
+ServiceMapHistory::UpdateLog::updatedSince(const Generation &gen) const {
+    std::vector<vespalib::string> result;
+    uint32_t skip = startGeneration.distance(gen);
+    uint32_t last = startGeneration.distance(currentGeneration);
+    for (uint32_t idx = skip; idx < last; ++idx) {
+        result.push_back(updates.peek(idx));
+    }
+    return result;
+}
+
+
+//-----------------------------------------------------------------------------
+
+ServiceMapHistory::ServiceMapHistory()
+  : _lock(),
+    _map(),
+    _waitList(),
+    _log()
+{}
+
+
+ServiceMapHistory::~ServiceMapHistory() {
+    notify_updated();
+}
+
+void ServiceMapHistory::notify_updated() {
+    WaitList waitList;
+    {
+        std::lock_guard guard(_lock);
+        std::swap(waitList, _waitList);
+    }
+    for (auto & [ handler, gen ] : waitList) {
+        handler->handle(makeDiffFrom(gen));
+    }
+}
+
+void ServiceMapHistory::asyncGenerationDiff(DiffCompletionHandler *handler, const Generation &fromGen) {
+    {
+        std::lock_guard guard(_lock);
+        if (fromGen == myGen()) {
+            _waitList.emplace_back(handler, fromGen);
+            return;
+        }
+    }
+    handler->handle(makeDiffFrom(fromGen));
+}
+
+bool ServiceMapHistory::cancel(DiffCompletionHandler *handler) {
+    std::lock_guard guard(_lock);
+    size_t removed = std::erase_if(_waitList, [=](const Waiter &elem){ return elem.first == handler; });
+    return (removed > 0);
+}
+
+void ServiceMapHistory::remove(const vespalib::string &name) {
+    {
+        std::lock_guard guard(_lock);
+        auto iter = _map.find(name);
+        if (iter == _map.end()) {
+            LOG(warning, "already removed: %s", name.c_str());
+            // already removed
+            return;
+        }
+        _map.erase(iter);
+        _log.add(name);
+    }
+    notify_updated();
+}
+
+void ServiceMapHistory::update(const ServiceMapping &mapping) {
+    {
+        std::lock_guard guard(_lock);
+        _map.insert_or_assign(mapping.name, mapping.spec);
+        _log.add(mapping.name);
+    }
+    notify_updated();
+}
+
+MapDiff ServiceMapHistory::makeDiffFrom(const Generation &fromGen) const {
+    std::lock_guard guard(_lock);
+    if (_log.isInRange(fromGen)) {
+        std::vector<vespalib::string> removes;
+        ServiceMappingList updates;
+        auto changes = _log.updatedSince(fromGen);
+        for (const vespalib::string & name : changes) {
+            if (_map.contains(name)) {
+                updates.emplace_back(name, _map.at(name));
+            } else {
+                removes.push_back(name);
+            }
+        }
+        return MapDiff(fromGen, removes, updates, myGen());
+    } else {
+        ServiceMappingList mappings;
+        for (const auto & [ name, spec ] : _map) {
+            mappings.emplace_back(name, spec);
+        }
+        return MapDiff(mappings, myGen());
+    }
+}
+
+//-----------------------------------------------------------------------------
+
+} // namespace slobrok
diff --git a/slobrok/src/vespa/slobrok/server/service_map_history.h b/slobrok/src/vespa/slobrok/server/service_map_history.h
new file mode 100644
index 00000000000..e1f20e0d8a4
--- /dev/null
+++ b/slobrok/src/vespa/slobrok/server/service_map_history.h
@@ -0,0 +1,97 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <vespa/vespalib/stllike/string.h>
+#include <vespa/vespalib/util/gencnt.h>
+#include <vespa/vespalib/util/arrayqueue.hpp>
+#include <map>
+#include <mutex>
+#include "service_mapping.h"
+#include "map_diff.h"
+
+namespace slobrok {
+
+/**
+ * @class ServiceMapHistory
+ * @brief API to generate incremental updates for a collection of name->spec mappings
+ **/
+
+class ServiceMapHistory
+{
+public:
+    using Generation = vespalib::GenCnt;
+
+    /** implement this interface to receive the result of an async generation diff */
+    class DiffCompletionHandler
+    {
+    public:
+        /**
+         * Handle the result of asyncGenerationDiff()
+         *
+         * @param updateDiff changes from the generation requested
+         **/
+        virtual void handle(MapDiff updateDiff) = 0;
+    protected:
+        virtual ~DiffCompletionHandler() {}
+    };
+
+private:
+    struct UpdateLog {
+        static constexpr uint32_t keep_items = 1000;
+        Generation startGeneration;
+        Generation currentGeneration;
+        vespalib::ArrayQueue<vespalib::string> updates;
+        UpdateLog();
+        ~UpdateLog();
+        void add(const vespalib::string &name);
+        bool isInRange(const Generation &gen) const;
+        std::vector<vespalib::string> updatedSince(const Generation &gen) const;
+    };
+    using Map = std::map<vespalib::string, vespalib::string>;
+    using Waiter = std::pair<DiffCompletionHandler *, Generation>;
+    using WaitList = std::vector<Waiter>;
+
+    mutable std::mutex _lock;
+    Map        _map;
+    WaitList   _waitList;
+    UpdateLog  _log;
+
+    void notify_updated();
+ 
+    const Generation &myGen() const { return _log.currentGeneration; }
+
+public:
+    ServiceMapHistory();
+    ~ServiceMapHistory();
+
+    /**
+     * Ask for notification when the history has changes newer than fromGen.
+     * Note that if there are any changes in the history already, the callback
+     * will happen immediately (inside asyncGenerationDiff).
+     **/
+    void asyncGenerationDiff(DiffCompletionHandler *handler, const Generation &fromGen);
+
+    /**
+     * Cancel pending notification.
+     * @return true if handler was canceled without calling handle() at all.
+     **/
+    bool cancel(DiffCompletionHandler *handler);
+
+    /** add or update name->spec mapping */
+    void update(const ServiceMapping &mapping);
+
+    /** remove mapping for name */
+    void remove(const vespalib::string &name);
+
+    /** For unit testing only: */
+    Generation currentGen() const { return myGen(); }
+
+private:
+    MapDiff makeDiffFrom(const Generation &fromGen) const;
+};
+
+//-----------------------------------------------------------------------------
+
+} // namespace slobrok
+
diff --git a/slobrok/src/vespa/slobrok/server/service_mapping.cpp b/slobrok/src/vespa/slobrok/server/service_mapping.cpp
new file mode 100644
index 00000000000..4f902a10e32
--- /dev/null
+++ b/slobrok/src/vespa/slobrok/server/service_mapping.cpp
@@ -0,0 +1,9 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "service_mapping.h"
+
+namespace slobrok {
+
+ServiceMapping::~ServiceMapping() = default;
+
+} // namespace slobrok
diff --git a/slobrok/src/vespa/slobrok/server/service_mapping.h b/slobrok/src/vespa/slobrok/server/service_mapping.h
new file mode 100644
index 00000000000..30c5307aa37
--- /dev/null
+++ b/slobrok/src/vespa/slobrok/server/service_mapping.h
@@ -0,0 +1,18 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <vespa/vespalib/stllike/string.h>
+#include <vector>
+
+namespace slobrok {
+
+struct ServiceMapping {
+    vespalib::string name;
+    vespalib::string spec;
+    ~ServiceMapping();
+};
+
+typedef std::vector<ServiceMapping> ServiceMappingList;
+
+}