Merge pull request #18804 from vespa-engine/arnej/add-external-mapping-monitor

Arnej/add external mapping monitor

8 files changed, 174 insertions, 2 deletions

diff --git a/slobrok/src/vespa/slobrok/server/CMakeLists.txt b/slobrok/src/vespa/slobrok/server/CMakeLists.txt
index 20f5bda4278..b4e7d5de1d5 100644
--- a/slobrok/src/vespa/slobrok/server/CMakeLists.txt
+++ b/slobrok/src/vespa/slobrok/server/CMakeLists.txt
@@ -11,6 +11,7 @@ vespa_add_library(slobrok_slobrokserver
     map_diff.cpp
     map_listener.cpp
     map_source.cpp
+    mapping_monitor.cpp
     metrics_producer.cpp
     mock_map_listener.cpp
     monitor.cpp
@@ -20,6 +21,7 @@ vespa_add_library(slobrok_slobrokserver
     remote_check.cpp
     remote_slobrok.cpp
     reserved_name.cpp
+    rpc_mapping_monitor.cpp
     rpc_server_manager.cpp
     rpc_server_map.cpp
     rpchooks.cpp
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 424ea82c6b5..8961e21f386 100644
--- a/slobrok/src/vespa/slobrok/server/local_rpc_monitor_map.h
+++ b/slobrok/src/vespa/slobrok/server/local_rpc_monitor_map.h
@@ -5,7 +5,6 @@
 #include "i_rpc_server_manager.h"
 #include "managed_rpc_server.h"
 #include "map_listener.h"
-#include "map_listener.h"
 #include "map_source.h"
 #include "named_service.h"
 #include "proxy_map_source.h"
diff --git a/slobrok/src/vespa/slobrok/server/mapping_monitor.cpp b/slobrok/src/vespa/slobrok/server/mapping_monitor.cpp
new file mode 100644
index 00000000000..359ba87bf22
--- /dev/null
+++ b/slobrok/src/vespa/slobrok/server/mapping_monitor.cpp
@@ -0,0 +1,8 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "mapping_monitor.h"
+
+namespace slobrok {
+
+} // namespace slobrok
+
diff --git a/slobrok/src/vespa/slobrok/server/mapping_monitor.h b/slobrok/src/vespa/slobrok/server/mapping_monitor.h
new file mode 100644
index 00000000000..035023f32fa
--- /dev/null
+++ b/slobrok/src/vespa/slobrok/server/mapping_monitor.h
@@ -0,0 +1,28 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include "service_mapping.h"
+#include <memory>
+#include <functional>
+
+namespace slobrok {
+
+struct MappingMonitorOwner {
+    virtual void up(const ServiceMapping& mapping) = 0;
+    virtual void down(const ServiceMapping& mapping) = 0;
+protected:
+    ~MappingMonitorOwner() = default;
+};
+
+struct MappingMonitor {
+    using UP = std::unique_ptr<MappingMonitor>;
+    virtual void start(const ServiceMapping& mapping) = 0;
+    virtual void stop(const ServiceMapping& mapping) = 0;
+    virtual ~MappingMonitor() = default;
+};
+
+using MappingMonitorFactory = std::function<MappingMonitor::UP(MappingMonitorOwner &)>;
+
+} // namespace slobrok
+
diff --git a/slobrok/src/vespa/slobrok/server/rpc_mapping_monitor.cpp b/slobrok/src/vespa/slobrok/server/rpc_mapping_monitor.cpp
new file mode 100644
index 00000000000..80c1a7d445a
--- /dev/null
+++ b/slobrok/src/vespa/slobrok/server/rpc_mapping_monitor.cpp
@@ -0,0 +1,68 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "rpc_mapping_monitor.h"
+
+#include <vespa/log/log.h>
+LOG_SETUP(".slobrok.server.rpc_mapping_monitor");
+
+namespace slobrok {
+
+void RpcMappingMonitor::DelayedTasks::PerformTask() {
+    std::vector<MUP> deleteAfterSwap;
+    std::swap(deleteAfterSwap, _deleteList);
+}
+
+RpcMappingMonitor::RpcMappingMonitor(FRT_Supervisor &orb, MappingMonitorOwner &owner)
+  : _orb(orb),
+    _delayedTasks(orb.GetScheduler()),
+    _map(),
+    _owner(owner)
+{}
+
+RpcMappingMonitor::~RpcMappingMonitor() = default;
+
+void RpcMappingMonitor::start(const ServiceMapping& mapping) {
+    LOG(spam, "start %s->%s", mapping.name.c_str(), mapping.spec.c_str());
+    LOG_ASSERT(_map.find(mapping) == _map.end());
+    _map.emplace(mapping,
+                 std::make_unique<ManagedRpcServer>(mapping.name, mapping.spec, *this));
+}
+
+void RpcMappingMonitor::stop(const ServiceMapping& mapping) {
+    LOG(spam, "stop %s->%s", mapping.name.c_str(), mapping.spec.c_str());
+    auto iter = _map.find(mapping);
+    LOG_ASSERT(iter != _map.end());
+    _delayedTasks.deleteLater(std::move(iter->second));
+    _map.erase(iter);
+}
+
+
+bool RpcMappingMonitor::active(const ServiceMapping &mapping, ManagedRpcServer *rpcsrv) const {
+    auto iter = _map.find(mapping);
+    if (iter == _map.end()) {
+        return false;
+    }
+    return iter->second.get() == rpcsrv;
+}
+        
+void RpcMappingMonitor::notifyFailedRpcSrv(ManagedRpcServer *rpcsrv, std::string errmsg) {
+    ServiceMapping mapping{rpcsrv->getName(), rpcsrv->getSpec()};
+    LOG(spam, "notifyFailed %s->%s", mapping.name.c_str(), mapping.spec.c_str());
+    if (active(mapping, rpcsrv)) {
+        LOG(debug, "service %s [at %s] failed: %s",
+            mapping.name.c_str(), mapping.spec.c_str(), errmsg.c_str());
+        _owner.down(mapping);
+    }
+}
+
+void RpcMappingMonitor::notifyOkRpcSrv(ManagedRpcServer *rpcsrv) {
+    ServiceMapping mapping{rpcsrv->getName(), rpcsrv->getSpec()};
+    LOG(spam, "notifyOk %s->%s", mapping.name.c_str(), mapping.spec.c_str());
+    if (active(mapping, rpcsrv)) {
+        LOG(debug, "service %s [at %s] up ok -> target",
+            mapping.name.c_str(), mapping.spec.c_str());
+        _owner.up(mapping);
+    }
+}
+
+} // namespace slobrok
diff --git a/slobrok/src/vespa/slobrok/server/rpc_mapping_monitor.h b/slobrok/src/vespa/slobrok/server/rpc_mapping_monitor.h
new file mode 100644
index 00000000000..2f05a459729
--- /dev/null
+++ b/slobrok/src/vespa/slobrok/server/rpc_mapping_monitor.h
@@ -0,0 +1,61 @@
+// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include "mapping_monitor.h"
+#include "i_rpc_server_manager.h"
+#include "managed_rpc_server.h"
+
+#include <vespa/fnet/frt/supervisor.h>
+#include <vespa/fnet/task.h>
+
+#include <vector>
+#include <memory>
+#include <map>
+
+namespace slobrok {
+
+class RpcMappingMonitor : public MappingMonitor,
+                          public IRpcServerManager
+{
+private:
+    using MUP = std::unique_ptr<ManagedRpcServer>;
+
+    using Map = std::map<ServiceMapping, MUP>;
+
+    class DelayedTasks : public FNET_Task {
+        std::vector<MUP>    _deleteList;
+    public:
+        void deleteLater(MUP rpcsrv) {
+            _deleteList.emplace_back(std::move(rpcsrv));
+            ScheduleNow();
+        }
+        void PerformTask() override;
+        DelayedTasks(FNET_Scheduler *scheduler)
+          : FNET_Task(scheduler),
+            _deleteList()
+        {}
+        ~DelayedTasks() { Kill(); }
+    };
+
+    FRT_Supervisor& _orb;
+    DelayedTasks _delayedTasks;
+    Map _map;
+    MappingMonitorOwner &_owner;
+
+    bool active(const ServiceMapping &mapping, ManagedRpcServer *rpcsrv) const;
+
+public:
+    RpcMappingMonitor(FRT_Supervisor &orb, MappingMonitorOwner &owner);
+    ~RpcMappingMonitor();
+
+    void start(const ServiceMapping& mapping) override;
+    void stop(const ServiceMapping& mapping) override;
+
+    void notifyFailedRpcSrv(ManagedRpcServer *rpcsrv, std::string errmsg) override;
+    void notifyOkRpcSrv(ManagedRpcServer *rpcsrv) override;
+    FRT_Supervisor *getSupervisor() override { return &_orb; }
+};
+
+} // namespace slobrok
+
diff --git a/slobrok/src/vespa/slobrok/server/service_map_mirror.cpp b/slobrok/src/vespa/slobrok/server/service_map_mirror.cpp
index 90deaa0f09c..746a5e7c76d 100644
--- a/slobrok/src/vespa/slobrok/server/service_map_mirror.cpp
+++ b/slobrok/src/vespa/slobrok/server/service_map_mirror.cpp
@@ -2,7 +2,7 @@
 
 #include "service_map_mirror.h"
 #include <vespa/log/log.h>
-LOG_SETUP(".slobrok.service_map_mirror");
+LOG_SETUP(".slobrok.server.service_map_mirror");
 
 namespace slobrok {
 
diff --git a/slobrok/src/vespa/slobrok/server/service_mapping.h b/slobrok/src/vespa/slobrok/server/service_mapping.h
index 6c540b8e9b5..6bbf910ee3e 100644
--- a/slobrok/src/vespa/slobrok/server/service_mapping.h
+++ b/slobrok/src/vespa/slobrok/server/service_mapping.h
@@ -16,6 +16,12 @@ struct ServiceMapping {
     bool operator== (const ServiceMapping &other) const {
         return name == other.name && spec == other.spec;
     }
+
+    bool operator< (const ServiceMapping &other) const {
+        if (name < other.name) return true;
+        if (other.name < name) return false;
+        return spec < other.spec;
+    }
 };
 
 typedef std::vector<ServiceMapping> ServiceMappingList;