configManager.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_dpm
 * @copyright (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 *
 * @brief DPM server config
 */
#include <configManager.hpp>
 
#include <iostream>
#include <stdexcept>
 
#include <boost/program_options.hpp>
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <rad/helper.hpp>
#include <yaml-cpp/yaml.h>
 
#include <daq/log4cplus.hpp>
 
namespace daq::dpm {
 
ConfigManager::ConfigManager(log4cplus::Logger logger)
    : m_logger(std::move(logger)), m_mgr(&m_config, m_logger) {
    m_mgr.Register(&Configuration::name, {"process-name", "Component instance ID"});
    m_mgr.Register(&Configuration::dataroot, {"dataroot", "ICS standard DATAROOT directory"});
    m_mgr.Register(&Configuration::workspace, {"workspace", "DPM merge workspace"});
    m_mgr.Register(&Configuration::rr_uri, {"rr-uri", "Request/reply URI"});
    m_mgr.Register(&Configuration::ps_uri, {"ps-uri", "Publish/subscribe URI"});
    m_mgr.Register(&Configuration::no_ipc, {"no-ipc", "Disables IPC interfaces"});
    m_mgr.Register(&Configuration::db_endpoint, {"db-endpoint", "REDIS 'host:port'"});
    m_mgr.Register(&Configuration::db_timeout, {"db-timeout", "Database timeout"});
    m_mgr.Register(&Configuration::log_level, {"log-level", "Log verbosity level"});
    m_mgr.Register(&Configuration::log_config, {"log-config", "Log configuration file"});
    m_mgr.Register(&Configuration::config_file, {"config-file", "YAML configuration file path"});
    m_mgr.Register(&Configuration::poll_once, {"poll-once", "Poll scheduler once"});
 
    m_mgr.Register(&Configuration::merge_bin, {"merge-bin", "DPM Merge application binary."});
    m_mgr.Register(&Configuration::rsync_bin, {"rsync-bin", "rsync application binary."});
 
    m_mgr.Register(&Configuration::limit_daq, {"limit-daq", "Concurrency limit: Number of DAQs"});
    m_mgr.Register(&Configuration::limit_merge,
                   {"limit-merge", "Concurrency limit: Number of merges"});
    m_mgr.Register(&Configuration::limit_net_send,
                   {"limit-net-send", "Concurrency limit: Number of network send tranfers."});
    m_mgr.Register(
        &Configuration::limit_net_receive,
        {"limit-net-receive", "Concurrency limit: Number of network receive transfers."});
 
    std::string db_addr = rad::Helper::GetEnvVar("DB_HOST");
    if (auto root = rad::Helper::GetEnvVar("DATAROOT"); !root.empty()) {
        m_mgr.Update(
            &Configuration::dataroot, root, {config::Origin::EnvironmentVariable, "$DATAROOT"});
    }
}
 
bool ConfigManager::ParseArguments(int argc, char* argv[]) {
    namespace po = boost::program_options;
    try {
        config::OriginInfo si{config::Origin::CommandLine, ""};
        po::options_description options_desc("Options");
        // clang-format off
        // Note be careful of providing default_value as it will be interpreted as if user provided
        // option!
        options_desc.add_options()
            ("help,h", "Print help messages")
            ("proc-name,n",
             po::value<std::string>(),
             "Process name")
            ("log-level,l",
             po::value<std::string>(),
             "Log level: ERROR, WARNING, STATE, EVENT, ACTION, INFO, DEBUG, TRACE")
            ("config,c",
             po::value<std::string>(),
             "Configuration filename")
            ("db-host,d",
             po::value<std::string>(),
             "In-memory DB host (ipaddr:port)")
            ("rr-uri",
             po::value<std::string>(),
             "Request/reply URI")
            ("ps-uri",
             po::value<std::string>(),
             "Publish/subscribe URI")
            ("workspace",
             po::value<std::string>(),
             "workspace path (relative paths are relateive to dataroot)")
            ("no-ipc",
             "Disable request/reply, publish/subscribe and OLDB interfaces for standalone mode")
            ("poll-once",
             "Poll scheduler once and then exit rather than running until asked to quit (this "
             "option also implies --no-ipc)")
            ;
        // clang-format on
        po::variables_map options_map;
        po::store(po::parse_command_line(argc, argv, options_desc), options_map);
        if (options_map.count("help")) {
            std::cerr << options_desc << std::endl;
            return false;
        }
        po::notify(options_map);
 
        if (options_map.count("config")) {
            auto value = options_map["config"].as<std::string>();
            m_mgr.Update(&Configuration::config_file, value, si);
        }
 
        if (options_map.count("log-level")) {
            // Note we cannot use log4cplus::LogLevel as it's an alias to int and we want to parse
            // a string using ostream, so we use the daq helper daq::LogLevel instead.
            auto ll_str = options_map["log-level"].as<std::string>();
            auto ll = boost::lexical_cast<LogLevel>(ll_str);
            m_mgr.Update(&Configuration::log_level, ll.value, si);
        }
 
        if (options_map.count("proc-name")) {
            auto value = options_map["proc-name"].as<std::string>();
            m_mgr.Update(&Configuration::name, value, si);
        }
 
        if (options_map.count("db-host")) {
            auto value = options_map["db-host"].as<std::string>();
            m_mgr.Update(&Configuration::db_endpoint, value, si);
        }
        if (options_map.count("rr-uri")) {
            auto value = options_map["rr-uri"].as<std::string>();
            m_mgr.Update(&Configuration::rr_uri, value, si);
        }
        if (options_map.count("ps-uri")) {
            auto value = options_map["ps-uri"].as<std::string>();
            m_mgr.Update(&Configuration::ps_uri, value, si);
        }
        if (options_map.count("workspace")) {
            auto value = options_map["workspace"].as<std::string>();
            m_mgr.Update(&Configuration::workspace, value, si);
        }
        if (options_map.count("no-ipc")) {
            m_mgr.Update(&Configuration::no_ipc, true, si);
        }
        if (options_map.count("poll-once")) {
            m_mgr.Update(&Configuration::poll_once, true, si);
            m_mgr.Update(
                &Configuration::no_ipc,
                true,
                {config::Origin::CommandLine, "Option is implicitly set due to `--poll-once`"});
        }
 
    } catch (...) {
        std::throw_with_nested(std::runtime_error("Failed to parse command line arguments"));
    }
    return true;
}
 
void ConfigManager::LoadConfig() try {
    std::string resolved = rad::Helper::FindFile(m_config.config_file);
    LOG4CPLUS_INFO(m_logger, "Loading configuration from " << resolved);
    if (resolved.empty()) {
        auto msg = fmt::format("Could not resolve config file {}", m_config.config_file);
        LOG4CPLUS_ERROR(m_logger, msg);
        throw std::invalid_argument(msg);
    }
    config::OriginInfo si{config::Origin::Configuration, resolved};
    auto config = YAML::LoadFile(resolved);
 
    if (auto const& node = config["cfg.dataroot"]; node) {
        m_mgr.Update(&Configuration::dataroot, node.as<std::string>(), si);
    }
    if (auto const& node = config["cfg.workspace"]; node) {
        m_mgr.Update(&Configuration::workspace, node.as<std::string>(), si);
    }
    if (auto const& node = config["cfg.log.properties"]; node) {
        m_mgr.Update(&Configuration::log_config, node.as<std::string>(), si);
    }
    if (auto const& node = config["cfg.req.endpoint"]; node) {
        m_mgr.Update(&Configuration::rr_uri, node.as<std::string>(), si);
    }
    if (auto const& node = config["cfg.pub.endpoint"]; node) {
        m_mgr.Update(&Configuration::ps_uri, node.as<std::string>(), si);
    }
    if (auto const& node = config["cfg.bin.merge"]; node) {
        m_mgr.Update(&Configuration::merge_bin, node.as<std::string>(), si);
    }
    if (auto const& node = config["cfg.bin.rsync"]; node) {
        m_mgr.Update(&Configuration::rsync_bin, node.as<std::string>(), si);
    }
 
    // Concurrency limits
    if (auto const& node = config["cfg.limit.daq"]; node) {
        m_mgr.Update(&Configuration::limit_daq, node.as<unsigned>(), si);
    }
    if (auto const& node = config["cfg.limit.merge"]; node) {
        m_mgr.Update(&Configuration::limit_merge, node.as<unsigned>(), si);
    }
    if (auto const& node = config["cfg.limit.net.receive"]; node) {
        m_mgr.Update(&Configuration::limit_net_receive, node.as<unsigned>(), si);
    }
    if (auto const& node = config["cfg.limit.net.send"]; node) {
        m_mgr.Update(&Configuration::limit_net_send, node.as<unsigned>(), si);
    }
 
} catch (...) {
    std::throw_with_nested(std::runtime_error(
        fmt::format("Failed to load configuration from {}", m_config.config_file)));
}
 
}  // namespace daq::dpm