add vespa-otelcol-start

12 files changed, 404 insertions, 0 deletions

diff --git a/logforwarder/CMakeLists.txt b/logforwarder/CMakeLists.txt
index 85406cba5de..b8f7fb3c416 100644
--- a/logforwarder/CMakeLists.txt
+++ b/logforwarder/CMakeLists.txt
@@ -7,4 +7,5 @@ vespa_define_module(
 
     APPS
     src/apps/vespa-logforwarder-start
+    src/apps/vespa-otelcol-start
 )
diff --git a/logforwarder/src/apps/vespa-otelcol-start/.gitignore b/logforwarder/src/apps/vespa-otelcol-start/.gitignore
new file mode 100644
index 00000000000..f2e64aa95d6
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/.gitignore
@@ -0,0 +1 @@
+vespa-otelcol-start
diff --git a/logforwarder/src/apps/vespa-otelcol-start/CMakeLists.txt b/logforwarder/src/apps/vespa-otelcol-start/CMakeLists.txt
new file mode 100644
index 00000000000..d95ce0584c9
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/CMakeLists.txt
@@ -0,0 +1,15 @@
+# Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(otelcol_start_app
+    SOURCES
+    main.cpp
+    cf-handler.cpp
+    child-handler.cpp
+    file-watcher.cpp
+    wrapper.cpp
+    OUTPUT_NAME vespa-otelcol-start
+    INSTALL bin
+    DEPENDS
+    config_cloudconfig
+    configdefinitions
+    vespalib
+)
diff --git a/logforwarder/src/apps/vespa-otelcol-start/cf-handler.cpp b/logforwarder/src/apps/vespa-otelcol-start/cf-handler.cpp
new file mode 100644
index 00000000000..9579271f88a
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/cf-handler.cpp
@@ -0,0 +1,67 @@
+// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "cf-handler.h"
+#include <vespa/config/common/configcontext.h>
+#include <vespa/config/common/configsystem.h>
+#include <vespa/config/common/exceptions.h>
+#include <vespa/config/helper/legacy.h>
+#include <vespa/config/subscription/configsubscriber.hpp>
+
+#include <unistd.h>
+
+#include <vespa/log/log.h>
+LOG_SETUP(".cf-handler");
+
+CfHandler::CfHandler(const std::string & configId)
+  : _subscriber(std::make_shared<config::ConfigContext>(*config::legacyConfigId2Spec(configId)))
+{}
+
+CfHandler::~CfHandler() = default;
+
+void CfHandler::subscribe(const std::string & configId, std::chrono::milliseconds timeout) {
+    LOG(info, "subscribe with config id: %s", configId.c_str());
+    std::string cfgId(config::legacyConfigId2ConfigId(configId));
+    _handle = _subscriber.subscribe<OpenTelemetryConfig>(cfgId, timeout);
+}
+
+void CfHandler::doConfigure() {
+    auto curConfig = _handle->getConfig();
+    if (_lastConfig && *curConfig == *_lastConfig) {
+        LOG(info, "same config as last");
+        return;
+    }
+    LOG(info, "new config, trigger restart");
+    _lastConfig = std::move(curConfig);
+    const OpenTelemetryConfig& config(*_lastConfig);
+    LOG(info, "watch %zu files", config.refPaths.size());
+    _fileWatcher.init(config.refPaths);
+    gotConfig(config);
+}
+
+void CfHandler::checkConfig() {
+    if (_subscriber.nextConfigNow()) {
+        doConfigure();
+    } else if (_fileWatcher.anyChanged()) {
+        LOG(info, "watched file updated, trigger restart");
+        const OpenTelemetryConfig& config(*_lastConfig);
+        gotConfig(config);
+    }
+}
+
+constexpr std::chrono::milliseconds CONFIG_TIMEOUT_MS(30 * 1000);
+
+void CfHandler::start(const std::string &configId) {
+    LOG(debug, "Reading configuration with id '%s'", configId.c_str());
+    try {
+        subscribe(configId, CONFIG_TIMEOUT_MS);
+    } catch (config::ConfigTimeoutException & ex) {
+        LOG(warning, "Timout getting config, please check your setup. Will exit and restart: %s", ex.getMessage().c_str());
+        std::_Exit(EXIT_FAILURE);
+    } catch (config::InvalidConfigException& ex) {
+        LOG(error, "Fatal: Invalid configuration, please check your setup: %s", ex.getMessage().c_str());
+        std::_Exit(EXIT_FAILURE);
+    } catch (config::ConfigRuntimeException& ex) {
+        LOG(error, "Fatal: Could not get config, please check your setup: %s", ex.getMessage().c_str());
+        std::_Exit(EXIT_FAILURE);
+    }
+}
diff --git a/logforwarder/src/apps/vespa-otelcol-start/cf-handler.h b/logforwarder/src/apps/vespa-otelcol-start/cf-handler.h
new file mode 100644
index 00000000000..7e8699caf85
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/cf-handler.h
@@ -0,0 +1,25 @@
+// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include "file-watcher.h"
+#include <vespa/config-open-telemetry.h>
+#include <vespa/config/subscription/configsubscriber.h>
+
+using cloud::config::OpenTelemetryConfig;
+
+class CfHandler {
+private:
+    FileWatcher _fileWatcher;
+    config::ConfigSubscriber _subscriber;
+    config::ConfigHandle<OpenTelemetryConfig>::UP _handle = {};
+    std::unique_ptr<OpenTelemetryConfig> _lastConfig = {};
+    time_t _lastCertFileChange = 0;
+    void subscribe(const std::string & configId, std::chrono::milliseconds timeout);
+    void doConfigure();
+public:
+    CfHandler(const std::string &configId);
+    virtual ~CfHandler();
+    void start(const std::string &configId);
+    void checkConfig();
+    virtual void gotConfig(const OpenTelemetryConfig&) = 0;
+};
diff --git a/logforwarder/src/apps/vespa-otelcol-start/child-handler.cpp b/logforwarder/src/apps/vespa-otelcol-start/child-handler.cpp
new file mode 100644
index 00000000000..46ac3bea60e
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/child-handler.cpp
@@ -0,0 +1,88 @@
+// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "child-handler.h"
+
+#include <stdio.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <signal.h>
+#include <sys/wait.h>
+#include <vector>
+#include <string>
+#include <cstdlib>
+
+#include <vespa/log/log.h>
+LOG_SETUP(".child-handler");
+
+ChildHandler::ChildHandler() : _childRunning(false), _childPid(0) {}
+
+ChildHandler::~ChildHandler() = default;
+
+bool ChildHandler::checkChild() {
+    if (! _childRunning) return true;
+    int waitStatus = 0;
+    int r = waitpid(_childPid, &waitStatus, WNOHANG);
+    if (r == 0) {
+        return false;
+    }
+    if (r < 0) {
+        perror("waitpid");
+        // XXX how to handle?
+        return false;
+    }
+    _childRunning = false;
+    if (WIFEXITED(waitStatus) && WEXITSTATUS(waitStatus) == 0) {
+        // all OK
+        LOG(info, "child ran ok, exit status 0");
+    } else if (WIFEXITED(waitStatus)) {
+        LOG(warning, "child failed (exit status %d)", WEXITSTATUS(waitStatus));
+    } else if (WIFSIGNALED(waitStatus)) {
+        if (_terminating) {
+            LOG(info, "child terminated (using signal %d)", WTERMSIG(waitStatus));
+        } else {
+            LOG(warning, "child failed (exit on signal %d)", WTERMSIG(waitStatus));
+        }
+    } else {
+        LOG(warning, "child failed (abnormal exit status %d)", waitStatus);
+    }
+    return true;
+}
+
+void ChildHandler::startChild(const std::string &progPath, const std::string &cfPath) {
+    _terminating = false;
+    LOG(info, "startChild '%s' '%s'", progPath.c_str(), cfPath.c_str());
+    pid_t child = fork();
+    if (child == -1) {
+        perror("fork()");
+        return;
+    }
+    if (child == 0) {
+        std::string cfArg{"--config=file:" + cfPath};
+        const char *cargv[] = { progPath.c_str(), cfArg.c_str(), nullptr };
+        execv(progPath.c_str(), const_cast<char **>(cargv));
+        // if execv fails:
+        perror(progPath.c_str());
+        std::_Exit(1);
+    }
+    LOG(info, "child running with pid %d", (int)child);
+    _childRunning = true;
+    _childPid = child;
+}
+
+void ChildHandler::stopChild() {
+    if (! _childRunning) return;
+    LOG(info, "stopChild");
+    _terminating = true;
+    kill(_childPid, SIGTERM);
+    for (int retry = 0; retry < 10; ++retry) {
+        if (checkChild()) return;
+        usleep(12500 + retry * 20000);
+    }
+    kill(_childPid, SIGKILL);
+    for (int retry = 0; retry < 10; ++retry) {
+        if (checkChild()) return;
+        usleep(12500 + retry * 20000);
+    }
+    LOG(error, "Could not terminete child process %d", _childPid);
+}
diff --git a/logforwarder/src/apps/vespa-otelcol-start/child-handler.h b/logforwarder/src/apps/vespa-otelcol-start/child-handler.h
new file mode 100644
index 00000000000..4fd72fc7682
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/child-handler.h
@@ -0,0 +1,18 @@
+// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include "cf-handler.h"
+#include <string>
+
+class ChildHandler {
+private:
+    bool _childRunning = false;
+    bool _terminating = false;
+    int _childPid = 0;
+public:
+    ChildHandler();
+    ~ChildHandler();
+    void startChild(const std::string &progPath, const std::string &cfFile);
+    void stopChild();
+    bool checkChild();
+};
diff --git a/logforwarder/src/apps/vespa-otelcol-start/file-watcher.cpp b/logforwarder/src/apps/vespa-otelcol-start/file-watcher.cpp
new file mode 100644
index 00000000000..f56a1d9d169
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/file-watcher.cpp
@@ -0,0 +1,37 @@
+// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "file-watcher.h"
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+namespace {
+
+time_t lastModTime(const vespalib::string &fn) {
+    if (fn.empty()) return 0;
+    struct stat info;
+    if (stat(fn.c_str(), &info) != 0) return 0;
+    return info.st_mtime;
+}
+
+} // namespace
+
+bool FileWatcher::anyChanged() {
+    bool result = false;
+    for (auto &entry : watchedFiles) {
+        time_t updated = lastModTime(entry.pathName);
+        if (updated != entry.seenModTime) {
+            result = true;
+            entry.seenModTime = updated;
+        }
+    }
+    return result;
+}
+
+void FileWatcher::init(const config::StringVector &pathList) {
+    watchedFiles.clear();
+    for (const auto& path : pathList) {
+        watchedFiles.emplace_back(path, lastModTime(path));
+    }
+}
diff --git a/logforwarder/src/apps/vespa-otelcol-start/file-watcher.h b/logforwarder/src/apps/vespa-otelcol-start/file-watcher.h
new file mode 100644
index 00000000000..0f50f6d90f7
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/file-watcher.h
@@ -0,0 +1,15 @@
+// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include <vespa/config-open-telemetry.h>
+
+class FileWatcher {
+    struct FileInfo {
+        vespalib::string pathName;
+        time_t seenModTime;
+    };
+    std::vector<FileInfo> watchedFiles;
+public:
+    bool anyChanged();
+    void init(const config::StringVector &pathList);
+};
diff --git a/logforwarder/src/apps/vespa-otelcol-start/main.cpp b/logforwarder/src/apps/vespa-otelcol-start/main.cpp
new file mode 100644
index 00000000000..2e3e0659707
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/main.cpp
@@ -0,0 +1,43 @@
+// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "wrapper.h"
+#include <csignal>
+#include <unistd.h>
+#include <vespa/vespalib/util/sig_catch.h>
+
+#include <vespa/defaults.h>
+#include <vespa/log/log.h>
+LOG_SETUP("vespa-otelcol-start");
+
+static void run(const char *configId) {
+    vespalib::SigCatch catcher;
+    Wrapper handler(configId);
+    handler.start(configId);
+    while (! catcher.receivedStopSignal()) {
+        handler.check();
+        usleep(125000); // Avoid busy looping;
+    }
+    handler.stop();
+};
+
+int main(int argc, char** argv) {
+    vespa::Defaults::bootstrap(argv[0]);
+    int c = -1;
+    const char *cfid = nullptr;
+    while ((c = getopt(argc, argv, "c:")) != -1) {
+        switch (c) {
+        case 'c':
+            cfid = optarg;
+            break;
+        default:
+            cfid = nullptr;
+            break;
+        }
+    }
+    if (cfid == nullptr) {
+        LOG(error, "Usage: %s -c <config-id>", argv[0]);
+        return EXIT_FAILURE;
+    }
+    run(cfid);
+    return 0;
+}
diff --git a/logforwarder/src/apps/vespa-otelcol-start/wrapper.cpp b/logforwarder/src/apps/vespa-otelcol-start/wrapper.cpp
new file mode 100644
index 00000000000..7228a3ea921
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/wrapper.cpp
@@ -0,0 +1,76 @@
+// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "wrapper.h"
+#include "child-handler.h"
+
+#include <dirent.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+
+#include <vespa/defaults.h>
+#include <vespa/log/log.h>
+LOG_SETUP(".wrapper");
+
+namespace {
+
+vespalib::string fixDir(const vespalib::string &parent, const vespalib::string &subdir) {
+    auto dirname = parent + "/" + subdir;
+    DIR *dp = opendir(dirname.c_str());
+    if (dp == NULL) {
+        if (errno != ENOENT || mkdir(dirname.c_str(), 0755) != 0) {
+            LOG(warning, "Could not create directory '%s'", dirname.c_str());
+            perror(dirname.c_str());
+        }
+    } else {
+        closedir(dp);
+    }
+    return dirname;
+}
+
+vespalib::string cfFilePath() {
+    vespalib::string path = vespa::Defaults::underVespaHome("var/db/vespa");
+    path = fixDir(path, "otelcol");
+    return path + "/" + "config.yaml";
+}
+
+void  writeConfig(const vespalib::string &config, const vespalib::string &path) {
+    LOG(info, "got config, writing %s", path.c_str());
+    vespalib::string tmpPath = path + ".new";
+    FILE *fp = fopen(tmpPath.c_str(), "w");
+    if (fp == NULL) {
+        LOG(warning, "could not open '%s' for write", tmpPath.c_str());
+        return;
+    }
+    fprintf(fp, "%s\n", config.c_str());
+    fclose(fp);
+    rename(tmpPath.c_str(), path.c_str());
+}
+
+} // namespace <unnamed>
+
+Wrapper::Wrapper(const std::string &configId)
+  : CfHandler(configId),
+    _childHandler()
+{}
+
+Wrapper::~Wrapper() = default;
+
+void Wrapper::stop() {
+    _childHandler.stopChild();
+}
+
+void Wrapper::check() {
+    checkConfig();
+    if (_childHandler.checkChild()) {
+        LOG(error, "Fatal: child process died unexpectedly");
+        std::_Exit(EXIT_FAILURE);
+    }
+}
+
+void Wrapper::gotConfig(const OpenTelemetryConfig& config) {
+    _childHandler.stopChild();
+    std::string progPath = vespa::Defaults::underVespaHome("sbin/otelcol-contrib");
+    std::string cfPath = cfFilePath();
+    writeConfig(config.yaml, cfPath);
+    _childHandler.startChild(progPath, cfPath);
+}
diff --git a/logforwarder/src/apps/vespa-otelcol-start/wrapper.h b/logforwarder/src/apps/vespa-otelcol-start/wrapper.h
new file mode 100644
index 00000000000..d6ff06a4f0d
--- /dev/null
+++ b/logforwarder/src/apps/vespa-otelcol-start/wrapper.h
@@ -0,0 +1,18 @@
+// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include "cf-handler.h"
+#include "child-handler.h"
+
+#include <string>
+
+class Wrapper : public CfHandler {
+private:
+    ChildHandler _childHandler;
+public:
+    Wrapper(const std::string &configId);
+    ~Wrapper();
+    void check();
+    void stop();
+    void gotConfig(const OpenTelemetryConfig& config) override;
+};