Merge pull request #8378 from vespa-engine/arnej/add-rpc-server-to-config-sentinel

add RPC server for start/stop of services

10 files changed, 331 insertions, 15 deletions

diff --git a/configd/src/apps/sentinel/CMakeLists.txt b/configd/src/apps/sentinel/CMakeLists.txt
index dceb4811d3d..6d41d674346 100644
--- a/configd/src/apps/sentinel/CMakeLists.txt
+++ b/configd/src/apps/sentinel/CMakeLists.txt
@@ -9,9 +9,13 @@ vespa_add_executable(configd_config-sentinel_app
     output-connection.cpp
     metrics.cpp
     state-api.cpp
+    cmdq.cpp
+    rpchooks.cpp
+    rpcserver.cpp
     OUTPUT_NAME vespa-config-sentinel
     INSTALL sbin
     DEPENDS
+    fnet
     configdefinitions
     staging_vespalib
 )
diff --git a/configd/src/apps/sentinel/cmdq.cpp b/configd/src/apps/sentinel/cmdq.cpp
new file mode 100644
index 00000000000..8fa3726c7f6
--- /dev/null
+++ b/configd/src/apps/sentinel/cmdq.cpp
@@ -0,0 +1,26 @@
+// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "cmdq.h"
+#include <vespa/fnet/frt/frt.h>
+
+namespace config::sentinel {
+
+Cmd::~Cmd()
+{
+    _req->Return();
+}
+
+void
+Cmd::retError(const char *errorString) const
+{
+    _req->SetError(FRTE_RPC_METHOD_FAILED, errorString);
+}
+
+void
+Cmd::retValue(const char *valueString) const
+{
+    FRT_Values *dst = _req->GetReturn();
+    dst->AddString(valueString);
+}
+
+}
diff --git a/configd/src/apps/sentinel/cmdq.h b/configd/src/apps/sentinel/cmdq.h
new file mode 100644
index 00000000000..df7dc9f241d
--- /dev/null
+++ b/configd/src/apps/sentinel/cmdq.h
@@ -0,0 +1,58 @@
+// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <memory>
+#include <mutex>
+#include <vector>
+
+class FRT_RPCRequest;
+
+namespace config::sentinel {
+
+class Cmd {
+public:
+    using UP = std::unique_ptr<Cmd>;
+    enum CmdType { LIST, START, STOP };
+
+    Cmd(FRT_RPCRequest *req, CmdType cmdType, const char *service = "")
+      : _req(req), _cmdType(cmdType), _serviceName(service)
+    {}
+
+    CmdType type() const { return _cmdType; }
+    const char *serviceName() const { return _serviceName; }
+
+    void retError(const char *errorString) const;
+    void retValue(const char *valueString) const;
+
+    ~Cmd();
+private:
+    FRT_RPCRequest *_req;
+    CmdType _cmdType;
+    const char *_serviceName;
+};
+
+class CommandQueue
+{
+private:
+    std::mutex _lock;
+    std::vector<Cmd::UP> _queue;
+public:
+    CommandQueue() = default;
+    ~CommandQueue() = default;
+
+    void enqueue(Cmd::UP cmd) {
+        std::lock_guard guard(_lock);
+        _queue.push_back(std::move(cmd));
+    }
+
+    std::vector<Cmd::UP> drain() {
+        std::vector<Cmd::UP> r;
+        std::lock_guard guard(_lock);
+        r.swap(_queue);
+        return r;
+    }
+
+};
+
+} // namespace config::sentinel
diff --git a/configd/src/apps/sentinel/config-handler.cpp b/configd/src/apps/sentinel/config-handler.cpp
index 325de3c232c..a62b74c6b4a 100644
--- a/configd/src/apps/sentinel/config-handler.cpp
+++ b/configd/src/apps/sentinel/config-handler.cpp
@@ -157,6 +157,10 @@ ConfigHandler::doConfigure()
         _stateApi.bound(_boundPort);
     }
 
