redis: add automatic trimming support for streams

diff --git a/doc/userguide/output/eve/eve-json-output.rst b/doc/userguide/output/eve/eve-json-output.rst
index 26a010a7a960fd1009d25e712d63ad03f8f943c2..269024e5eb746836726d27daf3833bd15d1f7769 100644 (file)
--- a/doc/userguide/output/eve/eve-json-output.rst
+++ b/doc/userguide/output/eve/eve-json-output.rst
@@ -46,6 +46,13 @@ Output types::
       #             ## publish is using a Redis channel. "channel" is an alias for publish
       #             ## xadd is using a Redis stream. "stream" is an alias for xadd
       #  key: suricata ## string denoting the key/channel/stream to use (default to suricata)
+      #  stream-maxlen: 100000        ## Automatically trims the stream length to at most
+                                      ## this number of events. Set to 0 to disable trimming.
+                                      ## Only used when mode is set to xadd/stream.
+      #  stream-trim-exact: false     ## Trim exactly to the maximum stream length above.
+                                      ## Default: use inexact trimming (inexact by a few
+                                      ## tens of items)
+                                      ## Only used when mode is set to xadd/stream.
       # Redis pipelining set up. This will enable to only do a query every
       # 'batch-size' events. This should lower the latency induced by network
       # connection at the cost of some memory. There is no flushing implemented

diff --git a/doc/userguide/partials/eve-log.yaml b/doc/userguide/partials/eve-log.yaml
index 2030a5b68f1683172c7587e86fb3e24c05c03b2a..4897ac5685daa9d1f5e976c8e7ccb6283d93f945 100644 (file)
--- a/doc/userguide/partials/eve-log.yaml
+++ b/doc/userguide/partials/eve-log.yaml
@@ -23,6 +23,13 @@ outputs:
       #             ## publish is using a Redis channel. "channel" is an alias for publish
       #             ## xadd is using a Redis stream. "stream" is an alias for xadd
       #  key: suricata ## string denoting the key/channel/stream to use (default to suricata)
+      #  stream-maxlen: 100000        ## Automatically trims the stream length to at most
+                                      ## this number of events. Set to 0 to disable trimming.
+                                      ## Only used when mode is set to xadd/stream.
+      #  stream-trim-exact: false     ## Trim exactly to the maximum stream length above.
+                                      ## Default: use inexact trimming (inexact by a few
+                                      ## tens of items)
+                                      ## Only used when mode is set to xadd/stream.
       # Redis pipelining set up. This will enable to only do a query every
       # 'batch-size' events. This should lower the latency induced by network
       # connection at the cost of some memory. There is no flushing implemented

diff --git a/src/util-log-redis.c b/src/util-log-redis.c
index ad48ab68ef43a7b34b282987a0171a4495547bf0..6d29caccbbefdb575fa69459ddfc58393113b6c3 100644 (file)
--- a/src/util-log-redis.c
+++ b/src/util-log-redis.c
@@ -34,18 +34,21 @@
 #include <event2/thread.h>
 #endif /* HAVE_LIBEVENT_PTHREADS */
 
-static const char * redis_lpush_cmd = "LPUSH";
-static const char * redis_rpush_cmd = "RPUSH";
-static const char * redis_publish_cmd = "PUBLISH";
+static const char *redis_lpush_cmd = "LPUSH";
+static const char *redis_rpush_cmd = "RPUSH";
+static const char *redis_publish_cmd = "PUBLISH";
 static const char *redis_xadd_cmd = "XADD";
-static const char * redis_default_key = "suricata";
-static const char * redis_default_server = "127.0.0.1";
+static const char *redis_default_key = "suricata";
+static const char *redis_default_server = "127.0.0.1";
 static const char *redis_default_format = "%s %s %s";
 static const char *redis_stream_format = "%s %s * eve %s";
+static const char *redis_stream_format_maxlen_tmpl = "%s %s MAXLEN %c %d * eve %s";
 
 static int SCConfLogReopenSyncRedis(LogFileCtx *log_ctx);
 static void SCLogFileCloseRedis(LogFileCtx *log_ctx);
 
+#define REDIS_MAX_STREAM_LENGTH_DEFAULT 100000
+
 /**
  * \brief SCLogRedisInit() - Initializes global stuff before threads
  */
