output/flush: Add EVE flushing logic

Add flushing logic driven off of the file contexts. This is a simpler
solution that removes the need for logger registration changes.

Overview:
Use the heartbeat-driven thread to periodically flush all registered EVE
contexts via a global flush list.

The global flush list is a mutex-protected TAILQ of LogFileFlushEntry
nodes; each node points to a LogFileCtx. Mutex = log_file_flush_mutex

Periodic flushing performed by a thread according to the
heartbeat.output-flush-interval [1,60]. LogFileFlushAll() is invoked to
initiate flushing of registered LogFileCtx structs; each struct's
fp_mutex is obtained while the flush occurs to synchronize with
LogFileWrite activity.

Interacts with file-rotation via the fp_mutex.

Deadlock prevention: the log_file_flush_mutex must be obtained before
the fp_mutex.

Issue: 8286
(cherry picked from commit a78911fce7)

src/log-flush.c

@@ -31,6 +31,7 @@
 #include "conf.h"
 #include "conf-yaml-loader.h"
 #include "util-privs.h"
+#include "util-logopenfile.h"
 
 /**
  * \brief Trigger detect threads to flush their output logs
@@ -119,7 +120,7 @@ error:
     return;
 }
 
-static int OutputFlushInterval(void)
+int OutputFlushInterval(void)
 {
     intmax_t output_flush_interval = 0;
     if (SCConfGetInt("heartbeat.output-flush-interval", &output_flush_interval) == 0) {
@@ -168,7 +169,7 @@ static void *LogFlusherWakeupThread(void *arg)
 
         if (++wait_count == flush_wait_count) {
             worker_flush_count++;
-            WorkerFlushLogs();
+            LogFileFlushAll();
             wait_count = 0;
         }
 

src/log-flush.h

@@ -23,4 +23,5 @@
 #ifndef SURICATA_LOG_FLUSH_H__
 #define SURICATA_LOG_FLUSH_H__
 void LogFlushThreads(void);
+int OutputFlushInterval(void);
 #endif /* SURICATA_LOG_FLUSH_H__ */

src/util-logopenfile.c

