Merge branch 'devel' into 'master'

merge devel

See merge request monitoring/ipfixcol2!23

Author: sedmicha

.gitignore

@@ -10,6 +10,9 @@ cmake-build-release/
 # JetBrains files
 .idea/
 
+# Clangd cache files
+.cache/
+
 # Prerequisites
 *.d
 

include/CMakeLists.txt

@@ -14,6 +14,7 @@ set(SUB_HEADERS
     ipfixcol2/message_garbage.h
     ipfixcol2/message_ipfix.h
     ipfixcol2/message_session.h
+    ipfixcol2/message_periodic.h
     ipfixcol2/plugins.h
     ipfixcol2/session.h
     ipfixcol2/utils.h

include/ipfixcol2.h

@@ -60,6 +60,7 @@
 #include <ipfixcol2/message_garbage.h>
 #include <ipfixcol2/message_session.h>
 #include <ipfixcol2/message_ipfix.h>
+#include <ipfixcol2/message_periodic.h>
 
 #include <ipfixcol2/plugins.h>
 #include <ipfixcol2/session.h>

include/ipfixcol2/message.h

@@ -77,6 +77,8 @@ enum ipx_msg_type {
     IPX_MSG_TERMINATE = (1 << 3),
     // An internal configuration message (only for internal usage)
     //IPX_MSG_CONFIG  = (1 << 4)
+    /** A periodic message                                                          */
+    IPX_MSG_PERIODIC = (1 << 5)
 };
 
 /** The data type of the base message                                               */
@@ -107,6 +109,7 @@ ipx_msg_destroy(ipx_msg_t *msg);
 #include <ipfixcol2/message_ipfix.h>
 #include <ipfixcol2/message_garbage.h>
 #include <ipfixcol2/message_session.h>
+#include <ipfixcol2/message_periodic.h>
 
 /**
  * \brief Cast from a base message to a session message
@@ -147,6 +150,19 @@ ipx_msg_base2ipfix(ipx_msg_t *msg)
     return (ipx_msg_ipfix_t *) msg;
 }
 
+/**
+ * \brief Cast base message to periodic message
+ *
+ * \param[in] msg   Pointer to the base message
+ * \return A pointer to a periodic message
+ */
+static inline ipx_msg_periodic_t *
+ipx_msg_base2periodic(ipx_msg_t *msg)
+{
+    assert(ipx_msg_get_type(msg) == IPX_MSG_PERIODIC);
+    return (ipx_msg_periodic_t *) msg;
+}
+
 /**@}*/
 
 #ifdef __cplusplus

include/ipfixcol2/message_ipfix.h

@@ -225,6 +225,15 @@ ipx_msg_ipfix_add_set_ref(struct ipx_msg_ipfix *msg);
 IPX_API struct ipx_ipfix_record *
 ipx_msg_ipfix_add_drec_ref(struct ipx_msg_ipfix **msg_ref);
 
+/**
+ * \brief Set the raw size of an IPFIX message.
+ *
+ * \param[in] msg IPFIX Message wrapper.
+ * \param[in] new_raw_size The new raw_size value.
+ */
+IPX_API void
+ipx_msg_ipfix_set_raw_size(ipx_msg_ipfix_t *msg, uint16_t new_raw_size);
+
 /**@}*/
 #ifdef __cplusplus
 }

include/ipfixcol2/message_periodic.h

@@ -0,0 +1,63 @@
+/**
+ * @file
+ * @author Adrian Duriska <[email protected]>
+ * @brief Periodic message
+ *
+ * Copyright: (C) 2024 CESNET, z.s.p.o.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#ifndef IPX_MESSAGE_PERIODIC_H
+#define IPX_MESSAGE_PERIODIC_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <ipfixcol2/message.h>
+
+typedef struct ipx_msg_periodic ipx_msg_periodic_t;
+
+/**
+ * \brief Cast periodic message to base message
+ *
+ * \param[in] msg   Pointer to the periodic message
+ * \return A pointer to a base message
+ */
+static inline ipx_msg_t *
+ipx_msg_periodic2base(ipx_msg_periodic_t *msg)
+{
+    return (ipx_msg_t *) msg;
+}
+
+/**
+ * \brief Get sequence number of periodic message
+ *
+ * \param[in] msg   Pointer to the periodic message
+ * \return Periodic messages sequence number
+ */
+IPX_API uint64_t
+ipx_msg_periodic_get_seq_num(ipx_msg_periodic_t *msg);
+
+/**
+ * \brief Get created time of periodic message
+ *
+ * \param[in] msg   Pointer to the periodic message
+ * \return Periodic messages created time
+ */
+IPX_API struct timespec
+ipx_msg_periodic_get_created(ipx_msg_periodic_t *msg);
+
+/**
+ * \brief Get last intermediate plugin processed time of periodic message
+ *
+ * \param[in] msg   Pointer to the periodic message
+ * \return Periodic messages last intermediate plugin processed time
+ */
+IPX_API struct timespec
+ipx_msg_periodic_get_last_processed(ipx_msg_periodic_t *msg);
+
+#ifdef __cplusplus
+}
+#endif
+#endif