@@ -525,8 +528,25 @@ int SCConfLogOpenRedis(ConfNode *redis_node, void *lf_ctx)
     } else if(!strcmp(redis_mode,"channel") || !strcmp(redis_mode,"publish")) {
         log_ctx->redis_setup.command = redis_publish_cmd;
     } else if (!strcmp(redis_mode, "stream") || !strcmp(redis_mode, "xadd")) {
+        int exact;
+        intmax_t maxlen;
         log_ctx->redis_setup.command = redis_xadd_cmd;
         log_ctx->redis_setup.format = redis_stream_format;
+        if (ConfGetChildValueBool(redis_node, "stream-trim-exact", &exact) == 0) {
+            exact = 0;
+        }
+        if (ConfGetChildValueInt(redis_node, "stream-maxlen", &maxlen) == 0) {
+            maxlen = REDIS_MAX_STREAM_LENGTH_DEFAULT;
+        }
+        if (maxlen > 0) {
+            /* we do not need a lot of space here since we only build another
+            format string, whose length is limited by the length of the
+            maxlen integer formatted as a string */
+            log_ctx->redis_setup.stream_format = SCCalloc(100, sizeof(char));
+            snprintf(log_ctx->redis_setup.stream_format, 100, redis_stream_format_maxlen_tmpl, "%s",
+                    "%s", exact ? '=' : '~', maxlen, "%s");
+            log_ctx->redis_setup.format = log_ctx->redis_setup.stream_format;
+        }
     } else {
         FatalError("Invalid redis mode: %s", redis_mode);
     }

diff --git a/src/util-log-redis.h b/src/util-log-redis.h
index f6d069555e588bd2a7f39c72777f42e9d89e4792..80c9da2c57b5b624deb3f072ef82ce5a285a5cc2 100644 (file)
--- a/src/util-log-redis.h
+++ b/src/util-log-redis.h
@@ -45,6 +45,7 @@ typedef struct RedisSetup_ {
     uint16_t  port;
     int is_async;
     int  batch_size;
+    char *stream_format;
 } RedisSetup;
 
 typedef struct SCLogRedisContext_ {

diff --git a/src/util-logopenfile.c b/src/util-logopenfile.c
index a6bdf45a3e00345dc200f0858103ed0056698f0f..2fdbc33c99a09bc696771f7b0e88383a3bea754d 100644 (file)
--- a/src/util-logopenfile.c
+++ b/src/util-logopenfile.c
@@ -916,6 +916,14 @@ int LogFileFreeCtx(LogFileCtx *lf_ctx)
         lf_ctx->filetype.filetype->Deinit(lf_ctx->filetype.init_data);
     }
 
+#ifdef HAVE_LIBHIREDIS
+    if (lf_ctx->type == LOGFILE_TYPE_REDIS) {
+        if (lf_ctx->redis_setup.stream_format != NULL) {
+            SCFree(lf_ctx->redis_setup.stream_format);
+        }
+    }
+#endif
+
     memset(lf_ctx, 0, sizeof(*lf_ctx));
     SCFree(lf_ctx);
 

diff --git a/suricata.yaml.in b/suricata.yaml.in
index 6b87db93b01cecf47f1dc8ce0631857696ea4b89..b8eb03ee8cdba09154f0c11cd9d47008e9f7921d 100644 (file)
--- a/suricata.yaml.in
+++ b/suricata.yaml.in
@@ -117,6 +117,13 @@ outputs:
       #             ## publish is using a Redis channel. "channel" is an alias for publish
       #             ## xadd is using a Redis stream. "stream" is an alias for xadd
       #  key: suricata ## string denoting the key/channel/stream to use (default to suricata)
+      #  stream-maxlen: 100000        ## Automatically trims the stream length to at most
+                                      ## this number of events. Set to 0 to disable trimming.
+                                      ## Only used when mode is set to xadd/stream.
+      #  stream-trim-exact: false     ## Trim exactly to the maximum stream length above.
+                                      ## Default: use inexact trimming (inexact by a few
+                                      ## tens of items)
+                                      ## Only used when mode is set to xadd/stream.
       # Redis pipelining set up. This will enable to only do a query every
       # 'batch-size' events. This should lower the latency induced by network
       # connection at the cost of some memory. There is no flushing implemented