@@ -1,4 +1,4 @@
-/* Copyright (C) 2007-2022 Open Information Security Foundation
+/* Copyright (C) 2007-2026 Open Information Security Foundation
  *
  * You can copy, redistribute or modify this Program under the terms of
  * the GNU General Public License version 2 as published by the Free
@@ -33,6 +33,7 @@
 #include "util-path.h"
 #include "util-misc.h"
 #include "util-time.h"
+#include "log-flush.h"
 
 #if defined(HAVE_SYS_UN_H) && defined(HAVE_SYS_SOCKET_H) && defined(HAVE_SYS_TYPES_H)
 #define BUILD_WITH_UNIXSOCKET
@@ -53,6 +54,11 @@ static bool LogFileNewThreadedCtx(LogFileCtx *parent_ctx, const char *log_path,
 // Threaded eve.json identifier
 static SC_ATOMIC_DECL_AND_INIT_WITH_VAL(uint16_t, eve_file_id, 1);
 
+/* Flush list for heartbeat-triggered flushing */
+static SCMutex log_file_flush_mutex = SCMUTEX_INITIALIZER;
+static TAILQ_HEAD(, LogFileFlushEntry_) log_file_flush_list = TAILQ_HEAD_INITIALIZER(
+        log_file_flush_list);
+
 #ifdef BUILD_WITH_UNIXSOCKET
 /** \brief connect to the indicated local stream socket, logging any errors
  *  \param path filesystem path to connect to
@@ -640,6 +646,10 @@ int SCConfLogOpenGeneric(
         if (rotate) {
             OutputRegisterFileRotationFlag(&log_ctx->rotation_flag);
         }
+        /* Register non-threaded regular files for direct heartbeat flushing */
+        if (!log_ctx->threaded && log_ctx->is_regular) {
+            LogFileRegisterForFlush(log_ctx);
+        }
     } else {
         SCLogError("Invalid entry for "
                    "%s.filetype.  Expected \"regular\" (default), \"unix_stream\", "
@@ -882,9 +892,15 @@ static bool LogFileNewThreadedCtx(LogFileCtx *parent_ctx, const char *log_path,
             goto error;
         }
         thread->is_regular = true;
-        thread->Write = SCLogFileWriteNoLock;
+        if (OutputFlushInterval() > 0) {
-        thread->Close = SCLogFileCloseNoLock;
+            thread->Write = SCLogFileWrite;
+            thread->Close = SCLogFileClose;
+        } else {
+            thread->Write = SCLogFileWriteNoLock;
+            thread->Close = SCLogFileCloseNoLock;
+        }
         OutputRegisterFileRotationFlag(&thread->rotation_flag);
+        LogFileRegisterForFlush(thread);
     } else if (parent_ctx->type == LOGFILE_TYPE_FILETYPE) {
         entry->slot_number = SC_ATOMIC_ADD(eve_file_id, 1);
         SCLogDebug("%s - thread %d [slot %d]", log_path, entry->internal_thread_id,
@@ -923,6 +939,11 @@ int LogFileFreeCtx(LogFileCtx *lf_ctx)
         SCReturnInt(0);
     }
 
+    /* Unregister from flush list first, before closing files.
+     * This ensures the heartbeat thread won't try to flush a context
+     * that's being destroyed. */
+    LogFileUnregisterForFlush(lf_ctx);
+
     if (lf_ctx->type == LOGFILE_TYPE_FILETYPE && lf_ctx->filetype.filetype->ThreadDeinit) {
         lf_ctx->filetype.filetype->ThreadDeinit(
                 lf_ctx->filetype.init_data, lf_ctx->filetype.thread_data);
@@ -988,6 +1009,86 @@ void LogFileFlush(LogFileCtx *file_ctx)
     file_ctx->Flush(file_ctx);
 }
 
+/**
+ * \brief Register a LogFileCtx for flush operations
+ *
+ * Adds a LogFileCtx to the global flush list so the heartbeat thread
+ * can flush it directly without using pseudo packets.
+ *
+ * \param ctx The LogFileCtx to register (must be LOGFILE_TYPE_FILE)
+ */
+void LogFileRegisterForFlush(LogFileCtx *ctx)
+{
+    if (!OutputFlushInterval()) {
+        SCLogDebug("heartbeat disabled; skipping flush registration");
+        return;
+    }
+
+    if (ctx == NULL || ctx->type != LOGFILE_TYPE_FILE) {
+        return;
+    }
+
+    LogFileFlushEntry *entry = SCMalloc(sizeof(LogFileFlushEntry));
+    if (entry == NULL) {
+        SCLogError("Unable to allocate memory for flush entry");
+        return;
+    }
+
+    entry->ctx = ctx;
+
+    SCMutexLock(&log_file_flush_mutex);
+    TAILQ_INSERT_TAIL(&log_file_flush_list, entry, entries);
+    SCMutexUnlock(&log_file_flush_mutex);
+}
+
+/**
+ * \brief Unregister a LogFileCtx from flush operations
+ *
+ * Removes a LogFileCtx from the global flush list.
+ *
+ * \param ctx The LogFileCtx to unregister
+ */
+void LogFileUnregisterForFlush(LogFileCtx *ctx)
+{
+    if (!OutputFlushInterval()) {
+        SCLogDebug("heartbeat disabled; skipping flush deregistration");
+        return;
+    }
+
+    if (ctx == NULL) {
+        return;
+    }
+
+    SCMutexLock(&log_file_flush_mutex);
+    LogFileFlushEntry *entry, *safe;
+    TAILQ_FOREACH_SAFE (entry, &log_file_flush_list, entries, safe) {
+        if (entry->ctx == ctx) {
+            TAILQ_REMOVE(&log_file_flush_list, entry, entries);
+            SCFree(entry);
+            break;
+        }
+    }
+    SCMutexUnlock(&log_file_flush_mutex);
+}
+
+/**
+ * \brief Flush all registered LogFileCtx instances
+ *
+ * Called by the heartbeat thread to flush all active file-based loggers.
+ * Iterates through the flush list and calls LogFileFlush on each context.
+ */
+void LogFileFlushAll(void)
+{
+    SCMutexLock(&log_file_flush_mutex);
+    LogFileFlushEntry *entry;
+    TAILQ_FOREACH (entry, &log_file_flush_list, entries) {
+        if (entry->ctx != NULL) {
+            LogFileFlush(entry->ctx);
+        }
+    }
+    SCMutexUnlock(&log_file_flush_mutex);
+}
+
 int LogFileWrite(LogFileCtx *file_ctx, MemBuffer *buffer)
 {
     if (file_ctx->type == LOGFILE_TYPE_FILE || file_ctx->type == LOGFILE_TYPE_UNIX_DGRAM ||

src/util-logopenfile.h

@@ -1,4 +1,4 @@
-/* Copyright (C) 2007-2021 Open Information Security Foundation
+/* Copyright (C) 2007-2026 Open Information Security Foundation
  *
  * You can copy, redistribute or modify this Program under the terms of
  * the GNU General Public License version 2 as published by the Free
@@ -68,6 +68,11 @@ typedef struct LogFileTypeCtx_ {
     void *thread_data;
 } LogFileTypeCtx;
 
+typedef struct LogFileFlushEntry_ {
+    struct LogFileCtx_ *ctx;
+    TAILQ_ENTRY(LogFileFlushEntry_) entries;
+} LogFileFlushEntry;
+
 /** Global structure for Output Context */
 typedef struct LogFileCtx_ {
     union {
@@ -184,4 +189,9 @@ int SCConfLogOpenGeneric(SCConfNode *conf, LogFileCtx *, const char *, int);
 int SCConfLogReopen(LogFileCtx *);
 bool SCLogOpenThreadedFile(const char *log_path, const char *append, LogFileCtx *parent_ctx);
 
+/* Flush list management functions */
+void LogFileRegisterForFlush(LogFileCtx *ctx);
+void LogFileUnregisterForFlush(LogFileCtx *ctx);
+void LogFileFlushAll(void);
+
 #endif /* SURICATA_UTIL_LOGOPENFILE_H */