Notice suppression clean up and notice/cluster integrtion fixes.

- Worker raised notices are printed a single time by the manager now.

- Cluster/notices integration cleaned up.

- New tests for cluster/notice integration.

- Notice suppression fixes and tests.

scripts/base/frameworks/cluster/main.bro

@@ -23,10 +23,10 @@ export {
 	const manager_events = /Drop::.*/ &redef;
 	
 	## Events raised by the proxies and handled by the manager.
-	const proxy_events = /Notice::notice/ &redef;
+	const proxy_events = /EMPTY/ &redef;
 	
 	## Events raised by workers and handled by the manager.
-	const worker_events = /(Notice::notice|TimeMachine::command|Drop::.*)/ &redef;
+	const worker_events = /(TimeMachine::command|Drop::.*)/ &redef;
 	
 	## Events sent by the control host (i.e. BroControl) when dynamically 
 	## connecting to a running instance to update settings or request data.

scripts/base/frameworks/cluster/nodes/manager.bro

@@ -8,8 +8,6 @@
 ##! This is where the cluster manager sets it's specific settings for other
 ##! frameworks and in the core.
 
-@load base/frameworks/notice
-
 @prefixes += cluster-manager
 
 # Load the script for local site configuration for the manager node.
@@ -23,10 +21,3 @@ redef Log::default_rotation_postprocessor_cmd = "archive-log";
 
 ## We're processing essentially *only* remote events.
 redef max_remote_events_processed = 10000;
-
-# Reraise remote notices locally.
-event Notice::notice(n: Notice::Info)
-	{
-	if ( is_remote_event() )
-		NOTICE(n);
-	}

scripts/base/frameworks/cluster/nodes/worker.bro

@@ -1,5 +1,3 @@
-@load base/frameworks/notice
-
 @prefixes += cluster-worker
 
 # Load the script for local site configuration for the worker nodes.
@@ -17,10 +15,3 @@ redef Log::default_rotation_postprocessor_cmd = "delete-log";
 ## Record all packets into trace file.
 # TODO: should we really be setting this to T?
 redef record_all_packets = T;
-
-# Workers need to have a filter for the notice log which doesn't 
-# do remote logging since we forward the notice event directly.
-event bro_init()
-	{
-	Log::disable_stream(Notice::LOG);
-	}

scripts/base/frameworks/notice/__load__.bro

@@ -11,3 +11,10 @@
 # There shouldn't be any defaul toverhead from loading these since they 
 # *should* only do anything when notices have the ACTION_EMAIL action applied.
 @load ./extend-email/hostnames
+
+# The cluster framework must be loaded first.
+@load base/frameworks/cluster
+
+@if ( Cluster::is_enabled() )
+@load ./cluster
+@endif

scripts/base/frameworks/notice/cluster.bro

@@ -0,0 +1,42 @@
+##! Implements notice functionality across clusters.
+
+@load base/frameworks/cluster
+
+module Notice;
+
+# Define the event used to transport notices on the cluster.
+global cluster_notice: event(n: Notice::Info);
+
+redef Cluster::manager_events += /Notice::begin_suppression/;
+redef Cluster::proxy_events += /Notice::cluster_notice/;
+redef Cluster::worker_events += /Notice::cluster_notice/;
+
+@if ( Cluster::local_node_type() != Cluster::MANAGER )
+event Notice::begin_suppression(n: Notice::Info)
+	{
+	suppressing[n$note, n$identifier] = n;
+	}
+	
+event Notice::notice(n: Notice::Info)
+	{
+	# Send the locally generated notice on to the manager.
+	event Notice::cluster_notice(n);
+	}
+
+event bro_init() &priority=3
+	{
+	# Workers and proxies need to disable the notice streams because notice 
+	# events are forwarded directly instead of being logged remotely.
+	Log::disable_stream(Notice::LOG);
+	Log::disable_stream(Notice::POLICY_LOG);
+	Log::disable_stream(Notice::ALARM_LOG);
+	}
+@endif
+
+@if ( Cluster::local_node_type() == Cluster::MANAGER )
+event Notice::cluster_notice(n: Notice::Info)
+	{
+	# Raise remotely received notices on the manager
+	NOTICE(n);
+	}
+@endif

scripts/base/frameworks/notice/main.bro

@@ -90,7 +90,6 @@ export {
 		
 		## These are policy items that returned T and applied their action
 		## to the notice.
-		## TODO: this can't take set() as a default. (bug)
 		policy_items:   set[count]     &log &optional;
 		
 		## By adding chunks of text into this element, other scripts can
@@ -125,7 +124,7 @@ export {
 		## This field indicates the length of time that this
 		## unique notice should be suppressed.  This field is automatically 
 		## filled out and should not be written to by any other script.
-		suppress_for:         interval      &log &default=default_suppression_interval;
+		suppress_for:        interval       &log &optional;
 	};
 	
 	## Ignored notice types.
@@ -157,7 +156,7 @@ export {
 		halt:     bool                             &log &default=F;
 		## This defines the length of time that this particular notice should
 		## be supressed.
-		suppress_for: interval                     &log &default=default_suppression_interval;
+		suppress_for: interval                     &log &optional;
 	};
 	
 	## This is the where the :bro:id:`Notice::policy` is defined.  All notice
@@ -245,9 +244,30 @@ export {
 	global internal_NOTICE: function(n: Notice::Info);
 }
 
+# This is used as a hack to implement per-item expiration intervals.
+function per_notice_suppression_interval(t: table[Notice::Type, string] of Notice::Info, idx: any): interval
+	{
+	local n: Notice::Type;
+	local s: string;
+	[n,s] = idx;
+	
+	local suppress_time = t[n,s]$suppress_for - (network_time() - t[n,s]$ts);
+	if ( suppress_time < 0secs )
+		suppress_time = 0secs;
+	
+	# If there is no more suppression time left, the notice needs to be sent
+	# to the end_suppression event.
+	if ( suppress_time == 0secs )
+		event Notice::end_suppression(t[n,s]);
+	
+	return suppress_time;
+	}
+
 # This is the internally maintained notice suppression table.  It's 
 # indexed on the Notice::Type and the $identifier field from the notice.
-global suppressing: table[Type, string] of Notice::Info = {} &synchronized;
+global suppressing: table[Type, string] of Notice::Info = {} 
+		&create_expire=0secs
+		&expire_func=per_notice_suppression_interval;
 
 # This is an internal variable used to store the notice policy ordered by
 # priority.
@@ -358,6 +378,7 @@ event notice(n: Notice::Info) &priority=-5
 		Log::write(Notice::LOG, n);
 	if ( ACTION_ALARM in n$actions )
 		Log::write(Notice::ALARM_LOG, n);
+	
 	# Normally suppress further notices like this one unless directed not to.
 	#  n$identifier *must* be specified for suppression to function at all.
 	if ( n?$identifier && 
@@ -367,10 +388,9 @@ event notice(n: Notice::Info) &priority=-5
 		{
 		suppressing[n$note, n$identifier] = n;
 		event Notice::begin_suppression(n);
-		schedule n$suppress_for { Notice::end_suppression(n) };
 		}
 	}
-
+	
 ## This determines if a notice is being suppressed.  It is only used 
 ## internally as part of the mechanics for the global NOTICE function.
 function is_being_suppressed(n: Notice::Info): bool
@@ -383,12 +403,6 @@ function is_being_suppressed(n: Notice::Info): bool
 	else
 		return F;
 	}
-
-event Notice::end_suppression(n: Notice::Info) &priority=-5
-	{
-	if ( [n$note, n$identifier] in suppressing )
-		delete suppressing[n$note, n$identifier];
-	}
 	
 # Executes a script with all of the notice fields put into the
 # new process' environment as "BRO_ARG_<field>" variables.
@@ -446,9 +460,6 @@ function apply_policy(n: Notice::Info)
 	if ( ! n?$policy_items )
 		n$policy_items = set();
 	
-	if ( ! n?$suppress_for )
-		n$suppress_for = default_suppression_interval;
-	
 	for ( i in ordered_policy )
 		{
 		if ( ordered_policy[i]$pred(n) )
@@ -456,12 +467,23 @@ function apply_policy(n: Notice::Info)
 			add n$actions[ordered_policy[i]$result];
 			add n$policy_items[int_to_count(i)];
 			
+			# If the predicate matched and there was a suppression interval, 
+			# apply it to the notice now.
+			if ( ordered_policy[i]?$suppress_for )
+				n$suppress_for = ordered_policy[i]$suppress_for;
+			
 			# If the policy item wants to halt policy processing, do it now!
 			if ( ordered_policy[i]$halt )
 				break;
 			}
 		}
 	
+	# Apply the suppression time after applying the policy so that policy
+	# items can give custom suppression intervals.  If there is no 
+	# suppression interval given yet, the default is applied.
+	if ( ! n?$suppress_for )
+		n$suppress_for = default_suppression_interval;
+	
 	# Delete the connection record if it's there so we aren't sending that
 	# to remote machines.  It can cause problems due to the size of the 
 	# connection record.