GH-780: Prevent log batches from indefinite buffering

Logs that got sent sparsely or burstily would get buffered for long
periods of time since the logic to flush them only does so on the next
log write.  In the worst case, a subsequent log write could never happen
and cause a log entry to be indefinitely buffered.

This fix introduces a recurring event/timer to simply flush all pending
logs at frequency of Broker::log_batch_interval.

scripts/base/frameworks/broker/main.zeek

@@ -379,6 +379,17 @@ export {
 
 module Broker;
 
+event Broker::log_flush() &priority=10
+	{
+	Broker::flush_logs();
+	schedule Broker::log_batch_interval { Broker::log_flush() };
+	}
+
+event zeek_init()
+	{
+	schedule Broker::log_batch_interval { Broker::log_flush() };
+	}
+
 event retry_listen(a: string, p: port, retry: interval)
 	{
 	listen(a, p, retry);

src/broker/Manager.cc

@@ -131,7 +131,6 @@ Manager::Manager(bool arg_reading_pcaps)
 	after_zeek_init = false;
 	peer_count = 0;
 	log_batch_size = 0;
-	log_batch_interval = 0;
 	log_topic_func = nullptr;
 	vector_of_data_type = nullptr;
 	log_id_type = nullptr;
@@ -147,7 +146,6 @@ void Manager::InitPostScript()
 	DBG_LOG(DBG_BROKER, "Initializing");
 
 	log_batch_size = get_option("Broker::log_batch_size")->AsCount();
-	log_batch_interval = get_option("Broker::log_batch_interval")->AsInterval();
 	default_log_topic_prefix =
 	    get_option("Broker::default_log_topic_prefix")->AsString()->CheckString();
 	log_topic_func = get_option("Broker::log_topic")->AsFunc();
@@ -580,8 +578,7 @@ bool Manager::PublishLogWrite(EnumVal* stream, EnumVal* writer, string path, int
 	auto& pending_batch = lb.msgs[topic];
 	pending_batch.emplace_back(msg.move_data());
 
-	if ( lb.message_count >= log_batch_size ||
+	if ( lb.message_count >= log_batch_size )
-	     (network_time - lb.last_flush >= log_batch_interval ) )
 		statistics.num_logs_outgoing += lb.Flush(bstate->endpoint, log_batch_size);
 
 	return true;
@@ -608,7 +605,6 @@ size_t Manager::LogBuffer::Flush(broker::endpoint& endpoint, size_t log_batch_si
 		}
 
 	auto rval = message_count;
-	last_flush = network_time;
 	message_count = 0;
 	return rval;
 	}
@@ -619,8 +615,9 @@ size_t Manager::FlushLogBuffers()
 	auto rval = 0u;
 
 	for ( auto& lb : log_buffers )
-		rval += lb.Flush(bstate->endpoint, log_batch_interval);
+		rval += lb.Flush(bstate->endpoint, log_batch_size);
 
+	statistics.num_logs_outgoing += rval;
 	return rval;
 	}
 

src/broker/Manager.h

@@ -354,7 +354,6 @@ private:
 	struct LogBuffer {
 		// Indexed by topic string.
 		std::unordered_map<std::string, broker::vector> msgs;
-		double last_flush;
 		size_t message_count;
 
 		size_t Flush(broker::endpoint& endpoint, size_t batch_size);
@@ -389,7 +388,6 @@ private:
 	int peer_count;
 
 	size_t log_batch_size;
-	double log_batch_interval;
 	Func* log_topic_func;
 	VectorType* vector_of_data_type;
 	EnumType* log_id_type;

testing/btest/Baseline/broker.remote_log_batch/recv.test.log

@@ -0,0 +1,11 @@
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	test
+#open	2020-02-05-20-53-15
+#fields	msg	num
+#types	string	count
+ping	0
+ping	1
+#close	2020-02-05-20-53-16

testing/btest/Baseline/broker.remote_log_batch/send.send.out

@@ -0,0 +1 @@
+Broker::peer_added, 127.0.0.1

testing/btest/Baseline/broker.remote_log_batch/send.test.log

@@ -0,0 +1,11 @@
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	test
+#open	2020-02-05-20-53-15
+#fields	msg	num
+#types	string	count
+ping	0
+ping	1
+#close	2020-02-05-20-53-16

testing/btest/Baseline/language.paraglob-serialization/recv.recv.out

@@ -9,4 +9,3 @@ is_remote should be T, and is, T
 receiver got ping number: 3
 [*, *ello, hello]
 is_remote should be T, and is, T
-[num_peers=1, num_stores=0, num_pending_queries=0, num_events_incoming=4, num_events_outgoing=3, num_logs_incoming=0, num_logs_outgoing=1, num_ids_incoming=0, num_ids_outgoing=0]

testing/btest/Baseline/plugins.hooks/output

@@ -276,7 +276,7 @@
 0.000000   MetaHookPost  CallFunction(Log::__create_stream, <frame>, (Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird])) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::__create_stream, <frame>, (X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509])) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::__create_stream, <frame>, (mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql])) -> <no result>
