A log write now raises the corresponding event.

The event has moved from the filters to the streams, and must now be
specificed when creating the stream. (Not clear yet whether that is a
indeed the right interface).

TODO.logging

@@ -2,12 +2,22 @@ List of the things not implemented yet:
 
     - Removing filters
     - Filter predicates
-    - Filter events
     - Dynamic path function.
     - Cluster-style remote_print
     - Rotation support
+    - Flushing support
     - Spawning writers in separate threads (not clear if we want that initially).
     - Not sure if the logging does the right thing with &optional and
       &default values. Needs testing.
+    - Seems we could do some of the filter-related type checks
+      currently done dynamically at startup via a TraversalCallback.
 
 There's probably more missing.
+
+Question:
+
+    * Is giving the record column twice to create_stream too
+	  redundant? (once directly, and once via the event)::
+
+        global ssh_log: event(rec: Log);
+        log_create_stream(LOG_SSH, SSH::Log, ssh_log);

policy/bro.init

@@ -308,19 +308,13 @@ type log_filter: record {
 	# extension is given; the writer will add whatever is
 	# appropiate.
 	path: string &optional;
-	path_func: function(id: string): string &optional;
+	path_func: function(id: Log_ID): string &optional;
 
 	# A subset of column names to record. If not given, all
 	# columns are recorded.
 	include: set[string] &optional;
 	exclude: set[string] &optional;
 
-	# An event that is raised whenever the filter is applied
-	# to an entry. The event receives the same parameter
-	# as the predicate. It will always be generated,
-	# independent of what the predicate returns.
-	ev: event(rec: any) &optional;
-
 	# The writer to use.
 	writer: Log_Writer &default=Log_default_writer;
 };

policy/test-logging.bro

@@ -16,12 +16,15 @@ export {
 	};
 }
 
+global ssh_log: event(rec: Log);
+
 event bro_init()
 {
 	# Create the stream.
 	# First argument is the ID for the stream.
 	# Second argument is the log record type.
-	log_create_stream(LOG_SSH, SSH::Log);
+	# Third argument is the log event, which must receive a single argument of type arg2.
+	log_create_stream(LOG_SSH, SSH::Log, ssh_log);
 
 	# Add a default filter that simply logs everything to "ssh.log" using the default writer.
 	Log_add_default_filter(LOG_SSH);
@@ -36,8 +39,10 @@ event bro_init()
 
     local cid = [$orig_h=1.2.3.4, $orig_p=1234/tcp, $resp_h=2.3.4.5, $resp_p=80/tcp];
 
+	local r: Log = [$t=network_time(), $id=cid, $status="success"];
+
 	# Log something.
-	log_write(LOG_SSH, [$t=network_time(), $id=cid, $status="success"]);
+	log_write(LOG_SSH, r);
 	log_write(LOG_SSH, [$t=network_time(), $id=cid, $status="failure", $country="US"]);
 	log_write(LOG_SSH, [$t=network_time(), $id=cid, $status="failure", $country="UK"]);
 	log_write(LOG_SSH, [$t=network_time(), $id=cid, $status="success", $country="BR"]);
@@ -45,10 +50,11 @@ event bro_init()
 	
 }
 
-#event log(rec: Log)
+event ssh_log(rec: Log)
-#	{
+	{
-#	print fmt("Ran the log handler from the same module.  Extracting time: %0.6f", rec$t);
+	print fmt("Ran the log handler from the same module.  Extracting time: %0.6f", rec$t);
-#	}
+	print rec;
+	}
 #
 #
 #module WHATEVER;

src/LogMgr.cc

@@ -1,5 +1,6 @@
 
 #include "LogMgr.h"
+#include "Event.h"
 #include "EventHandler.h"
 #include "NetVar.h"
 
