Return weird if a log line is over a configurable size limit

scripts/base/init-bare.zeek

@@ -3747,6 +3747,14 @@ export {
 	## .. :zeek:see:`Log::flush_interval`
 	const write_buffer_size = 1000 &redef;
 
+	## Maximum size of a message that can be sent to a remote logger or logged
+	## locally. If this limit is met, report a ``log_line_too_large`` weird and drop
+	## the log entry. This isn't necessarily the full size of a line that might be
+	## written to a log, but a general representation of the size as the log record is
+	## serialized for writing. The size of end result from serialization might be
+	## higher than this limit, but it prevents runaway-sized log entries from causing
+	## problems.
+	const max_log_record_size = 1024*1024*64 &redef;
 }
 
 module POP3;

src/logging/Manager.cc

@@ -245,8 +245,11 @@ struct Manager::WriterInfo {
     string instantiating_filter;
 
     std::shared_ptr<telemetry::Counter> total_writes;
+    std::shared_ptr<telemetry::Counter> total_discarded_writes;
 
-    WriterInfo(std::shared_ptr<telemetry::Counter> total_writes) : total_writes(std::move(total_writes)) {}
+    WriterInfo(std::shared_ptr<telemetry::Counter> total_writes,
+               std::shared_ptr<telemetry::Counter> total_discarded_writes)
+        : total_writes(std::move(total_writes)), total_discarded_writes(std::move(total_discarded_writes)) {}
 };
 
 struct Manager::Stream {
@@ -484,7 +487,11 @@ Manager::Manager()
           telemetry_mgr
               ->CounterFamily("zeek", "log-writer-writes", {"writer", "module", "stream", "filter-name", "path"},
                               "Total number of log writes passed to a concrete log writer not vetoed by stream or "
-                              "filter policies.")) {
+                              "filter policies.")),
+      total_log_writer_discarded_writes_family(
+          telemetry_mgr->CounterFamily("zeek", "log-writer-discarded-writes",
+                                       {"writer", "module", "stream", "filter-name", "path"},
+                                       "Total number of log writes discarded due to size limitations.")) {
     rotations_pending = 0;
 }
 
