UniRec output: add splitBiflow option

Note: To preserve current behaviour the option is enabled by default.

Author: Lukas955

doc/data/configs/tcp2unirec.xml

@@ -1,5 +1,5 @@
 <!--
-  Receive flow data over TCP, convert them into UniRec format and send via 
+  Receive flow data over TCP, convert them into UniRec format and send via
   TCP TRAP communication interface (port 8000).
 -->
 <ipfixcol2>
@@ -23,6 +23,7 @@
       <params>
         <!-- UniRec template -->
         <uniRecFormat>TIME_FIRST,TIME_LAST,SRC_IP,DST_IP,PROTOCOL,?SRC_PORT,?DST_PORT,?TCP_FLAGS,PACKETS,BYTES</uniRecFormat>
+        <splitBiflow>true</splitBiflow>
         <!-- TRAP interface configuration -->
         <trapIfcCommon>
           <timeout>HALF_WAIT</timeout>

extra_plugins/output/unirec/README.rst

@@ -50,6 +50,8 @@ except the one you would like to use.
         <plugin>unirec</plugin>
         <params>
             <uniRecFormat>TIME_FIRST,TIME_LAST,SRC_IP,DST_IP,PROTOCOL,?SRC_PORT,?DST_PORT,?TCP_FLAGS,PACKETS,BYTES</uniRecFormat>
+            <splitBiflow>false</splitBiflow>
+
             <trapIfcCommon>
                 <timeout>NO_WAIT</timeout>
                 <buffer>true</buffer>
@@ -92,6 +94,10 @@ Parameters
     fields is defined in `unirec-element.txt <config/unirec-elements.txt>`_ file.
     Example value: "DST_IP,SRC_IP,BYTES,DST_PORT,?TCP_FLAGS,SRC_PORT,PROTOCOL".
 
+:``splitBiflow``:
+    In case of Biflow records, split the record to two unidirectional flow records. Non-biflow
+    records are unaffected. [values: true/false, default: true]
+
 :``trapIfcCommon``:
     The following parameters can be used with any type of a TRAP interface. There are parameters
     of the interface that are normally let default. However, it is possible to override them
@@ -247,9 +253,6 @@ with increased verbosity level i.e. ``ipfixcol2 -v``.
 Notes
 -----
 
-Bidirectional flows are not currently supported by UniRec, therefore, biflow records are
-automatically split into two unidirectional flow records during conversion.
-
 When multiple IPFIX Information Elements are mapped to the same UniRec field and those IPFIX fields
 are present in an IPFIX record, the last field occurrence (in the appropriate IPFIX Template)
 is converted to the UniRec field.

extra_plugins/output/unirec/src/configuration.c