@@ -23,7 +24,6 @@ struct LogMgr::Filter {
 	string name;
 	Func* pred;
 	Func* path_func;
-	EventHandlerPtr* event;
 	string path;
 	LogWriterDefinition* writer;
 
@@ -40,6 +40,7 @@ struct LogMgr::Filter {
 struct LogMgr::Stream {
 	string name;
 	RecordType* columns;
+	EventHandlerPtr event;
 	list<Filter*> filters;
 
 	~Stream();
@@ -72,7 +73,7 @@ LogMgr::~LogMgr()
 		delete *s;
 	}
 
-bool LogMgr::CreateStream(EnumVal* stream_id, RecordType* columns)
+bool LogMgr::CreateStream(EnumVal* stream_id, RecordType* columns, EventHandlerPtr handler)
 	{
 	// TODO: Should check that the record has only supported types.
 
@@ -91,9 +92,10 @@ bool LogMgr::CreateStream(EnumVal* stream_id, RecordType* columns)
 	streams[idx] = new Stream;
 	streams[idx]->name = stream_id->Type()->AsEnumType()->Lookup(idx);
 	streams[idx]->columns = columns;
+	streams[idx]->event = handler;
 	columns->Ref();
 
-	DBG_LOG(DBG_LOGGING, "Created new logging stream '%s'", streams[idx]->name.c_str());
+	DBG_LOG(DBG_LOGGING, "Created new logging stream '%s', raising event %s", streams[idx]->name.c_str(), streams[idx]->event->Name());
 
 	return true;
 	}
@@ -242,12 +244,6 @@ bool LogMgr::AddFilter(EnumVal* stream_id, RecordVal* fval)
 	filter->pred = path_func ? path_func->AsFunc() : 0;
 	filter->writer = ld;
 
-	if ( event )
-		{
-		// TODO: Implement
-		filter->event = 0;
-		}
-
 	// Build the list of fields that the filter wants included, including
 	// potentially rolling out fields.
 	Val* include = fval->Lookup(rtype->FieldOffset("include"));
@@ -295,7 +291,6 @@ bool LogMgr::AddFilter(EnumVal* stream_id, RecordVal* fval)
 	DBG_LOG(DBG_LOGGING, "   writer    : %s", ld->name);
 	DBG_LOG(DBG_LOGGING, "   path      : %s", filter->path.c_str());
 	DBG_LOG(DBG_LOGGING, "   path_func : %s", (filter->path_func ? "set" : "not set"));
-	DBG_LOG(DBG_LOGGING, "   event     : %s", (filter->event ? "set" : "not set"));
 	DBG_LOG(DBG_LOGGING, "   pred      : %s", (filter->pred ? "set" : "not set"));
 
 	for ( int i = 0; i < filter->num_fields; i++ )
@@ -344,6 +339,14 @@ bool LogMgr::Write(EnumVal* stream_id, RecordVal* columns)
 		return false;
 		}
 
+	// Raise the log event.
+	if ( stream->event )
+		{
+		val_list* vl = new val_list;
+		vl->append(columns->Ref());
+		mgr.QueueEvent(stream->event, vl, SOURCE_LOCAL);
+		}
+
 	// Send to each of our filters.
 	for ( list<Filter*>::iterator i = stream->filters.begin(); i != stream->filters.end(); ++i )
 		{
@@ -351,12 +354,6 @@ bool LogMgr::Write(EnumVal* stream_id, RecordVal* columns)
 
 		string path = filter->path;
 
-		if ( filter->event )
-			{
-			// XXX Raise event here.
-			// TODO: Actually, the filter should be an attribute of the stream, right?
-			}
-
 		if ( filter->pred )
 			{
 			// XXX Check predicate here.

src/LogMgr.h

@@ -5,6 +5,7 @@
 #define LOGMGR_H
 
 #include "Val.h"
+#include "EventHandler.h"
 
 // One value per writer type we have.
 namespace LogWriterType {
@@ -54,7 +55,7 @@ public:
 
     // These correspond to the BiFs visible on the scripting layer. The
     // actual BiFs just forward here.
-    bool CreateStream(EnumVal* stream_id, RecordType* columns);
+    bool CreateStream(EnumVal* stream_id, RecordType* columns, EventHandlerPtr handler);
     bool AddFilter(EnumVal* stream_id, RecordVal* filter);
     bool RemoveFilter(EnumVal* stream_id, StringVal* filter);
     bool Write(EnumVal* stream_id, RecordVal* columns);

src/bro.bif

@@ -488,7 +488,7 @@ function logging_log%(index: string, rec: any%): any
 #include "LogMgr.h"
 %%}
 
-function log_create_stream%(id: Log_ID, columns: any%) : bool
+function log_create_stream%(id: Log_ID, columns: any, ev: any%) : bool
     %{
     if ( columns->Type()->Tag() != TYPE_TYPE )
 	    {
@@ -502,7 +502,31 @@ function log_create_stream%(id: Log_ID, columns: any%) : bool
 		return new Val(0, TYPE_BOOL);
 		}
 
-	bool result = log_mgr->CreateStream(id->AsEnumVal(), columns->Type()->AsTypeType()->Type()->AsRecordType());
+    if ( ev->Type()->Tag() != TYPE_FUNC || ! ev->Type()->AsFuncType()->IsEvent() )
+		{
+		run_time("event argument is not of event type");
+		return new Val(0, TYPE_BOOL);
+		}
+
+	FuncType* et = ev->Type()->AsFuncType();
+	type_list* args = et->ArgTypes()->Types();
+
+    if ( args->length() != 1 )
+		{
+		run_time("event must take a single argument");
+		return new Val(0, TYPE_BOOL);
+		}
+
+	if ( ! same_type((*args)[0], columns->Type()->AsTypeType()->Type()) )
+		{
+		run_time("event argument type does not match log record type");
+		return new Val(0, TYPE_BOOL);
+		}
+
+	EventHandler* handler = event_registry->Lookup(ev->AsFunc()->GetID()->Name());
+	assert(handler);
+
+	bool result = log_mgr->CreateStream(id->AsEnumVal(), columns->Type()->AsTypeType()->Type()->AsRecordType(), handler);
 	return new Val(result, TYPE_BOOL);
     %}