MON-63843-agent-linux-streaming

agent/CMakeLists.txt

@@ -101,6 +101,7 @@ add_custom_command(
 add_library(centagent_lib STATIC
     ${SRC_DIR}/agent.grpc.pb.cc 
     ${SRC_DIR}/agent.pb.cc 
+    ${SRC_DIR}/bireactor.cc
     ${SRC_DIR}/check.cc
     ${SRC_DIR}/check_exec.cc
     ${SRC_DIR}/opentelemetry/proto/collector/metrics/v1/metrics_service.grpc.pb.cc
@@ -110,6 +111,8 @@ add_library(centagent_lib STATIC
     ${SRC_DIR}/opentelemetry/proto/resource/v1/resource.pb.cc
     ${SRC_DIR}/config.cc
     ${SRC_DIR}/scheduler.cc
+    ${SRC_DIR}/streaming_client.cc
+    ${SRC_DIR}/streaming_server.cc
 )
 
 include_directories(
@@ -136,7 +139,9 @@ target_link_libraries(
   centreon_process
   -L${Boost_LIBRARY_DIR_RELEASE}
   boost_program_options
-  fmt::fmt)
+  fmt::fmt
+  stdc++fs
+  )
 
 target_precompile_headers(${CENTREON_AGENT}  REUSE_FROM centagent_lib)
 

agent/inc/com/centreon/agent/bireactor.hh

@@ -0,0 +1,84 @@
+/**
+ * Copyright 2024 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]
+ */
+
+#ifndef CENTREON_AGENT_BIREACTOR_HH
+#define CENTREON_AGENT_BIREACTOR_HH
+
+#include "agent.grpc.pb.h"
+
+namespace com::centreon::agent {
+
+template <class bireactor_class>
+class bireactor
+    : public bireactor_class,
+      public std::enable_shared_from_this<bireactor<bireactor_class>> {
+ private:
+  static std::set<std::shared_ptr<bireactor>> _instances;
+  static std::mutex _instances_m;
+
+  bool _write_pending;
+  std::deque<std::shared_ptr<MessageFromAgent>> _write_queue;
+  std::shared_ptr<MessageToAgent> _read_current;
+
+  const std::string_view _class_name;
+
+  const std::string _peer;
+
+ protected:
+  std::shared_ptr<boost::asio::io_context> _io_context;
+  std::shared_ptr<spdlog::logger> _logger;
+
+  bool _alive;
+  mutable std::mutex _protect;
+
+ public:
+  bireactor(const std::shared_ptr<boost::asio::io_context>& io_context,
+            const std::shared_ptr<spdlog::logger>& logger,
+            const std::string_view& class_name,
+            const std::string& peer);
+
+  virtual ~bireactor();
+
+  static void register_stream(const std::shared_ptr<bireactor>& strm);
+
+  void start_read();
+
+  void start_write();
+  void write(const std::shared_ptr<MessageFromAgent>& request);
+
+  // bireactor part
+  void OnReadDone(bool ok) override;
+
+  virtual void on_incomming_request(
+      const std::shared_ptr<MessageToAgent>& request) = 0;
+
+  virtual void on_error() = 0;
+
+  void OnWriteDone(bool ok) override;
+
+  // server version
+  void OnDone();
+  // client version
+  void OnDone(const ::grpc::Status& /*s*/);
+
+  virtual void shutdown();
+};
+
+}  // namespace com::centreon::agent
+
+#endif

agent/inc/com/centreon/agent/streaming_client.hh

@@ -0,0 +1,109 @@
+/**
+ * Copyright 2024 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]
+ */
+
+#ifndef CENTREON_AGENT_STREAMING_CLIENT_HH
+#define CENTREON_AGENT_STREAMING_CLIENT_HH
+
+#include "com/centreon/common/grpc/grpc_client.hh"
+
+#include "bireactor.hh"
+#include "scheduler.hh"
+
+namespace com::centreon::agent {
+
+class streaming_client;
+
+class client_reactor
+    : public bireactor<
+          ::grpc::ClientBidiReactor<MessageFromAgent, MessageToAgent>> {
+  std::weak_ptr<streaming_client> _parent;
+  ::grpc::ClientContext _context;
+
+ public:
+  client_reactor(const std::shared_ptr<boost::asio::io_context>& io_context,
+                 const std::shared_ptr<spdlog::logger>& logger,
+                 const std::shared_ptr<streaming_client>& parent,
+                 const std::string& peer);
+
+  std::shared_ptr<client_reactor> shared_from_this() {
+    return std::static_pointer_cast<client_reactor>(
+        bireactor<::grpc::ClientBidiReactor<MessageFromAgent, MessageToAgent>>::
+            shared_from_this());
+  }
+
+  ::grpc::ClientContext& get_context() { return _context; }
+
+  void on_incomming_request(
+      const std::shared_ptr<MessageToAgent>& request) override;
+
+  void on_error() override;
+
+  void shutdown() override;
+};
+
+/**
+ * @brief this object not only manages connection to engine, but also embed
+ * check scheduler
+ *
+ */
+class streaming_client : public common::grpc::grpc_client_base,
+                         public std::enable_shared_from_this<streaming_client> {
+  std::shared_ptr<boost::asio::io_context> _io_context;
+  std::shared_ptr<spdlog::logger> _logger;
+  std::string _supervised_host;
+
+  std::unique_ptr<AgentService::Stub> _stub;
+
+  std::shared_ptr<client_reactor> _reactor;
+  std::shared_ptr<scheduler> _sched;
+
+  std::mutex _protect;
+
+  void _create_reactor();
+
+  void _start();
+
+  void _send(const std::shared_ptr<MessageFromAgent>& request);
+
+ public:
+  streaming_client(const std::shared_ptr<boost::asio::io_context>& io_context,
+                   const std::shared_ptr<spdlog::logger>& logger,
+                   const std::shared_ptr<common::grpc::grpc_config>& conf,
+                   const std::string& supervised_host);
+
+  static std::shared_ptr<streaming_client> load(
+      const std::shared_ptr<boost::asio::io_context>& io_context,
+      const std::shared_ptr<spdlog::logger>& logger,
+      const std::shared_ptr<common::grpc::grpc_config>& conf,
+      const std::string& supervised_host);
+
+  void on_incomming_request(const std::shared_ptr<client_reactor>& caller,
+                            const std::shared_ptr<MessageToAgent>& request);
+  void on_error(const std::shared_ptr<client_reactor>& caller);
+
+  void shutdown();
+
+  // use only for tests
+  engine_to_agent_request_ptr get_last_message_to_agent() const {
+    return _sched->get_last_message_to_agent();
+  }
+};
+
+}  // namespace com::centreon::agent
+
+#endif

agent/inc/com/centreon/agent/streaming_server.hh

@@ -0,0 +1,73 @@
+/**
+ * Copyright 2024 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]
+ */
+
+#ifndef CENTREON_AGENT_STREAMING_SERVER_HH
+#define CENTREON_AGENT_STREAMING_SERVER_HH
+
+#include "com/centreon/common/grpc/grpc_server.hh"
+
+#include "bireactor.hh"
+#include "scheduler.hh"
+
+namespace com::centreon::agent {
+
+class server_reactor;
+
+/**
+ * @brief grpc engine to agent server (reverse connection)
+ * It accept only one connection at a time
+ * If another connection occurs, previous connection is shutdown
+ * This object is both grpc server and grpc service
+ */
+class streaming_server : public common::grpc::grpc_server_base,
+                         public std::enable_shared_from_this<streaming_server>,
+                         public ReversedAgentService::Service {
+  std::shared_ptr<boost::asio::io_context> _io_context;
+  std::shared_ptr<spdlog::logger> _logger;
+  std::string _supervised_host;
+
+  /** active engine to agent connection*/
+  std::shared_ptr<server_reactor> _incoming;
+
+  mutable std::mutex _protect;
+
+  void _start();
+
+ public:
+  streaming_server(const std::shared_ptr<boost::asio::io_context>& io_context,
+                   const std::shared_ptr<spdlog::logger>& logger,
+                   const std::shared_ptr<common::grpc::grpc_config>& conf,
+                   const std::string& supervised_host);
+
+  ~streaming_server();
+
+  static std::shared_ptr<streaming_server> load(
+      const std::shared_ptr<boost::asio::io_context>& io_context,
+      const std::shared_ptr<spdlog::logger>& logger,
+      const std::shared_ptr<common::grpc::grpc_config>& conf,
+      const std::string& supervised_host);
+
+  ::grpc::ServerBidiReactor<MessageToAgent, MessageFromAgent>* Import(
+      ::grpc::CallbackServerContext* context);
+
+  void shutdown();
+};
+
+}  // namespace com::centreon::agent
+
+#endif