-0.000000   MetaHookPost  CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1579727603.636084, node=zeek, filter=ip or not ip, init=T, success=T])) -> <no result>
+0.000000   MetaHookPost  CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1580932537.132762, node=zeek, filter=ip or not ip, init=T, success=T])) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::add_default_filter, <frame>, (Broker::LOG)) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::add_default_filter, <frame>, (Cluster::LOG)) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::add_default_filter, <frame>, (Config::LOG)) -> <no result>
@@ -457,7 +457,7 @@
 0.000000   MetaHookPost  CallFunction(Log::create_stream, <frame>, (Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird])) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::create_stream, <frame>, (X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509])) -> <no result>
 0.000000   MetaHookPost  CallFunction(Log::create_stream, <frame>, (mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql])) -> <no result>
-0.000000   MetaHookPost  CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1579727603.636084, node=zeek, filter=ip or not ip, init=T, success=T])) -> <no result>
+0.000000   MetaHookPost  CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1580932537.132762, node=zeek, filter=ip or not ip, init=T, success=T])) -> <no result>
 0.000000   MetaHookPost  CallFunction(NetControl::check_plugins, <frame>, ()) -> <no result>
 0.000000   MetaHookPost  CallFunction(NetControl::init, <null>, ()) -> <no result>
 0.000000   MetaHookPost  CallFunction(Notice::want_pp, <frame>, ()) -> <no result>
@@ -1180,7 +1180,7 @@
 0.000000   MetaHookPre   CallFunction(Log::__create_stream, <frame>, (Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird]))
 0.000000   MetaHookPre   CallFunction(Log::__create_stream, <frame>, (X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509]))
 0.000000   MetaHookPre   CallFunction(Log::__create_stream, <frame>, (mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql]))
-0.000000   MetaHookPre   CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1579727603.636084, node=zeek, filter=ip or not ip, init=T, success=T]))
+0.000000   MetaHookPre   CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1580932537.132762, node=zeek, filter=ip or not ip, init=T, success=T]))
 0.000000   MetaHookPre   CallFunction(Log::add_default_filter, <frame>, (Broker::LOG))
 0.000000   MetaHookPre   CallFunction(Log::add_default_filter, <frame>, (Cluster::LOG))
 0.000000   MetaHookPre   CallFunction(Log::add_default_filter, <frame>, (Config::LOG))
@@ -1361,7 +1361,7 @@
 0.000000   MetaHookPre   CallFunction(Log::create_stream, <frame>, (Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird]))
 0.000000   MetaHookPre   CallFunction(Log::create_stream, <frame>, (X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509]))
 0.000000   MetaHookPre   CallFunction(Log::create_stream, <frame>, (mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql]))
-0.000000   MetaHookPre   CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1579727603.636084, node=zeek, filter=ip or not ip, init=T, success=T]))
+0.000000   MetaHookPre   CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1580932537.132762, node=zeek, filter=ip or not ip, init=T, success=T]))
 0.000000   MetaHookPre   CallFunction(NetControl::check_plugins, <frame>, ())
 0.000000   MetaHookPre   CallFunction(NetControl::init, <null>, ())
 0.000000   MetaHookPre   CallFunction(Notice::want_pp, <frame>, ())
@@ -2083,7 +2083,7 @@
 0.000000 | HookCallFunction Log::__create_stream(Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird])
 0.000000 | HookCallFunction Log::__create_stream(X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509])
 0.000000 | HookCallFunction Log::__create_stream(mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql])
-0.000000 | HookCallFunction Log::__write(PacketFilter::LOG, [ts=1579727603.636084, node=zeek, filter=ip or not ip, init=T, success=T])
+0.000000 | HookCallFunction Log::__write(PacketFilter::LOG, [ts=1580932537.132762, node=zeek, filter=ip or not ip, init=T, success=T])
 0.000000 | HookCallFunction Log::add_default_filter(Broker::LOG)
 0.000000 | HookCallFunction Log::add_default_filter(Cluster::LOG)
 0.000000 | HookCallFunction Log::add_default_filter(Config::LOG)
@@ -2264,7 +2264,7 @@
 0.000000 | HookCallFunction Log::create_stream(Weird::LOG, [columns=Weird::Info, ev=Weird::log_weird, path=weird])
 0.000000 | HookCallFunction Log::create_stream(X509::LOG, [columns=X509::Info, ev=X509::log_x509, path=x509])
 0.000000 | HookCallFunction Log::create_stream(mysql::LOG, [columns=MySQL::Info, ev=MySQL::log_mysql, path=mysql])
-0.000000 | HookCallFunction Log::write(PacketFilter::LOG, [ts=1579727603.636084, node=zeek, filter=ip or not ip, init=T, success=T])
+0.000000 | HookCallFunction Log::write(PacketFilter::LOG, [ts=1580932537.132762, node=zeek, filter=ip or not ip, init=T, success=T])
 0.000000 | HookCallFunction NetControl::check_plugins()
 0.000000 | HookCallFunction NetControl::init()
 0.000000 | HookCallFunction Notice::want_pp()
@@ -2705,27 +2705,35 @@
 0.000000 | HookLoadFile  base<...>/xmpp
 0.000000 | HookLoadFile  base<...>/zeek.bif.zeek
 0.000000 | HookLogInit   packet_filter 1/1 {ts (time), node (string), filter (string), init (bool), success (bool)}
