config.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup server
 * @copyright ESO - European Southern Observatory
 * @author
 *
 * @brief Config class source file.
 */
 
#include "config.hpp"
 
#include <iostream>
#include <optional>
 
#include <rad/assert.hpp>
#include <rad/config.hpp>
#include <rad/exceptions.hpp>
#include <rad/helper.hpp>
 
#include <boost/program_options.hpp>
#include <fmt/chrono.h>
#include <fmt/format.h>
 
#include <daq/config/manager.hpp>
 
#include "logger.hpp"
 
namespace bpo = boost::program_options;
 
template <>
struct daq::config::Formatter<std::filesystem::path> {
    static auto Format(std::filesystem::path const &t) {
        return t.c_str();
    }
};
 
namespace server {
 
Config::Config()
    : m_mgr(GetLogger())
    , m_proc_name(CONFIG_DEFAULT_PROCNAME)
    , m_log_level(CONFIG_DEFAULT_LOG_LEVEL)
    , m_log_properties()
    , m_config_filename(CONFIG_DEFAULT_FILENAME)
    , m_scxml_filename(CONFIG_DEFAULT_SCXML_FILENAME)
    , m_db_prefix(CONFIG_DEFAULT_OLDB_URI_PREFIX + "/ocm")
    , m_db_timeout_sec(CONFIG_DEFAULT_DB_TIMEOUT_SEC)
    , m_req_endpoint(CONFIG_DEFAULT_REQ_ENDPOINT)
    , m_pub_endpoint(CONFIG_DEFAULT_PUB_ENDPOINT)
    , m_out_path()
    , m_workspace(CONFIG_DEFAULT_WORKSPACE) {
    RAD_TRACE(GetLogger());
    m_mgr.Register(&m_instrument_id, {"cfg/instrument_id", "Instrument id"});
    m_mgr.Register(&m_proc_name, {rad::KEY_CONFIG_PROCNAME, "Component instance ID"});
    m_mgr.Register(&m_log_level, {rad::KEY_CONFIG_LOG_LEVEL, "Log verbosity level"});
    m_mgr.Register(&m_log_properties, {rad::KEY_CONFIG_LOG_PROPERTIES, "Log configuration file"});
    m_mgr.Register(&m_config_filename, {rad::KEY_CONFIG_FILENAME, "YAML configuration file path"});
 
    m_mgr.Register(&m_scxml_filename,
                   {rad::KEY_CONFIG_SM_SCXML, "State machine SCXML model file path"});
    m_mgr.Register(&m_db_prefix, {rad::KEY_CONFIG_OLDB_URI_PREFIX, "CII OLDB URI prefix"});
    m_mgr.Register(&m_db_timeout_sec,
                   {rad::KEY_CONFIG_OLDB_CONN_TIMEOUT, "CII OLDB timeout (sec)"});
    m_mgr.Register(&m_out_path, {KEY_CONFIG_DATAROOT, "ICS standard DATAROOT directory"});
    m_mgr.Register(&m_req_endpoint, {rad::KEY_CONFIG_REQ_ENDPOINT, "Request/reply URI"});
    m_mgr.Register(&m_pub_endpoint, {"cfg/pub_endpoint", "Publish/subscribe URI"});
    m_mgr.Register(&m_workspace, {"cfg/daq/workspace", "OCM DAQ workspace"});
 
    m_mgr.Register(&m_dpm_params.rr_uri, {"cfg/dpm/req_endpoint", "DPM request/reply URI"});
    m_mgr.Register(&m_dpm_params.ps_uri, {"cfg/dpm/pub_endpoint", "DPM publish/subscribe URI"});
    m_mgr.Register(&m_dpm_params.timeout, {"cfg/dpm/timeout_sec", "DPM timeout (sec)"});
 
    m_mgr.Register(&m_stale_acquiring,
                   {"cfg/daq/stale_acquiring_hours",
                    "Age when DAQ acquisitions are considered stale (hours)"});
    m_mgr.Register(
        &m_stale_merging,
        {"cfg/daq/stale_merging_hours", "Age when DAQ merges are considered stale (hours)"});
    m_mgr.Register(&m_dictionary_names,
                   {"cfg/dictionaries", "Specifies the FITS keyword dictionaries to load"});
 
    /*
     * Read environment variables.
     */
    if (auto val = rad::Helper::GetEnvVar(CONFIG_ENVVAR_OUT_PATH); !val.empty()) {
        m_mgr.Update(&m_out_path, val, {daq::config::Origin::EnvironmentVariable, "$DATAROOT"});
    }
}
 
bool Config::ParseOptions(int argc, char *argv[]) {
    RAD_TRACE(GetLogger());
 
    /*
     * Define command line options.
     */
    bpo::options_description options_desc("Options");
    // clang-format off
    options_desc.add_options()
        ("help,h", "Print help messages")
        ("proc-name,n",
         bpo::value<std::string>(),
         "Process name")
        ("log-level,l",
         bpo::value<std::string>(),
         "Log level: ERROR, WARNING, STATE, EVENT, ACTION, INFO, DEBUG, TRACE")
        ("config,c",
         bpo::value<std::string>(),
         "Configuration filename")
        ("db-host,d",
         bpo::value<std::string>(),
         "*deprecated* In-memory DB host (ipaddr:port)");
    // clang-format on
 
    // Parse the options.
    try {
        bpo::variables_map options_map;
        bpo::store(bpo::parse_command_line(argc, argv, options_desc), options_map);
        if (options_map.count("help")) {
            std::cout << options_desc << std::endl;
            return false;
        }
 
        /*
         * Throws on error, so do after help in case
         * there are any problems.
         */
        bpo::notify(options_map);
 
        daq::config::OriginInfo si{daq::config::Origin::CommandLine, ""};
 
        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>();
            m_mgr.Update(&m_log_level, ll_str, si);
        }
 
        if (options_map.count("proc-name")) {
            auto value = options_map["proc-name"].as<std::string>();
            if (m_mgr.Update(&m_proc_name, value, si)) {
                // Since proc-name updated we also update default for db-prefix
                m_mgr.Update(&m_db_prefix,
                             CONFIG_DEFAULT_OLDB_URI_PREFIX + "/" + m_proc_name,
                             {daq::config::Origin::Default, ""});
            }
        }
 
        if (options_map.count("config")) {
            auto value = options_map["config"].as<std::string>();
            m_mgr.Update(&m_config_filename, value, si);
        }
        if (options_map.count("db-host")) {
            LOG4CPLUS_WARN(GetLogger(),
                           "Command line parameter --db-host|-d is deprecated and only provided "
                           "for compatibility");
        }
    } catch (bpo::error &e) {
        std::cerr << "ERROR: " << e.what() << std::endl << std::endl;
        std::cerr << options_desc << std::endl;
        throw std::invalid_argument(e.what());
    }
 