@@ -496,6 +503,7 @@ Manager::~Manager() {
 void Manager::InitPostScript() {
     rotation_format_func = id::find_func("Log::rotation_format_func");
     log_stream_policy_hook = id::find_func("Log::log_stream_policy");
+    max_log_record_size = id::find_val("Log::max_log_record_size")->AsCount();
 }
 
 WriterBackend* Manager::CreateBackend(WriterFrontend* frontend, EnumVal* tag) {
@@ -1140,7 +1148,14 @@ bool Manager::WriteToFilters(const Manager::Stream* stream, zeek::RecordValPtr c
         assert(info);
 
         // Alright, can do the write now.
-        auto rec = RecordToLogRecord(stream, filter, columns.get());
+        size_t total_size = 0;
+        auto rec = RecordToLogRecord(stream, filter, columns.get(), total_size);
+
+        if ( total_size > max_log_record_size ) {
+            reporter->Weird("log_record_too_large", util::fmt("%s", stream->name.c_str()));
+            w->second->total_discarded_writes->Inc();
+            continue;
+        }
 
         if ( zeek::plugin_mgr->HavePluginForHook(zeek::plugin::HOOK_LOG_WRITE) ) {
             // The current HookLogWrite API takes a threading::Value**.
@@ -1373,7 +1388,7 @@ bool Manager::SetMaxDelayQueueSize(const EnumValPtr& id, zeek_uint_t queue_size)
     return true;
 }
 
-threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty) {
+threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty, size_t& total_size) {
     if ( ! val )
         return {ty->Tag(), false};
 
@@ -1381,7 +1396,10 @@ threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty) {
 
     switch ( lval.type ) {
         case TYPE_BOOL:
-        case TYPE_INT: lval.val.int_val = val->AsInt(); break;
+        case TYPE_INT:
+            lval.val.int_val = val->AsInt();
+            total_size += sizeof(lval.val.int_val);
+            break;
 
         case TYPE_ENUM: {
             const char* s = ty->AsEnumType()->Lookup(val->AsInt());
@@ -1398,13 +1416,16 @@ threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty) {
                 lval.val.string_val.data = util::copy_string("", 0);
                 lval.val.string_val.length = 0;
             }
+
+            total_size += lval.val.string_val.length;
+
             break;
         }
 
-        case TYPE_COUNT: {
+        case TYPE_COUNT:
             lval.val.uint_val = val->AsCount();
+            total_size += sizeof(lval.val.uint_val);
             break;
-        }
 
         case TYPE_PORT: {
             auto p = val->AsCount();
@@ -1420,16 +1441,26 @@ threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty) {
 
             lval.val.port_val.port = p & ~PORT_SPACE_MASK;
             lval.val.port_val.proto = pt;
+            total_size += lval.val.port_val.size();
             break;
         }
 
-        case TYPE_SUBNET: val->AsSubNet()->Get().ConvertToThreadingValue(&lval.val.subnet_val); break;
+        case TYPE_SUBNET:
+            val->AsSubNet()->Get().ConvertToThreadingValue(&lval.val.subnet_val);
+            total_size += lval.val.subnet_val.size();
+            break;
 
-        case TYPE_ADDR: val->AsAddr()->Get().ConvertToThreadingValue(&lval.val.addr_val); break;
+        case TYPE_ADDR:
+            val->AsAddr()->Get().ConvertToThreadingValue(&lval.val.addr_val);
+            total_size += lval.val.addr_val.size();
+            break;
 
         case TYPE_DOUBLE:
         case TYPE_TIME:
-        case TYPE_INTERVAL: lval.val.double_val = val->AsDouble(); break;
+        case TYPE_INTERVAL:
+            lval.val.double_val = val->AsDouble();
+            total_size += sizeof(lval.val.double_val);
+            break;
 
         case TYPE_STRING: {
             const String* s = val->AsString()->AsString();
@@ -1438,6 +1469,7 @@ threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty) {
 
             lval.val.string_val.data = buf;
             lval.val.string_val.length = s->Len();
+            total_size += lval.val.string_val.length;
             break;
         }
 
@@ -1447,6 +1479,7 @@ threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty) {
             auto len = strlen(s);
             lval.val.string_val.data = util::copy_string(s, len);
             lval.val.string_val.length = len;
+            total_size += lval.val.string_val.length;
             break;
         }
 
@@ -1458,6 +1491,7 @@ threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty) {
             auto len = strlen(s);
             lval.val.string_val.data = util::copy_string(s, len);
             lval.val.string_val.length = len;
+            total_size += lval.val.string_val.length;
             break;
         }
 
@@ -1474,14 +1508,15 @@ threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty) {
             auto& set_t = tbl_t->GetIndexTypes()[0];
             bool is_managed = ZVal::IsManagedType(set_t);
 
-            lval.val.set_val.size = set->Length();
+            zeek_int_t set_length = set->Length();
-            lval.val.set_val.vals = new threading::Value*[lval.val.set_val.size];
+            lval.val.set_val.vals = new threading::Value*[set_length];
 
-            for ( zeek_int_t i = 0; i < lval.val.set_val.size; i++ ) {
+            for ( zeek_int_t i = 0; i < set_length && total_size < max_log_record_size; i++ ) {
                 std::optional<ZVal> s_i = ZVal(set->Idx(i), set_t);
-                lval.val.set_val.vals[i] = new threading::Value(ValToLogVal(s_i, set_t.get()));
+                lval.val.set_val.vals[i] = new threading::Value(ValToLogVal(s_i, set_t.get(), total_size));
                 if ( is_managed )
                     ZVal::DeleteManagedType(*s_i);
+                lval.val.set_val.size++;
             }
 
             break;
@@ -1489,14 +1524,15 @@ threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty) {
 
         case TYPE_VECTOR: {
             VectorVal* vec = val->AsVector();
-            lval.val.vector_val.size = vec->Size();
+            zeek_int_t vec_length = vec->Size();
-            lval.val.vector_val.vals = new threading::Value*[lval.val.vector_val.size];
+            lval.val.vector_val.vals = new threading::Value*[vec_length];
 
             auto& vv = vec->RawVec();
             auto& vt = vec->GetType()->Yield();
 
-            for ( zeek_int_t i = 0; i < lval.val.vector_val.size; i++ ) {
+            for ( zeek_int_t i = 0; i < vec_length && total_size < max_log_record_size; i++ ) {
-                lval.val.vector_val.vals[i] = new threading::Value(ValToLogVal(vv[i], vt.get()));
+                lval.val.vector_val.vals[i] = new threading::Value(ValToLogVal(vv[i], vt.get(), total_size));
+                lval.val.vector_val.size++;
             }
 
             break;
@@ -1508,7 +1544,8 @@ threading::Value Manager::ValToLogVal(std::optional<ZVal>& val, Type* ty) {
     return lval;
 }
 
-detail::LogRecord Manager::RecordToLogRecord(const Stream* stream, Filter* filter, RecordVal* columns) {
+detail::LogRecord Manager::RecordToLogRecord(const Stream* stream, Filter* filter, RecordVal* columns,
+                                             size_t& total_size) {
     RecordValPtr ext_rec;
 
     if ( filter->num_ext_fields > 0 ) {
@@ -1558,7 +1595,11 @@ detail::LogRecord Manager::RecordToLogRecord(const Stream* stream, Filter* filte
         }
 
         if ( val )
-            vals.emplace_back(ValToLogVal(val, vt));
+            vals.emplace_back(ValToLogVal(val, vt, total_size));
+
+        if ( total_size > max_log_record_size ) {
+            return {};
+        }
     }
 
     return vals;
@@ -1607,7 +1648,8 @@ WriterFrontend* Manager::CreateWriter(EnumVal* id, EnumVal* writer, WriterBacken
                                                        {"filter-name", instantiating_filter},
                                                        {"path", info->path}};
 
-    WriterInfo* winfo = new WriterInfo(zeek::log_mgr->total_log_writer_writes_family->GetOrAdd(labels));
+    WriterInfo* winfo = new WriterInfo(zeek::log_mgr->total_log_writer_writes_family->GetOrAdd(labels),
+                                       zeek::log_mgr->total_log_writer_discarded_writes_family->GetOrAdd(labels));
     winfo->type = writer->Ref()->AsEnumVal();
     winfo->writer = nullptr;
     winfo->open_time = run_state::network_time;

src/logging/Manager.h

@@ -422,8 +422,8 @@ private:
     bool TraverseRecord(Stream* stream, Filter* filter, RecordType* rt, TableVal* include, TableVal* exclude,
                         const std::string& path, const std::list<int>& indices);
 
-    detail::LogRecord RecordToLogRecord(const Stream* stream, Filter* filter, RecordVal* columns);
+    detail::LogRecord RecordToLogRecord(const Stream* stream, Filter* filter, RecordVal* columns, size_t& total_size);
-    threading::Value ValToLogVal(std::optional<ZVal>& val, Type* ty);
+    threading::Value ValToLogVal(std::optional<ZVal>& val, Type* ty, size_t& total_size);
 
     Stream* FindStream(EnumVal* id);
     void RemoveDisabledWriters(Stream* stream);
@@ -445,12 +445,14 @@ private:
     bool DelayCompleted(Manager::Stream* stream, detail::DelayInfo& delay_info);
 
     std::vector<Stream*> streams; // Indexed by stream enum.
-    int rotations_pending;        // Number of rotations not yet finished.
+    int rotations_pending = 0;    // Number of rotations not yet finished.
     FuncPtr rotation_format_func;
     FuncPtr log_stream_policy_hook;
+    size_t max_log_record_size = 0;
 
     std::shared_ptr<telemetry::CounterFamily> total_log_stream_writes_family;
     std::shared_ptr<telemetry::CounterFamily> total_log_writer_writes_family;
+    std::shared_ptr<telemetry::CounterFamily> total_log_writer_discarded_writes_family;
 
     zeek_uint_t last_delay_token = 0;
     std::vector<detail::WriteContext> active_writes;

src/threading/SerialTypes.h

@@ -137,6 +137,7 @@ struct Value {
     struct port_t {
         zeek_uint_t port;
         TransportProto proto;
+        constexpr size_t size() { return sizeof(port) + sizeof(proto); }
     };
 
     struct addr_t {
@@ -145,6 +146,8 @@ struct Value {
             struct in_addr in4;
             struct in6_addr in6;
         } in;
+
+        constexpr size_t size() { return sizeof(in) + sizeof(IPFamily); }
     };
 
     // A small note for handling subnet values: Subnet values emitted from
@@ -157,6 +160,7 @@ struct Value {
     struct subnet_t {
         addr_t prefix;
         uint8_t length;
+        constexpr size_t size() { return prefix.size() + sizeof(length); }
     };
 
     /**

testing/btest/Baseline/scripts.base.frameworks.logging.length-checking/logger.test.log

@@ -0,0 +1,11 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	test
+#open XXXX-XX-XX-XX-XX-XX
+#fields	strings
+#types	vector[string]
+a,b,c
+#close XXXX-XX-XX-XX-XX-XX

testing/btest/Baseline/scripts.base.frameworks.logging.length-checking/logger.weird.log

@@ -0,0 +1,11 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	weird
+#open XXXX-XX-XX-XX-XX-XX
+#fields	ts	uid	id.orig_h	id.orig_p	id.resp_h	id.resp_p	name	addl	notice	peer	source
+#types	time	string	addr	port	addr	port	string	string	bool	string	string
+XXXXXXXXXX.XXXXXX	-	-	-	-	-	log_record_too_large	Test::LOG	F	zeek	-
+#close XXXX-XX-XX-XX-XX-XX

testing/btest/Baseline/scripts.base.frameworks.logging.length-checking/worker-2.test.log

@@ -0,0 +1,11 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	test
+#open XXXX-XX-XX-XX-XX-XX
+#fields	strings
+#types	vector[string]
+a,b,c
+#close XXXX-XX-XX-XX-XX-XX

testing/btest/Baseline/scripts.base.frameworks.logging.length-checking/worker-2.weird.log

@@ -0,0 +1,11 @@
+### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	weird
+#open XXXX-XX-XX-XX-XX-XX
+#fields	ts	uid	id.orig_h	id.orig_p	id.resp_h	id.resp_p	name	addl	notice	peer	source
+#types	time	string	addr	port	addr	port	string	string	bool	string	string
+XXXXXXXXXX.XXXXXX	-	-	-	-	-	log_record_too_large	Test::LOG	F	zeek	-
+#close XXXX-XX-XX-XX-XX-XX

testing/btest/Baseline/scripts.base.frameworks.logging.telemetry/telemetry.log

@@ -13,4 +13,7 @@ XXXXXXXXXX.XXXXXX	zeek	counter	zeek_log_stream_writes_total	module,stream	HTTP,H
 XXXXXXXXXX.XXXXXX	zeek	counter	zeek_log_writer_writes_total	filter_name,module,path,stream,writer	default,Conn,conn,Conn::LOG,Log::WRITER_ASCII	30.0
 XXXXXXXXXX.XXXXXX	zeek	counter	zeek_log_writer_writes_total	filter_name,module,path,stream,writer	default,DNS,dns,DNS::LOG,Log::WRITER_ASCII	23.0
 XXXXXXXXXX.XXXXXX	zeek	counter	zeek_log_writer_writes_total	filter_name,module,path,stream,writer	default,HTTP,http,HTTP::LOG,Log::WRITER_ASCII	10.0
+XXXXXXXXXX.XXXXXX	zeek	counter	zeek_log_writer_discarded_writes_total	filter_name,module,path,stream,writer	default,Conn,conn,Conn::LOG,Log::WRITER_ASCII	0.0
+XXXXXXXXXX.XXXXXX	zeek	counter	zeek_log_writer_discarded_writes_total	filter_name,module,path,stream,writer	default,DNS,dns,DNS::LOG,Log::WRITER_ASCII	0.0
+XXXXXXXXXX.XXXXXX	zeek	counter	zeek_log_writer_discarded_writes_total	filter_name,module,path,stream,writer	default,HTTP,http,HTTP::LOG,Log::WRITER_ASCII	0.0
 #close XXXX-XX-XX-XX-XX-XX

testing/btest/scripts/base/frameworks/logging/length-checking.zeek

@@ -0,0 +1,114 @@
+# @TEST-GROUP: broker
+#
+# @TEST-DOC: Limit the size of log lines that can be written.
+#
+# @TEST-PORT: BROKER_PORT
+#
+# @TEST-EXEC: btest-bg-run logger "zeek -b ../logger.zeek"
+# @TEST-EXEC: btest-bg-run worker-1 "zeek -b ../worker-1.zeek"
+# @TEST-EXEC: btest-bg-run worker-2 "zeek -b ../worker-2.zeek"
+#
+# @TEST-EXEC: btest-bg-wait -k 10
+# @TEST-EXEC: btest-diff logger/weird.log
+# @TEST-EXEC: btest-diff logger/test.log
+# @TEST-EXEC: btest-diff worker-2/weird.log
+# @TEST-EXEC: btest-diff worker-2/test.log
+
+# @TEST-START-FILE common.zeek
+@load base/frameworks/notice/weird
+
+module Test;
+
+export {
+	redef enum Log::ID += { LOG };
+
+	type Info: record {
+		strings: vector of string &log;
+	};
+}
+
+# Limit log lines to 1MB.
+redef Log::max_log_record_size = 1024 * 1024;
+
+redef Broker::disable_ssl = T;
+
+event zeek_init()
+	{
+	Log::create_stream(LOG, [$columns=Info, $path="test"]);
+	}
+# @TEST-END-FILE
+
+# @TEST-START-FILE logger.zeek
+@load ./common.zeek
+
+redef Log::enable_remote_logging = F;
+redef Log::enable_local_logging = T;
+
+event zeek_init()
+	{
+	Broker::subscribe("zeek/logs");
+	Broker::listen("127.0.0.1", to_port(getenv("BROKER_PORT")));
+	}
+
+global peers_lost = 0;
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+	{
+	print "peer lost";
+	++peers_lost;
+	if ( peers_lost == 2 )
+		terminate();
+	}
+# @TEST-END-FILE
+
+# @TEST-START-FILE worker.zeek
+@load ./common.zeek
+
+event zeek_init()
+	{
+	Broker::peer("127.0.0.1", to_port(getenv("BROKER_PORT")));
+	}
+
+event do_write()
+	{
+	local s = "AAAAAAAAAAAAAAAAAAAA"; # 20 bytes
+	local s100 = s + s + s + s + s;
+	local s1000 = s100 + s100 + s100 + s100 + s100 + s100 + s100 + s100 + s100 + s100;
+
+	local rec = Test::Info();
+	local i = 0;
+	while ( ++i <= ( 1000 * 1000 ) )
+		{
+		rec$strings += s1000;
+		}
+
+	Log::write(Test::LOG, rec);
+
+	local rec2 = Test::Info();
+	rec2$strings += "a";
+	rec2$strings += "b";
+	rec2$strings += "c";
+
+	Log::write(Test::LOG, rec2);
+
+	terminate();
+	}
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+	{
+	print "new_peer", msg;
+	schedule 1sec { do_write() };
+	}
+# @TEST-END-FILE
+
+# @TEST-START-FILE worker-1.zeek
+@load ./worker.zeek
+redef Log::enable_remote_logging = T;
+redef Log::enable_local_logging = F;
+# @TEST-END-FILE worker-1.zeek
+
+# @TEST-START-FILE worker-2.zeek
+@load ./worker.zeek
+redef Log::enable_remote_logging = F;
+redef Log::enable_local_logging = T;
+# @TEST-END-FILE worker-2.zeek