Core: add support for Data Record extension [WIP]

Author: Lukas955

include/ipfixcol2/message_ipfix.h

@@ -98,9 +98,12 @@ 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];
 };
 

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();
     }
 

src/core/configurator/extensions.cpp

@@ -1,21 +1,81 @@
-
+/**
+ * @file src/code/configurator/extensions.cpp
+ * @author Lukas Hutak ([email protected])
+ * @brief Manager of Data Record extensions
+ * @date February 2020
+ */
+
+/* Copyright (C) 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
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in
+ *    the documentation and/or other materials provided with the
+ *    distribution.
+ * 3. Neither the name of the Company nor the names of its contributors
+ *    may be used to endorse or promote products derived from this
+ *    software without specific prior written permission.
+ *
+ * ALTERNATIVELY, provided that this notice is retained in full, this
+ * product may be distributed under the terms of the GNU General Public
+ * License (GPL) version 2 or later, in which case the provisions
+ * of the GPL apply INSTEAD OF those given above.
+ *
+ * This software is provided ``as is'', and any express or implied
+ * warranties, including, but not limited to, the implied warranties of
+ * merchantability and fitness for a particular purpose are disclaimed.
+ * In no event shall the company or contributors be liable for any
+ * direct, indirect, incidental, special, exemplary, or consequential
+ * damages (including, but not limited to, procurement of substitute
+ * goods or services; loss of use, data, or profits; or business
+ * interruption) however caused and on any theory of liability, whether
+ * in contract, strict liability, or tort (including negligence or
+ * otherwise) arising in any way out of the use of this software, even
+ * if advised of the possibility of such damage.
+ *
+ */
 
 #include <algorithm>
 #include "extensions.hpp"
 
-/** Identification of this component (for log) */
-static const char *comp_str = "Extensions";
+extern "C" {
+#include "../message_ipfix.h"
+#include "../verbose.h"
+}
 
+/** Identification of this component (for log) */
+static const char *comp_str = "Configurator (extensions)";
+
+/**
+ * \brief Register extension producer or dependency
+ *
+ * \param[in] name Name of the plugin instance
+ * \param[in] pos  Position of the instance in the pipeline
+ * \param[in] ext  Reference to extension definition of the plugin (read only)
+ */
 void
-ipx_cfg_extensions::add_extension(ipx_instance *inst, unsigned int idx, struct ipx_ctx_ext *ext)
+ipx_cfg_extensions::add_extension(const char *name, unsigned int pos, const struct ipx_ctx_ext *ext)
 {
     auto ext_info = &m_extensions[ext->data_type][ext->data_name];
     std::vector<struct plugin_rec> &vec_plugins = (ext->etype == IPX_EXTENSION_PRODUCER)
         ? ext_info->producers
         : ext_info->consumers;
-    vec_plugins.push_back({inst->get_name(), idx, ext});
+    vec_plugins.push_back({name, pos, ext});
 }
 
