Core: Implementation of periodic message generation

Author: Adrian Duriska

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

@@ -42,6 +42,7 @@ set(CORE_SOURCE
     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";
 
@@ -61,6 +67,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 +96,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 +149,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 +175,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
 }

src/core/configurator/instance_outmgr.cpp

@@ -58,7 +58,6 @@ static const struct ipx_ctx_callbacks output_mgr_callbacks = {
     nullptr  // No feedback
 };
 
-
 ipx_instance_outmgr::ipx_instance_outmgr(uint32_t bsize)
     : ipx_instance_intermediate("Output manager", &output_mgr_callbacks, bsize)
 {

src/core/context.c

@@ -45,6 +45,7 @@
 #include <pthread.h>
 #include <errno.h>
 #include <signal.h>
+#include <sys/time.h>
 #if defined(__OpenBSD__) || defined(__FreeBSD__)
 #include <pthread_np.h>
 #else
@@ -58,6 +59,7 @@
 #include "fpipe.h"
 #include "ring.h"
 #include "message_ipfix.h"
+#include "message_periodic.h"
 #include "configurator/cpipe.h"
 
 /** Identification of this component (for log) */
@@ -191,7 +193,7 @@ ipx_ctx_create(const char *name, const struct ipx_ctx_callbacks *callbacks)
     ctx->cfg_system.vlevel = ipx_verb_level_get();
     ctx->cfg_system.rec_size = IPX_MSG_IPFIX_BASE_REC_SIZE;
     ctx->cfg_system.msg_mask_selected = 0; // No messages to process selected
-    ctx->cfg_system.msg_mask_allowed = IPX_MSG_IPFIX | IPX_MSG_SESSION;
+    ctx->cfg_system.msg_mask_allowed = IPX_MSG_IPFIX | IPX_MSG_SESSION | IPX_MSG_PERIODIC;
     ctx->cfg_system.term_msg_cnt = 1; // By default, wait for 1 termination message
 
     ctx->cfg_extension.items = NULL;
@@ -816,14 +818,14 @@ thread_handle_rc(struct ipx_ctx *ctx, int rc)
         // No more data -> stop the collector
         IPX_CTX_DEBUG(ctx, "The instance has signalized end-of-file/stream.", '\0');
         ipx_ctx_processing_set(ctx, false);
-        ipx_cpipe_send_term(ctx, IPX_CPIPE_TYPE_TERM_SLOW);
+        ipx_cpipe_send(ctx, IPX_CPIPE_TYPE_TERM_SLOW);
         break;
     case IPX_ERR_DENIED:
         // Fatal error -> stop the collector as fast as possible
         IPX_CTX_ERROR(ctx, "ipx_plugin_get()/ipx_plugin_process() failed! The collector cannot "
             "work properly anymore!", '\0');
         ipx_ctx_processing_set(ctx, false);
-        ipx_cpipe_send_term(ctx, IPX_CPIPE_TYPE_TERM_FAST);
+        ipx_cpipe_send(ctx, IPX_CPIPE_TYPE_TERM_FAST);
         break;
     default:
         IPX_CTX_ERROR(ctx, "ipx_plugin_get()/ipx_plugin_process() returned unexpected return "
@@ -876,6 +878,13 @@ thread_input_process_pipe(struct ipx_ctx *ctx)
         return IPX_OK;
     }
 
+    if (msg_type == IPX_MSG_PERIODIC) {
+        ipx_msg_periodic_t *periodic_message = ipx_msg_base2periodic(msg_ptr);
+        ipx_msg_periodic_update_last_processed(periodic_message);
+        ipx_ctx_msg_pass(ctx, msg_ptr);
+        return IPX_OK;
+    }
+
     if (msg_type == IPX_MSG_TERMINATE) {
         // Destroy the instance (usually produce garbage messages, etc)
         const char *plugin_name = ctx->plugin_cbs->info->name;
@@ -954,6 +963,8 @@ thread_intermediate(void *arg)
     ipx_msg_t *msg_ptr;
     enum ipx_msg_type msg_type;
 
+    uint64_t waiting_for_seq = 0;
+
     bool terminate = false;
     while (!terminate) {
         // Get a new message for the buffer
@@ -978,6 +989,16 @@ thread_intermediate(void *arg)
             }
         }
 
+        if (msg_type == IPX_MSG_PERIODIC) {
+            ipx_msg_periodic_t *periodic_message = ipx_msg_base2periodic(msg_ptr);
+            if (ipx_msg_periodic_get_seq_num(periodic_message) != waiting_for_seq) {
+                ipx_msg_periodic_destroy(periodic_message);
+                continue;
+            }
+            waiting_for_seq++;
+            ipx_msg_periodic_update_last_processed(periodic_message);
+        }
+
         if (!ipx_ctx_processing_get(ctx)
                 && (msg_type == IPX_MSG_IPFIX || msg_type == IPX_MSG_SESSION)) {
             // Data processing is disabled -> drop IPFIX and Session messages
@@ -1048,6 +1069,11 @@ thread_output(void *arg)
             thread_handle_rc(ctx, rc);
         }
 
+        if (msg_type == IPX_MSG_PERIODIC) {
+            ipx_msg_periodic_t *periodic_message = ipx_msg_base2periodic(msg_ptr);
+            ipx_msg_periodic_update_last_processed(periodic_message);
+        }
+
         if (msg_type == IPX_MSG_TERMINATE) {
             ipx_msg_terminate_t *terminate_msg = ipx_msg_base2terminate(msg_ptr);
             enum ipx_msg_terminate_type type = ipx_msg_terminate_get_type(terminate_msg);

src/core/message_base.c

@@ -42,6 +42,7 @@
 #include <ipfixcol2.h>
 #include "message_base.h"
 #include "message_terminate.h"
+#include "message_periodic.h"
 
 // Get the type of a message for the collector pipeline
 enum ipx_msg_type

src/core/message_terminate.c

@@ -77,7 +77,7 @@ void
 ipx_msg_terminate_destroy(ipx_msg_terminate_t *msg)
 {
     ipx_msg_header_destroy((ipx_msg_t *) msg);
-    ipx_cpipe_send_term(NULL, IPX_CPIPE_TYPE_TERM_DONE);
+    ipx_cpipe_send(NULL, IPX_CPIPE_TYPE_TERM_DONE);
     free(msg);
 }