IPFIX output: doc update (some parameter names changed), avoid flushing templates to the first IPFIX File

Author: Lukas955

src/plugins/output/ipfix/README.rst

@@ -41,8 +41,8 @@ Example configuration
             <useLocalTime>false</useLocalTime>
             <windowSize>300</windowSize>
             <alignWindows>true</alignWindows>
-            <skipUnknownDataSets>true</skipUnknownDataSets>
-            <splitOnExportTime>false</splitOnExportTime>
+            <preserveOriginal>false</preserveOriginal>
+            <rotateOnExportTime>false</rotateOnExportTime>
         </params>
     </output>
 
@@ -63,27 +63,36 @@ Parameters
     is "0", all flow will be stored into a single file. [default: 0]
 
 :``alignWindows``:
-    Align file rotation with next N minute interval (true/false). [default: true]
-
-:``skipUnknownDataSets``:
-    Specifies whether Data Sets with unknown (Options) Template should be left
-    out from the output file. Sequence numbers and message lengths will be
-    adjusted accordingly. [default: false]
-
-:``splitOnExportTime``:
+    Align file rotation with next N minute interval. [values: true/false,
+    default: true]
+
+:``preserveOriginal``:
+    Preserve and store original IPFIX Messages as received by the plugin i.e.
+    do not modify the Messages or their sequence numbers. If this option is
+    disabled (recommended), Data Sets with unknown (Options) Templates are
+    left out from the file and sequence numbers are adjusted accordingly
+    (starts from 0), which should produce warning-free stream of IPFIX
+    Messages. However, if the original IPFIX Messages are preserved, some
+    warnings (e.g. missing (Options) Templates, unexpected sequence number,
+    etc.) might be produced during replay and even errors might raise when
+    there is an ODID collision and the main Transport Session is replaced.
+    Use with caution. [values: true/false, default: false]
+
+:``rotateOnExportTime``:
     Specifies whether files should be rotated based on IPFIX Export Time
     (i.e. timestamp from IPFIX Message header) instead of system time.
-    Warning: If the plugin receives flow records from multiple exporters at time
-    the rotation could be unsteady. [default: false]
+    If enabled, the Export Time is used also for creating filenames.
+    Warning: If the plugin receives flow records from multiple exporters at
+    time the rotation could be unsteady. [default: false]
 
 Note
 ----
 
 After a new IPFIX File is created, all previously seen and still valid (Options)
 Templates of the each ODID are written to the file once the first IPFIX Message
-corresponding to the ODID arrives with at least one successfully parsed data record.
-This is necessary so each file can be used independently of the (Options) Template
-definitions from previous files.
+corresponding to the ODID arrives with at least one successfully parsed data
+record. This is necessary so each file can be used independently of the
+(Options) Template definitions from previous files.
 
 ``ipfixsend2`` tool doesn't support sequential reading of multiple IPFIX Files
 right now. However, there is a workaround - you can merge multiple IPFIX Files

src/plugins/output/ipfix/doc/ipfixcol2-ipfix-output.7.rst

@@ -7,6 +7,7 @@ IPFIX (output plugin)
 ---------------------
 
 :Author: Michal Sedlák ([email protected])
+:Author: Lukáš Huták ([email protected])
 :Date:   2019-03-28
 :Copyright: Copyright © 2019 CESNET, z.s.p.o.
 :Version: 1.0

src/plugins/output/ipfix/src/Config.cpp

@@ -50,7 +50,7 @@ enum params_xml_nodes {
     PARAM_USE_LOCALTIME,
     PARAM_WINDOW_SIZE,
     PARAM_ALIGN_WINDOWS,
-    PARAM_SKIP_UNKNOWN_DATASETS,
+    PARAM_PRESERVE_ORIGINAL,
     PARAM_SPLIT_ON_EXPORT_TIME
 };
 
@@ -61,8 +61,8 @@ static const struct fds_xml_args args_params[] = {
     FDS_OPTS_ELEM(PARAM_USE_LOCALTIME, "useLocalTime", FDS_OPTS_T_BOOL, FDS_OPTS_P_OPT),
     FDS_OPTS_ELEM(PARAM_WINDOW_SIZE,   "windowSize",   FDS_OPTS_T_UINT, FDS_OPTS_P_OPT),
     FDS_OPTS_ELEM(PARAM_ALIGN_WINDOWS, "alignWindows", FDS_OPTS_T_BOOL, FDS_OPTS_P_OPT),
-    FDS_OPTS_ELEM(PARAM_SKIP_UNKNOWN_DATASETS, "skipUnknownDataSets", FDS_OPTS_T_BOOL, FDS_OPTS_P_OPT),
-    FDS_OPTS_ELEM(PARAM_SPLIT_ON_EXPORT_TIME,  "splitOnExportTime",   FDS_OPTS_T_BOOL, FDS_OPTS_P_OPT),
+    FDS_OPTS_ELEM(PARAM_PRESERVE_ORIGINAL,    "preserveOriginal",   FDS_OPTS_T_BOOL, FDS_OPTS_P_OPT),
+    FDS_OPTS_ELEM(PARAM_SPLIT_ON_EXPORT_TIME, "rotateOnExportTime", FDS_OPTS_T_BOOL, FDS_OPTS_P_OPT),
     FDS_OPTS_END
 };
 
