added test for local rpc monitor map

7 files changed, 300 insertions, 10 deletions

diff --git a/slobrok/CMakeLists.txt b/slobrok/CMakeLists.txt
index 6acbf5d5134..c6c6313cf68 100644
--- a/slobrok/CMakeLists.txt
+++ b/slobrok/CMakeLists.txt
@@ -19,6 +19,7 @@ vespa_define_module(
     TESTS
     src/tests/backoff
     src/tests/configure
+    src/tests/local_rpc_monitor_map
     src/tests/mirrorapi
     src/tests/registerapi
     src/tests/service_map_history
diff --git a/slobrok/src/tests/local_rpc_monitor_map/CMakeLists.txt b/slobrok/src/tests/local_rpc_monitor_map/CMakeLists.txt
new file mode 100644
index 00000000000..aa30021939c
--- /dev/null
+++ b/slobrok/src/tests/local_rpc_monitor_map/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_local_rpc_monitor_map_test_app TEST
+    SOURCES
+    local_rpc_monitor_map_test.cpp
+    DEPENDS
+    slobrok_slobrokserver
+    GTest::GTest
+)
+vespa_add_test(NAME slobrok_local_rpc_monitor_map_test_app COMMAND slobrok_local_rpc_monitor_map_test_app)
diff --git a/slobrok/src/tests/local_rpc_monitor_map/local_rpc_monitor_map_test.cpp b/slobrok/src/tests/local_rpc_monitor_map/local_rpc_monitor_map_test.cpp
new file mode 100644
index 00000000000..40f8baa32a9
--- /dev/null
+++ b/slobrok/src/tests/local_rpc_monitor_map/local_rpc_monitor_map_test.cpp
@@ -0,0 +1,267 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/vespalib/gtest/gtest.h>
+#include <vespa/slobrok/server/local_rpc_monitor_map.h>
+#include <vespa/vespalib/util/stringfmt.h>
+#include <vespa/vespalib/util/time.h>
+#include <vespa/fnet/scheduler.h>
+#include <map>
+
+using namespace vespalib;
+using namespace slobrok;
+using vespalib::make_string_short::fmt;
+
+struct MapCall {
+    vespalib::string name;
+    ServiceMapping mapping;
+    ServiceMapping old;
+    static MapCall add(const ServiceMapping &m) { return {"add", m, {"",""}}; }
+    static MapCall remove(const ServiceMapping &m) { return {"remove", m, {"",""}}; }
+    static MapCall update(const ServiceMapping &o, const ServiceMapping &m) { return {"update", m, o}; }
+    void check(const MapCall &rhs) const {
+        EXPECT_EQ(name, rhs.name);
+        EXPECT_EQ(mapping, rhs.mapping);
+        EXPECT_EQ(old, rhs.old);
+    }
+    ~MapCall();
+};
+MapCall::~MapCall() = default;
+
+struct MonitorCall {
+    vespalib::string name;
+    ServiceMapping mapping;
+    bool hurry;
+    static MonitorCall start(const ServiceMapping &m, bool h) { return {"start", m, h}; }
+    static MonitorCall stop(const ServiceMapping &m) { return {"stop", m, false}; }
+    void check(const MonitorCall &rhs) const {
+        EXPECT_EQ(name, rhs.name);
+        EXPECT_EQ(mapping, rhs.mapping);
+        EXPECT_EQ(hurry, rhs.hurry);
+    }
+    ~MonitorCall();
+};
+MonitorCall::~MonitorCall() = default;
+
+template <typename Call>
+class CallLog {
+private:
+    std::vector<Call> _calls;
+    size_t _checked;
+public:
+    CallLog() noexcept : _calls(), _checked(0) {}
+    ~CallLog() { EXPECT_EQ(_calls.size(), _checked); }
+    void log(Call call) { _calls.push_back(call); }
+    void expect(std::initializer_list<Call> list) {
+        ASSERT_EQ(list.size(), (_calls.size() - _checked));
+        for (const auto &call: list) {
+            call.check(_calls[_checked++]);
+        }
+    }
+};
+
+struct MapLog : CallLog<MapCall>, MapListener {
+    void add(const ServiceMapping &mapping) override {
+        log(MapCall::add(mapping));
+    }
+    void remove(const ServiceMapping &mapping) override {
+        log(MapCall::remove(mapping));
+    }
+    void update(const ServiceMapping &old_mapping,
+                const ServiceMapping &new_mapping) override
+    {
+        log(MapCall::update(old_mapping, new_mapping));
+    }
+};
+
+struct MonitorLog : CallLog<MonitorCall>, MappingMonitor {
+    void start(const ServiceMapping& mapping, bool hurry) override {
+        log(MonitorCall::start(mapping, hurry));
+    }
+    void stop(const ServiceMapping& mapping) override {
+        log(MonitorCall::stop(mapping));
+    }
+};
+
+struct MyMappingMonitor : MappingMonitor {
+    MonitorLog &monitor;
+    MyMappingMonitor(MonitorLog &m) : monitor(m) {}
+    void start(const ServiceMapping& mapping, bool hurry) override {
+        monitor.start(mapping, hurry);
+    }
+    void stop(const ServiceMapping& mapping) override {
+        monitor.stop(mapping);
+    }
+};
+
+struct LocalRpcMonitorMapTest : public ::testing::Test {
+    steady_time time;
+    FNET_Scheduler scheduler;
+    MonitorLog monitor_log;
+    MapLog map_log;
+    LocalRpcMonitorMap map;
+    std::unique_ptr<MapSubscription> subscription;
+    ServiceMapping mapping;
+    ServiceMapping mapping_conflict;
+    LocalRpcMonitorMapTest()
+      : time(duration::zero()),
+        scheduler(&time, &time), monitor_log(), map_log(),
+        map(&scheduler, [this](auto &owner)
+            {
+                EXPECT_EQ(&owner, &map);
+                return std::make_unique<MyMappingMonitor>(monitor_log);
+            }),
+        subscription(MapSubscription::subscribe(map.dispatcher(), map_log)),
+        mapping("dummy_service", "dummy_spec"),
+        mapping_conflict("dummy_service", "conflicting_dummy_spec")
+    {}
+    void tick(duration elapsed = FNET_Scheduler::tick_ms) {
+        time += elapsed;
+        scheduler.CheckTasks();
+    }
+    void add_mapping(const ServiceMapping &m, bool is_up) {
+        map.add(m); // <- add from consensus map
+        monitor_log.expect({});
+        tick(0ms); // <- process delayed add event
+        monitor_log.expect({MonitorCall::start(m, false)});
+        map_log.expect({});
+        if (is_up) {
+            map.up(m); // <- up from monitor
+            map_log.expect({MapCall::add(m)});
+        } else {
+            map.down(m); // <- down from monitor
+            map_log.expect({});
+        }
+    }
+    void flip_up_state(const ServiceMapping &m, bool was_up, size_t cnt) {
+        for (size_t i = 0; i < cnt; ++i) {
+            if (was_up) {
+                map.up(m);
+                map_log.expect({});
+                map.down(m);
+                map_log.expect({MapCall::remove(m)});
+            } else {
+                map.down(m);
+                map_log.expect({});
+                map.up(m);
+                map_log.expect({MapCall::add(m)});
+            }
+            was_up = !was_up;
+        }
+        monitor_log.expect({});
+    }
+    void remove_mapping(const ServiceMapping &m, bool was_up) {
+        map.remove(m); // <- remove from consensus map
+        monitor_log.expect({});
+        tick(0ms); // <- process delayed remove event
+        monitor_log.expect({MonitorCall::stop(m)});
+        if (was_up) {
+            map_log.expect({MapCall::remove(m)});
+        } else {
+            map_log.expect({});
+        }
+    }
+    ~LocalRpcMonitorMapTest();
+};
+LocalRpcMonitorMapTest::~LocalRpcMonitorMapTest() = default;
+
+struct MyAddLocalHandler : LocalRpcMonitorMap::AddLocalCompletionHandler {
+    std::unique_ptr<OkState> &state;
+    bool &handler_deleted;
+    MyAddLocalHandler(std::unique_ptr<OkState> &s, bool &hd)
+      : state(s), handler_deleted(hd) {} 
+    void doneHandler(OkState result) override {
+        state = std::make_unique<OkState>(result);
+    }
+    ~MyAddLocalHandler() override {
+        handler_deleted = true;
+    }
+};
+
+TEST_F(LocalRpcMonitorMapTest, external_add_remove_while_up) {
+    add_mapping(mapping, true);
+    remove_mapping(mapping, true);
+}
+
+TEST_F(LocalRpcMonitorMapTest, external_add_remove_while_down) {
+    add_mapping(mapping, false);
+    remove_mapping(mapping, false);
+}
+
+TEST_F(LocalRpcMonitorMapTest, server_up_down_up_down) {
+    add_mapping(mapping, true);
+    flip_up_state(mapping, true, 3);
+    remove_mapping(mapping, false);
+}
+
+TEST_F(LocalRpcMonitorMapTest, server_down_up_down_up) {
+    add_mapping(mapping, false);
+    flip_up_state(mapping, false, 3);
+    remove_mapping(mapping, true);
+}
+
+TEST_F(LocalRpcMonitorMapTest, multi_mapping) {
+    ServiceMapping m1("dummy_service1", "dummy_spec1");
+    ServiceMapping m2("dummy_service2", "dummy_spec2");
+    ServiceMapping m3("dummy_service3", "dummy_spec3");
+    add_mapping(m1, true);
+    add_mapping(m2, false);
+    add_mapping(m3, true);
+    flip_up_state(m1, true, 3);
+    flip_up_state(m2, false, 3);
+    flip_up_state(m3, true, 3);
+    remove_mapping(m1, false);
+    remove_mapping(m2, true);
+    remove_mapping(m3, false);
+}
+
+TEST_F(LocalRpcMonitorMapTest, local_add_ok) {
+    std::unique_ptr<OkState> state;
+    bool handler_deleted;
+    map.addLocal(mapping, std::make_unique<MyAddLocalHandler>(state, handler_deleted));
+    monitor_log.expect({MonitorCall::start(mapping, true)});
+    map_log.expect({});
+    map.up(mapping);
+    monitor_log.expect({});
+    map_log.expect({MapCall::add(mapping)});
+    ASSERT_TRUE(state);
+    EXPECT_TRUE(state->ok());
+    ASSERT_TRUE(handler_deleted);
+}
+
+TEST_F(LocalRpcMonitorMapTest, local_add_already_up) {
+    std::unique_ptr<OkState> state;
+    bool handler_deleted;
+    add_mapping(mapping, true);
+    map.addLocal(mapping, std::make_unique<MyAddLocalHandler>(state, handler_deleted));
+    monitor_log.expect({});
+    map_log.expect({});
+    ASSERT_TRUE(state);
+    EXPECT_TRUE(state->ok());
+    ASSERT_TRUE(handler_deleted);
+}
+
+TEST_F(LocalRpcMonitorMapTest, local_add_already_down) {
+    std::unique_ptr<OkState> state;
+    bool handler_deleted;
+    add_mapping(mapping, false);
+    map.addLocal(mapping, std::make_unique<MyAddLocalHandler>(state, handler_deleted));
+    monitor_log.expect({});
+    map_log.expect({});
+    ASSERT_TRUE(state);
+    EXPECT_TRUE(state->ok());
+    ASSERT_TRUE(handler_deleted);
+}
+
+TEST_F(LocalRpcMonitorMapTest, local_add_conflict) {
+    std::unique_ptr<OkState> state;
+    bool handler_deleted;
+    add_mapping(mapping, true);
+    map.addLocal(mapping_conflict, std::make_unique<MyAddLocalHandler>(state, handler_deleted));
+    monitor_log.expect({});
+    map_log.expect({});
+    ASSERT_TRUE(state);
+    EXPECT_TRUE(state->failed());
+    ASSERT_TRUE(handler_deleted);
+}
+
+GTEST_MAIN_RUN_ALL_TESTS()
diff --git a/slobrok/src/vespa/slobrok/server/local_rpc_monitor_map.cpp b/slobrok/src/vespa/slobrok/server/local_rpc_monitor_map.cpp
index 454d123eead..73d355b7645 100644
--- a/slobrok/src/vespa/slobrok/server/local_rpc_monitor_map.cpp
+++ b/slobrok/src/vespa/slobrok/server/local_rpc_monitor_map.cpp
@@ -25,9 +25,9 @@ void LocalRpcMonitorMap::DelayedTasks::PerformTask() {
     }
 }
 
