Merge remote-tracking branch 'origin/fds_output' into devel

include/CMakeLists.txt

@@ -16,6 +16,7 @@ set(SUB_HEADERS
     ipfixcol2/message_session.h
     ipfixcol2/plugins.h
     ipfixcol2/session.h
+    ipfixcol2/utils.h
     ipfixcol2/verbose.h
     "${PROJECT_BINARY_DIR}/include/ipfixcol2/api.h"
 )

include/ipfixcol2.h

@@ -63,6 +63,7 @@
 
 #include <ipfixcol2/plugins.h>
 #include <ipfixcol2/session.h>
+#include <ipfixcol2/utils.h>
 #include <ipfixcol2/verbose.h>
 
 

include/ipfixcol2/utils.h

@@ -0,0 +1,70 @@
+/**
+ * @file   include/ipfixcol2/utils.h
+ * @author Lukas Hutak <[email protected]>
+ * @brief  Auxiliary utilities for plugins (header file)
+ * @date   June 2019
+ *
+ * Copyright(c) 2019 CESNET z.s.p.o.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#ifndef IPX_UTILS_H
+#define IPX_UTILS_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <ipfixcol2/api.h>
+#include <sys/stat.h> // mkdir file permissions
+
+/**
+* \defgroup ipxSource Transport session identification
+* \ingroup publicAPIs
+* \brief Transport session interface
+*
+* Data types and API functions for identification and management of Transport
+* session identification. The Exporting Process uses the Transport Session to
+* send messages from multiple _independent_ Observation Domains to the
+* Collecting Process. Moreover, in case of SCTP session messages are also send
+* over _independent_ streams.
+*
+* Following structures represents Transport session between Exporting process
+* and Collecting Process. However, proper processing of flows also requires
+* distinguishing Observation Domain IDs and Stream identifications out of
+* scope of these structures.
+*
+* @{
+*/
+
+/**
+ * @brief Default file permission of newly created directories
+ * @note Read/write/execute for a user and his group, read/execute for others.
+ */
+#define IPX_UTILS_MKDIR_DEF (S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH)
+
+/**
+ * @brief Create recursively a directory
+ *
+ * @note
+ *   File permission @p mode only affects newly created directories. In other
+ *   words, if a directory (or subdirectory) already exists, file permission
+ *   bits @p mode are not applied.
+ * @note
+ *   The function is implemented as "recursive" wrapper over standard mkdir
+ *   function. See man 3 mkdir for more information.
+ * @param[in] path Full directory path to create
+ * @param[in] mode The file permission bits of the new directories
+ *   (see default value #IPX_UTILS_MKDIR_DEF)
+ * @return #IPX_OK on success
+ * @return #IPX_ERR_DENIED otherwise and errno is set appropriately.
+ */
+IPX_API int
+ipx_utils_mkdir(const char *path, mode_t mode);
+
+/**@}*/
+
+#ifdef __cplusplus
+}
+#endif
+#endif // IPX_UTILS_H

src/core/utils.c

@@ -43,10 +43,15 @@
 #undef _GNU_SOURCE
 #define _POSIX_C_SOURCE 200809L
 
+
+#include <errno.h>
+#include <limits.h>
+#include <stdlib.h>
 #include <string.h>
-#include <stdio.h>
+
 #include "utils.h"
 
+
 int
 ipx_strerror_fn(int errnum, char *buffer, size_t buffer_size)
 {
@@ -57,4 +62,86 @@ ipx_strerror_fn(int errnum, char *buffer, size_t buffer_size)
 
     snprintf(buffer, buffer_size, "strerror_r() failed: Unable process error code %d!", errnum);
     return IPX_ERR_ARG;
-}
+}
+
+int
+ipx_utils_mkdir(const char *path, mode_t mode)
+{
+    const char ch_slash = '/';
+    bool add_slash = false;
+
+    // Check the parameter
+    size_t len = strlen(path);
+    if (path[len - 1] != ch_slash) {
+        len++; // We have to add another slash
+        add_slash = true;
+    }
+
+    if (len > PATH_MAX - 1) {
+        errno = ENAMETOOLONG;
+        return IPX_ERR_DENIED;
+    }
+
+    // Make a copy
+    char *path_cpy = malloc((len + 1) * sizeof(char)); // +1 for '\0'
+    if (!path_cpy) {
+        errno = ENOMEM;
+        return IPX_ERR_DENIED;
+    }
+
+    strcpy(path_cpy, path);
+    if (add_slash) {
+        path_cpy[len - 1] = ch_slash;
+        path_cpy[len] = '\0';
+    }
+
+    struct stat info;
+    char *pos;
+
+    // Create directories from the beginning
+    for (pos = path_cpy + 1; *pos; pos++) {
+        // Find a slash
+        if (*pos != ch_slash) {
+            continue;
+        }
+
+        *pos = '\0'; // Temporarily truncate pathname
+
+        // Check if a subdirectory exists
+        if (stat(path_cpy, &info) == 0) {
+            // Check if the "info" is about directory
+            if (!S_ISDIR(info.st_mode)) {
+                free(path_cpy);
+                errno = ENOTDIR;
+                return IPX_ERR_DENIED;
+            }
+
+            // Fix the pathname and continue with the next subdirectory
+            *pos = ch_slash;
+            continue;
+        }
+
+        // Errno is filled by stat()
+        if (errno != ENOENT) {
+            int errno_cpy = errno;
+            free(path_cpy);
+            errno = errno_cpy;
+            return IPX_ERR_DENIED;
+        }
+
+        // Required directory doesn't exist -> create new one
+        if (mkdir(path_cpy, mode) != 0 && errno != EEXIST) {
+            // Failed (by the way, EEXIST because of race condition i.e.
+            // multiple applications creating the same folder)
+            int errno_cpy = errno;
+            free(path_cpy);
+            errno = errno_cpy;
+            return IPX_ERR_DENIED;
+        }
+
+        *pos = ch_slash;
+    }
+
+    free(path_cpy);
+    return IPX_OK;
+}