include/ipfixcol2/plugins.h

@@ -381,6 +381,7 @@ ipx_ctx_msg_pass(ipx_ctx_t *ctx, ipx_msg_t *msg);
  * types of messages:
  * - ::IPX_MSG_IPFIX (IPFIX Message)
  * - ::IPX_MSG_SESSION (Transport Session Message)
+ * - ::IPX_MSG_PERIODIC (Periodic Message)
  *
  * If \p mask_new is non-NULL, the new subscription mask is installed from \p mask_new.
  * If \p mask_old is non-NULL, the previous mask is saved in \p mask_old.

src/core/CMakeLists.txt

@@ -41,6 +41,8 @@ set(CORE_SOURCE
     message_garbage.c
     message_ipfix.c
     message_ipfix.h
+    message_periodic.c
+    message_periodic.h
     message_session.c
     message_terminate.c
     message_terminate.h

src/core/configurator/configurator.cpp

@@ -46,12 +46,14 @@
 #include <cstdlib>
 #include <dlfcn.h>
 #include <signal.h>
+#include <sys/time.h>
 
 #include "configurator.hpp"
 #include "extensions.hpp"
 
 extern "C" {
 #include "../message_terminate.h"
+#include "../message_periodic.h"
 #include "../plugin_parser.h"
 #include "../plugin_output_mgr.h"
 #include "../verbose.h"
@@ -88,7 +90,7 @@ termination_handler(int sig)
     cnt++;
 
     // Send a termination request to the configurator
-    int rc = ipx_cpipe_send_term(NULL, request_type);
+    int rc = ipx_cpipe_send(NULL, request_type);
     if (rc != IPX_OK) {
         static const char *msg = "ERROR: Signal handler: failed to send a termination request";
         write(STDOUT_FILENO, msg, strlen(msg));
@@ -568,6 +570,22 @@ ipx_configurator::termination_send_msg()
     m_term_sent = m_running_inputs.size();
 }
 
+void
+ipx_configurator::periodic_send_msg(uint32_t *periodic_message_sequence)
+{
+    for (auto &input : m_running_inputs) {
+        ipx_msg_periodic_t *msg = ipx_msg_periodic_create(*periodic_message_sequence);
+        if (!msg) {
+            IPX_ERROR(comp_str, "Can't create periodic message!", '\0');
+            termination_send_msg();
+            return;
+        }
+
+        ipx_fpipe_write(input->get_feedback(), ipx_msg_periodic2base(msg));
+    }
+    (*periodic_message_sequence)++;
+}
+
 int
 ipx_configurator::run(ipx_controller *ctrl)
 {
@@ -595,6 +613,9 @@ ipx_configurator::run(ipx_controller *ctrl)
     // Collector is running -> process termination/reconfiguration requests
     m_state = STATUS::RUNNING;
     bool terminate = false;
+
+    uint32_t periodic_message_sequence = 0;
+
     while (!terminate) {
         struct ipx_cpipe_req req;
         if (ipx_cpipe_receive(&req) != IPX_OK) {
@@ -609,6 +630,9 @@ ipx_configurator::run(ipx_controller *ctrl)
         case IPX_CPIPE_TYPE_TERM_DONE:
             terminate = termination_handle(req, ctrl);
             break;
+        case IPX_CPIPE_TYPE_PERIODIC:
+            periodic_send_msg(&periodic_message_sequence);
+            break;
         default:
             IPX_ERROR(comp_str, "Ignoring unknown configuration request!", '\0');
             continue;

src/core/configurator/configurator.hpp

@@ -150,6 +150,9 @@ class ipx_configurator {
     void
     termination_send_msg();
 
+    void
+    periodic_send_msg(uint32_t *periodic_message_sequence);
+
     void
     termination_stop_all();
     void