Implement zstandard commit compression

5 files changed, 64 insertions, 9 deletions

diff --git a/commit.c b/commit.c
index a19aefe..12cc027 100644
--- a/commit.c
+++ b/commit.c
@@ -1,9 +1,47 @@
 #include <errno.h>
 #include <string.h>
+#include <zstd.h>
 #include "commit.h"
 
+static void nfl_commit_default(FILE* f, nflog_entry_t* store, uint32_t store_size);
+static void nfl_commit_lz4(FILE* f, nflog_entry_t* store, uint32_t store_size);
+static void nfl_commit_zstd(FILE* f, nflog_entry_t* store, uint32_t store_size);
+
+typedef void (*nflog_commit_run_table_t)(FILE* f, nflog_entry_t* store, uint32_t size);
+static const nflog_commit_run_table_t commit_run_table[] = {
+    nfl_commit_default,
+    nfl_commit_lz4,
+    nfl_commit_zstd
+};
+
 void nfl_commit_init() {
+    /* TODO */
+}
+
+static void nfl_commit_default(FILE* f, nflog_entry_t* store, uint32_t store_size) {
+    uint32_t written;
+    written = fwrite(store, 1, store_size, f);
+    ERR(written != store_size, strerror(errno));
+}
 
+static void nfl_commit_lz4(FILE* f, nflog_entry_t* store, uint32_t store_size) {
+    /* TODO */
+}
+
+static void nfl_commit_zstd(FILE* f, nflog_entry_t* store, uint32_t store_size) {
+    size_t const bufsize = ZSTD_compressBound(store_size);
+    void* buf;
+
+    ERR((buf = malloc(bufsize)), NULL);
+
+    size_t const csize = ZSTD_compress(buf, bufsize, store, store_size, 1);
+    if (ZSTD_isError(csize)) {
+        fprintf(stderr, "zstd error: %s \n", ZSTD_getErrorName(csize));
+        exit(8);
+    }
+
+    nfl_commit_default(f, buf, bufsize);
+    free(buf);
 }
 
 void nfl_commit_worker(nflog_header_t* header, nflog_entry_t* store, const char* filename) {
@@ -19,8 +57,7 @@ void nfl_commit_worker(nflog_header_t* header, nflog_entry_t* store, const char*
 
     // commit store
     uint32_t store_size = sizeof(nflog_entry_t) * header->max_n_entries;
-    written = fwrite(store, 1, store_size, f);
-    ERR(written != store_size, strerror(errno));
+    commit_run_table[header->compression_opt](f, store, store_size);
 
     // Do fsync ?
     fclose(f);
diff --git a/commit.h b/commit.h
index 315cf8e..d61fbe2 100644
--- a/commit.h
+++ b/commit.h
@@ -1,7 +1,7 @@
 #ifndef _COMMIT_H
 #define _COMMIT_H
 
-#include "main.h"
+#include "common.h"
 void nfl_commit_init();
 void nfl_commit_worker(nflog_header_t* header, nflog_entry_t* store, const char* filename);
 
diff --git a/common.c b/common.c
index 3e3b9ad..b0f59b0 100644
--- a/common.c
+++ b/common.c
@@ -80,5 +80,19 @@ const char *nfl_format_output(nflog_entry_t *entry) {
           entry->timestamp, dest_ip,
           entry->protocol == IPPROTO_TCP ? "TCP" : "UDP",
           entry->uid, entry->sport, entry->dport);
-    return strdup(out);
+}
+
+int nfl_setup_compression(const char *flag, enum nflog_compression_t *opt) {
+    if(flag == NULL) {
+        *opt=COMPRESS_NONE;
+    } else if(!strcmp(flag, "zstd") || !strcmp(flag, "zstandard")) {
+        *opt=COMPRESS_ZSTD;
+    } else if(!strcmp(flag, "lz4")) {
+        *opt=COMPRESS_LZ4;
+    } else {
+        fprintf(stderr, "Unknown compression algorithm: %s\n", flag);
+        return 0;
+    }
+
+    return 1;
 }
diff --git a/configure.ac b/configure.ac
index 805cb4f..f84c323 100644
--- a/configure.ac
+++ b/configure.ac
@@ -22,6 +22,9 @@ AC_SEARCH_LIBS(nflog_open, netfilter_log)
 AC_CHECK_HEADERS(pthread.h)
 AC_SEARCH_LIBS(pthread_create, pthread)
 
+AC_CHECK_HEADERS(zstd.h)
+AC_SEARCH_LIBS(ZSTD_compress, zstd)
+
 AC_CONFIG_FILES([Makefile])
 
 AC_OUTPUT
diff --git a/main.h b/main.h
index f4c4a38..df54e06 100644
--- a/main.h
+++ b/main.h
@@ -78,15 +78,15 @@
 #define MAX_TRUNK_ID (80)
 #define STORAGE_PREFIX "nflog_storage"
 
-enum nflog_flag_t {
-    COMPRESS_NONE = 1,
-    COMPRESS_LZ4  = 2,
-    COMPRESS_ZSTD = 4
+enum nflog_compression_t {
+    COMPRESS_NONE,
+    COMPRESS_LZ4,
+    COMPRESS_ZSTD
 };
 
 typedef struct __attribute__((packed)) _nflog_header_t {
 	uint16_t                   cksum;                /*     0     4 */
-	enum nflog_flag_t          flag;                 /*     0     4 */
+	enum nflog_compression_t   compression_opt;      /*     0     4 */
 	uint32_t                   id;                   /*     4     4 */
 	uint32_t                   n_entries;            /*     8     4 */
 	uint32_t                   max_n_entries;        /*    12     4 */
@@ -130,6 +130,7 @@ typedef struct _nflog_global_t {
     sem_t* nfl_commit_queue;
     uint16_t nfl_group_id;
     const char* storage_dir;
+    enum nflog_compression_t compression_opt;
 } nflog_global_t;
 
 typedef struct _nflog_state_t {