+/**
+ * @brief Check extension definition
+ *
+ * The function checks if there is exactly one producer and that the producer is placed
+ * before all consumers in the collector pipeline.
+ * @param[in] ident Identification of the extension (for log only)
+ * @param[in] rec   Internal extension record
+ * @throw runtime_error if any condition is broken
+ */
 void
 ipx_cfg_extensions::check_dependencies(const std::string &ident, const struct ext_rec &rec)
 {
@@ -58,81 +118,122 @@ ipx_cfg_extensions::check_dependencies(const std::string &ident, const struct ex
     // Check if the producer is placed before all consumers
     const struct plugin_rec &producer_rec = rec.producers.front();
     const auto min_cons = std::min_element(rec.consumers.cbegin(), rec.consumers.cend());
-    if (producer_rec.plugin_idx > min_cons->plugin_idx) {
+    if (producer_rec.inst_pos > min_cons->inst_pos) {
         throw std::runtime_error("Instance '" + producer_rec.name + "', which is a provider "
             "of Data Record extension " + ident + ", is placed in the collector pipeline "
             "after '" + min_cons->name + "' instance, which depends on the extension. "
             "Please, swap the order of the plugin instances");
     }
 }
 
-
 void
-ipx_cfg_extensions::resolve(std::vector<ipx_instance *> &plugins)
+ipx_cfg_extensions::register_instance(ipx_ctx_t *ctx, size_t pos)
 {
-    // Add extensions and dependencies from all plugins
-    unsigned int pos = 0;
-    for (auto &it : plugins) {
-        if (dynamic_cast<ipx_instance_input *>(it)) {
-            // Input plugins cannot register extensions
-            continue;
-        }
-
-        struct ipx_ctx_ext *arr_ptr = nullptr;
-        size_t arr_size = 0;
+    const char *inst_name = ipx_ctx_name_get(ctx);
+    struct ipx_ctx_ext *arr_ptr = nullptr;
+    size_t arr_size = 0;
 
-        std::tie(arr_ptr, arr_size) = it->get_extensions();
-        for (size_t i = 0; i < arr_size; ++i) {
-            add_extension(it, pos, &arr_ptr[i]);
-        }
-
-        if (!dynamic_cast<ipx_instance_output *>(it)) {
-            // Increment position for all plugins except outputs
-            pos++;
-        }
-    };
+    if (m_resolved) {
+        throw std::runtime_error("(internal) Instance extensions " + std::string(inst_name)
+            + " cannot be registered anymore as extension depencies have been resolved!");
+    }
 
-    if (m_extensions.empty()) {
-        // No extensions, no dependencies
-        return;
+    // Register all extensions
+    ipx_ctx_ext_defs(ctx, &arr_ptr, &arr_size);
+    for (size_t i = 0; i < arr_size; ++i) {
+        add_extension(inst_name, pos, &arr_ptr[i]);
     }
+}
 
+void
+ipx_cfg_extensions::resolve()
+{
     size_t offset = 0;
     uint64_t mask = 1U;
 
+    if (m_resolved) {
+        return;
+    }
+
     for (auto &ext_type : m_extensions) {
         for (auto &ext_name : ext_type.second) {
+            // Check extension dependencies and update its description
+            struct ext_rec &ext = ext_name.second;
+            std::string ident = "'" + ext_type.first + "/" + ext_name.first + "'";
+
             if (!mask) {
                 // No more bits in the mask!
                 throw std::runtime_error("Maximum number of Data Record extensions has been reached!");
             }
 
-            struct ext_rec &ext = ext_name.second;
-            assert(ext.producers.size() == 1 && "Exactly one producer");
-
             // Check the extension
-            std::string ident = "'" + ext_type.first + "/" + ext_name.first + "'";
-            check_dependencies(ident, ext_name.second);
+            check_dependencies(ident, ext);
+            assert(ext.producers.size() == 1 && "Exactly one producer");
 
             // Determine size, offset and bitset mask
             ext.size = ext.producers[0].rec->size;
             ext.offset = offset;
             ext.mask = mask;
 
             // Align the offset to multiple of 8
-            ext.offset += (ext.size % 8U == 0) ? ext.size : (((ext.size / 8U) + 1U) * 8U);
+            ext.offset += (ext.size % 8U == 0) ? (ext.size) : (((ext.size / 8U) + 1U) * 8U);
             ext.mask <<= 1U;
-
-            IPX_DEBUG(comp_str, "Extension %s registered (size: %zu, offset: %zu)",
-                ext.size, ext.offset);
         }
     }
 
+    m_size_total = offset;
+    m_resolved = true;
+}
+
+void
+ipx_cfg_extensions::update_instance(ipx_ctx_t *ctx)
+{
+    struct ipx_ctx_ext *arr_ptr = nullptr;
+    size_t arr_size = 0;
 
+    if (!m_resolved) {
+        throw std::runtime_error("(internal) Extensions hasn't been resolved yet!");
+    }
 
-    // TODO: save total size of all extensions
+    // For all instance extensions and dependencies
+    ipx_ctx_ext_defs(ctx, &arr_ptr, &arr_size);
+    for (size_t i = 0; i < arr_size; ++i) {
+        // Find the extension definition and update instance parameters
+        struct ipx_ctx_ext *ext = &arr_ptr[i];
 
+        const auto &exts_by_type = m_extensions.at(ext->data_type);
+        const auto &ext_def = exts_by_type.at(ext->data_name);
 
+        // In case of the producer, the extension size must be still the same
+        assert(ext->etype != IPX_EXTENSION_PRODUCER || ext->size == ext_def.size);
+        ext->mask = ext_def.mask;
+        ext->offset = ext_def.offset;
+        ext->size = ext_def.size;
+    }
 
+    // Update size of the Data Record in the plugin context
+    ipx_ctx_recsize_set(ctx, IPX_MSG_IPFIX_BASE_REC_SIZE + m_size_total);
 }
 
+void
+ipx_cfg_extensions::list_extensions()
+{
+    if (!m_resolved) {
+        throw std::runtime_error("(internal) Extensions hasn't been resolved yet!");
+    }
+
+    if (m_extensions.empty()) {
+        IPX_DEBUG(comp_str, "No Data Record extensions!");
+        return;
+    }
+
+    for (auto &ext_type : m_extensions) {
+        for (auto &ext_name : ext_type.second) {
+            struct ext_rec &ext = ext_name.second;
+            std::string ident = "'" + ext_type.first + "/" + ext_name.first + "'";
+
+            IPX_DEBUG(comp_str, "Data Record extension %s (size: %zu, offset: %zu, consumers: %zu)",
+                ident.c_str(), ext.size, ext.offset, ext.consumers.size());
+        }
+    }
+}