-0.000000 | HookLogWrite  packet_filter [ts=1579727603.636084, node=zeek, filter=ip or not ip, init=T, success=T]
+0.000000 | HookLogWrite  packet_filter [ts=1580932537.132762, node=zeek, filter=ip or not ip, init=T, success=T]
 0.000000 | HookQueueEvent NetControl::init()
 0.000000 | HookQueueEvent filter_change_tracking()
 0.000000 | HookQueueEvent zeek_init()
 1362692526.869344   MetaHookPost  BroObjDtor(<void ptr>) -> <void>
+1362692526.869344   MetaHookPost  CallFunction(Broker::__flush_logs, <frame>, ()) -> <no result>
+1362692526.869344   MetaHookPost  CallFunction(Broker::flush_logs, <frame>, ()) -> <no result>
+1362692526.869344   MetaHookPost  CallFunction(Broker::log_flush, <null>, ()) -> <no result>
 1362692526.869344   MetaHookPost  CallFunction(ChecksumOffloading::check, <null>, ()) -> <no result>
 1362692526.869344   MetaHookPost  CallFunction(filter_change_tracking, <null>, ()) -> <no result>
 1362692526.869344   MetaHookPost  CallFunction(get_net_stats, <frame>, ()) -> <no result>
 1362692526.869344   MetaHookPost  CallFunction(new_connection, <null>, ([id=[orig_h=141.142.228.5, orig_p=59856<...>/tcp], orig=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=c8:bc:c8:96:d2:a0], resp=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=00:10:db:88:d2:ef], start_time=1362692526.869344, duration=0.0, service={}, history=, uid=CHhAvVGS1DHFjwGM9, tunnel=<uninitialized>, vlan=<uninitialized>, inner_vlan=<uninitialized>, successful=F, dpd=<uninitialized>, dpd_state=<uninitialized>, conn=<uninitialized>, extract_orig=F, extract_resp=F, thresholds=<uninitialized>, dce_rpc=<uninitialized>, dce_rpc_state=<uninitialized>, dce_rpc_backing=<uninitialized>, dhcp=<uninitialized>, dnp3=<uninitialized>, dns=<uninitialized>, dns_state=<uninitialized>, ftp=<uninitialized>, ftp_data_reuse=F, ssl=<uninitialized>, http=<uninitialized>, http_state=<uninitialized>, irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])) -> <no result>
 1362692526.869344   MetaHookPost  DrainEvents() -> <void>
+1362692526.869344   MetaHookPost  QueueEvent(Broker::log_flush()) -> false
 1362692526.869344   MetaHookPost  QueueEvent(ChecksumOffloading::check()) -> false
 1362692526.869344   MetaHookPost  QueueEvent(filter_change_tracking()) -> false
 1362692526.869344   MetaHookPost  QueueEvent(new_connection([id=[orig_h=141.142.228.5, orig_p=59856<...>/tcp], orig=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=c8:bc:c8:96:d2:a0], resp=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=00:10:db:88:d2:ef], start_time=1362692526.869344, duration=0.0, service={}, history=, uid=CHhAvVGS1DHFjwGM9, tunnel=<uninitialized>, vlan=<uninitialized>, inner_vlan=<uninitialized>, successful=F, dpd=<uninitialized>, dpd_state=<uninitialized>, conn=<uninitialized>, extract_orig=F, extract_resp=F, thresholds=<uninitialized>, dce_rpc=<uninitialized>, dce_rpc_state=<uninitialized>, dce_rpc_backing=<uninitialized>, dhcp=<uninitialized>, dnp3=<uninitialized>, dns=<uninitialized>, dns_state=<uninitialized>, ftp=<uninitialized>, ftp_data_reuse=F, ssl=<uninitialized>, http=<uninitialized>, http_state=<uninitialized>, irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])) -> false
 1362692526.869344   MetaHookPost  SetupAnalyzerTree(1362692526.869344(1362692526.869344) TCP 141.142.228.5:59856 -> 192.150.187.43:80) -> <void>
 1362692526.869344   MetaHookPost  UpdateNetworkTime(1362692526.869344) -> <void>
 1362692526.869344   MetaHookPre   BroObjDtor(<void ptr>)
+1362692526.869344   MetaHookPre   CallFunction(Broker::__flush_logs, <frame>, ())
+1362692526.869344   MetaHookPre   CallFunction(Broker::flush_logs, <frame>, ())
+1362692526.869344   MetaHookPre   CallFunction(Broker::log_flush, <null>, ())
 1362692526.869344   MetaHookPre   CallFunction(ChecksumOffloading::check, <null>, ())
 1362692526.869344   MetaHookPre   CallFunction(filter_change_tracking, <null>, ())
 1362692526.869344   MetaHookPre   CallFunction(get_net_stats, <frame>, ())
 1362692526.869344   MetaHookPre   CallFunction(new_connection, <null>, ([id=[orig_h=141.142.228.5, orig_p=59856<...>/tcp], orig=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=c8:bc:c8:96:d2:a0], resp=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=00:10:db:88:d2:ef], start_time=1362692526.869344, duration=0.0, service={}, history=, uid=CHhAvVGS1DHFjwGM9, tunnel=<uninitialized>, vlan=<uninitialized>, inner_vlan=<uninitialized>, successful=F, dpd=<uninitialized>, dpd_state=<uninitialized>, conn=<uninitialized>, extract_orig=F, extract_resp=F, thresholds=<uninitialized>, dce_rpc=<uninitialized>, dce_rpc_state=<uninitialized>, dce_rpc_backing=<uninitialized>, dhcp=<uninitialized>, dnp3=<uninitialized>, dns=<uninitialized>, dns_state=<uninitialized>, ftp=<uninitialized>, ftp_data_reuse=F, ssl=<uninitialized>, http=<uninitialized>, http_state=<uninitialized>, irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>]))
 1362692526.869344   MetaHookPre   DrainEvents()
