Change Intel framework to round-robin insertion events across proxies

scripts/base/frameworks/intel/cluster.bro

@@ -54,7 +54,11 @@ event Cluster::node_up(name: string, id: string)
 # has to be distributed.
 event Intel::new_item(item: Item) &priority=5
 	{
+	if ( Cluster::proxy_pool$alive_count == 0 )
 		Broker::publish(indicator_topic, Intel::insert_indicator, item);
+	else
+		Cluster::relay_rr(Cluster::proxy_pool, "Intel::new_item_relay_rr",
+		                  indicator_topic, Intel::insert_indicator, item);
 	}
 
 # Handling of item insertion triggered by remote node.

testing/btest/Baseline/scripts.base.frameworks.intel.cluster-transparency-with-proxy/manager-1..stdout

@@ -0,0 +1,5 @@
+new_item triggered for 1.2.3.4 by manager on manager-1
+insert_item: 4.3.2.1 inserted by worker-2
+new_item triggered for 4.3.2.1 by worker-2 on manager-1
+insert_item: 123.123.123.123 inserted by worker-1
+new_item triggered for 123.123.123.123 by worker-1 on manager-1

testing/btest/Baseline/scripts.base.frameworks.intel.cluster-transparency-with-proxy/manager-1.intel.log

@@ -0,0 +1,10 @@
+#separator \x09
+#set_separator	,
+#empty_field	(empty)
+#unset_field	-
+#path	intel
+#open	2018-02-27-17-03-26
+#fields	ts	uid	id.orig_h	id.orig_p	id.resp_h	id.resp_p	seen.indicator	seen.indicator_type	seen.where	seen.node	matched	sources	fuid	file_mime_type	file_desc
+#types	time	string	addr	port	addr	port	string	enum	enum	string	set[enum]	set[string]	string	string	string
+1519751006.478387	-	-	-	-	-	123.123.123.123	Intel::ADDR	Intel::IN_ANYWHERE	worker-2	Intel::ADDR	worker-1	-	-	-
+#close	2018-02-27-17-03-26

testing/btest/Baseline/scripts.base.frameworks.intel.cluster-transparency-with-proxy/worker-1..stdout

@@ -0,0 +1,4 @@
+new_indicator: 1.2.3.4 inserted by manager
+new_indicator: 4.3.2.1 inserted by worker-2
+new_item triggered for 123.123.123.123 by worker-1 on worker-1
+new_indicator: 123.123.123.123 inserted by worker-1

testing/btest/Baseline/scripts.base.frameworks.intel.cluster-transparency-with-proxy/worker-2..stdout

@@ -0,0 +1,5 @@
+new_indicator: 1.2.3.4 inserted by manager
+new_item triggered for 4.3.2.1 by worker-2 on worker-2
+new_indicator: 4.3.2.1 inserted by worker-2
+new_indicator: 123.123.123.123 inserted by worker-1
+Doing a lookup

testing/btest/scripts/base/frameworks/intel/cluster-transparency-with-proxy.bro

@@ -0,0 +1,90 @@
+# @TEST-SERIALIZE: comm
+#
+# @TEST-EXEC: btest-bg-run manager-1 BROPATH=$BROPATH:.. CLUSTER_NODE=manager-1 bro %INPUT
+# @TEST-EXEC: btest-bg-run proxy-1  BROPATH=$BROPATH:.. CLUSTER_NODE=proxy-1 bro %INPUT
+# @TEST-EXEC: btest-bg-run worker-1  BROPATH=$BROPATH:.. CLUSTER_NODE=worker-1 bro %INPUT
+# @TEST-EXEC: btest-bg-run worker-2  BROPATH=$BROPATH:.. CLUSTER_NODE=worker-2 bro %INPUT
+# @TEST-EXEC: btest-bg-wait -k 10
+# @TEST-EXEC: TEST_DIFF_CANONIFIER=$SCRIPTS/diff-sort btest-diff manager-1/.stdout
+# @TEST-EXEC: TEST_DIFF_CANONIFIER=$SCRIPTS/diff-sort btest-diff worker-1/.stdout
+# @TEST-EXEC: TEST_DIFF_CANONIFIER=$SCRIPTS/diff-sort btest-diff worker-2/.stdout
+# @TEST-EXEC: btest-diff manager-1/intel.log
+
+@TEST-START-FILE cluster-layout.bro
+redef Cluster::nodes = {
+	["manager-1"] = [$node_type=Cluster::MANAGER, $ip=127.0.0.1, $p=37757/tcp],
+	["worker-1"]  = [$node_type=Cluster::WORKER,  $ip=127.0.0.1, $p=37760/tcp, $manager="manager-1"],
+	["worker-2"]  = [$node_type=Cluster::WORKER,  $ip=127.0.0.1, $p=37761/tcp, $manager="manager-1"],
+	["proxy-1"]  = [$node_type=Cluster::PROXY,  $ip=127.0.0.1, $p=37762/tcp, $manager="manager-1"],
+};
+@TEST-END-FILE
+
+module Intel;
+
+redef Log::default_rotation_interval=0sec;
+
+event Cluster::node_up(name: string, id: string)
+	{
+	# Insert the data once both workers are connected.
+	if ( Cluster::local_node_type() == Cluster::MANAGER && Cluster::worker_count == 2 && Cluster::proxy_pool$alive_count == 1 )
+		{
+		Intel::insert([$indicator="1.2.3.4", $indicator_type=Intel::ADDR, $meta=[$source="manager"]]);
+		}
+	}
+
+global worker2_data = 0;
+global sent_data = F;
+# Watch for new indicators send to workers.
+event Intel::insert_indicator(item: Intel::Item)
+	{
+	print fmt("new_indicator: %s inserted by %s", item$indicator, item$meta$source);
+
+	if ( ! sent_data )
+		{
+		# We wait to insert data here because we can now be sure the 
+		# full cluster is constructed.
+		sent_data = T;
+		if ( Cluster::node == "worker-1" )
+			Intel::insert([$indicator="123.123.123.123", $indicator_type=Intel::ADDR, $meta=[$source="worker-1"]]);
+		if ( Cluster::node == "worker-2" )
+			Intel::insert([$indicator="4.3.2.1", $indicator_type=Intel::ADDR, $meta=[$source="worker-2"]]);
+		}
+
+	# We're forcing worker-2 to do a lookup when it has three intelligence items
+	# which were distributed over the cluster (data inserted locally is resent).
+	if ( Cluster::node == "worker-2" )
+		{
+		++worker2_data;
+		if ( worker2_data == 3 )
+			{
+			# Now that everything is inserted, see if we can match on the data inserted
+			# by worker-1.
+			print "Doing a lookup";
+			Intel::seen([$host=123.123.123.123, $where=Intel::IN_ANYWHERE]);
+			}
+		}
+	}
+
+# Watch for remote inserts sent to the manager.
+event Intel::insert_item(item: Intel::Item)
+	{
+	print fmt("insert_item: %s inserted by %s", item$indicator, item$meta$source);
+	}
+
+# Watch for new items.
+event Intel::new_item(item: Intel::Item)
+	{
+	print fmt("new_item triggered for %s by %s on %s", item$indicator,
+			item$meta$source, Cluster::node);
+	}
+
+event Intel::log_intel(rec: Intel::Info)
+	{
+	terminate();
+	}
+
+event Cluster::node_down(name: string, id: string)
+	{
+	# Cascading termination
+	terminate();
+	}