Merge branch 'periodic-message' into 'devel'

Periodic message

See merge request monitoring/ipfixcol2!15

Author: sedmicha

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_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

src/core/configurator/cpipe.c

@@ -14,16 +14,22 @@
 #include <fcntl.h>
 #include <errno.h>
 #include <limits.h> // PIPE_BUF
+#include <sys/timerfd.h>
+#include <poll.h>
 
 #include <ipfixcol2.h>
 
 #include "cpipe.h"
 #include "../verbose.h"
 
+/// Periodic message delay in milliseconds
+#define IPX_PERIODIC_MESSAGE_DELAY 100
 /// Invalid file descriptor value
 #define INVALID_FD (-1)
 /// Configuration pipe - cpipe_fd[0] for read, cpipe_fd[1] for write
 static int cpipe_fd[2] = {INVALID_FD, INVALID_FD};
+/// Periodic timer file descriptor
+static int periodic_timer_fd = INVALID_FD;
 /// Identification of the module (just for log!)
 static const char *module = "Configuration pipe";
 
@@ -33,7 +39,8 @@ static_assert(sizeof(struct ipx_cpipe_req) <= PIPE_BUF, "non-atomic write!");
 int
 ipx_cpipe_init()
 {
-    assert(cpipe_fd[0] == INVALID_FD && cpipe_fd[1] == INVALID_FD && "Already initialized!");
+    assert(cpipe_fd[0] == INVALID_FD && cpipe_fd[1] == INVALID_FD && cpipe_fd[2] == INVALID_FD
+        && "Already initialized!");
     int rc;
     const char *err_str;
 
@@ -61,6 +68,21 @@ ipx_cpipe_init()
         return IPX_ERR_DENIED;
     }
 
+    periodic_timer_fd = timerfd_create(CLOCK_MONOTONIC, 0);
+    if (periodic_timer_fd == -1) {
+        ipx_strerror(errno, err_str);
+        IPX_ERROR(module, "timerfd_create() failed: %s", err_str);
+        ipx_cpipe_destroy();
+        return IPX_ERR_DENIED;
+    }
+    struct itimerspec timer_settings;
+    struct timespec interval;
+    interval.tv_nsec = (IPX_PERIODIC_MESSAGE_DELAY % 1000) * 1000000;
+    interval.tv_sec = IPX_PERIODIC_MESSAGE_DELAY / 1000;
+    timer_settings.it_interval = interval;
+    timer_settings.it_value = interval;
+    timerfd_settime(periodic_timer_fd, 0, &timer_settings, NULL);
+
     return IPX_OK;
 }
 
@@ -75,16 +97,31 @@ ipx_cpipe_destroy()
         close(cpipe_fd[i]);
         cpipe_fd[i] = INVALID_FD;
     }
+    if (periodic_timer_fd != INVALID_FD) {
+        close(periodic_timer_fd);
+        periodic_timer_fd = INVALID_FD;
+    }
 }
 
 int
 ipx_cpipe_receive(struct ipx_cpipe_req *msg)
 {
     const size_t buffer_size = sizeof(*msg);
     size_t buffer_read = 0;
+    uint64_t periodic_timer_buf;
+
+    struct pollfd fds;
+    fds.fd = periodic_timer_fd;
+    fds.events = POLLIN;
 
     errno = 0;
     while (buffer_read < buffer_size) {
+        if (poll(&fds, 1, 0) != -1) {
+            // Reset timer expiration counter
+            if (read(periodic_timer_fd, &periodic_timer_buf, sizeof(periodic_timer_buf)) != -1) {
+                ipx_cpipe_send(NULL, IPX_CPIPE_TYPE_PERIODIC);
+            }
+        }
         uint8_t *ptr = ((uint8_t *) msg) + buffer_read;
         ssize_t rc = read(cpipe_fd[0], ptr, buffer_size - buffer_read);
         if (rc > 0) {
@@ -113,16 +150,15 @@ ipx_cpipe_receive(struct ipx_cpipe_req *msg)
 }
 
 int
-ipx_cpipe_send_term(ipx_ctx_t *ctx, enum ipx_cpipe_type type)
+ipx_cpipe_send(ipx_ctx_t *ctx, enum ipx_cpipe_type type)
 {
     // WARNING: Keep on mind that this function can be called from signal handler!
 
     // In case we change 'errno' (e.g. write())
     int errno_backup = errno;
 
-    if (type != IPX_CPIPE_TYPE_TERM_SLOW
-            && type != IPX_CPIPE_TYPE_TERM_FAST
-            && type != IPX_CPIPE_TYPE_TERM_DONE) {
+    if (type != IPX_CPIPE_TYPE_TERM_SLOW && type != IPX_CPIPE_TYPE_TERM_FAST
+        && type != IPX_CPIPE_TYPE_TERM_DONE && type != IPX_CPIPE_TYPE_PERIODIC) {
         return IPX_ERR_ARG;
     }
 
@@ -140,4 +176,3 @@ ipx_cpipe_send_term(ipx_ctx_t *ctx, enum ipx_cpipe_type type)
     errno = errno_backup;
     return (rc == -1) ? IPX_ERR_DENIED : IPX_OK;
 }
-

src/core/configurator/cpipe.h

@@ -83,11 +83,13 @@ enum ipx_cpipe_type {
      *
      * Sending this request before termination of all plugin instances is considered as fatal.
      */
-    IPX_CPIPE_TYPE_TERM_DONE        ///< Terminate request - complete
+    IPX_CPIPE_TYPE_TERM_DONE,        ///< Terminate request - complete
 
     // Proposed types for the future runtime reconfiguration
     // IPX_CPIPE_TYPE_RECONF_START,
     // IPX_CPIPE_TYPE_RECONF_DONE
+
+    IPX_CPIPE_TYPE_PERIODIC
 };
 
 /// Configuration request
@@ -152,7 +154,7 @@ ipx_cpipe_receive(struct ipx_cpipe_req *msg);
  * @return #IPX_ERR_DENIED if the request failed to be sent
  */
 int
-ipx_cpipe_send_term(ipx_ctx_t *ctx, enum ipx_cpipe_type type);
+ipx_cpipe_send(ipx_ctx_t *ctx, enum ipx_cpipe_type type);
 
 #ifdef __cplusplus
 }