+1362692526.869344   MetaHookPre   QueueEvent(Broker::log_flush())
 1362692526.869344   MetaHookPre   QueueEvent(ChecksumOffloading::check())
 1362692526.869344   MetaHookPre   QueueEvent(filter_change_tracking())
 1362692526.869344   MetaHookPre   QueueEvent(new_connection([id=[orig_h=141.142.228.5, orig_p=59856<...>/tcp], orig=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=c8:bc:c8:96:d2:a0], resp=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=00:10:db:88:d2:ef], start_time=1362692526.869344, duration=0.0, service={}, history=, uid=CHhAvVGS1DHFjwGM9, tunnel=<uninitialized>, vlan=<uninitialized>, inner_vlan=<uninitialized>, successful=F, dpd=<uninitialized>, dpd_state=<uninitialized>, conn=<uninitialized>, extract_orig=F, extract_resp=F, thresholds=<uninitialized>, dce_rpc=<uninitialized>, dce_rpc_state=<uninitialized>, dce_rpc_backing=<uninitialized>, dhcp=<uninitialized>, dnp3=<uninitialized>, dns=<uninitialized>, dns_state=<uninitialized>, ftp=<uninitialized>, ftp_data_reuse=F, ssl=<uninitialized>, http=<uninitialized>, http_state=<uninitialized>, irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>]))
@@ -2733,16 +2741,20 @@
 1362692526.869344   MetaHookPre   UpdateNetworkTime(1362692526.869344)
 1362692526.869344  | HookBroObjDtor
 1362692526.869344  | HookUpdateNetworkTime 1362692526.869344
+1362692526.869344 | HookCallFunction Broker::__flush_logs()
+1362692526.869344 | HookCallFunction Broker::flush_logs()
+1362692526.869344 | HookCallFunction Broker::log_flush()
 1362692526.869344 | HookCallFunction ChecksumOffloading::check()
 1362692526.869344 | HookCallFunction filter_change_tracking()
 1362692526.869344 | HookCallFunction get_net_stats()
 1362692526.869344 | HookCallFunction new_connection([id=[orig_h=141.142.228.5, orig_p=59856<...>/tcp], orig=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=c8:bc:c8:96:d2:a0], resp=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=00:10:db:88:d2:ef], start_time=1362692526.869344, duration=0.0, service={}, history=, uid=CHhAvVGS1DHFjwGM9, tunnel=<uninitialized>, vlan=<uninitialized>, inner_vlan=<uninitialized>, successful=F, dpd=<uninitialized>, dpd_state=<uninitialized>, conn=<uninitialized>, extract_orig=F, extract_resp=F, thresholds=<uninitialized>, dce_rpc=<uninitialized>, dce_rpc_state=<uninitialized>, dce_rpc_backing=<uninitialized>, dhcp=<uninitialized>, dnp3=<uninitialized>, dns=<uninitialized>, dns_state=<uninitialized>, ftp=<uninitialized>, ftp_data_reuse=F, ssl=<uninitialized>, http=<uninitialized>, http_state=<uninitialized>, irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])
 1362692526.869344 | HookDrainEvents
+1362692526.869344 | HookQueueEvent Broker::log_flush()
 1362692526.869344 | HookQueueEvent ChecksumOffloading::check()
 1362692526.869344 | HookQueueEvent filter_change_tracking()
 1362692526.869344 | HookQueueEvent new_connection([id=[orig_h=141.142.228.5, orig_p=59856<...>/tcp], orig=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=c8:bc:c8:96:d2:a0], resp=[size=0, state=0, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=00:10:db:88:d2:ef], start_time=1362692526.869344, duration=0.0, service={}, history=, uid=CHhAvVGS1DHFjwGM9, tunnel=<uninitialized>, vlan=<uninitialized>, inner_vlan=<uninitialized>, successful=F, dpd=<uninitialized>, dpd_state=<uninitialized>, conn=<uninitialized>, extract_orig=F, extract_resp=F, thresholds=<uninitialized>, dce_rpc=<uninitialized>, dce_rpc_state=<uninitialized>, dce_rpc_backing=<uninitialized>, dhcp=<uninitialized>, dnp3=<uninitialized>, dns=<uninitialized>, dns_state=<uninitialized>, ftp=<uninitialized>, ftp_data_reuse=F, ssl=<uninitialized>, http=<uninitialized>, http_state=<uninitialized>, irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])
 1362692526.869344 | HookSetupAnalyzerTree 1362692526.869344(1362692526.869344) TCP 141.142.228.5:59856 -> 192.150.187.43:80
