redis: add automatic trimming support for streams

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

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

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);
     }

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_ {

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);
 

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