Core [extensions]: add incomplete implementation of extensions of Data Record [WIP]

Author: Lukas955

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
@@ -105,6 +104,8 @@ struct ipx_ipfix_record {
     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
@@ -397,7 +398,7 @@ IPX_API const fds_iemgr_t *
 ipx_ctx_iemgr_get(ipx_ctx_t *ctx);
 
 /**
- * \brief Register an extension of Data Records (Intermediate and Output plugins ONLY!)
+ * \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,
@@ -406,28 +407,32 @@ ipx_ctx_iemgr_get(ipx_ctx_t *ctx);
  *
  * 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!
+ * 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 and Output plugins.
+ *   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 Size of memory required for the extension (in bytes)
- * \param[out] rext Internal description of the extension
+ * \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)
+ * \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_NOMEM if the maximum number of extensions has been reached
+ * \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_rext_producer(ipx_ctx_t *ctx, const char *type, const char *name, size_t size,
-    ipx_ctx_rext **rext);
+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!)
@@ -450,37 +455,50 @@ ipx_ctx_rext_producer(ipx_ctx_t *ctx, const char *type, const char *name, size_t
  * \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] rext Internal description of the extension
+ * \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_NOMEM if the maximum number of extensions has been reached
+ * \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_rext_consumer(ipx_ctx_t *ctx, const char *type, const char *name, ipx_ctx_rext_t **rext);
+ipx_ctx_ext_consumer(ipx_ctx_t *ctx, const char *type, const char *name, ipx_ctx_ext_t **ext);
 
 /**
  * \brief Get size of an extension
  *
  * Information about the extension are not available during ipx_plugin_init().
- * \param[in]  rext Internal description of the extension
+ * \param[in]  ext  Internal description of the extension
  * \param[out] size Size of the extension
  * \return #IPX_OK on success
  * \return #IPX_ERR_DENIED if the information is not available
  */
 IPX_API int
-ipx_ctx_rext_size(ipx_ctx_rext *rext, size_t *size);
+ipx_ctx_ext_size(ipx_ctx_ext_t *ext, size_t *size);
 
 /**
  * \brief Get data of an extension
  *
- * \param[in] rext     Internal description of the extension
+ * \note
+ *   In case of a consumer, the function can return NULL if the extension hasn't been filled
+ *   by the producer!
+ * \param[in] ext      Internal description of the extension
  * \param[in] ipx_drec Data Record with extensions
- * \return Pointer to the extension
+ * \return Pointer to the extension or NULL
  */
 IPX_API void *
-ipx_ctx_rext_get(ipx_ctx_rext *rext, struct ipx_ipfix_record *ipx_drec);
+ipx_ctx_ext_get(ipx_ctx_ext_t *ext, struct ipx_ipfix_record *ipx_drec);
+
+/**
+ * \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] ipx_drec Data Record with extensions
+ */
+IPX_API void
+ipx_ctx_ext_set_filled(ipx_ctx_ext_t *ext, struct ipx_ipfix_record *ipx_drec);
 
 /**
  * @}

src/core/CMakeLists.txt

@@ -28,6 +28,8 @@ set(CORE_SOURCE
     api.c
     context.c
     context.h
+    extension.c
+    extension.h
     fpipe.c
     fpipe.h
     message_base.c

src/core/context.c

@@ -47,6 +47,7 @@
 #include <sys/prctl.h>
 
 #include "context.h"
+#include "extension.h"
 #include "verbose.h"
 #include "utils.h"
 #include "fpipe.h"
@@ -150,6 +151,13 @@ struct ipx_ctx {
          */
         unsigned int term_msg_cnt;
     } cfg_system; /**< System configuration                                                      */
+
+    struct {
+        /** Array of extensions (producers and consumers)                                        */
+        struct ipx_ctx_ext *items;
+        /** Size of extension definitions in the array                                           */
+        size_t items_cnt;
+    } cfg_extension; /**< Extension configuration                                                */
 };
 
 ipx_ctx_t *