-1362692526.869344 | RequestObjDtor ChecksumOffloading::check()
+1362692526.869344 | RequestObjDtor Broker::log_flush()
 1362692526.939084   MetaHookPost  CallFunction(connection_established, <null>, ([id=[orig_h=141.142.228.5, orig_p=59856<...>/tcp], orig=[size=0, state=4, num_pkts=1, num_bytes_ip=64, flow_label=0, l2_addr=c8:bc:c8:96:d2:a0], resp=[size=0, state=4, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=00:10:db:88:d2:ef], start_time=1362692526.869344, duration=0.06974, service={}, history=Sh, uid=CHhAvVGS1DHFjwGM9, tunnel=<uninitialized>, vlan=<uninitialized>, inner_vlan=<uninitialized>, successful=F, dpd=<uninitialized>, dpd_state=<uninitialized>, conn=<uninitialized>, extract_orig=F, extract_resp=F, thresholds=<uninitialized>, dce_rpc=<uninitialized>, dce_rpc_state=<uninitialized>, dce_rpc_backing=<uninitialized>, dhcp=<uninitialized>, dnp3=<uninitialized>, dns=<uninitialized>, dns_state=<uninitialized>, ftp=<uninitialized>, ftp_data_reuse=F, ssl=<uninitialized>, http=<uninitialized>, http_state=<uninitialized>, irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])) -> <no result>
 1362692526.939084   MetaHookPost  DrainEvents() -> <void>
 1362692526.939084   MetaHookPost  QueueEvent(connection_established([id=[orig_h=141.142.228.5, orig_p=59856<...>/tcp], orig=[size=0, state=4, num_pkts=1, num_bytes_ip=64, flow_label=0, l2_addr=c8:bc:c8:96:d2:a0], resp=[size=0, state=4, num_pkts=0, num_bytes_ip=0, flow_label=0, l2_addr=00:10:db:88:d2:ef], start_time=1362692526.869344, duration=0.06974, service={}, history=Sh, uid=CHhAvVGS1DHFjwGM9, tunnel=<uninitialized>, vlan=<uninitialized>, inner_vlan=<uninitialized>, successful=F, dpd=<uninitialized>, dpd_state=<uninitialized>, conn=<uninitialized>, extract_orig=F, extract_resp=F, thresholds=<uninitialized>, dce_rpc=<uninitialized>, dce_rpc_state=<uninitialized>, dce_rpc_backing=<uninitialized>, dhcp=<uninitialized>, dnp3=<uninitialized>, dns=<uninitialized>, dns_state=<uninitialized>, ftp=<uninitialized>, ftp_data_reuse=F, ssl=<uninitialized>, http=<uninitialized>, http_state=<uninitialized>, irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])) -> false
@@ -3139,6 +3151,9 @@
 1362692527.080828   MetaHookPre   UpdateNetworkTime(1362692527.080828)
 1362692527.080828  | HookUpdateNetworkTime 1362692527.080828
 1362692527.080828 | HookDrainEvents
+1362692527.080972   MetaHookPost  CallFunction(Broker::__flush_logs, <frame>, ()) -> <no result>
+1362692527.080972   MetaHookPost  CallFunction(Broker::flush_logs, <frame>, ()) -> <no result>
+1362692527.080972   MetaHookPost  CallFunction(Broker::log_flush, <null>, ()) -> <no result>
 1362692527.080972   MetaHookPost  CallFunction(ChecksumOffloading::check, <null>, ()) -> <no result>
 1362692527.080972   MetaHookPost  CallFunction(Conn::conn_state, <frame>, ([id=[orig_h=141.142.228.5, orig_p=59856<...>/plain], current_entity=<uninitialized>, orig_mime_depth=1, resp_mime_depth=1], http_state=[pending={}, current_request=1, current_response=1, trans_depth=1], irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>], tcp)) -> <no result>
 1362692527.080972   MetaHookPost  CallFunction(Conn::determine_service, <frame>, ([id=[orig_h=141.142.228.5, orig_p=59856<...>/plain], current_entity=<uninitialized>, orig_mime_depth=1, resp_mime_depth=1], http_state=[pending={}, current_request=1, current_response=1, trans_depth=1], irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])) -> <no result>
@@ -3167,6 +3182,7 @@
 1362692527.080972   MetaHookPost  DrainEvents() -> <void>
 1362692527.080972   MetaHookPost  LogInit(Log::WRITER_ASCII, default, true, true, conn(1362692527.080972,0.0,0.0), 21, {ts (time), uid (string), id.orig_h (addr), id.orig_p (port), id.resp_h (addr), id.resp_p (port), proto (enum), service (string), duration (interval), orig_bytes (count), resp_bytes (count), conn_state (string), local_orig (bool), local_resp (bool), missed_bytes (count), history (string), orig_pkts (count), orig_ip_bytes (count), resp_pkts (count), resp_ip_bytes (count), tunnel_parents (set[string])}) -> <void>
 1362692527.080972   MetaHookPost  LogWrite(Log::WRITER_ASCII, default, conn(1362692527.080972,0.0,0.0), 21, {ts (time), uid (string), id.orig_h (addr), id.orig_p (port), id.resp_h (addr), id.resp_p (port), proto (enum), service (string), duration (interval), orig_bytes (count), resp_bytes (count), conn_state (string), local_orig (bool), local_resp (bool), missed_bytes (count), history (string), orig_pkts (count), orig_ip_bytes (count), resp_pkts (count), resp_ip_bytes (count), tunnel_parents (set[string])}, <void ptr>) -> true