-LocalRpcMonitorMap::LocalRpcMonitorMap(FRT_Supervisor &supervisor,
+LocalRpcMonitorMap::LocalRpcMonitorMap(FNET_Scheduler *scheduler,
                                        MappingMonitorFactory mappingMonitorFactory)
-  : _delayedTasks(supervisor.GetScheduler(), *this),
+  : _delayedTasks(scheduler, *this),
     _map(),
     _dispatcher(),
     _history(),
@@ -83,7 +83,7 @@ ServiceMapHistory & LocalRpcMonitorMap::history() {
 }
 
 void LocalRpcMonitorMap::addLocal(const ServiceMapping &mapping,
-                                  std::unique_ptr<ScriptCommand> inflight)
+                                  std::unique_ptr<AddLocalCompletionHandler> inflight)
 {
     LOG(debug, "try local add: mapping %s->%s",
         mapping.name.c_str(), mapping.spec.c_str());
diff --git a/slobrok/src/vespa/slobrok/server/local_rpc_monitor_map.h b/slobrok/src/vespa/slobrok/server/local_rpc_monitor_map.h
index 3b2c74648d2..33a1109d915 100644
--- a/slobrok/src/vespa/slobrok/server/local_rpc_monitor_map.h
+++ b/slobrok/src/vespa/slobrok/server/local_rpc_monitor_map.h
@@ -27,6 +27,13 @@ namespace slobrok {
 class LocalRpcMonitorMap : public MapListener,
                            public MappingMonitorOwner
 {
+public:
+    // Interface used to signal the result of addLocal
+    struct AddLocalCompletionHandler {
+        virtual void doneHandler(OkState result) = 0;
+        virtual ~AddLocalCompletionHandler() {}
+    };
+
 private:
     enum class EventType { ADD, REMOVE };
 
@@ -66,12 +73,12 @@ private:
     struct PerService {
         bool up;
         bool localOnly;
-        std::unique_ptr<ScriptCommand> inflight;
+        std::unique_ptr<AddLocalCompletionHandler> inflight;
         vespalib::string spec;
     };
 
     PerService localService(const ServiceMapping &mapping,
-                            std::unique_ptr<ScriptCommand> inflight)
+                            std::unique_ptr<AddLocalCompletionHandler> inflight)
     {
         return PerService{
             .up = false,
@@ -109,13 +116,13 @@ private:
         ServiceMapping mapping;
         bool up;
         bool localOnly;
-        std::unique_ptr<ScriptCommand> inflight;
+        std::unique_ptr<AddLocalCompletionHandler> inflight;
     };
 
     RemovedData removeFromMap(Map::iterator iter);
 
 public:
-    LocalRpcMonitorMap(FRT_Supervisor &supervisor,
+    LocalRpcMonitorMap(FNET_Scheduler *scheduler,
                        MappingMonitorFactory mappingMonitorFactory);
     ~LocalRpcMonitorMap();
 
@@ -124,7 +131,7 @@ public:
 
     /** for use by register API, will call doneHandler() on inflight script */
     void addLocal(const ServiceMapping &mapping,
-                  std::unique_ptr<ScriptCommand> inflight);
+                  std::unique_ptr<AddLocalCompletionHandler> inflight);
 
     void add(const ServiceMapping &mapping) override;
     void remove(const ServiceMapping &mapping) override;
diff --git a/slobrok/src/vespa/slobrok/server/rpchooks.cpp b/slobrok/src/vespa/slobrok/server/rpchooks.cpp
index b2fb7f2ca89..d40fd93afd6 100644
--- a/slobrok/src/vespa/slobrok/server/rpchooks.cpp
+++ b/slobrok/src/vespa/slobrok/server/rpchooks.cpp
@@ -37,6 +37,12 @@ public:
     ~MetricsReport() override { Kill(); }
 };
 
+struct ScriptCommandWrapper : LocalRpcMonitorMap::AddLocalCompletionHandler {
+    ScriptCommand script;
+    ScriptCommandWrapper(ScriptCommand &&script_in) : script(std::move(script_in)) {}
+    void doneHandler(OkState result) override { script.doneHandler(result); } 
+};
+
 } // namespace <unnamed>
 
 //-----------------------------------------------------------------------------
@@ -242,7 +248,7 @@ RPCHooks::rpc_registerRpcServer(FRT_RPCRequest *req)
         // TODO: run only this path, and complete the request instead of ignoring
         auto script = ScriptCommand::makeIgnoreCmd(_env, dName, dSpec);
         ServiceMapping mapping{dName, dSpec};
-        _env.localMonitorMap().addLocal(mapping, std::make_unique<ScriptCommand>(std::move(script)));
+        _env.localMonitorMap().addLocal(mapping, std::make_unique<ScriptCommandWrapper>(std::move(script)));
     }
     // is this already OK?
     if (_rpcsrvmanager.alreadyManaged(dName, dSpec)) {
diff --git a/slobrok/src/vespa/slobrok/server/sbenv.cpp b/slobrok/src/vespa/slobrok/server/sbenv.cpp
index 19396934e67..9fb5511948c 100644
--- a/slobrok/src/vespa/slobrok/server/sbenv.cpp
+++ b/slobrok/src/vespa/slobrok/server/sbenv.cpp
@@ -113,7 +113,7 @@ SBEnv::SBEnv(const ConfigShim &shim, bool useNewConsensusLogic)
       _health(),
       _metrics(_rpcHooks, *_transport),
       _components(),
-      _localRpcMonitorMap(*_supervisor,
+      _localRpcMonitorMap(getScheduler(),
                           [this] (MappingMonitorOwner &owner) {
                               return std::make_unique<RpcMappingMonitor>(*_supervisor, owner);
                           }),