diff --git a/src/decode.h b/src/decode.h index cef221f3ca..1f0be7cc22 100644 --- a/src/decode.h +++ b/src/decode.h @@ -1342,12 +1342,8 @@ void DecodeUnregisterCounters(void); #define PKT_FIRST_ALERTS BIT_U32(29) #define PKT_FIRST_TAG BIT_U32(30) -#define PKT_PSEUDO_LOG_FLUSH BIT_U32(31) /**< Detect/log flush for protocol upgrade */ - /** \brief return 1 if the packet is a pseudo packet */ -#define PKT_IS_PSEUDOPKT(p) \ - ((p)->flags & (PKT_PSEUDO_STREAM_END|PKT_PSEUDO_DETECTLOG_FLUSH)) -#define PKT_IS_FLUSHPKT(p) ((p)->flags & (PKT_PSEUDO_LOG_FLUSH)) +#define PKT_IS_PSEUDOPKT(p) ((p)->flags & (PKT_PSEUDO_STREAM_END | PKT_PSEUDO_DETECTLOG_FLUSH)) #define PKT_SET_SRC(p, src_val) ((p)->pkt_src = src_val) diff --git a/src/detect-engine.c b/src/detect-engine.c index 2aac613754..9c4d031328 100644 --- a/src/detect-engine.c +++ b/src/detect-engine.c @@ -2232,38 +2232,6 @@ int DetectEngineInspectPktBufferGeneric( } } -/** \internal - * \brief inject a pseudo packet into each detect thread - * if the thread should flush its output logs. - */ -void InjectPacketsForFlush(ThreadVars **detect_tvs, int no_of_detect_tvs) -{ - /* inject a fake packet if the detect thread that needs it. This function - * is called when a heartbeat log-flush request has been made - * and it should process a pseudo packet and flush its output logs - * to speed the process. */ -#if DEBUG - int count = 0; -#endif - for (int i = 0; i < no_of_detect_tvs; i++) { - if (detect_tvs[i]) { // && detect_tvs[i]->inq != NULL) { - Packet *p = PacketGetFromAlloc(); - if (p != NULL) { - SCLogDebug("Injecting pkt for tv %s[i=%d] %d", detect_tvs[i]->name, i, count++); - p->flags |= PKT_PSEUDO_STREAM_END; - p->flags |= PKT_PSEUDO_LOG_FLUSH; - PKT_SET_SRC(p, PKT_SRC_DETECT_RELOAD_FLUSH); - PacketQueue *q = detect_tvs[i]->stream_pq; - SCMutexLock(&q->mutex_q); - PacketEnqueue(q, p); - SCCondSignal(&q->cond_q); - SCMutexUnlock(&q->mutex_q); - } - } - } - SCLogDebug("leaving: thread notification count = %d", count); -} - /** \internal * \brief inject a pseudo packet into each detect thread * -that doesn't use the new det_ctx yet diff --git a/src/detect-engine.h b/src/detect-engine.h index 5aca85a688..a2cd0be4d6 100644 --- a/src/detect-engine.h +++ b/src/detect-engine.h @@ -210,6 +210,4 @@ void DetectLowerSetupCallback( void DeStateRegisterTests(void); -/* packet injection */ -void InjectPacketsForFlush(ThreadVars **detect_tvs, int no_of_detect_tvs); #endif /* SURICATA_DETECT_ENGINE_H */ diff --git a/src/flow-worker.c b/src/flow-worker.c index 8a4d73821b..0d1bb86c3d 100644 --- a/src/flow-worker.c +++ b/src/flow-worker.c @@ -73,8 +73,6 @@ typedef struct FlowWorkerThreadData_ { SC_ATOMIC_DECLARE(DetectEngineThreadCtxPtr, detect_thread); - SC_ATOMIC_DECLARE(bool, flush_ack); - void *output_thread; /* Output thread data. */ void *output_thread_flow; /* Output thread data. */ @@ -567,15 +565,6 @@ static TmEcode FlowWorker(ThreadVars *tv, Packet *p, void *data) SCLogDebug("packet %" PRIu64, PcapPacketCntGet(p)); - if ((PKT_IS_FLUSHPKT(p))) { - SCLogDebug("thread %s flushing", tv->printable_name); - OutputLoggerFlush(tv, p, fw->output_thread); - /* Ack if a flush was requested */ - bool notset = false; - SC_ATOMIC_CAS(&fw->flush_ack, notset, true); - return TM_ECODE_OK; - } - /* handle Flow */ if (det_ctx != NULL && det_ctx->de_ctx->PreFlowHook != NULL) { const uint8_t action = det_ctx->de_ctx->PreFlowHook(tv, det_ctx, p); @@ -759,18 +748,6 @@ void *FlowWorkerGetThreadData(void *flow_worker) return (FlowWorkerThreadData *)flow_worker; } -bool FlowWorkerGetFlushAck(void *flow_worker) -{ - FlowWorkerThreadData *fw = flow_worker; - return SC_ATOMIC_GET(fw->flush_ack) == true; -} - -void FlowWorkerSetFlushAck(void *flow_worker) -{ - FlowWorkerThreadData *fw = flow_worker; - SC_ATOMIC_SET(fw->flush_ack, false); -} - const char *ProfileFlowWorkerIdToString(enum ProfileFlowWorkerId fwi) { switch (fwi) { diff --git a/src/flow-worker.h b/src/flow-worker.h index 6bdea55193..476f296ff4 100644 --- a/src/flow-worker.h +++ b/src/flow-worker.h @@ -33,8 +33,6 @@ const char *ProfileFlowWorkerIdToString(enum ProfileFlowWorkerId fwi); void FlowWorkerReplaceDetectCtx(void *flow_worker, void *detect_ctx); void *FlowWorkerGetDetectCtxPtr(void *flow_worker); void *FlowWorkerGetThreadData(void *flow_worker); -bool FlowWorkerGetFlushAck(void *flow_worker); -void FlowWorkerSetFlushAck(void *flow_worker); void TmModuleFlowWorkerRegister (void); diff --git a/src/log-flush.c b/src/log-flush.c index cebfa44ebe..bfe22e1a0c 100644 --- a/src/log-flush.c +++ b/src/log-flush.c @@ -1,4 +1,4 @@ -/* Copyright (C) 2024 Open Information Security Foundation +/* Copyright (C) 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 @@ -23,10 +23,8 @@ #include "suricata-common.h" #include "suricata.h" -#include "detect.h" -#include "detect-engine.h" -#include "flow-worker.h" #include "log-flush.h" +#include "util-logopenfile.h" #include "tm-threads.h" #include "conf.h" #include "conf-yaml-loader.h" @@ -34,90 +32,15 @@ #include "util-logopenfile.h" /** - * \brief Trigger detect threads to flush their output logs + * \brief Trigger flush of all registered log files * * This function is intended to be called at regular intervals to force - * buffered log data to be persisted + * buffered log data to be persisted. With the new design, this simply calls + * LogFileFlushAll() which directly flushes all registered file contexts. */ static void WorkerFlushLogs(void) { - SCEnter(); - - /* count detect threads in use */ - uint32_t no_of_detect_tvs = TmThreadCountThreadsByTmmFlags(TM_FLAG_FLOWWORKER_TM); - /* can be zero in unix socket mode */ - if (no_of_detect_tvs == 0) { - return; - } - - /* prepare swap structures */ - void *fw_threads[no_of_detect_tvs]; - ThreadVars *detect_tvs[no_of_detect_tvs]; - memset(fw_threads, 0x00, (no_of_detect_tvs * sizeof(void *))); - memset(detect_tvs, 0x00, (no_of_detect_tvs * sizeof(ThreadVars *))); - - /* start by initiating the log flushes */ - - uint32_t i = 0; - SCMutexLock(&tv_root_lock); - /* get reference to tv's and setup fw_threads array */ - for (ThreadVars *tv = tv_root[TVT_PPT]; tv != NULL; tv = tv->next) { - if ((tv->tmm_flags & TM_FLAG_FLOWWORKER_TM) == 0) { - continue; - } - for (TmSlot *s = tv->tm_slots; s != NULL; s = s->slot_next) { - TmModule *tm = TmModuleGetById(s->tm_id); - if (!(tm->flags & TM_FLAG_FLOWWORKER_TM)) { - continue; - } - - if (suricata_ctl_flags != 0) { - SCMutexUnlock(&tv_root_lock); - goto error; - } - - fw_threads[i] = FlowWorkerGetThreadData(SC_ATOMIC_GET(s->slot_data)); - if (fw_threads[i]) { - FlowWorkerSetFlushAck(fw_threads[i]); - SCLogDebug("Setting flush-ack for thread %s[i=%d]", tv->printable_name, i); - detect_tvs[i] = tv; - } - - i++; - break; - } - } - BUG_ON(i != no_of_detect_tvs); - - SCMutexUnlock(&tv_root_lock); - - SCLogDebug("Creating flush pseudo packets for %d threads", no_of_detect_tvs); - InjectPacketsForFlush(detect_tvs, no_of_detect_tvs); - - uint32_t threads_done = 0; -retry: - for (i = 0; i < no_of_detect_tvs; i++) { - if (suricata_ctl_flags != 0) { - threads_done = no_of_detect_tvs; - break; - } - SleepMsec(1); - if (fw_threads[i] && FlowWorkerGetFlushAck(fw_threads[i])) { - SCLogDebug("thread slot %d has ack'd flush request", i); - threads_done++; - } else if (detect_tvs[i]) { - SCLogDebug("thread slot %d not yet ack'd flush request", i); - TmThreadsCaptureBreakLoop(detect_tvs[i]); - } - } - if (threads_done < no_of_detect_tvs) { - threads_done = 0; - SleepMsec(250); - goto retry; - } - -error: - return; + LogFileFlushAll(); } int OutputFlushInterval(void)