+1362692527.080972   MetaHookPost  QueueEvent(Broker::log_flush()) -> false
 1362692527.080972   MetaHookPost  QueueEvent(ChecksumOffloading::check()) -> false
 1362692527.080972   MetaHookPost  QueueEvent(connection_state_remove([id=[orig_h=141.142.228.5, orig_p=59856<...>/plain], current_entity=<uninitialized>, orig_mime_depth=1, resp_mime_depth=1], http_state=[pending={}, current_request=1, current_response=1, trans_depth=1], irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])) -> false
 1362692527.080972   MetaHookPost  QueueEvent(filter_change_tracking()) -> false
@@ -3174,6 +3190,9 @@
 1362692527.080972   MetaHookPost  QueueEvent(successful_connection_remove([id=[orig_h=141.142.228.5, orig_p=59856<...>/plain], current_entity=<uninitialized>, orig_mime_depth=1, resp_mime_depth=1], http_state=[pending={}, current_request=1, current_response=1, trans_depth=1], irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])) -> false
 1362692527.080972   MetaHookPost  QueueEvent(zeek_done()) -> false
 1362692527.080972   MetaHookPost  UpdateNetworkTime(1362692527.080972) -> <void>
+1362692527.080972   MetaHookPre   CallFunction(Broker::__flush_logs, <frame>, ())
+1362692527.080972   MetaHookPre   CallFunction(Broker::flush_logs, <frame>, ())
+1362692527.080972   MetaHookPre   CallFunction(Broker::log_flush, <null>, ())
 1362692527.080972   MetaHookPre   CallFunction(ChecksumOffloading::check, <null>, ())
 1362692527.080972   MetaHookPre   CallFunction(Conn::conn_state, <frame>, ([id=[orig_h=141.142.228.5, orig_p=59856<...>/plain], current_entity=<uninitialized>, orig_mime_depth=1, resp_mime_depth=1], http_state=[pending={}, current_request=1, current_response=1, trans_depth=1], irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>], tcp))
 1362692527.080972   MetaHookPre   CallFunction(Conn::determine_service, <frame>, ([id=[orig_h=141.142.228.5, orig_p=59856<...>/plain], current_entity=<uninitialized>, orig_mime_depth=1, resp_mime_depth=1], http_state=[pending={}, current_request=1, current_response=1, trans_depth=1], irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>]))
@@ -3202,6 +3221,7 @@
 1362692527.080972   MetaHookPre   DrainEvents()
 1362692527.080972   MetaHookPre   LogInit(Log::WRITER_ASCII, default, true, true, conn(1362692527.080972,0.0,0.0), 21, {ts (time), uid (string), id.orig_h (addr), id.orig_p (port), id.resp_h (addr), id.resp_p (port), proto (enum), service (string), duration (interval), orig_bytes (count), resp_bytes (count), conn_state (string), local_orig (bool), local_resp (bool), missed_bytes (count), history (string), orig_pkts (count), orig_ip_bytes (count), resp_pkts (count), resp_ip_bytes (count), tunnel_parents (set[string])})
 1362692527.080972   MetaHookPre   LogWrite(Log::WRITER_ASCII, default, conn(1362692527.080972,0.0,0.0), 21, {ts (time), uid (string), id.orig_h (addr), id.orig_p (port), id.resp_h (addr), id.resp_p (port), proto (enum), service (string), duration (interval), orig_bytes (count), resp_bytes (count), conn_state (string), local_orig (bool), local_resp (bool), missed_bytes (count), history (string), orig_pkts (count), orig_ip_bytes (count), resp_pkts (count), resp_ip_bytes (count), tunnel_parents (set[string])}, <void ptr>)
+1362692527.080972   MetaHookPre   QueueEvent(Broker::log_flush())
 1362692527.080972   MetaHookPre   QueueEvent(ChecksumOffloading::check())
 1362692527.080972   MetaHookPre   QueueEvent(connection_state_remove([id=[orig_h=141.142.228.5, orig_p=59856<...>/plain], current_entity=<uninitialized>, orig_mime_depth=1, resp_mime_depth=1], http_state=[pending={}, current_request=1, current_response=1, trans_depth=1], irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>]))
 1362692527.080972   MetaHookPre   QueueEvent(filter_change_tracking())
@@ -3210,6 +3230,9 @@
 1362692527.080972   MetaHookPre   QueueEvent(zeek_done())
 1362692527.080972   MetaHookPre   UpdateNetworkTime(1362692527.080972)
 1362692527.080972  | HookUpdateNetworkTime 1362692527.080972
+1362692527.080972 | HookCallFunction Broker::__flush_logs()
+1362692527.080972 | HookCallFunction Broker::flush_logs()
+1362692527.080972 | HookCallFunction Broker::log_flush()
 1362692527.080972 | HookCallFunction ChecksumOffloading::check()
 1362692527.080972 | HookCallFunction Conn::conn_state([id=[orig_h=141.142.228.5, orig_p=59856<...>/plain], current_entity=<uninitialized>, orig_mime_depth=1, resp_mime_depth=1], http_state=[pending={}, current_request=1, current_response=1, trans_depth=1], irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>], tcp)
 1362692527.080972 | HookCallFunction Conn::determine_service([id=[orig_h=141.142.228.5, orig_p=59856<...>/plain], current_entity=<uninitialized>, orig_mime_depth=1, resp_mime_depth=1], http_state=[pending={}, current_request=1, current_response=1, trans_depth=1], irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])
