Merge remote-tracking branch 'origin/topic/johanna/gh-214-notice-on-workers'

* origin/topic/johanna/gh-214-notice-on-workers:
  Change notices to be processed on worker.

Fixes GH-214

CHANGES

@@ -1,4 +1,18 @@
 
+2.6-534 | 2019-06-28 11:48:41 -0700
+
+  * Change notices to be processed on worker. (Johanna Amann, Corelight)
+
+    In the past they were processed on the manager - which requires big
+    records to be sent around.
+
+    This has a potential of incompatibilities if someone relied on global
+    state for notice processing.
+
+    Also may prevent notice de-duplication due to expected race
+    condition of suppression messages taking time to proaogate out
+    to all cluster nodes.
+
 2.6-531 | 2019-06-27 12:09:08 -0700
 
   * GH-375: Remove the BroFile cache (Johanna Amann, Corelight)

NEWS

@@ -338,6 +338,15 @@ Changed Functionality
   passed to any other functions for further processing. The remainder of the
   ``ocsp_response_bytes`` is unchanged.
 
+- For performance reasons, processing of notices is now always
+  performed by the node on which the notice is raised rather than
+  the centralized Manager node.  This has potential incompatibilities
+  for those that relied on global state for notice policy processing.
+  It also introduces an expected race condition that may cause multiple
+  notices of the same kind that are generated within a short timespan
+  of each other on separate cluster nodes to all be logged rather
+  than suppressed and de-duplicated into a single notice.
+
 Removed Functionality
 ---------------------
 

VERSION

@@ -1 +1 @@
-2.6-531
+2.6-534

doc

@@ -1 +1 @@
-Subproject commit f50f35e5026b6e8c59f0ed6ab39de4a4e950d88e
+Subproject commit d57b3ad38b2b1ecfd9f1d51699b5fcb785178bf7

scripts/base/frameworks/notice/main.zeek

@@ -273,6 +273,18 @@ export {
 	## identifier: The identifier string of the notice that should be suppressed.
 	global begin_suppression: event(ts: time, suppress_for: interval, note: Type, identifier: string);
 
+	## This is an internal event that is used to broadcast the begin_suppression
+	## event over a cluster.
+	##
+	## ts: time indicating then when the notice to be suppressed occured.
+	##
+	## suppress_for: length of time that this notice should be suppressed.
+	##
+	## note: The :zeek:type:`Notice::Type` of the notice.
+	##
+	## identifier: The identifier string of the notice that should be suppressed.
+	global manager_begin_suppression: event(ts: time, suppress_for: interval, note: Type, identifier: string);
+
 	## A function to determine if an event is supposed to be suppressed.
 	##
 	## n: The record containing the notice in question.
@@ -314,17 +326,8 @@ export {
 	## rec: The record containing notice data before it is logged.
 	global log_notice: event(rec: Info);
 
-	## This is an internal wrapper for the global :zeek:id:`NOTICE`
+	## This is an internal function to populate policy records.
-	## function; disregard.
+	global apply_policy: function(n: Notice::Info);
-	##
-	## n: The record of notice data.
-	global internal_NOTICE: function(n: Notice::Info);
-
-	## This is the event used to transport notices on the cluster.
-	##
-	## n: The notice information to be sent to the cluster manager for
-	##    further processing.
-	global cluster_notice: event(n: Notice::Info);
 }
 
 module GLOBAL;
@@ -334,17 +337,11 @@ function NOTICE(n: Notice::Info)
 	if ( Notice::is_being_suppressed(n) )
 		return;
 
-	@if ( Cluster::is_enabled() )
+	# Fill out fields that might be empty and do the policy processing.
-		if ( Cluster::local_node_type() == Cluster::MANAGER )
+	Notice::apply_policy(n);
-			Notice::internal_NOTICE(n);
+
-		else
+	# Generate the notice event with the notice.
-			{
+	hook Notice::notice(n);
-			n$peer_name = n$peer_descr = Cluster::node;
-			Broker::publish(Cluster::manager_topic, Notice::cluster_notice, n);
-			}
-	@else
-		Notice::internal_NOTICE(n);
-	@endif
 	}
 
 module Notice;