    return true;
}
 
void Config::LoadConfig(const std::string &filename) {
    RAD_TRACE(GetLogger());
    m_config.emplace();
 
    std::string config_filename = filename;
    if (config_filename == "") {
        config_filename = m_config_filename;
    }
 
    // resolve filename
    std::string resolved_config_filename = rad::Helper::FindFile(config_filename);
    if (resolved_config_filename.size() == 0) {
        LOG4CPLUS_ERROR(GetLogger(), "Cannot find <" << config_filename << ">");
        char const *cfgpath = std::getenv("CFGPATH");
        if (cfgpath == nullptr) {
            cfgpath = "{unset}";
        }
        throw rad::Exception("Cannot find " + config_filename, fmt::format("$CFGPATH={}", cfgpath));
    } else {
        LOG4CPLUS_DEBUG(GetLogger(),
                        "Loading configuration from <" << resolved_config_filename << ">");
    }
    try {
        using daq::config::GetParamAs;
        auto ifs = std::ifstream(resolved_config_filename);
        ifs.exceptions(std::ios_base::badbit);
 
        m_config->UpdateFromStream(ifs);
        auto const &root = m_config->GetInstanceRootNode();
        daq::config::OriginInfo si{daq::config::Origin::Configuration, resolved_config_filename};
 
        if (auto value = GetParamAs<std::string>(rad::KEY_CONFIG_LOG_PROPERTIES, root); value) {
            m_mgr.Update(&m_log_properties, *value, si);
        }
 
        if (auto value = GetParamAs<std::string>(rad::KEY_CONFIG_REQ_ENDPOINT, root); value) {
            auto updated = m_mgr.Update(&m_req_endpoint, *value, si);
            if (updated && m_req_endpoint.empty()) {
                throw rad::Exception("Invalid configuration",
                                     fmt::format("invalid configuration for '{}': '{}' request "
                                                 "endpoint must be non-empty and have a trailing "
                                                 "path separator '/'.",
                                                 rad::KEY_CONFIG_REQ_ENDPOINT,
                                                 m_req_endpoint));
            }
        }
        if (auto value = GetParamAs<std::string>(KEY_CONFIG_PUB_ENDPOINT, root); value) {
            auto updated = m_mgr.Update(&m_pub_endpoint, *value, si);
            if (updated && m_pub_endpoint.empty()) {
                throw rad::Exception("Invalid configuration",
                                     fmt::format("invalid configuration for '{}': '{}' publish "
                                                 "endpoint must be non-empty and have a trailing "
                                                 "path separator '/'.",
                                                 KEY_CONFIG_PUB_ENDPOINT,
                                                 m_pub_endpoint));
            }
        }
 
        if (auto value = GetParamAs<std::string>(rad::KEY_CONFIG_OLDB_URI_PREFIX, root); value) {
            m_mgr.Update(&m_db_prefix, *value, si);
        }
 
        if (auto value = GetParamAs<int>(rad::KEY_CONFIG_OLDB_CONN_TIMEOUT, root); value) {
            m_mgr.Update(&m_db_timeout_sec, *value, si);
        }
 
        if (auto value = GetParamAs<std::string>(rad::KEY_CONFIG_SM_SCXML, root); value) {
            m_mgr.Update(&m_scxml_filename, *value, si);
        }
 
        if (auto value = GetParamAs<std::string>(KEY_CONFIG_INSTRUMENT_ID, root); value) {
            m_mgr.Update(&m_instrument_id, *value, si);
        } else {
            LOG4CPLUS_ERROR(
                GetLogger(),
                "Required configuration parameter <" << KEY_CONFIG_INSTRUMENT_ID << "> missing");
            throw rad::Exception(
                "Invalid configuration: Required parameter empty: " + KEY_CONFIG_INSTRUMENT_ID,
                filename);
        }
        if (auto value = GetParamAs<std::string>(KEY_CONFIG_DATAROOT, root); value) {
            m_mgr.Update(&m_out_path, *value, si);
        }
        // Workspace
        if (auto value = GetParamAs<std::string>(KEY_CONFIG_WORKSPACE, root); value) {
            m_mgr.Update(&m_workspace, *value, si);
        }
        LOG4CPLUS_DEBUG(GetLogger(), "Resolved workspace = <" << GetWorkspace() << ">");
        // Stale daqs from workspace
        if (auto value = GetParamAs<int>(KEY_CONFIG_STALE_DAQ_ACQUIRING, root); value) {
            m_mgr.Update(&m_stale_acquiring, std::chrono::hours(*value), si);
        }
        if (auto value = GetParamAs<int>(KEY_CONFIG_STALE_DAQ_MERGING, root)) {
            m_mgr.Update(&m_stale_merging, std::chrono::hours(*value), si);
        }
        // DPM
        if (auto value = GetParamAs<std::string>(KEY_CONFIG_DPM_REQ_ENDPOINT, root); value) {
            m_mgr.Update(&m_dpm_params.rr_uri, *value, si);
        } else {
            LOG4CPLUS_ERROR(
                GetLogger(),
                "Required configuration parameter <" << KEY_CONFIG_DPM_REQ_ENDPOINT << "> missing");
            throw rad::Exception(
                "Invalid configuration: Required parameter empty: " + KEY_CONFIG_DPM_REQ_ENDPOINT,
                filename);
        }
        if (auto value = GetParamAs<std::string>(KEY_CONFIG_DPM_PUB_ENDPOINT, root); value) {
            m_mgr.Update(&m_dpm_params.ps_uri, *value, si);
        } else {
            LOG4CPLUS_ERROR(
                GetLogger(),
                "Required configuration parameter <" << KEY_CONFIG_DPM_PUB_ENDPOINT << "> missing");
            throw rad::Exception(
                "Invalid configuration: Required parameter empty: " + KEY_CONFIG_DPM_PUB_ENDPOINT,
                filename);
        }
        if (auto value = GetParamAs<int>(KEY_CONFIG_DPM_TIMEOUT_SEC, root); value) {
            m_mgr.Update(&m_dpm_params.timeout, std::chrono::seconds(*value), si);
        }
        if (auto value = GetParamAs<std::vector<std::string>>(KEY_CONFIG_DICTIONARIES, root);
            value) {
            m_mgr.Update(&m_dictionary_names, *value, si);
        }
    } catch (std::exception const &e) {
        LOG4CPLUS_ERROR(GetLogger(), e.what());
        throw rad::Exception("Failed to load configuration", filename);
    }
 
    LOG4CPLUS_DEBUG(GetLogger(), "Loaded configuration file <" << m_config_filename << ">");
}
 