@@ -3238,6 +3261,7 @@
 1362692527.080972 | HookDrainEvents
 1362692527.080972 | HookLogInit   conn 1/1 {ts (time), uid (string), id.orig_h (addr), id.orig_p (port), id.resp_h (addr), id.resp_p (port), proto (enum), service (string), duration (interval), orig_bytes (count), resp_bytes (count), conn_state (string), local_orig (bool), local_resp (bool), missed_bytes (count), history (string), orig_pkts (count), orig_ip_bytes (count), resp_pkts (count), resp_ip_bytes (count), tunnel_parents (set[string])}
 1362692527.080972 | HookLogWrite  conn [ts=1362692526.869344, uid=CHhAvVGS1DHFjwGM9, id.orig_h=141.142.228.5, id.orig_p=59856, id.resp_h=192.150.187.43, id.resp_p=80, proto=tcp, service=http, duration=0.211484, orig_bytes=136, resp_bytes=5007, conn_state=SF, local_orig=<uninitialized>, local_resp=<uninitialized>, missed_bytes=0, history=ShADadFf, orig_pkts=7, orig_ip_bytes=512, resp_pkts=7, resp_ip_bytes=5379, tunnel_parents=<uninitialized>]
+1362692527.080972 | HookQueueEvent Broker::log_flush()
 1362692527.080972 | HookQueueEvent ChecksumOffloading::check()
 1362692527.080972 | HookQueueEvent connection_state_remove([id=[orig_h=141.142.228.5, orig_p=59856<...>/plain], current_entity=<uninitialized>, orig_mime_depth=1, resp_mime_depth=1], http_state=[pending={}, current_request=1, current_response=1, trans_depth=1], irc=<uninitialized>, krb=<uninitialized>, modbus=<uninitialized>, mysql=<uninitialized>, ntlm=<uninitialized>, ntp=<uninitialized>, radius=<uninitialized>, rdp=<uninitialized>, rfb=<uninitialized>, sip=<uninitialized>, sip_state=<uninitialized>, snmp=<uninitialized>, smb_state=<uninitialized>, smtp=<uninitialized>, smtp_state=<uninitialized>, socks=<uninitialized>, ssh=<uninitialized>, syslog=<uninitialized>])
 1362692527.080972 | HookQueueEvent filter_change_tracking()

testing/btest/Baseline/scripts.policy.misc.dump-events/all-events-no-args.log

@@ -1,6 +1,7 @@
          0.000000 zeek_init
          0.000000 filter_change_tracking
          0.000000 NetControl::init
+1254722767.492060 Broker::log_flush
 1254722767.492060 ChecksumOffloading::check
 1254722767.492060 filter_change_tracking
 1254722767.492060 new_connection
@@ -19,6 +20,7 @@
 1254722768.219663 smtp_reply
 1254722768.224809 protocol_confirmation
 1254722768.224809 smtp_request
+1254722768.565386 Broker::log_flush
 1254722768.566183 smtp_reply
 1254722768.566183 smtp_reply
 1254722768.566183 smtp_reply
@@ -30,6 +32,7 @@
 1254722768.911655 smtp_request
 1254722769.253544 smtp_reply
 1254722769.254118 smtp_request
+1254722769.613798 Broker::log_flush
 1254722769.613798 smtp_reply
 1254722769.614414 smtp_request
 1254722769.956765 smtp_reply
@@ -37,6 +40,7 @@
 1254722770.319708 smtp_reply
 1254722770.320203 smtp_request
 1254722770.320203 mime_begin_entity
+1254722770.661679 Broker::log_flush
 1254722770.661679 smtp_reply
 1254722770.692743 mime_one_header
 1254722770.692743 mime_one_header
@@ -86,6 +90,7 @@
 1254722770.692804 file_over_new_connection
 1254722770.695115 new_connection
 1254722771.494181 file_sniff
+1254722771.834595 Broker::log_flush
 1254722771.858334 mime_end_entity
 1254722771.858334 get_file_handle
 1254722771.858334 file_state_remove
@@ -97,12 +102,15 @@
 1254722771.858334 get_file_handle
 1254722771.858334 smtp_request
 1254722772.248789 smtp_reply
+1254722774.763825 Broker::log_flush
 1254722774.763825 smtp_request
 1254722775.105467 smtp_reply
+1254722776.690444 Broker::log_flush
 1254722776.690444 new_connection
 1437831776.764391 ChecksumOffloading::check
 1437831776.764391 connection_state_remove
 1437831776.764391 successful_connection_remove
+1437831776.764391 Broker::log_flush
 1437831776.764391 connection_state_remove
 1437831776.764391 successful_connection_remove
 1437831776.764391 connection_state_remove
@@ -111,6 +119,7 @@
 1437831776.764391 successful_connection_remove
 1437831776.764391 filter_change_tracking
 1437831776.764391 new_connection
+1437831787.856895 Broker::log_flush
 1437831787.856895 new_connection
 1437831787.861602 connection_established
 1437831787.867142 smtp_reply
@@ -155,9 +164,11 @@
 1437831787.905375 get_file_handle
 1437831787.905375 smtp_request
 1437831787.914113 smtp_reply
+1437831798.533593 Broker::log_flush
 1437831798.533593 new_connection
 1437831799.262632 new_connection
 1437831799.461152 new_connection