+    if (!_rpcServer || config.port.rpc != _rpcServer->getPort()) {
+        _rpcServer = std::make_unique<RpcServer>(config.port.rpc, _cmdQ);
+    }
+
     LOG(debug, "ConfigHandler::configure() %d config elements, tenant(%s), application(%s), instance(%s)",
         (int)config.service.size(), config.application.tenant.c_str(), config.application.name.c_str(),
         config.application.instance.c_str());
@@ -288,6 +292,15 @@ ConfigHandler::handleOutputs()
 void
 ConfigHandler::handleCommands()
 {
+    {
+        // handle RPC commands
+        std::vector<Cmd::UP> got = _cmdQ.drain();
+        for (const Cmd::UP & cmd : got) {
+            handleCmd(*cmd);
+        }
+        // implicit return via Cmd destructor
+    }
+
     // Accept new command connections, and read commands.
     int fd;
     struct sockaddr_storage sad;
@@ -362,6 +375,59 @@ splitCommand(char *line, char *&cmd, char *&args)
     args = line;
 }
 
+void
+ConfigHandler::handleCmd(const Cmd& cmd)
+{
+    switch (cmd.type()) {
+    case Cmd::LIST:
+        {
+            char retbuf[65536];
+            size_t left = 65536;
+            size_t pos = 0;
+            for (ServiceMap::iterator it(_services.begin()), mt(_services.end()); it != mt; it++) {
+                Service *service = it->second.get();
+                const SentinelConfig::Service& config = service->serviceConfig();
+                int sz = snprintf(retbuf + pos, left,
+                                  "%s state=%s mode=%s pid=%d exitstatus=%d id=\"%s\"\n",
+                                  service->name().c_str(), service->stateName(),
+                                  service->isAutomatic() ? "AUTO" : "MANUAL",
+                                  service->pid(), service->exitStatus(),
+                                  config.id.c_str());
+                pos += sz;
+                left -= sz;
+                if (left <= 0) break;
+            }
+            cmd.retValue(retbuf);
+        }
+        break;
+    case Cmd::START:
+        {
+            Service *service = serviceByName(cmd.serviceName());
+            if (service == nullptr) {
+                cmd.retError("Cannot find named service");
+                return;
+            }
+            service->setAutomatic(true);
+            if (! service->isRunning()) {
+                service->start();
+            }
+        }
+        break;
+    case Cmd::STOP:
+        {
+            Service *service = serviceByName(cmd.serviceName());
+            if (service == nullptr) {
+                cmd.retError("Cannot find named service");
+                return;
+            }
+            service->setAutomatic(false);
+            if (service->isRunning()) {
+                service->terminate(true, false);
+            }
+        }
+        break;
+    }
+}
 
 void
 ConfigHandler::handleCommand(CommandConnection *c)
diff --git a/configd/src/apps/sentinel/config-handler.h b/configd/src/apps/sentinel/config-handler.h
index eb22b7fada1..e33bbf1c8da 100644
--- a/configd/src/apps/sentinel/config-handler.h
+++ b/configd/src/apps/sentinel/config-handler.h
@@ -4,6 +4,8 @@
 #include "service.h"
 #include "metrics.h"
 #include "state-api.h"
+#include "cmdq.h"
+#include "rpcserver.h"
 #include <vespa/config-sentinel.h>
 #include <vespa/config/config.h>
 #include <sys/types.h>
@@ -29,6 +31,8 @@ private:
     ServiceMap _services;
     std::list<CommandConnection *> _connections;
     std::list<OutputConnection *> _outputConnections;
+    CommandQueue _cmdQ;
+    std::unique_ptr<RpcServer> _rpcServer;
     int _boundPort;
     int _commandSocket;
     StartMetrics _startMetrics;
@@ -41,6 +45,7 @@ private:
     Service *serviceByName(const vespalib::string & name);
     void handleCommands();
     void handleCommand(CommandConnection *c);
+    void handleCmd(const Cmd& cmd);
     void handleOutputs();
     void handleChildDeaths();
 