@@ -177,6 +185,9 @@ ipx_ctx_create(const char *name, const struct ipx_ctx_callbacks *callbacks)
     ctx->cfg_system.msg_mask_allowed = IPX_MSG_IPFIX | IPX_MSG_SESSION;
     ctx->cfg_system.term_msg_cnt = 1; // By default, wait for 1 termination message
 
+    ctx->cfg_extension.items = NULL;
+    ctx->cfg_extension.items_cnt = 0;
+
     if (callbacks == NULL) {
         // Dummy context for testing
         ctx->permissions = IPX_CP_MSG_PASS;
@@ -215,6 +226,13 @@ ipx_ctx_destroy(ipx_ctx_t *ctx)
         ctx->pipeline.dst = tmp;
     }
 
+    // Destroy all extensions
+    for (size_t idx = 0; idx < ctx->cfg_extension.items_cnt; ++idx) {
+        struct ipx_ctx_ext *ext = &ctx->cfg_extension.items[idx];
+        ipx_ctx_ext_destroy(ext);
+    }
+    free(ctx->cfg_extension.items);
+
     free(ctx->name);
     free(ctx);
 }
@@ -358,6 +376,108 @@ ipx_ctx_ring_dst_set(ipx_ctx_t *ctx, ipx_ring_t *ring)
 }
 
 