@@ -521,6 +518,9 @@ hook Notice::notice(n: Notice::Info) &priority=-5
 	     n$suppress_for != 0secs )
 		{
 		event Notice::begin_suppression(n$ts, n$suppress_for, n$note, n$identifier);
+@if ( Cluster::is_enabled() && Cluster::local_node_type() != Cluster::MANAGER )
+		event Notice::manager_begin_suppression(n$ts, n$suppress_for, n$note, n$identifier);
+@endif
 		}
 	}
 
@@ -531,15 +531,27 @@ event Notice::begin_suppression(ts: time, suppress_for: interval, note: Type,
 	suppressing[note, identifier] = suppress_until;
 	}
 
+@if ( Cluster::is_enabled() && Cluster::local_node_type() == Cluster::MANAGER )
 event zeek_init()
 	{
-	if ( ! Cluster::is_enabled() )
-		return;
-
 	Broker::auto_publish(Cluster::worker_topic, Notice::begin_suppression);
 	Broker::auto_publish(Cluster::proxy_topic, Notice::begin_suppression);
 	}
 
+event Notice::manager_begin_suppression(ts: time, suppress_for: interval, note: Type,
+								identifier: string)
+	{
+	event Notice::begin_suppression(ts, suppress_for, note, identifier);
+	}
+@endif
+
+@if ( Cluster::is_enabled() && Cluster::local_node_type() != Cluster::MANAGER )
+event zeek_init()
+	{
+	Broker::auto_publish(Cluster::manager_topic, Notice::manager_begin_suppression);
+	}
+@endif
+
 function is_being_suppressed(n: Notice::Info): bool
 	{
 	if ( n?$identifier && [n$note, n$identifier] in suppressing )
@@ -605,6 +617,14 @@ function apply_policy(n: Notice::Info)
 	if ( ! n?$ts )
 		n$ts = network_time();
 
+@if ( Cluster::is_enabled() )
+	if ( ! n?$peer_name )
+		n$peer_name = Cluster::node;
+
+	if ( ! n?$peer_descr )
+		n$peer_descr = Cluster::node;
+@endif
+
 	if ( n?$f )
 		populate_file_info(n$f, n);
 
@@ -652,28 +672,4 @@ function apply_policy(n: Notice::Info)
 	# suppression interval given yet, the default is applied.
 	if ( ! n?$suppress_for )
 		n$suppress_for = default_suppression_interval;
-
-	# Delete the connection and file records if they're there so we
-	# aren't sending that to remote machines.  It can cause problems
-	# due to the size of those records.
-	if ( n?$conn )
-		delete n$conn;
-	if ( n?$iconn )
-		delete n$iconn;
-	if ( n?$f )
-		delete n$f;
-	}
-
-function internal_NOTICE(n: Notice::Info)
-	{
-	# Fill out fields that might be empty and do the policy processing.
-	apply_policy(n);
-
-	# Generate the notice event with the notice.
-	hook Notice::notice(n);
-	}
-
-event Notice::cluster_notice(n: Notice::Info)
-	{
-	NOTICE(n);
 	}

testing/btest/scripts/base/frameworks/notice/cluster.zeek

@@ -33,9 +33,15 @@ event delayed_notice()
 		NOTICE([$note=Test_Notice, $msg="test notice!"]);
 	}
 
+event terminate_me()
+	{
+	terminate();
+	}
+
 event ready()
 	{
 	schedule 1secs { delayed_notice() };
+	schedule 2secs { terminate_me() };
 	}
 
 @if ( Cluster::local_node_type() == Cluster::MANAGER )
@@ -50,7 +56,7 @@ event Cluster::node_up(name: string, id: string)
 		Broker::publish(Cluster::worker_topic, ready);
 	}
 
-event Notice::log_notice(rec: Notice::Info)
+event Cluster::node_down(name: string, id: string)
 	{
 	terminate();
 	}