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 <fstream>
#include <iostream>
#include <stdexcept>
 
#include <boost/program_options.hpp>
#include <fmt/chrono.h>
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <fmt/std.h>
#include <rad/helper.hpp>
 
#include <daq/log4cplus.hpp>
 
namespace daq::dpm {
 
ConfigManager::ConfigManager(log4cplus::Logger logger)
    : m_logger(std::move(logger)), m_mgr(m_logger) {
    m_mgr.Register(&m_config.name, {"cfg/procname", "Component instance ID"});
    m_mgr.Register(&m_config.dataroot, {"cfg/dataroot", "ICS standard DATAROOT directory"});
    m_mgr.Register(&m_config.workspace, {"cfg/daq/workspace", "DPM merge workspace"});
    m_mgr.Register(&m_config.rr_uri, {"cfg/req_endpoint", "Request/reply URI"});
    m_mgr.Register(&m_config.ps_uri, {"cfg/pub_endpoint", "Publish/subscribe URI"});
    m_mgr.Register(&m_config.no_ipc, {"cfg/no_ipc", "Disables IPC interfaces"});
    m_mgr.Register(&m_config.db_prefix,
                   {"cfg/oldb_uri_prefix", "CII OLDB prefix 'cii.oldb:/elt/..'"});
    m_mgr.Register(&m_config.db_timeout, {"cfg/oldb_conn_timeout", "CII OLDB connection timeout"});
    m_mgr.Register(&m_config.log_level, {"cfg/log_level", "Log verbosity level"});
    m_mgr.Register(&m_config.log_config, {"cfg/log_properties", "Log configuration file"});
    m_mgr.Register(&m_config.config_file, {"cfg/filename", "YAML configuration file path"});
    m_mgr.Register(&m_config.poll_once, {"cfg/poll_once", "Poll scheduler once"});
 
    m_mgr.Register(&m_config.merge_bin, {"cfg/bin_merge", "DPM Merge application binary."});
    m_mgr.Register(&m_config.rsync_bin, {"cfg/bin_rsync", "rsync application binary."});
 
    m_mgr.Register(&m_config.limit_daq, {"cfg/limits/daq", "Concurrency limit: Number of DAQs"});
    m_mgr.Register(&m_config.limit_merge,
                   {"cfg/limits/merge", "Concurrency limit: Number of merges"});
    m_mgr.Register(&m_config.limit_net_send,
                   {"cfg/limits/net_send", "Concurrency limit: Number of network send tranfers."});
    m_mgr.Register(
        &m_config.limit_net_receive,
        {"cfg/limits/net_receive", "Concurrency limit: Number of network receive transfers."});
 
    if (auto root = rad::Helper::GetEnvVar("DATAROOT"); !root.empty()) {
        m_mgr.Update(&m_config.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(&m_config.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(&m_config.log_level, ll.value, si);
        }
 
        if (options_map.count("proc-name")) {
            auto value = options_map["proc-name"].as<std::string>();
            if (m_mgr.Update(&m_config.name, value, si)) {
                // Since proc-name updated we also update default for db-prefix
                m_mgr.Update(&m_config.db_prefix,
                             "cii.oldb:/elt/" + m_config.name,
                             {daq::config::Origin::Default, "Updated default using proc-name"});
            }
        }
 
        if (options_map.count("db-host")) {
            LOG4CPLUS_WARN(m_logger,
                           "Command line parameter --db-host|-d is deprecated and only provided "
                           "for compatibility");
        }
        if (options_map.count("rr-uri")) {
            auto value = options_map["rr-uri"].as<std::string>();
            m_mgr.Update(&m_config.rr_uri, value, si);
        }
        if (options_map.count("ps-uri")) {
            auto value = options_map["ps-uri"].as<std::string>();
            m_mgr.Update(&m_config.ps_uri, value, si);
        }
        if (options_map.count("workspace")) {
            auto value = options_map["workspace"].as<std::string>();
            m_mgr.Update(&m_config.workspace, value, si);
        }
        if (options_map.count("no-ipc")) {
            m_mgr.Update(&m_config.no_ipc, true, si);
        }
        if (options_map.count("poll-once")) {
            m_mgr.Update(&m_config.poll_once, true, si);
            m_mgr.Update(
                &m_config.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};
    using daq::config::GetParamAs;
    auto ifs = std::ifstream(resolved);
    ifs.exceptions(std::ios_base::badbit);
 
    elt::configng::CiiConfigDocument config;
    config.UpdateFromStream(ifs);
    auto const& root = config.GetInstanceRootNode();
 
    using daq::config::GetParamAs;
    UpdateFromConfig<std::string>(&m_config.dataroot, root, si);
    UpdateFromConfig<std::string>(&m_config.workspace, root, si);
    UpdateFromConfig<std::string>(&m_config.log_config, root, si);
    UpdateFromConfig<std::string>(&m_config.rr_uri, root, si);
    UpdateFromConfig<std::string>(&m_config.ps_uri, root, si);
    UpdateFromConfig<std::string>(&m_config.merge_bin, root, si);
    UpdateFromConfig<std::string>(&m_config.rsync_bin, root, si);
    // OLDB
    UpdateFromConfig<std::string>(&m_config.db_prefix, root, si);
    UpdateFromConfig<int>(&m_config.db_timeout, root, si);
    // Concurrency limits
    UpdateFromConfig<int>(&m_config.limit_daq, root, si);
    UpdateFromConfig<int>(&m_config.limit_merge, root, si);
    UpdateFromConfig<int>(&m_config.limit_net_receive, root, si);
    UpdateFromConfig<int>(&m_config.limit_net_send, root, si);
 
} catch (...) {
    std::throw_with_nested(std::runtime_error(
        fmt::format("Failed to load configuration from {}", m_config.config_file)));
}
 
}  // namespace daq::dpm