Merge pull request #21 from CESNET/hutak-extensions

Data Record extensions

Author: Lukas955

CMakeLists.txt

@@ -14,7 +14,7 @@ endif()
 
 # Versions and other informations
 set(IPFIXCOL_VERSION_MAJOR 2)
-set(IPFIXCOL_VERSION_MINOR 1)
+set(IPFIXCOL_VERSION_MINOR 2)
 set(IPFIXCOL_VERSION_PATCH 0)
 set(IPFIXCOL_VERSION
     ${IPFIXCOL_VERSION_MAJOR}.${IPFIXCOL_VERSION_MINOR}.${IPFIXCOL_VERSION_PATCH})

include/ipfixcol2/message_ipfix.h

@@ -51,7 +51,6 @@ extern "C" {
 
 #include <ipfixcol2/api.h>
 #include "session.h"
-#include "plugins.h"
 
 /**
  * \defgroup ipxMsgIPFIX IPFIX message
@@ -99,12 +98,17 @@ struct ipx_ipfix_set {
  * \brief Data record (record + extensions)
  */
 struct ipx_ipfix_record {
-    /** Data record information                                               */
+    /** Data record information                                                 */
     struct fds_drec rec;
-    /** Start of reserved space for registered extensions (filled by plugins) */
+
+    /** Bit mask of filled extensions (set by producers)                        */
+    uint64_t ext_mask;
+    /** Start of reserved space for registered extensions (filled by producers) */
     uint8_t ext[1];
 };
 
+#include "plugins.h"
+
 /**
  * \brief Create an empty wrapper around IPFIX (or NetFlow) Message
  *

include/ipfixcol2/plugins.h

@@ -2,10 +2,10 @@
  * \file include/ipfixcol2/plugins.h
  * \author Lukas Hutak <[email protected]>
  * \brief IPFIXcol plugin interface and functions (header file)
- * \date 2018
+ * \date 2018-2020
  */
 
-/* Copyright (C) 2018 CESNET, z.s.p.o.
+/* Copyright (C) 2018-2020 CESNET, z.s.p.o.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions
@@ -49,12 +49,13 @@ extern "C" {
 /** Internal plugin context                                                                    */
 typedef struct ipx_ctx ipx_ctx_t;
 /** Internal data structure that represents IPFIXcol record extension                          */
-typedef struct ipx_ctx_rext ipx_ctx_rext_t;
+typedef struct ipx_ctx_ext ipx_ctx_ext_t;
 
 #include <ipfixcol2/api.h>
 #include <ipfixcol2/verbose.h>
 #include <ipfixcol2/session.h>
 #include <ipfixcol2/message.h>
+#include <ipfixcol2/message_ipfix.h>
 
 /**
  * \defgroup pluginAPI Plugin API
@@ -396,6 +397,101 @@ ipx_ctx_subscribe(ipx_ctx_t *ctx, const ipx_msg_mask_t *mask_new, ipx_msg_mask_t
 IPX_API const fds_iemgr_t *
 ipx_ctx_iemgr_get(ipx_ctx_t *ctx);
 
+/**
+ * \brief Register an extension of Data Records (Intermediate plugins ONLY!)
+ *
+ * Reserve space for metadata that will be part of each Data Record. The purpose of extension
+ * it is to add non-flow information which can be useful during record processing. For example,
+ * one plugin can add some labels and one or more plugins further in the pipeline can use them
+ * later.
+ *
+ * Structure or data type of the extension is up to the producer. Nevertheless, the producer and
+ * all consumers must use the same. The producer is also RESPONSIBLE for filling content of the
+ * extension to EACH Data Record in the IPFIX Message! After filling the extension, function
+ * ipx_ctx_ext_set_filled() must be called to mark the extension memory as filled. Otherwise,
+ * consumers are not able get its content.
+ *
+ * One plugin instance can register multiple extensions.
+ * \warning
+ *   This function can be called only during ipx_plugin_init() of Intermediate plugins.
+ * \note
+ *   Only single plugin instance at time can produce extension with the given combination
+ *   of the \p type and the \p name.
+ * \param[in]  ctx  Plugin context
+ * \param[in]  type Identification of the extension type (e.g. "profiles-v1")
+ * \param[in]  name Identification of the extension name (e.g. "main_profiles")
+ * \param[in]  size Non-zero size of memory required for the extension (in bytes)
+ * \param[out] ext  Internal description of the extension
+ *
+ * \return #IPX_OK on success
+ * \return #IPX_ERR_ARG if the \p type or \p name are not valid (i.e. empty or NULL) or \p size
+ *   is zero.
+ * \return #IPX_ERR_DENIED if the plugin doesn't have permission to register extension
+ * \return #IPX_ERR_EXISTS if the extension or dependency has been already registered by this plugin
+ * \return #IPX_ERR_NOMEM if a memory allocation failure has occurred
+ */
+IPX_API int
+ipx_ctx_ext_producer(ipx_ctx_t *ctx, const char *type, const char *name, size_t size,
+    ipx_ctx_ext_t **ext);
+
+/**
+ * @brief Add dependency on an extension of Data Records (Intermediate and Output plugins ONLY!)
+ *
+ * Register dependency on an extension. This will make sure that the required extension is
+ * available for EACH Data Record during ipx_plugin_process() and that there is a particular
+ * producer earlier in the processing pipeline.
+ *
+ * One plugin instance can register multiple dependencies.
+ *
+ * \warning
+ *   This function can be called only during ipx_plugin_init() of Intermediate and Output plugins.
+ * \note
+ *   If the function has succeeded, it doesn't mean that there is particular extension producer.
+ *   Since dependencies are resolved later during configuration of the collector, startup
+ *   process will be interrupted if all requirements are not met.
+ * \note
+ *   The plugin instance CANNOT add dependency on an extension which it is producing.
+ *
+ * \param[in]  ctx  Plugin context
+ * \param[in]  type Identification of the extension type (e.g. "profiles-v1")
+ * \param[in]  name Identification of the extension name (e.g. "main_profiles")
+ * \param[out] ext  Internal description of the extension
+ *
+ * \return #IPX_OK on success (see notes)
+ * \return #IPX_ERR_ARG if the \p type or \p name are not valid (i.e. empty or NULL)
+ * \return #IPX_ERR_DENIED if the plugin doesn't have permission to register dependency
+ * \return #IPX_ERR_EXISTS if the dependency or extension has been already registered by this plugin
+ * \return #IPX_ERR_NOMEM if a memory allocation failure has occurred
+ */
+IPX_API int
+ipx_ctx_ext_consumer(ipx_ctx_t *ctx, const char *type, const char *name, ipx_ctx_ext_t **ext);
+
+/**
+ * \brief Get an extension
+ *
+ * In case of a producer of the extension, it always returns #IPX_OK and fills the pointer and
+ * size. If the producer decides to fill the extension, it also must call ipx_ctx_ext_set_filled().
+ * Otherwise, consumers will not be able to get its content.
+ *
+ * \param[in]  ext  Internal description of the extension
+ * \param[in]  drec Data Record with extensions
+ * \param[out] data Pointer to extension data
+ * \param[out] size Size of the extensions (bytes)
+ * \return #IPX_OK on success
+ * \return #IPX_ERR_NOTFOUND if the extension hasn't been filled by its producer
+ */
+IPX_API int
+ipx_ctx_ext_get(ipx_ctx_ext_t *ext, struct ipx_ipfix_record *drec, void **data, size_t *size);
+
+/**
+ * \brief Set the extension of a Data Record as filled (ONLY for the producer of the extension)
+ *
+ * \param[in] ext  Internal description of the extension
+ * \param[in] drec Data Record with extensions
+ */
+IPX_API void
+ipx_ctx_ext_set_filled(ipx_ctx_ext_t *ext, struct ipx_ipfix_record *drec);
+
 /**
  * @}
  * @}

pkg/deb/templates/changelog

@@ -1,5 +1,5 @@
-ipfixcol2 (2.1.0-1) unstable; urgency=low
+ipfixcol2 (2.2.0-1) unstable; urgency=low
 
   * Initial release.
 
- -- Lukas Hutak <[email protected]>  Thu, 25 Oct 2018 10:17:56 +0200
+ -- Lukas Hutak <[email protected]>  Sun, 23 Feb 2020 14:42:00 +0100

src/core/CMakeLists.txt

@@ -4,6 +4,8 @@ set(CORE_SOURCE
     configurator/config_file.hpp
     configurator/configurator.cpp
     configurator/configurator.hpp
+    configurator/extensions.cpp
+    configurator/extensions.hpp
     configurator/instance.hpp
     configurator/instance_input.cpp
     configurator/instance_input.hpp
@@ -28,6 +30,8 @@ set(CORE_SOURCE
     api.c
     context.c
     context.h
+    extension.c
+    extension.h
     fpipe.c
     fpipe.h
     message_base.c

src/core/configurator/configurator.cpp

@@ -46,6 +46,7 @@
 #include <signal.h>
 
 #include "configurator.hpp"
+#include "extensions.hpp"
 
 extern "C" {
 #include "../message_terminate.h"
@@ -249,16 +250,40 @@ ipx_configurator::start(const ipx_config_model &model)
 
     IPX_DEBUG(comp_str, "All instances have been successfully initialized.", '\0');
 
-    // Phase 4. Start threads of all plugins
+    // Phase 4. Register and resolved Data Record extensions and dependencies
+    ipx_cfg_extensions ext_mgr;
+    size_t pos = 0; // Position of an instance in the collector pipeline
+
+    for (auto &input : inputs) {
+        input->extensions_register(&ext_mgr, pos);
+    }
+
+    pos++;
+    for (auto &inter : inters) {
+        inter->extensions_register(&ext_mgr, pos);
+        pos++;
+    }
+
+    for (auto &output : outputs) {
+        output->extensions_register(&ext_mgr, pos);
+    }
+
+    ext_mgr.resolve();
+    ext_mgr.list_extensions();
+
+    // Phase 5. Start threads of all plugins and update definitions of extensions
     for (auto &output : outputs) {
+        output->extensions_resolve(&ext_mgr);
         output->start();
     }
 
     for (auto &inter : inters) {
+        inter->extensions_resolve(&ext_mgr);
         inter->start();
     }
 
     for (auto &input : inputs) {
+        input->extensions_resolve(&ext_mgr);
         input->start();
     }