const std::string &Config::GetMsgReplierEndpoint() const {
    RAD_TRACE(GetLogger());
    return m_req_endpoint;
}
 
const std::string &Config::GetPubEndpoint() const {
    RAD_TRACE(GetLogger());
    return m_pub_endpoint;
}
 
const std::string &Config::GetDbPrefix() const {
    RAD_TRACE(GetLogger());
    if (m_db_prefix.empty()) {
        return m_proc_name;
    } else {
        return m_db_prefix;
    }
}
 
std::chrono::seconds Config::GetDbTimeout() const {
    return std::chrono::seconds(m_db_timeout_sec);
}
 
const std::string &Config::GetSmScxmlFilename() const {
    RAD_TRACE(GetLogger());
    return m_scxml_filename;
}
 
const std::string &Config::GetConfigFilename() const {
    RAD_TRACE(GetLogger());
    return m_config_filename;
}
 
const std::string &Config::GetProcName() const {
    RAD_TRACE(GetLogger());
    return m_proc_name;
}
 
const std::string &Config::GetLogLevel() const {
    RAD_TRACE(GetLogger());
    return m_log_level;
}
 
const std::string &Config::GetLogProperties() const {
    RAD_TRACE(GetLogger());
    return m_log_properties;
}
 
std::vector<std::string> const &Config::GetDictionaryNames() const {
    return m_dictionary_names;
}
 
daq::DpmClientParams const &Config::GetDpmClientParams() const {
    return m_dpm_params;
}
 
}  // namespace server