+1437831799.610433 Broker::log_flush
 1437831799.610433 connection_established
 1437831799.611764 ssl_extension_server_name
 1437831799.611764 ssl_extension
@@ -216,6 +227,7 @@
 1437831800.045701 ssl_plaintext_data
 1437831800.045701 ssl_established
 1437831800.217854 net_done
+1437831800.217854 Broker::log_flush
 1437831800.217854 filter_change_tracking
 1437831800.217854 connection_state_remove
 1437831800.217854 successful_connection_remove

testing/btest/broker/remote_event.zeek

@@ -87,7 +87,6 @@ event ping(msg: string, n: count)
 
         if ( n == events_to_recv )
                 {
-		print get_broker_stats();
                 terminate();
                 return;
                 }
@@ -96,4 +95,9 @@ event ping(msg: string, n: count)
         Broker::publish("zeek/event/my_topic", e);
         }
 
+event zeek_done()
+	{
+	print get_broker_stats();
+	}
+
 @TEST-END-FILE

testing/btest/broker/remote_event_any.zeek

@@ -92,7 +92,6 @@ event ping(msg: string, n: any)
 
         if ( (n as count) == events_to_recv )
                 {
-		print get_broker_stats();
                 terminate();
                 return;
                 }
@@ -104,4 +103,9 @@ event ping(msg: string, n: any)
 			Broker::publish("zeek/event/my_topic", pong, msg, n);
         }
 
+event zeek_done()
+	{
+	print get_broker_stats();
+	}
+
 @TEST-END-FILE

testing/btest/broker/remote_event_ssl_auth.zeek

@@ -247,7 +247,6 @@ event ping(msg: string, n: count)
 
         if ( n == events_to_recv )
                 {
-		print get_broker_stats();
                 terminate();
                 return;
                 }
@@ -256,4 +255,9 @@ event ping(msg: string, n: count)
         Broker::publish("zeek/event/my_topic", e);
         }
 
+event zeek_done()
+	{
+	print get_broker_stats();
+	}
+
 @TEST-END-FILE

testing/btest/broker/remote_log_batch.zeek

@@ -0,0 +1,105 @@
+# @TEST-PORT: BROKER_PORT
+
+# @TEST-EXEC: btest-bg-run recv "zeek -B broker -b ../recv.zeek >recv.out"
+# @TEST-EXEC: btest-bg-run send "zeek -B broker -b ../send.zeek >send.out"
+
+# @TEST-EXEC: btest-bg-wait 45
+# @TEST-EXEC: btest-diff recv/recv.out
+# @TEST-EXEC: btest-diff recv/test.log
+# @TEST-EXEC: btest-diff send/send.out
+# @TEST-EXEC: btest-diff send/test.log
+
+@TEST-START-FILE common.zeek
+
+redef exit_only_after_terminate = T;
+
+module Test;
+
+export {
+	redef enum Log::ID += { LOG };
+
+	type Info: record {
+		msg: string &log;
+		nolog: string &default="no";
+		num: count &log;
+	};
+}
+
+event zeek_init() &priority=5
+	{
+	Log::create_stream(Test::LOG, [$columns=Test::Info]);
+	}
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    terminate();
+    }
+
+event quit()
+	{
+	terminate();
+	}
+
+@TEST-END-FILE
+
+@TEST-START-FILE recv.zeek
+
+
+@load ./common
+
+event zeek_init()
+	{
+	Broker::subscribe("zeek/");
+	Broker::listen("127.0.0.1", to_port(getenv("BROKER_PORT")));
+	}
+
+event Broker::peer_removed(endpoint: Broker::EndpointInfo, msg: string)
+	{
+	terminate();
+	}
+
+@TEST-END-FILE
+
+@TEST-START-FILE send.zeek
+
+@load ./common
+
+event zeek_init()
+	{
+	Broker::peer("127.0.0.1", to_port(getenv("BROKER_PORT")));
+	}
+
+global n = 0;
+global done = F;
+
+event die()
+	{
+	terminate();
+	}
+
+event do_write()
+	{
+	Log::write(Test::LOG, [$msg = "ping", $num = n]);
+	++n;
+
+	if ( n == 1 )
+		schedule .1secs { do_write() };
+	else
+		done = T;
+	}
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "Broker::peer_added", endpoint$network$address;
+    event do_write();
+    }
+
+module Broker;
+
+event Broker::log_flush()
+	{
+	if ( done )
+		Broker::publish("zeek/quit", quit);
+	}
+
+@TEST-END-FILE

testing/btest/broker/remote_log_types.zeek

@@ -91,6 +91,8 @@ function foo(i : count) : string
 		return "Bar";
 	}
 
+global done = F;
+
 event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
 	{
 	print "Broker::peer_added", endpoint$network$address;
@@ -118,7 +120,15 @@ event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
 		$f=foo
 		]);
 
-	Broker::publish("zeek/", quit_receiver);
+	done = T;
+	}
+
+module Broker;
+
+event Broker::log_flush()
+	{
+	if ( done )
+		Broker::publish("zeek/", quit_receiver);
 	}
 
 @TEST-END-FILE

testing/btest/language/paraglob-serialization.zeek

@@ -88,7 +88,6 @@ event ping(msg: opaque of paraglob, n: count)
     print "is_remote should be T, and is", is_remote_event();
     if ( n > events_to_recv )
       {
-      print get_broker_stats();
       terminate();
       return;
       }