Merge pull request #11 from norrisjeremy/devel

JSON output: add support for GZIP compression of files

Author: Lukas955

README.rst

@@ -72,7 +72,7 @@ Second, install build dependencies of the collector
 
 .. code-block::
 
-    yum install gcc gcc-c++ cmake make python-docutils
+    yum install gcc gcc-c++ cmake make python-docutils zlib-devel
     # Optionally: doxygen pkg-config
 
 * Note: latest systems (e.g. Fedora) use ``dnf`` instead of ``yum``.
@@ -83,7 +83,7 @@ Second, install build dependencies of the collector
 
 .. code-block::
 
-    apt-get install gcc g++ cmake make python-docutils
+    apt-get install gcc g++ cmake make python-docutils libz-dev
     # Optionally: doxygen pkg-config
 
 Finally, build and install the collector:

src/plugins/output/json/CMakeLists.txt

@@ -15,6 +15,10 @@ add_library(json-output MODULE
     src/Server.hpp
 )
 
+find_package(ZLIB REQUIRED)
+include_directories(${ZLIB_INCLUDE_DIRS})
+target_link_libraries(json-output ${ZLIB_LIBRARIES})
+
 install(
     TARGETS json-output
     LIBRARY DESTINATION "${INSTALL_DIR_LIB}/ipfixcol2/"

src/plugins/output/json/README.rst

@@ -119,6 +119,7 @@ Don't forget to remove (or comment) outputs that you don't want to use!
                     <prefix>json.</prefix>
                     <timeWindow>300</timeWindow>
                     <timeAlignment>yes</timeAlignment>
+                    <compression>none</compression>
                 </file>
             </outputs>
         </params>
@@ -223,6 +224,12 @@ outputs can be used at the same time if the outputs are not in collision with ea
         Specifies the time interval in seconds to rotate files [minimum 60, default 300]
     :``timeAlignment``:
          Align file rotation with next N minute interval (yes/no).
+    :``compression``:
+        Data compression helps to significantly reduce size of output files.
+        Following compression algorithms are available:
+
+        :``none``: Compression disabled [default]
+        :``gzip``: GZIP compression
 
 :``print``:
     Write data on standard output.

src/plugins/output/json/src/Config.cpp

@@ -84,7 +84,8 @@ enum params_xml_nodes {
     FILE_PATH,         /**< Path specification format       */
     FILE_PREFIX,       /**< File prefix                     */
     FILE_WINDOW,       /**< Window interval                 */
-    FILE_ALIGN         /**< Window alignment                */
+    FILE_ALIGN,        /**< Window alignment                */
+    FILE_COMPRESS      /**< Compression                     */
 };
 
 /** Definition of the \<print\> node  */
@@ -118,6 +119,7 @@ static const struct fds_xml_args args_file[] = {
     FDS_OPTS_ELEM(FILE_PREFIX, "prefix",        FDS_OPTS_T_STRING, 0),
     FDS_OPTS_ELEM(FILE_WINDOW, "timeWindow",    FDS_OPTS_T_UINT,   0),
     FDS_OPTS_ELEM(FILE_ALIGN,  "timeAlignment", FDS_OPTS_T_BOOL,   0),
+    FDS_OPTS_ELEM(FILE_COMPRESS, "compression", FDS_OPTS_T_STRING, FDS_OPTS_P_OPT),
     FDS_OPTS_END
 };
 
@@ -339,6 +341,7 @@ Config::parse_file(fds_xml_ctx_t *file)
     struct cfg_file output;
     output.window_align = true;
     output.window_size = 300;
+    output.m_calg = calg::NONE;
 
     const struct fds_xml_cont *content;
     while (fds_xml_next(file, &content) != FDS_EOC) {
@@ -368,6 +371,18 @@ Config::parse_file(fds_xml_ctx_t *file)
             assert(content->type == FDS_OPTS_T_BOOL);
             output.window_align = content->val_bool;
             break;
+        case FILE_COMPRESS:
+            // Compression method
+            assert(content->type == FDS_OPTS_T_STRING);
+            if (strcasecmp(content->ptr_string, "none") == 0) {
+                output.m_calg = calg::NONE;
+            } else if (strcasecmp(content->ptr_string, "gzip") == 0) {
+                output.m_calg = calg::GZIP;
+            } else {
+                const std::string inv_str = content->ptr_string;
+                throw std::invalid_argument("Unknown compression algorithm '" + inv_str + "'");
+            }
+            break;
         default:
             throw std::invalid_argument("Unexpected element within <file>!");
         }

src/plugins/output/json/src/Config.hpp

@@ -104,6 +104,11 @@ struct cfg_server : cfg_output {
     bool blocking;
 };
 
+enum class calg {
+    NONE, ///< Do not use compression
+    GZIP  ///< GZIP compression
+};
+
 /** Configuration of file writer                                                                 */
 struct cfg_file : cfg_output {
     /** Path pattern                                                                             */
@@ -114,6 +119,8 @@ struct cfg_file : cfg_output {
     uint32_t window_size;
     /** Enable/disable window alignment                                                          */
     bool window_align;
+    /** Compression algorithm                                                                    */
+    calg m_calg;
 };
 
 /** Parsed configuration of an instance                                                          */

src/plugins/output/json/src/File.cpp

@@ -52,6 +52,7 @@
 #include <sys/stat.h>
 #include <unistd.h>
 #include <climits>
+#include <zlib.h>
 
 /**
  * \brief Class constructor
@@ -69,6 +70,7 @@ File::File(const struct cfg_file &cfg, ipx_ctx_t *ctx) : Output(cfg.name, ctx)
     _thread->storage_path = cfg.path_pattern;
     _thread->file_prefix = cfg.prefix;
     _thread->window_size = cfg.window_size;
+    _thread->m_calg = cfg.m_calg;
     time(&_thread->window_time);
 
     if (cfg.window_size < _WINDOW_MIN_SIZE) {
@@ -88,8 +90,8 @@ File::File(const struct cfg_file &cfg, ipx_ctx_t *ctx) : Output(cfg.name, ctx)
     }
 
     // Create directory & first file
-    FILE *new_file = file_create(ctx, _thread->storage_path, _thread->file_prefix,
-        _thread->window_time);
+    void *new_file = file_create(ctx, _thread->storage_path, _thread->file_prefix,
+        _thread->window_time, _thread->m_calg);
     if (!new_file) {
         delete _thread;
         throw std::runtime_error("(File output) Failed to create a time window file.");
@@ -99,28 +101,44 @@ File::File(const struct cfg_file &cfg, ipx_ctx_t *ctx) : Output(cfg.name, ctx)
 
     pthread_rwlockattr_t attr;
     if (pthread_rwlockattr_init(&attr) != 0) {
-        fclose(_thread->file);
+        if (_thread->m_calg == calg::GZIP) {
+            gzclose((gzFile)_thread->file);
+        } else {
+            fclose((FILE *)_thread->file);
+        }
         delete _thread;
         throw std::runtime_error("(File output) Rwlockattr initialization failed!");
     }
 
     if (pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP) != 0) {
-        fclose(_thread->file);
+        if (_thread->m_calg == calg::GZIP) {
+            gzclose((gzFile)_thread->file);
+        } else {
+            fclose((FILE *)_thread->file);
+        }
         pthread_rwlockattr_destroy(&attr);
         delete _thread;
         throw std::runtime_error("(File output) Rwlockattr setkind failed!");
     }
 
     if (pthread_rwlock_init(&_thread->rwlock, &attr) != 0) {
-        fclose(_thread->file);
+        if (_thread->m_calg == calg::GZIP) {
+            gzclose((gzFile)_thread->file);
+        } else {
+            fclose((FILE *)_thread->file);
+        }
         pthread_rwlockattr_destroy(&attr);
         delete _thread;
         throw std::runtime_error("(File output) Rwlock initialization failed!");
     }
 
     pthread_rwlockattr_destroy(&attr);
     if (pthread_create(&_thread->thread, NULL, &File::thread_window, _thread) != 0) {
-        fclose(_thread->file);
+        if (_thread->m_calg == calg::GZIP) {
+            gzclose((gzFile)_thread->file);
+        } else {
+            fclose((FILE *)_thread->file);
+        }
         pthread_rwlock_destroy(&_thread->rwlock);
         delete _thread;
         throw std::runtime_error("(File output) Failed to start a thread for changing time "
@@ -141,7 +159,11 @@ File::~File()
         pthread_rwlock_destroy(&_thread->rwlock);
 
         if (_thread->file) {
-            fclose(_thread->file);
+            if (_thread->m_calg == calg::GZIP) {
+                gzclose((gzFile)_thread->file);
+            } else {
+                fclose((FILE *)_thread->file);
+            }
         }
 
         delete _thread;
@@ -177,12 +199,16 @@ File::thread_window(void *context)
         // New time window
         pthread_rwlock_wrlock(&data->rwlock);
         if (data->file) {
-            fclose(data->file);
+            if (data->m_calg == calg::GZIP) {
+                gzclose((gzFile)data->file);
+            } else {
+                fclose((FILE *)data->file);
+            }
             data->file = nullptr;
         }
 
         data->window_time += data->window_size;
-        FILE *file = file_create(data->ctx, data->storage_path, data->file_prefix, data->window_time);
+        void *file = file_create(data->ctx, data->storage_path, data->file_prefix, data->window_time, data->m_calg);
         if (!file) {
             IPX_CTX_ERROR(data->ctx, "(File output) Failed to create a time window file.", '\0');
         }
@@ -209,7 +235,11 @@ File::process(const char *str, size_t len)
     pthread_rwlock_rdlock(&_thread->rwlock);
     if (_thread->file) {
         // Store the record
-        fwrite(str, len, 1, _thread->file);
+        if (_thread->m_calg == calg::GZIP) {
+            gzfwrite(str, len, 1, (gzFile)_thread->file);
+        } else {
+            fwrite(str, len, 1, (FILE *)_thread->file);
+        }
     }
     pthread_rwlock_unlock(&_thread->rwlock);
     return IPX_OK;
@@ -220,7 +250,11 @@ File::flush()
 {
     pthread_rwlock_rdlock(&_thread->rwlock);
     if (_thread->file) {
-        fflush(_thread->file);
+        if (_thread->m_calg == calg::GZIP) {
+            gzflush((gzFile)_thread->file, Z_SYNC_FLUSH);
+        } else {
+            fflush((FILE *)_thread->file);
+        }
     }
     pthread_rwlock_unlock(&_thread->rwlock);
 }
@@ -342,9 +376,9 @@ File::dir_create(ipx_ctx_t *ctx, const std::string &path)
  * \param[in] tm     Timestamp
  * \return On success returns pointer to the file, Otherwise returns NULL.
  */
-FILE *
+void *
 File::file_create(ipx_ctx_t *ctx, const std::string &tmplt, const std::string &prefix,
-    const time_t &tm)
+    const time_t &tm, calg m_calg)
 {
     char file_fmt[20];
 
@@ -372,8 +406,15 @@ File::file_create(ipx_ctx_t *ctx, const std::string &tmplt, const std::string &p
         return NULL;
     }
 
-    std::string file_name = directory + prefix + file_fmt;
-    FILE *file = fopen(file_name.c_str(), "a");
+    std::string file_name;
+    void *file;
+    if (m_calg == calg::GZIP) {
+        file_name = directory + prefix + file_fmt + ".gz";
+        file = gzopen(file_name.c_str(), "a9");
+    } else {
+        file_name = directory + prefix + file_fmt;
+        file = fopen(file_name.c_str(), "a");
+    }
     if (!file) {
         // Failed to create a flow file
         char buffer[128];

src/plugins/output/json/src/File.hpp

@@ -45,7 +45,6 @@
 #include <atomic>
 #include <string>
 #include <ctime>
-#include <cstdio>
 
 #include <pthread.h>
 #include "Storage.hpp"
@@ -79,8 +78,9 @@ class File : public Output {
         time_t window_time;          /**< Current time window        */
         std::string storage_path;    /**< Storage path (template)    */
         std::string file_prefix;     /**< File prefix                */
+        calg m_calg;                 /**< Compression                */
 
-        FILE *file;                  /**< File descriptor            */
+        void *file;                  /**< File descriptor            */
     } thread_ctx_t;
 
     /** Thread for changing time windows */
@@ -92,8 +92,8 @@ class File : public Output {
     // Create a directory for a time window
     static int dir_create(ipx_ctx_t *ctx, const std::string &path);
     // Create a file for a time window
-    static FILE *file_create(ipx_ctx_t *ctx, const std::string &tmplt, const std::string &prefix,
-        const time_t &tm);
+    static void *file_create(ipx_ctx_t *ctx, const std::string &tmplt, const std::string &prefix,
+        const time_t &tm, calg m_calg);
     // Window changer
     static void *thread_window(void *context);
 };