Mirror/zeek
1
0
Fork 0
mirror of https://github.com/zeek/zeek.git synced 2025-10-02 06:38:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

logging: Dedicated log flush timer

Browse source 
Log flushing is currently triggered based on the threading heartbeat timer
of WriterBackends and the hard-coded WRITE_BUFFER_SIZE 1000.

This change introduces a separate timer that is managed by the logger
manager instead of piggy-backing on the heartbeat timer, as well as a
const &redef for the buffer size.

This allows to modify the log flush frequency and batch size independently
of the threading heartbeat interval. Later, this will allow to re-use the
buffering and flushing logic of writer frontends for non-Broker cluster
backends, too.

One change here is that even frontends that do not have a backend will
be flushed regularly. This is wanted for non-Broker backends and should be
very cheap. Possibly, Broker can piggy back on this timer down the road, too,
rather than using its own script-level timer (see Broker::log_flush()).
This commit is contained in:
Arne Welzel 2024-09-27 11:37:23 +02:00
parent 77b9510c8a
commit 0d925e935e
11 changed files with 87 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

27
scripts/base/init-bare.zeek
View file

@ -2903,6 +2903,33 @@ export {
} # end export
module Log;
export {
## Default interval for flushing the write buffers of all
## enabled log streams.
##
## In earlier Zeek releases this was governed by :zeek:see:`Threading::heartbeat_interval`.
## For Broker, see also :zeek::see:`Broker::log_batch_interval`.
##
## .. :zeek:see:`Log::flush`
## .. :zeek:see:`Log::set_buf`
## .. :zeek:see:`Log::write_buffer_size`
const flush_interval = 1.0sec &redef;
## Default maximum size of the log write buffer per filter/path pair.
## If this many log writes are buffered, the writer frontend flushes
## its writes to its backend before flush_interval expires.
##
## In earlier Zeek releases this was hard-coded to 1000.
##
## .. :zeek:see:`Log::flush`
## .. :zeek:see:`Log::set_buf`
## .. :zeek:see:`Log::flush_interval`
const write_buffer_size = 1000 &redef;
} # end export
module POP3;
export {

1
src/Timer.cc
View file

@ -50,6 +50,7 @@ const char* TimerNames[] = {
"ThreadHeartbeat",
"UnknownProtocolExpire",
"LogDelayExpire",
"LogFlushWriteBufferTimer",
};
const char* timer_type_to_string(TimerType type) { return TimerNames[type]; }

3
src/Timer.h
View file

@ -57,8 +57,9 @@ enum TimerType : uint8_t {
TIMER_THREAD_HEARTBEAT,
TIMER_UNKNOWN_PROTOCOL_EXPIRE,
TIMER_LOG_DELAY_EXPIRE,
TIMER_LOG_FLUSH_WRITE_BUFFER,
};
constexpr int NUM_TIMER_TYPES = int(TIMER_LOG_DELAY_EXPIRE) + 1;
constexpr int NUM_TIMER_TYPES = int(TIMER_LOG_FLUSH_WRITE_BUFFER) + 1;
extern const char* timer_type_to_string(TimerType type);

3
src/const.bif
View file

@ -29,3 +29,6 @@ const Tunnel::ip_tunnel_timeout: interval;
const Tunnel::validate_vxlan_checksums: bool;
const Threading::heartbeat_interval: interval;
const Log::flush_interval: interval;
const Log::write_buffer_size: count;

36
src/logging/Manager.cc
View file

@ -16,6 +16,7 @@
#include "zeek/NetVar.h"
#include "zeek/OpaqueVal.h"
#include "zeek/RunState.h"
#include "zeek/Timer.h"
#include "zeek/Type.h"
#include "zeek/broker/Manager.h"
#include "zeek/input.h"
@ -39,6 +40,21 @@ extern zeek::OpaqueTypePtr log_delay_token_type;
namespace zeek::logging {
namespace detail {
// A timer that regularly flushes the write buffer of all WriterFrontends.
class LogFlushWriteBufferTimer : public zeek::detail::Timer {
public:
explicit LogFlushWriteBufferTimer(double t) : Timer(t, zeek::detail::TIMER_LOG_FLUSH_WRITE_BUFFER) {}
void Dispatch(double t, bool is_expire) override {
zeek::log_mgr->FlushAllWriteBuffers();
if ( ! is_expire )
zeek::log_mgr->StartLogFlushTimer();
}
};
using DelayTokenType = zeek_uint_t;
class DelayInfo;
@ -414,7 +430,6 @@ void Manager::Stream::DispatchDelayExpiredTimer(double t, bool is_expire) {
ScheduleLogDelayExpiredTimer(delay_queue.front()->ExpireTime());
}
Manager::Manager()
: plugin::ComponentManager<logging::Component>("Log", "Writer"),
total_log_stream_writes_family(telemetry_mgr->CounterFamily("zeek", "log-stream-writes", {"module", "stream"},
@ -598,6 +613,9 @@ bool Manager::CreateStream(EnumVal* id, RecordVal* sval) {
DBG_LOG(DBG_LOGGING, "Created new logging stream '%s', raising event %s", streams[idx]->name.c_str(),
event ? streams[idx]->event->Name() : "<none>");
if ( ! log_flush_timer )
StartLogFlushTimer();
return true;
}
@ -2033,4 +2051,20 @@ bool Manager::FinishedRotation(WriterFrontend* writer, const char* new_name, con
return result;
}
void Manager::FlushAllWriteBuffers() {
for ( const auto* s : zeek::log_mgr->streams ) {
if ( ! s ) // may store nullptr
continue;
for ( const auto& [_, info] : s->writers )
info->writer->FlushWriteBuffer();
}
}
void Manager::StartLogFlushTimer() {
double next_t = zeek::run_state::network_time + BifConst::Log::flush_interval;
log_flush_timer = new detail::LogFlushWriteBufferTimer(next_t);
zeek::detail::timer_mgr->Add(log_flush_timer);
}
} // namespace zeek::logging

12
src/logging/Manager.h
View file

@ -32,6 +32,8 @@ class RotationTimer;
namespace detail {
class LogFlushWriteBufferTimer;
class DelayInfo;
using WriteIdx = uint64_t;
@ -348,6 +350,7 @@ protected:
friend class RotationFinishedMessage;
friend class RotationFailedMessage;
friend class RotationTimer;
friend class detail::LogFlushWriteBufferTimer;
// Instantiates a new WriterBackend of the given type (note that
// doing so creates a new thread!).
@ -364,6 +367,12 @@ protected:
bool FinishedRotation(WriterFrontend* writer, const char* new_name, const char* old_name, double open, double close,
bool success, bool terminating);
// Flush write buffers of all writers.
void FlushAllWriteBuffers();
// Start the regular log flushing timer.
void StartLogFlushTimer();
private:
struct Filter;
struct Stream;
@ -404,6 +413,9 @@ private:
zeek_uint_t last_delay_token = 0;
std::vector<detail::WriteContext> active_writes;
// Timer for flushing write buffers of frontends.
detail::LogFlushWriteBufferTimer* log_flush_timer = nullptr;
};
} // namespace logging

12
src/logging/WriterBackend.cc
View file

@ -46,17 +46,6 @@ private:
bool terminating;
};
class FlushWriteBufferMessage final : public threading::OutputMessage<WriterFrontend> {
public:
FlushWriteBufferMessage(WriterFrontend* writer)
: threading::OutputMessage<WriterFrontend>("FlushWriteBuffer", writer) {}
bool Process() override {
Object()->FlushWriteBuffer();
return true;
}
};
class DisableMessage final : public threading::OutputMessage<WriterFrontend> {
public:
DisableMessage(WriterFrontend* writer) : threading::OutputMessage<WriterFrontend>("Disable", writer) {}
@ -305,7 +294,6 @@ bool WriterBackend::OnHeartbeat(double network_time, double current_time) {
if ( Failed() )
return true;
SendOut(new FlushWriteBufferMessage(frontend));
return DoHeartbeat(network_time, current_time);
}

2
src/logging/WriterFrontend.cc
View file

@ -89,7 +89,7 @@ private:
WriterFrontend::WriterFrontend(const WriterBackend::WriterInfo& arg_info, EnumVal* arg_stream, EnumVal* arg_writer,
bool arg_local, bool arg_remote)
: write_buffer(detail::WriteBuffer(WRITER_BUFFER_SIZE)) {
: write_buffer(detail::WriteBuffer(BifConst::Log::write_buffer_size)) {
stream = arg_stream;
writer = arg_writer;
Ref(stream);

4
src/logging/WriterFrontend.h
View file

@ -261,9 +261,7 @@ protected:
int num_fields; // The number of log fields.
const threading::Field* const* fields; // The log fields.
// Buffer for bulk writes.
static const int WRITER_BUFFER_SIZE = 1000;
detail::WriteBuffer write_buffer; // Buffer of size WRITER_BUFFER_SIZE.
detail::WriteBuffer write_buffer; // Buffer for bulk writes.
};
} // namespace zeek::logging

4
testing/btest/Baseline/language.expire_func-copy/output
View file

@ -18,10 +18,10 @@
@XXXXXXXXXX.XXXXXX expired [orig_h=fe80::20c:29ff:febd:6f01, orig_p=5353/udp, resp_h=ff02::fb, resp_p=5353/udp]
@XXXXXXXXXX.XXXXXX expired [orig_h=172.16.238.1, orig_p=49657/tcp, resp_h=172.16.238.131, resp_p=80/tcp]
@XXXXXXXXXX.XXXXXX expired [orig_h=172.16.238.1, orig_p=49656/tcp, resp_h=172.16.238.131, resp_p=22/tcp]
@XXXXXXXXXX.XXXXXX expired [orig_h=172.16.238.131, orig_p=45126/udp, resp_h=172.16.238.2, resp_p=53/udp]
@XXXXXXXXXX.XXXXXX expired [orig_h=172.16.238.1, orig_p=49659/tcp, resp_h=172.16.238.131, resp_p=21/tcp]
@XXXXXXXXXX.XXXXXX expired copy [orig_h=172.16.238.131, orig_p=45126/udp, resp_h=172.16.238.2, resp_p=53/udp]
@XXXXXXXXXX.XXXXXX expired copy [orig_h=172.16.238.1, orig_p=49659/tcp, resp_h=172.16.238.131, resp_p=21/tcp]
@XXXXXXXXXX.XXXXXX expired [orig_h=172.16.238.131, orig_p=45126/udp, resp_h=172.16.238.2, resp_p=53/udp]
@XXXXXXXXXX.XXXXXX expired [orig_h=172.16.238.1, orig_p=49659/tcp, resp_h=172.16.238.131, resp_p=21/tcp]
@XXXXXXXXXX.XXXXXX expired [orig_h=172.16.238.131, orig_p=53102/udp, resp_h=172.16.238.2, resp_p=53/udp]
@XXXXXXXXXX.XXXXXX expired [orig_h=172.16.238.131, orig_p=48621/udp, resp_h=172.16.238.2, resp_p=53/udp]
@XXXXXXXXXX.XXXXXX expired [orig_h=172.16.238.131, orig_p=33109/udp, resp_h=172.16.238.2, resp_p=53/udp]

6
testing/btest/Baseline/scripts.base.frameworks.netcontrol.rule-added-hook-2/netcontrol.log
View file

@ -16,10 +16,10 @@ XXXXXXXXXX.XXXXXX 2 NetControl::RULE ADD NetControl::REQUESTED NetControl::DROP
XXXXXXXXXX.XXXXXX 2 NetControl::RULE ADD NetControl::REQUESTED NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-2
XXXXXXXXXX.XXXXXX 2 NetControl::RULE ADD NetControl::SUCCEEDED NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-1
XXXXXXXXXX.XXXXXX 2 NetControl::RULE ADD NetControl::EXISTS NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-2
XXXXXXXXXX.XXXXXX 2 NetControl::RULE EXPIRE NetControl::TIMEOUT NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-1
XXXXXXXXXX.XXXXXX 2 NetControl::RULE REMOVE NetControl::REQUESTED NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-1
XXXXXXXXXX.XXXXXX 2 NetControl::RULE EXPIRE NetControl::TIMEOUT NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-2
XXXXXXXXXX.XXXXXX 2 NetControl::RULE REMOVE NetControl::REQUESTED NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-2
XXXXXXXXXX.XXXXXX 2 NetControl::RULE REMOVE NetControl::SUCCEEDED NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-1
XXXXXXXXXX.XXXXXX 2 NetControl::RULE EXPIRE NetControl::TIMEOUT NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-1
XXXXXXXXXX.XXXXXX 2 NetControl::RULE REMOVE NetControl::REQUESTED NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-1
XXXXXXXXXX.XXXXXX 2 NetControl::RULE REMOVE NetControl::SUCCEEDED NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-2
XXXXXXXXXX.XXXXXX 2 NetControl::RULE REMOVE NetControl::SUCCEEDED NetControl::DROP NetControl::FORWARD NetControl::ADDRESS 192.168.18.50/32 - - 0 1.000000 - plugin-1
#close XXXX-XX-XX-XX-XX-XX