diff --git a/src/log-flush.c b/src/log-flush.c index 8da1157df7..cebfa44ebe 100644 --- a/src/log-flush.c +++ b/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; } diff --git a/src/log-flush.h b/src/log-flush.h index e201942b2e..8676e75745 100644 --- a/src/log-flush.h +++ b/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__ */ diff --git a/src/util-logopenfile.c b/src/util-logopenfile.c index 52170113a0..4e8511b63f 100644 --- a/src/util-logopenfile.c +++ b/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; - thread->Close = SCLogFileCloseNoLock; + if (OutputFlushInterval() > 0) { + 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 || diff --git a/src/util-logopenfile.h b/src/util-logopenfile.h index e12df01af6..09846ce869 100644 --- a/src/util-logopenfile.h +++ b/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 */