add test

agent/src/main.cc

@@ -35,6 +35,8 @@ static std::shared_ptr<streaming_client> _streaming_client;
 
 static std::shared_ptr<streaming_server> _streaming_server;
 
+static asio::signal_set _signals(*g_io_context, SIGTERM, SIGUSR1, SIGUSR2);
+
 static void signal_handler(const boost::system::error_code& error,
                            int signal_number) {
   if (!error) {
@@ -58,6 +60,7 @@ static void signal_handler(const boost::system::error_code& error,
         }
         break;
     }
+    _signals.async_wait(signal_handler);
   }
 }
 
@@ -118,18 +121,29 @@ int main(int argc, char* argv[]) {
       "{} or increase by kill -USR2 {}",
       getpid(), getpid());
 
-  sigignore(SIGPIPE);
-
   po::variables_map vm;
-  po::store(po::parse_command_line(argc, argv, desc), vm);
-  po::notify(vm);
 
-  if (vm.count("help")) {
-    std::cout << desc << "\n";
-    return 1;
+  try {
+    po::store(po::parse_command_line(argc, argv, desc), vm);
+    po::notify(vm);
+
+    if (vm.count("help")) {
+      std::cout << desc << "\n";
+      return 1;
+    }
+
+    if (vm.count("config-file")) {
+      po::store(po::parse_config_file(
+                    vm["config-file"].as<std::string>().c_str(), desc),
+                vm);
+    }
+
+  } catch (const std::exception& e) {
+    SPDLOG_ERROR("fail to parse arguments {}", e.what());
+    return 2;
   }
 
-  const std::string logger_name = "centeon-agent";
+  const std::string logger_name = "centreon-agent";
 
   auto logger_file = [&]() {
     if (vm.count("logger-file")) {
@@ -175,13 +189,18 @@ int main(int argc, char* argv[]) {
                      getpid(), getpid());
   std::shared_ptr<com::centreon::common::grpc::grpc_config> conf;
   std::string supervised_host;
+
   try {
-    asio::signal_set signals(*g_io_context, SIGTERM, SIGUSR1, SIGUSR2);
-    signals.async_wait(signal_handler);
+    // ignored but mandatory because of forks
+    _signals.add(SIGPIPE);
+
+    _signals.async_wait(signal_handler);
     if (!vm.count("endpoint")) {
       SPDLOG_CRITICAL(
           "endpoint param is mandatory (represents where to connect or where "
           "to listen example: 127.0.0.1:4317)");
+      // in order to avoid service restart non stop
+      std::this_thread::sleep_for(std::chrono::seconds(10));
       return -1;
     }
     std::string host_port = vm["endpoint"].as<std::string>();

agent/src/streaming_client.cc

@@ -129,6 +129,7 @@ void streaming_client::_start() {
  *
  */
 void streaming_client::_create_reactor() {
+  std::lock_guard l(_protect);
   if (_reactor) {
     _reactor->shutdown();
   }

clib/inc/com/centreon/process.hh

@@ -105,7 +105,7 @@ class process {
           bool in_stream = true,
           bool out_stream = true,
           bool err_stream = true);
-  virtual ~process() noexcept;
+  virtual ~process();
   process(const process&) = delete;
   process& operator=(const process&) = delete;
   // void enable_stream(stream s, bool enable);
@@ -129,6 +129,6 @@ class process {
   void set_timeout(bool timeout);
 };
 
-}
+}  // namespace com::centreon
 
 #endif  // !CC_PROCESS_POSIX_HH

clib/src/process.cc

@@ -1,20 +1,20 @@
 /**
-* Copyright 2012-2013,2020-2021 Centreon
-*
-* Licensed under the Apache License, Version 2.0 (the "License");
-* you may not use this file except in compliance with the License.
-* You may obtain a copy of the License at
-*
-*     http://www.apache.org/licenses/LICENSE-2.0
-*
-* Unless required by applicable law or agreed to in writing, software
-* distributed under the License is distributed on an "AS IS" BASIS,
-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-* See the License for the specific language governing permissions and
-* limitations under the License.
-*
-* For more information : [email protected]
-*/
+ * Copyright 2012-2013,2020-2021 Centreon
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * For more information : [email protected]
+ */
 
 #include <sstream>
 
@@ -63,7 +63,7 @@ process::process(process_listener* listener,
 /**
  *  Destructor.
  */
-process::~process() noexcept {
+process::~process() {
   std::unique_lock<std::mutex> lock(_lock_process);
   _kill(SIGKILL);
   _cv_process_running.wait(lock, [this] { return !_is_running(); });
@@ -632,6 +632,7 @@ void process::_kill(int sig) {
       throw basic_error() << "could not terminate process " << _process << ": "
                           << msg;
     }
+    _process = 0;
   }
 }
 

engine/modules/opentelemetry/CMakeLists.txt

@@ -95,7 +95,8 @@ target_link_libraries(opentelemetry
 
 add_dependencies(opentelemetry
   pb_open_telemetry_lib
-  pb_neb_lib)
+  pb_neb_lib
+  engine_rpc)
 
 target_include_directories(opentelemetry PRIVATE
   "${MODULE_DIR}/inc/com/centreon/engine/modules/opentelemetry"

engine/modules/opentelemetry/src/centreon_agent/agent_config.cc

@@ -75,8 +75,8 @@ agent_config::agent_config(const rapidjson::Value& json_config_v) {
   _export_period = file_content.get_unsigned("export_period", 60);
   _check_timeout = file_content.get_unsigned("_check_timeout", 30);
 
-  if (file_content.has_member("agent_reverse_server")) {
-    const auto& reverse_array = file_content.get_member("agent_reverse_server");
+  if (file_content.has_member("reverse_connections")) {
+    const auto& reverse_array = file_content.get_member("reverse_connections");
     for (auto conf_iter = reverse_array.Begin();
          conf_iter != reverse_array.End(); ++conf_iter) {
       _agent_grpc_reverse_conf.insert(

engine/modules/opentelemetry/src/centreon_agent/agent_reverse_client.cc

@@ -57,6 +57,15 @@ void agent_reverse_client::update(const agent_config::pointer& new_conf) {
   absl::MutexLock l(&_agents_m);
 
   auto connection_iterator = _agents.begin();
+
+  if (!new_conf) {
+    while (connection_iterator != _agents.end()) {
+      _shutdown_connection(connection_iterator);
+      connection_iterator = _agents.erase(connection_iterator);
+    }
+    return;
+  }
+
   auto conf_iterator = new_conf->get_agent_grpc_reverse_conf().begin();
 
   while (connection_iterator != _agents.end() &&

engine/modules/opentelemetry/src/centreon_agent/agent_service.cc

@@ -85,7 +85,14 @@ agent_service::agent_service(
     : _io_context(io_context),
       _conf(conf),
       _metric_handler(handler),
-      _logger(logger) {}
+      _logger(logger) {
+  if (!_conf) {
+    _conf = std::make_shared<agent_config>(60, 100, 10, 30);
+    SPDLOG_LOGGER_INFO(logger,
+                       "no centreon_agent configuration given => we use a "
+                       "default configuration ");
+  }
+}
 
 /**
  * @brief prefered way to construct an agent_service

engine/modules/opentelemetry/src/host_serv_extractor.cc

@@ -87,17 +87,26 @@ host_serv_attributes_extractor::host_serv_attributes_extractor(
       [](po::options_description& desc) {
         desc.add_options()(
             "host_path", po::value<std::string>(),
-            "where to find host name. Example: "
-            "resourceMetrics.scopeMetrics.metrics.dataPoints.attributes.host");
-        desc.add_options()("service_path", po::value<std::string>(),
-                           "where to find service description. Example: "
-                           "resourceMetrics.scopeMetrics.metrics.dataPoints."
-                           "attributes.service");
+            "where to find host name. Example:\n"
+            "resource_metrics.scopeMetrics.metrics.dataPoints.attributes.host\n"
+            "or\n"
+            "resource_metrics.resource.attributes.host\n"
+            "or\n"
+            "resource_metrics.scope_metrics.scope.attributes.host");
+        desc.add_options()(
+            "service_path", po::value<std::string>(),
+            "where to find service description. Example:\n"
+            "resource_metrics.scope_metrics.data.data_points.attributes."
+            "service\n"
+            "or\n"
+            "resource_metrics.resource.attributes.service\n"
+            "or\n"
+            "resource_metrics.scope_metrics.scope.attributes.service");
       });
 
   static auto parse_path = [](const std::string& path, attribute_owner& attr,
                               std::string& key) {
-    static re2::RE2 path_extractor("\\.(\\w+)\\.attributes\\.(\\w+)");
+    static re2::RE2 path_extractor("(?i)\\.(\\w+)\\.attributes\\.([\\.\\w]+)");
     std::string sz_attr;
     if (!RE2::PartialMatch(path, path_extractor, &sz_attr, &key)) {
       throw exceptions::msg_fmt(

engine/modules/opentelemetry/src/otl_config.cc

@@ -64,6 +64,10 @@ static constexpr std::string_view _grpc_config_schema(R"(
         "telegraf_conf_server": {
             "description": "http(s) telegraf config server",
             "type": "object"
+        },
+        "centreon_agent": {
+            "description": "config of centreon_agent",
+            "type": "object"
         }
       }, 
       "type" : "object"

engine/src/configuration/applier/state.cc

@@ -460,6 +460,7 @@ void applier::state::_apply(configuration::state const& new_cfg) {
   config->log_level_downtimes(new_cfg.log_level_downtimes());
   config->log_level_comments(new_cfg.log_level_comments());
   config->log_level_macros(new_cfg.log_level_macros());
+  config->log_level_otl(new_cfg.log_level_otl());
   config->use_true_regexp_matching(new_cfg.use_true_regexp_matching());
   config->use_send_recovery_notifications_anyways(
       new_cfg.use_send_recovery_notifications_anyways());
@@ -1168,10 +1169,10 @@ void applier::state::apply_log_config(configuration::state& new_cfg) {
     broker_sink->set_level(spdlog::level::info);
     log_cfg.add_custom_sink(broker_sink);
 
-    log_cfg.apply_custom_sinks({"functions", "config", "events", "checks",
-                                "notifications", "eventbroker",
-                                "external_command", "commands", "downtimes",
-                                "comments", "macros", "process", "runtime"});
+    log_cfg.apply_custom_sinks(
+        {"functions", "config", "events", "checks", "notifications",
+         "eventbroker", "external_command", "commands", "downtimes", "comments",
+         "macros", "process", "runtime", "otl"});
     log_cfg.set_level("functions", new_cfg.log_level_functions());
     log_cfg.set_level("config", new_cfg.log_level_config());
     log_cfg.set_level("events", new_cfg.log_level_events());
@@ -1185,6 +1186,7 @@ void applier::state::apply_log_config(configuration::state& new_cfg) {
     log_cfg.set_level("macros", new_cfg.log_level_macros());
     log_cfg.set_level("process", new_cfg.log_level_process());
     log_cfg.set_level("runtime", new_cfg.log_level_runtime());
+    log_cfg.set_level("otel", new_cfg.log_level_otl());
     if (has_already_been_loaded)
       log_cfg.allow_only_atomic_changes(true);
     log_v2::instance().apply(log_cfg);

engine/tests/opentelemetry/agent_to_engine_test.cc

@@ -28,6 +28,7 @@
 #include "opentelemetry/proto/collector/metrics/v1/metrics_service.grpc.pb.h"
 #include "opentelemetry/proto/metrics/v1/metrics.pb.h"
 
+#include "com/centreon/engine/contact.hh"
 #include "com/centreon/engine/host.hh"
 #include "com/centreon/engine/service.hh"
 
@@ -77,6 +78,13 @@ class agent_to_engine_test : public TestEngine {
     spdlog::default_logger()->set_level(spdlog::level::trace);
     ::fmt::formatter< ::opentelemetry::proto::collector::metrics::v1::
                           ExportMetricsServiceRequest>::json_grpc_format = true;
+    timeperiod::timeperiods.clear();
+    contact::contacts.clear();
+    host::hosts.clear();
+    host::hosts_by_id.clear();
+    service::services.clear();
+    service::services_by_id.clear();
+
     init_config_state();
 
     configuration::applier::connector conn_aply;
@@ -307,8 +315,10 @@ TEST_F(agent_to_engine_test, server_send_conf_to_agent_and_receive_metrics) {
       serv_2_found = true;
     } else if (compare_to_expected_host_metric(*to_compare)) {
       host_metric_found = true;
-    } else
+    } else {
+      SPDLOG_ERROR("bad resource metric: {}", to_compare->DebugString());
       ASSERT_TRUE(false);
+    }
   }
   ASSERT_TRUE(host_metric_found);
   ASSERT_TRUE(serv_1_found);

engine/tests/test_engine.cc

@@ -141,7 +141,7 @@ configuration::host TestEngine::new_configuration_host(
   hst.parse("contacts", contacts.c_str());
 
   configuration::command cmd("hcmd");
-  cmd.parse("command_line", "echo 0");
+  cmd.parse("command_line", "/bin/echo 0");
   if (!connector.empty()) {
     cmd.parse("connector", connector.data());
   }
@@ -193,7 +193,9 @@ configuration::service TestEngine::new_configuration_service(
     svc.set_host_id(12);
 
   configuration::command cmd(fmt::format("cmd_serv_{}", svc_id));
-  cmd.parse("command_line", "echo -n 'output| metric=$ARG1$;50;75 metric2=30ms;50:75;75:80;0;100'");
+  cmd.parse("command_line",
+            "/bin/echo -n 'output| metric=$ARG1$;50;75 "
+            "metric2=30ms;50:75;75:80;0;100'");
   if (!connector.empty()) {
     cmd.parse("connector", connector.data());
   }