@@ -85,6 +85,7 @@ struct ifc_common {
 /*
  *  <params>
  *      <uniRecFormat>DST_IP,SRC_IP,BYTES,DST_PORT,?TCP_FLAGS,SRC_PORT,PROTOCOL</uniRecFormat>
+ *      <splitBiflow>true</splitBiflow>
  *      <trapIfcCommon>                                                           <!-- optional -->
  *          <timeout>NO_WAIT</timeout>                                            <!-- optional -->
  *          <buffer>true</buffer>                                                 <!-- optional -->
@@ -121,6 +122,7 @@ struct ifc_common {
 enum params_xml_nodes {
     // Main parameters
     NODE_UNIREC_FMT = 1,
+    NODE_BIFLOW_SPLIT,
     NODE_TRAP_COMMON,
     NODE_TRAP_SPEC,
     // TRAP common parameters
@@ -205,6 +207,7 @@ static const struct fds_xml_args args_trap_common[] = {
 static const struct fds_xml_args args_params[] = {
     FDS_OPTS_ROOT("params"),
     FDS_OPTS_ELEM(NODE_UNIREC_FMT,     "uniRecFormat",  FDS_OPTS_T_STRING, 0),
+    FDS_OPTS_ELEM(NODE_BIFLOW_SPLIT,   "splitBiflow",   FDS_OPTS_T_BOOL,   FDS_OPTS_P_OPT),
     FDS_OPTS_NESTED(NODE_TRAP_COMMON,  "trapIfcCommon", args_trap_common,  FDS_OPTS_P_OPT),
     FDS_OPTS_NESTED(NODE_TRAP_SPEC,    "trapIfcSpec",   args_trap_spec,    0),
     FDS_OPTS_END
@@ -775,7 +778,10 @@ cfg_parse_params(ipx_ctx_t *ctx, fds_xml_ctx_t *root, struct conf_params *cfg)
 {
     int rc;
 
-    // Prepare default TRAP common parameters
+    // Set default values
+    cfg->biflow_split = true;
+
+    // Set default TRAP common parameters
     struct ifc_common common;
     common.autoflush = DEF_IFC_AUTOFLUSH;
     common.buffer =    DEF_IFC_BUFFER;
@@ -795,6 +801,11 @@ cfg_parse_params(ipx_ctx_t *ctx, fds_xml_ctx_t *root, struct conf_params *cfg)
                 return IPX_ERR_NOMEM;
             }
             break;
+        case NODE_BIFLOW_SPLIT:
+            // Split biflow
+            assert(content->type == FDS_OPTS_T_BOOL);
+            cfg->biflow_split = content->val_bool;
+            break;
         case NODE_TRAP_SPEC:
             // TRAP output interface specifier
             assert(content->type == FDS_OPTS_T_CONTEXT);

extra_plugins/output/unirec/src/configuration.h

@@ -59,6 +59,8 @@ struct conf_params {
     char *unirec_spec;
     /** The same as \ref conf_params.unirec_spec, however, question marks have been removed  */
     char *unirec_fmt;
+    /** Split biflow record to 2 unidirectional flows                                        */
+    bool biflow_split;
 };
 
 /**

extra_plugins/output/unirec/src/unirecplugin.c

@@ -78,7 +78,7 @@ IPX_API struct ipx_plugin_info ipx_plugin_info = {
     // Configuration flags (reserved for future use)
     .flags = 0,
     // Plugin version string (like "1.2.3")
-    .version = "2.1.0",
+    .version = "2.2.0",
     // Minimal IPFIXcol version string (like "1.2.3")
     .ipx_min = "2.0.0"
 };
@@ -354,6 +354,7 @@ ipx_plugin_process(ipx_ctx_t *ctx, void *cfg, ipx_msg_t *msg)
 {
     (void) ctx;
     struct conf_unirec *conf = (struct conf_unirec *) cfg;
+    const bool split_enabled = conf->params->biflow_split;
     IPX_CTX_DEBUG(ctx, "Received a new message to process.");
 
     uint16_t msg_size = 0;
@@ -367,10 +368,11 @@ ipx_plugin_process(ipx_ctx_t *ctx, void *cfg, ipx_msg_t *msg)
     for (uint32_t i = 0; i < rec_cnt; i++) {
         // Get a pointer to the next record
         struct ipx_ipfix_record *ipfix_rec = ipx_msg_ipfix_get_drec(ipfix, i);
-        bool biflow = (ipfix_rec->rec.tmplt->flags & FDS_TEMPLATE_BIFLOW) != 0;
+        bool rec_is_biflow = (ipfix_rec->rec.tmplt->flags & FDS_TEMPLATE_BIFLOW) != 0;
+        bool biflow_split = split_enabled && rec_is_biflow;
 
-        // Fill record
-        uint16_t flags = biflow ? FDS_DREC_BIFLOW_FWD : 0; // In case of biflow, forward fields only
+        // Fill record (Note: in case of biflow split, forward fields only)
+        uint16_t flags = biflow_split ? (FDS_DREC_BIFLOW_FWD | FDS_DREC_REVERSE_SKIP) : 0;
         msg_data = translator_translate(conf->trans, &ipfix_rec->rec, flags, &msg_size);
         if (!msg_data) {
             // Nothing to send
@@ -380,12 +382,12 @@ ipx_plugin_process(ipx_ctx_t *ctx, void *cfg, ipx_msg_t *msg)
         IPX_CTX_DEBUG(ctx, "Send via TRAP IFC.");
         trap_ctx_send(conf->trap_ctx, 0, msg_data, msg_size);
 
-        // Is it biflow? Send the reverse direction
-        if (!biflow) {
+        // Is it biflow and split is enabled? Send the reverse direction
+        if (!biflow_split) {
             continue;
         }
 
-        flags = FDS_DREC_BIFLOW_REV;
+        flags = FDS_DREC_BIFLOW_REV | FDS_DREC_REVERSE_SKIP;
         msg_data = translator_translate(conf->trans, &ipfix_rec->rec, flags, &msg_size);
         if (!msg_data) {
             // Nothing to send