+// -------------------------------------------------------------------------------------------------
+
+/**
+ * @brief Create a new extension record
+ *
+ * First, make sure that the extension hasn't been previously added. Then append a new
+ * (uninitialized) record to the array of extensions.
+ *
+ * @param[in] ctx  Plugin context
+ * @param[in] type Identification of extension type
+ * @param[in] name Identification of extension name
+ * @return #IPX_OK on success
+ * @return #IPX_ERR_EXISTS if the extension has been already defined
+ * @return #IPX_ERR_ARG if @p type or @p name are not valid strings
+ * @return #IPX_ERR_NOMEM in case of a memory allocation error
+ */
+static int
+ipx_ctx_ext_create(ipx_ctx_t *ctx, const char *type, const char *name)
+{
+    size_t new_size, alloc_size;
+    struct ipx_ctx_ext *new_items = NULL;
+
+    if (!type || !name) {
+        return IPX_ERR_ARG;
+    }
+
+    // Try to find it
+    for (size_t idx = 0; idx < ctx->cfg_extension.items_cnt; ++idx) {
+        const struct ipx_ctx_ext *rec = &ctx->cfg_extension.items[idx];
+        if (strcmp(type, rec->data_type) != 0 || strcmp(name, rec->data_name) != 0) {
+            continue;
+        }
+
+        return IPX_ERR_EXISTS;
+    }
+
+    // Create a new record
+    new_size = ctx->cfg_extension.items_cnt + 1U;
+    alloc_size = new_size * sizeof(struct ipx_ctx_ext);
+    new_items = realloc(ctx->cfg_extension.items, alloc_size);
+    if (!new_items) {
+        IPX_CTX_ERROR(ctx, "%s(): Memory allocation failed", __func__);
+        return IPX_ERR_NOMEM;
+    }
+
+    ctx->cfg_extension.items = new_items;
+    ctx->cfg_extension.items_cnt = new_size;
+    return IPX_OK;
+}
+
+int
+ipx_ctx_ext_producer(ipx_ctx_t *ctx, const char *type, const char *name, size_t size,
+    ipx_ctx_ext_t **ext)
+{
+    struct ipx_ctx_ext *rec = NULL;
+    int rc = IPX_OK;
+
+    // Check permissions (only intermediate plugins during initialization)
+    if (ctx->type != IPX_PT_INTERMEDIATE || ctx->state != IPX_CS_NEW) {
+        return IPX_ERR_DENIED;
+    }
+
+    if ((rc = ipx_ctx_ext_create(ctx, type, name)) != IPX_OK) {
+        return rc;
+    }
+
+    rec = &ctx->cfg_extension.items[ctx->cfg_extension.items_cnt - 1];
+    if ((rc = ipx_ctx_ext_init(rec, IPX_EXTENSION_PRODUCER, type, name, size)) != IPX_OK) {
+        ctx->cfg_extension.items_cnt--; // The added extension is not valid!
+        return rc;
+    }
+
+    *ext = rec;
+    return IPX_OK;
+}
+
+int
+ipx_ctx_ext_consumer(ipx_ctx_t *ctx, const char *type, const char *name, ipx_ctx_ext_t **ext)
+{
+    struct ipx_ctx_ext *rec = NULL;
+    int rc = IPX_OK;
+
+    // Check permissions (only intermediate and output plugins during initialization) + duplicities
+    if ((ctx->type != IPX_PT_INTERMEDIATE && ctx->type != IPX_PT_OUTPUT)
+            || ctx->state != IPX_CS_NEW) {
+        return IPX_ERR_DENIED;
+    }
+
+    if ((rc = ipx_ctx_ext_create(ctx, type, name)) != IPX_OK) {
+        return rc;
+    }
+
+    rec = &ctx->cfg_extension.items[ctx->cfg_extension.items_cnt - 1];
+    if ((rc = ipx_ctx_ext_init(rec, IPX_EXTENSION_CONSUMER, type, name, 0)) != IPX_OK) {
+        ctx->cfg_extension.items_cnt--; // The added extension is not valid!
+        return rc;
+    }
+
+    *ext = rec;
+    return IPX_OK;
+}
+
 // -------------------------------------------------------------------------------------------------
 
 /**

src/core/extension.c

@@ -0,0 +1,76 @@
+
+#include "extension.h"
+#include <ipfixcol2/plugins.h>
+#include <stdlib.h>
+#include <string.h>
+
+int
+ipx_ctx_ext_size(ipx_ctx_ext_t *ext, size_t *size)
+{
+    if (!ext->mask) {
+        // Mask is not filled -> initialization hasn't been completed yet
+        return IPX_ERR_DENIED;
+    }
+
+    *size = ext->size;
+    return IPX_OK;
+}
+
+void *
+ipx_ctx_ext_get(ipx_ctx_ext_t *ext, struct ipx_ipfix_record *ipx_drec)
+{
+    if (ext->etype == IPX_EXTENSION_CONSUMER) {
+        // TODO: check if the memory has been filled (only for consumers)
+        return NULL;
+    }
+
+    return (ipx_drec->ext + ext->offset);
+}
+
+void
+ipx_ctx_ext_set_filled(ipx_ctx_ext_t *ext, struct ipx_ipfix_record *ipx_drec)
+{
+    if (ext->etype != IPX_EXTENSION_PRODUCER) {
+        return; // Not allowed!
+    }
+
+    // TODO: label the extension as filled!
+}
+
+int
+ipx_ctx_ext_init(struct ipx_ctx_ext *ext, enum ipx_extension etype, const char *data_type,
+    const char *data_name, size_t size)
+{
+    // Check parameters
+    if (!data_name || !data_type || strlen(data_type) == 0 || strlen(data_name) == 0) {
+        return IPX_ERR_ARG;
+    }
+    if (etype != IPX_EXTENSION_CONSUMER && etype != IPX_EXTENSION_PRODUCER) {
+        return IPX_ERR_ARG;
+    }
+    if (etype == IPX_EXTENSION_PRODUCER && size == 0) {
+        return IPX_ERR_ARG;
+    }
+
+    ext->etype = etype;
+    ext->size = (etype == IPX_EXTENSION_PRODUCER) ? size : 0;
+    ext->data_type = strdup(data_type);
+    ext->data_name = strdup(data_name);
+    if (!ext->data_type || !ext->data_name) {
+        free(ext->data_type);
+        free(ext->data_name);
+        return IPX_ERR_NOMEM;
+    }
+
+    ext->offset = 0;
+    ext->mask = 0;
+    return IPX_OK;
+}
+
+void
+ipx_ctx_ext_destroy(struct ipx_ctx_ext *ext)
+{
+    free(ext->data_name);
+    free(ext->data_type);
+}
+