@@ -71,7 +71,7 @@ void Config::set_defaults() {
     use_localtime = false;
     window_size = 0;
     align_windows = true;
-    skip_unknown_datasets = false;
+    preserve_original = false;
     split_on_export_time = false;
 }
 
@@ -96,9 +96,9 @@ void Config::parse_params(fds_xml_ctx_t *params)
             assert(content->type == FDS_OPTS_T_BOOL);
             align_windows = content->val_bool;
             break;
-        case PARAM_SKIP_UNKNOWN_DATASETS:
+        case PARAM_PRESERVE_ORIGINAL:
             assert(content->type == FDS_OPTS_T_BOOL);
-            skip_unknown_datasets = content->val_bool;
+            preserve_original = content->val_bool;
             break;
         case PARAM_SPLIT_ON_EXPORT_TIME:
             assert(content->type == FDS_OPTS_T_BOOL);

src/plugins/output/ipfix/src/Config.hpp

@@ -74,8 +74,8 @@ class Config {
     uint64_t window_size;
     /// Rotate files on multiple of time interval
     bool align_windows;
-    /// Skip Data Sets with undefined templates
-    bool skip_unknown_datasets;
+    /// Preserve original IPFIX Message (i.e. don't skip Data Sets with undefined templates)
+    bool preserve_original;
     /// Split on IPFIX Export Time instead on system time
     bool split_on_export_time;
 

src/plugins/output/ipfix/src/IPFIXOutput.cpp

@@ -70,22 +70,23 @@ IPFIXOutput::should_start_new_file(std::time_t current_time)
         return false;
     }
 