diff --git a/configd/src/apps/sentinel/rpchooks.cpp b/configd/src/apps/sentinel/rpchooks.cpp
new file mode 100644
index 00000000000..6e271db3977
--- /dev/null
+++ b/configd/src/apps/sentinel/rpchooks.cpp
@@ -0,0 +1,64 @@
+// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "rpchooks.h"
+#include "cmdq.h"
+#include <vespa/fnet/frt/frt.h>
+
+#include <vespa/log/log.h>
+LOG_SETUP(".rpchooks");
+
+namespace config::sentinel {
+
+RPCHooks::~RPCHooks() = default;
+
+
+void
+RPCHooks::initRPC(FRT_Supervisor *supervisor)
+{
+    FRT_ReflectionBuilder rb(supervisor);
+
+    //-------------------------------------------------------------------------
+    rb.DefineMethod("sentinel.ls", "", "s",
+                    FRT_METHOD(RPCHooks::rpc_listServices), this);
+    rb.MethodDesc("list services");
+    rb.ReturnDesc("status", "Status for services");
+    //-------------------------------------------------------------------------
+    rb.DefineMethod("sentinel.service.stop", "s", "",
+                    FRT_METHOD(RPCHooks::rpc_stopService), this);
+    rb.MethodDesc("stop a service");
+    //-------------------------------------------------------------------------
+    rb.DefineMethod("sentinel.service.start", "s", "",
+                    FRT_METHOD(RPCHooks::rpc_startService), this);
+    rb.MethodDesc("stop a service");
+    //-------------------------------------------------------------------------
+}
+
+void
+RPCHooks::rpc_listServices(FRT_RPCRequest *req)
+{
+    LOG(debug, "got listservices");
+    req->Detach();
+    _commands.enqueue(std::make_unique<Cmd>(req, Cmd::LIST));
+}
+
+void
+RPCHooks::rpc_stopService(FRT_RPCRequest *req)
+{
+    FRT_Values &args  = *req->GetParams();
+    const char *srvNM = args[0]._string._str;
+    LOG(debug, "got stopservice '%s'", srvNM);
+    req->Detach();
+    _commands.enqueue(std::make_unique<Cmd>(req, Cmd::STOP, srvNM));
+}
+
+void
+RPCHooks::rpc_startService(FRT_RPCRequest *req)
+{
+    FRT_Values &args  = *req->GetParams();
+    const char *srvNM = args[0]._string._str;
+    LOG(debug, "got startservice '%s'", srvNM);
+    req->Detach();
+    _commands.enqueue(std::make_unique<Cmd>(req, Cmd::START, srvNM));
+}
+
+} // namespace slobrok
diff --git a/configd/src/apps/sentinel/rpchooks.h b/configd/src/apps/sentinel/rpchooks.h
new file mode 100644
index 00000000000..9375dc895be
--- /dev/null
+++ b/configd/src/apps/sentinel/rpchooks.h
@@ -0,0 +1,39 @@
+// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <vespa/fnet/frt/invokable.h>
+#include <memory>
+
+class FNET_Task;
+class FRT_Supervisor;
+
+namespace config::sentinel {
+
+class CommandQueue;
+
+/**
+ * @class RPCHooks
+ * @brief The FNET-RPC interface to a config sentinel
+ *
+ * Contains methods for receiveing and unpacking requests,
+ * invoking the right internal method, and (in most cases)
+ * packaging and returning the result of the request.
+ **/
+class RPCHooks : public FRT_Invokable
+{
+private:
+    CommandQueue &_commands;
+
+public:
+    RPCHooks(CommandQueue &commands) : _commands(commands) {}
+    ~RPCHooks() override;
+
+    void initRPC(FRT_Supervisor *supervisor);
+private:
+    void rpc_listServices(FRT_RPCRequest *req);
+    void rpc_stopService(FRT_RPCRequest *req);
+    void rpc_startService(FRT_RPCRequest *req);
+};
+
+} // namespace config::sentinel
diff --git a/configd/src/apps/sentinel/rpcserver.cpp b/configd/src/apps/sentinel/rpcserver.cpp
new file mode 100644
index 00000000000..b719d012304
--- /dev/null
+++ b/configd/src/apps/sentinel/rpcserver.cpp
@@ -0,0 +1,29 @@
+// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "rpcserver.h"
+
+#include <vespa/log/log.h>
+LOG_SETUP(".rpcserver");
+
+namespace config::sentinel {
+
+RpcServer::RpcServer(int portNumber, CommandQueue &cmdQ)
+    : _supervisor(),
+      _rpcHooks(cmdQ),
+      _port(portNumber)
+{
+    _rpcHooks.initRPC(&_supervisor);
+    if (_supervisor.Listen(portNumber)) {
+        LOG(config, "listening on port %d", portNumber);
+        _supervisor.Start();
+    } else {
+        LOG(error, "unable to listen to port %d", portNumber);
+    }
+}
+
+RpcServer::~RpcServer()
+{
+    _supervisor.ShutDown(true);
+}
+
+} // namespace config::sentinel
diff --git a/configd/src/apps/sentinel/rpcserver.h b/configd/src/apps/sentinel/rpcserver.h
new file mode 100644
index 00000000000..f295975f224
--- /dev/null
+++ b/configd/src/apps/sentinel/rpcserver.h
@@ -0,0 +1,27 @@
+// Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <memory>
+
+#include "cmdq.h"
+#include "rpchooks.h"
+#include <vespa/fnet/frt/supervisor.h>
+
+namespace config::sentinel {
+
+class RpcServer
+{
+private:
+    FRT_Supervisor _supervisor;
+    RPCHooks _rpcHooks;
+    int _port;
+
+public:
+    RpcServer(int port, CommandQueue &cmdQ);
+    ~RpcServer();
+
+    int getPort() const { return _port; }
+};
+
+} // namespace config::sentinel
diff --git a/configd/src/apps/sentinel/sentinel.cpp b/configd/src/apps/sentinel/sentinel.cpp
index bb05f9e40ad..023b96ce96c 100644
--- a/configd/src/apps/sentinel/sentinel.cpp
+++ b/configd/src/apps/sentinel/sentinel.cpp
@@ -65,7 +65,7 @@ main(int argc, char **argv)
     try {
         handler.subscribe(configId, CONFIG_TIMEOUT_MS);
     } catch (ConfigTimeoutException & ex) {
-        LOG(warning, "Timout getting config, please check your setup. Will exit and restart: %s", ex.getMessage().c_str());
+        LOG(warning, "Timeout getting config, please check your setup. Will exit and restart: %s", ex.getMessage().c_str());
         EV_STOPPING("config-sentinel", ex.what());
         exit(EXIT_FAILURE);
     } catch (InvalidConfigException& ex) {
@@ -88,25 +88,23 @@ main(int argc, char **argv)
             LOG(warning, "Configuration problem: (ignoring): %s",
                 ex.what());
         }
-        if (!vespalib::SignalHandler::CHLD.check()) {
-            int maxNum = 0;
-            fd_set fds;
-            FD_ZERO(&fds);
-            handler.updateActiveFdset(&fds, &maxNum);
-
-            struct timeval tv;
-            tv.tv_sec = 1;
-            tv.tv_usec = 0;
-
-            if (!vespalib::SignalHandler::CHLD.check()) {
-                select(maxNum, &fds, nullptr, nullptr, &tv);
-            }
+        if (vespalib::SignalHandler::CHLD.check()) {
+            continue;
         }
+        int maxNum = 0;
+        fd_set fds;
+        FD_ZERO(&fds);
+        handler.updateActiveFdset(&fds, &maxNum);
 
         struct timeval tv;
+        tv.tv_sec = 1;
+        tv.tv_usec = 0;
+
+        select(maxNum, &fds, nullptr, nullptr, &tv);
+
         gettimeofday(&tv, nullptr);
         double delta = tv.tv_sec - lastTv.tv_sec
-                       + 1e-6 * tv.tv_usec - lastTv.tv_usec;
+                       + 1e-6 * (tv.tv_usec - lastTv.tv_usec);
         if (delta < 0.01) {
             usleep(12500); // Avoid busy looping;
         }