aiden
/
suricata
mirror of https://github.com/OISF/suricata
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

output/flush: Remove pkt-based flush logic

Browse Source 
Remove packet-based flush logic in favor of simpler solution

Issue: 8286
pull/15091/head
Jeff Lucovsky 5 months ago committed by Victor Julien
parent a78911fce7
commit d0ba1c4c5e
6 changed files with 7 additions and 147 deletions
Show Stats Download Patch File Download Diff File

6
src/decode.h
Unescape Escape View File

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

32
src/detect-engine.c
Unescape Escape View File

@ -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

2
src/detect-engine.h
Unescape Escape View File

@ -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 */

23
src/flow-worker.c
Unescape Escape View File

@ -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) {

2
src/flow-worker.h
Unescape Escape View File

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

89
src/log-flush.c
Unescape Escape View File

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

Write Preview
Loading…
Cancel
Save