-    uint32_t time_difference = current_time - file_start_time;
-    if (time_difference >= config->window_size) {
-        return true;
-    } else {
-        return false;
-    }
+    return (current_time >= file_start_time + time_t(config->window_size));
 }
 
 /**
- * @brief Create a new output file
- * @param[in] current_time Current export time
- * @throw runtime_error if the file cannot be created
+ * \brief Create a new output file
+ * \param[in] current_time Current export time
+ * \throw runtime_error if the file cannot be created
  */
 void
 IPFIXOutput::new_file(const std::time_t current_time)
 {
+    // Require (Options) Templates definitions to be added (only if this is not the first file)
+    bool add_tmplts = (output_file != nullptr);
+
+    // Close the previous file, if exists
+    close_file();
+
     // Get the timestamp of the file to create
     if (config->align_windows) {
         // Round down to the nearest multiple of window size
@@ -98,7 +99,7 @@ IPFIXOutput::new_file(const std::time_t current_time)
     std::tm *ts_res = config->use_localtime
         ? localtime_r(&file_start_time, &ts_now)
         : gmtime_r(&file_start_time, &ts_now);
-    if (ts_res == NULL) {
+    if (ts_res == nullptr) {
         // An error has occurred
         const char *err_str;
         ipx_strerror(errno, err_str);
@@ -141,7 +142,7 @@ IPFIXOutput::new_file(const std::time_t current_time)
 
     // Consider all Templates as undefined
     for (auto &odid_pair : odid_contexts) {
-        odid_pair.second.needs_to_write_templates = true;
+        odid_pair.second.needs_to_write_templates = add_tmplts;
     }
 
     IPX_CTX_INFO(plugin_context, "New output file created: %s", filename);
@@ -221,7 +222,7 @@ write_templates_cb(const struct fds_template *tmplt, void *data)
     assert(tmplt->type == FDS_TYPE_TEMPLATE || tmplt->type == FDS_TYPE_TEMPLATE_OPTS);
 
     auto *ctx = reinterpret_cast<struct write_templates_aux *>(data);
-    constexpr size_t MSG_SIZE = 1400; // Create only IPFIX Message with at most this length  // TODO
+    constexpr size_t MSG_SIZE = 1400; // Create only IPFIX Message with at most this length
 
     // First, check if the template can fit into the current message
     size_t size_needed = 0;
@@ -317,13 +318,15 @@ IPFIXOutput::get_odid(uint32_t odid, const ipx_session *session)
 {
     auto *odid_ctx = &odid_contexts[odid];
     if (odid_ctx->session == nullptr) {
-        // Grant right to this Transport Session to write into thw file with the given ODID...
+        // Grant this Transport Session access to write into the file with the given ODID...
         odid_ctx->session = session;
+        IPX_CTX_INFO(plugin_context, "[ODID: %" PRIu32 "] '%s' has been granted access to write to "
+            "the file with the given ODID.", odid, session->ident);
         return odid_ctx;
     }
 
     if (odid_ctx->session == session) {
-        // This is already known Transport Session with right to write into the file
+        // This is already known Transport Session with access to write into the file
         return odid_ctx;
     }
 
@@ -376,7 +379,6 @@ IPFIXOutput::on_ipfix_message(ipx_msg_ipfix *message)
     assert(time_now >= 0 && sizeof(time_now) >= sizeof(msg_etime));
 
     if (should_start_new_file(time_now)) {
-        close_file();
         new_file(time_now); // This will make sure that templates will be written to the file
     }
 
@@ -390,13 +392,13 @@ IPFIXOutput::on_ipfix_message(ipx_msg_ipfix *message)
 
     // Write all (Options) Templates, if required
     if (tsnap != nullptr && odid_context->needs_to_write_templates) {
-        uint32_t new_sn = (config->skip_unknown_datasets) ? odid_context->sequence_number : msg_seq;
+        uint32_t new_sn = (config->preserve_original) ? msg_seq : odid_context->sequence_number;
         write_templates(tsnap, msg_odid, msg_etime, new_sn);
         odid_context->needs_to_write_templates = false;
     }
 
     // If we don't have to look for unknown Data Sets, just copy the whole message -> FAST PATH
-    if (!config->skip_unknown_datasets) {
+    if (config->preserve_original) {
         std::fwrite(msg_hdr, msg_size, 1, output_file);
         return;
     }
@@ -477,7 +479,7 @@ IPFIXOutput::remove_session(const struct ipx_session *session)
 {
     auto it = odid_contexts.begin();
     while (it != odid_contexts.end()) {
-        // Has this session right to write to the file?
+        // Has this session access to write to the file?
         struct odid_context_s &ctx = it->second;
         if (ctx.session != session) {
             // Remove it from colliding sessions, if present
@@ -488,7 +490,7 @@ IPFIXOutput::remove_session(const struct ipx_session *session)
 
         // This is the Transport Session with rights to write to the file with given ODID
         if (ctx.colliding_sessions.empty()) {
-            // Only session in the context -> remove whole context
+            // Only session in the context -> remove the whole context
             it = odid_contexts.erase(it);
             continue;
         }
@@ -516,7 +518,7 @@ IPFIXOutput::on_session_message(ipx_msg_session *message)
 
     case IPX_MSG_SESSION_CLOSE:
         // Free up ODIDs used by this session allowing them to be used again
-        remove_session(session); // TODO: consider blocking ODID until new file is started + reset seqnum!
+        remove_session(session);
         break;
     }
 }

src/plugins/output/ipfix/src/IPFIXOutputPlugin.cpp

@@ -90,11 +90,11 @@ ipx_plugin_init(ipx_ctx_t *ctx, const char *params)
         ipx_ctx_private_set(ctx, instance);
 
     } catch (std::exception &ex) {
-        IPX_CTX_ERROR(ctx, "IPFIXOutput plugin: %s", ex.what());
+        IPX_CTX_ERROR(ctx, "%s", ex.what());
         return IPX_ERR_DENIED;
 
     } catch (...) {
-        IPX_CTX_ERROR(ctx, "IPFIXOutput plugin: Unexpected exception has occurred!", '\0');
+        IPX_CTX_ERROR(ctx, "Unexpected exception has occurred!", '\0');
         return IPX_ERR_DENIED;
     }
 
@@ -127,11 +127,11 @@ ipx_plugin_process(ipx_ctx_t *ctx, void *cfg, ipx_msg_t *msg)
         }
 
     } catch (std::exception &ex) {
-        IPX_CTX_ERROR(ctx, "IPFIXOutput plugin: %s", ex.what());
+        IPX_CTX_ERROR(ctx, "%s", ex.what());
         return IPX_ERR_DENIED;
 
     } catch (...) {
-        IPX_CTX_ERROR(ctx, "IPFIXOutput plugin: Unexpected exception has occurred!", '\0');
+        IPX_CTX_ERROR(ctx, "Unexpected exception has occurred!", '\0');
         return IPX_ERR_DENIED;
     }