src/plugins/output/fds/CMakeLists.txt

@@ -0,0 +1,31 @@
+# Create a linkable module
+add_library(fds-output MODULE
+    src/Config.cpp
+    src/Config.hpp
+    src/Exception.hpp
+    src/fds.cpp
+    src/Storage.cpp
+    src/Storage.hpp
+)
+
+install(
+    TARGETS fds-output
+    LIBRARY DESTINATION "${INSTALL_DIR_LIB}/ipfixcol2/"
+)
+
+if (ENABLE_DOC_MANPAGE)
+    # Build a manual page
+    set(SRC_FILE "${CMAKE_CURRENT_SOURCE_DIR}/doc/ipfixcol2-fds-output.7.rst")
+    set(DST_FILE "${CMAKE_CURRENT_BINARY_DIR}/ipfixcol2-fds-output.7")
+
+    add_custom_command(TARGET fds-output PRE_BUILD
+        COMMAND ${RST2MAN_EXECUTABLE} --syntax-highlight=none ${SRC_FILE} ${DST_FILE}
+        DEPENDS ${SRC_FILE}
+        VERBATIM
+    )
+
+    install(
+        FILES "${DST_FILE}"
+        DESTINATION "${INSTALL_DIR_MAN}/man7"
+    )
+endif()

src/plugins/output/fds/README.rst

@@ -0,0 +1,69 @@
+Flow Data Storage (output plugin)
+=================================
+
+The plugin converts and stores IPFIX Data Records into FDS file format. The file
+is based on IPFIX, therefore, it provides highly-effective way for long-term
+storage and stores complete flow records (including all Enterprise-specific
+fields, biflow, etc.) together with identification of the flow exporters who
+exported these records.
+
+All data are stored into flat files, which are automatically rotated and renamed
+every N minutes (by default 5 minutes).
+
+Example configuration
+---------------------
+
+.. code-block:: xml
+
+    <output>
+        <name>FDS output</name>
+        <plugin>fds</plugin>
+        <params>
+            <storagePath>/tmp/ipfixcol2/fds/</storagePath>
+            <compression>none</compression>
+            <dumpInterval>
+                <timeWindow>300</timeWindow>
+                <align>yes</align>
+            </dumpInterval>
+        </params>
+    </output>
+
+Parameters
+----------
+
+:``storagePath``:
+    The path element specifies the storage directory for data files. Keep on
+    mind that the path must exist in your system. Otherwise, no files are stored.
+    All files will be stored based on the configuration using the following
+    template: ``<storagePath>/YYYY/MM/DD/flows.<ts>.fds`` where ``YYYY/MM/DD``
+    means year/month/day and ``<ts>`` represents a UTC timestamp in
+    format ``YYMMDDhhmmss``.
+
+:``compression``:
+    Data compression helps to significantly reduce size of output files.
+    Following compression algorithms are available:
+
+    :``none``: Compression disabled [default]
+    :``lz4``:  LZ4 compression (very fast, slightly worse compression ration)
+    :``zstd``: ZSTD compression (slightly slower, good compression ration)
+
+:``dumpInterval``:
+    Configuration of output files rotation.
+
+    :``timeWindow``:
+        Specifies time interval in seconds to rotate files i.e. close the current
+        file and create a new one. [default: 300]
+
+    :``align``:
+        Align file rotation with next N minute interval. For example, if enabled
+        and window size is 5 minutes long, files will be created at 0, 5, 10, etc.
+        [values: yes/no, default: yes]
+
+:``asyncIO``:
+    Allows to use asynchronous I/O for writing to the file. Usually when parts
+    of the file are being written, the process is blocked on synchronous I/O
+    and waits for the operation to complete. However, asynchronous I/O allows
+    the plugin to simultaneously write to file and process flow records, which
+    significantly improves overall performance. (Note: a pool of service
+    threads shared among instances of FDS plugin might be created).
+    [values: true/false, default: true]

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

@@ -0,0 +1,20 @@
+======================
+ ipfixcol2-fds-output
+======================
+
+---------------------------------
+Flow Data Storage (output plugin)
+---------------------------------
+
+:Author: Lukáš Huták ([email protected])
+:Date:   2019-07-01
+:Copyright: Copyright © 2019 CESNET, z.s.p.o.
+:Version: 2.0
+:Manual section: 7
+:Manual group: IPFIXcol collector
+
+Description
+-----------
+
+.. include:: ../README.rst
+   :start-line: 3