From 3919a35b9ba0e8a57a125556981b36663e39acb5 Mon Sep 17 00:00:00 2001
From: Seth Hall <seth@icir.org>
Date: Mon, 15 Aug 2011 15:57:48 -0400
Subject: [PATCH] Metrics framework update.  Mostly to make metrics work on
 clusters.

- Metrics now work on cluster deployments with no caveats.  It should be
  completely transparent.  Intermediate updates to speed some detection
  will come later.
---
 scripts/base/frameworks/cluster/__load__.bro  |   2 +-
 scripts/base/frameworks/cluster/main.bro      |  43 ++++-
 .../frameworks/cluster/setup-connections.bro  |   3 +-
 scripts/base/frameworks/metrics/__load__.bro  |  10 +
 scripts/base/frameworks/metrics/cluster.bro   | 146 +++++++++++++++
 scripts/base/frameworks/metrics/main.bro      | 172 ++++++++----------
 .../base/frameworks/metrics/non-cluster.bro   |  17 ++
 scripts/base/init-default.bro                 |   4 +-
 scripts/base/utils/site.bro                   |  10 +-
 .../frameworks/metrics/conn-example.bro       |   4 +-
 .../frameworks/metrics/http-example.bro       |  18 +-
 .../policy/frameworks/metrics/ssl-example.bro |   6 +-
 .../manager-1.metrics.log                     |   4 +
 .../metrics.log                               |   4 +
 .../notice.log                                |   4 +
 .../frameworks/metrics/basic-cluster.bro      |  38 ++++
 .../btest/policy/frameworks/metrics/basic.bro |  16 ++
 .../policy/frameworks/metrics/notice.bro      |  23 +++
 18 files changed, 407 insertions(+), 117 deletions(-)
 create mode 100644 scripts/base/frameworks/metrics/cluster.bro
 create mode 100644 scripts/base/frameworks/metrics/non-cluster.bro
 create mode 100644 testing/btest/Baseline/policy.frameworks.metrics.basic-cluster/manager-1.metrics.log
 create mode 100644 testing/btest/Baseline/policy.frameworks.metrics.basic/metrics.log
 create mode 100644 testing/btest/Baseline/policy.frameworks.metrics.notice/notice.log
 create mode 100644 testing/btest/policy/frameworks/metrics/basic-cluster.bro
 create mode 100644 testing/btest/policy/frameworks/metrics/basic.bro
 create mode 100644 testing/btest/policy/frameworks/metrics/notice.bro

diff --git a/scripts/base/frameworks/cluster/__load__.bro b/scripts/base/frameworks/cluster/__load__.bro
index 76eb2b820e..03262d3d75 100644
--- a/scripts/base/frameworks/cluster/__load__.bro
+++ b/scripts/base/frameworks/cluster/__load__.bro
@@ -19,7 +19,7 @@ redef peer_description = Cluster::node;
 
 @load ./setup-connections
 
-# Don't start the listening process until we're a bit more sure that the
+# Don't load the listening script until we're a bit more sure that the
 # cluster framework is actually being enabled.
 @load frameworks/communication/listen-clear
 
diff --git a/scripts/base/frameworks/cluster/main.bro b/scripts/base/frameworks/cluster/main.bro
index 0fc793e7f5..f6066e5800 100644
--- a/scripts/base/frameworks/cluster/main.bro
+++ b/scripts/base/frameworks/cluster/main.bro
@@ -47,6 +47,25 @@ export {
 		time_machine: string      &optional;
 	};
 	
+	## This function can be called at any time to determine if the cluster
+	## framework is being enabled for this run.
+	global is_enabled: function(): bool;
+	
+	## This function can be called at any time to determine what type of
+	## cluster node the current Bro instance is going to be acting as.
+	## :bro:id:`is_enabled` should be called first to find out if this is
+	## actually going to be a cluster node.
+	global local_node_type: function(): NodeType;
+	
+	## This gives the value for the number of workers currently connected to,
+	## and it's maintained internally by the cluster framework.  It's 
+	## primarily intended for use by managers to find out how many workers 
+	## should be responding to requests.
+	global worker_count: count = 0;
+	
+	## The cluster layout definition.  This should be placed into a filter
+	## named cluster-layout.bro somewhere in the BROPATH.  It will be 
+	## automatically loaded if the CLUSTER_NODE environment variable is set.
 	const nodes: table[string] of Node = {} &redef;
 	
 	## This is usually supplied on the command line for each instance
@@ -54,7 +73,29 @@ export {
 	const node = getenv("CLUSTER_NODE") &redef;
 }
 
-event bro_init()
+function is_enabled(): bool
+	{
+	return (node != "");
+	}
+
+function local_node_type(): NodeType
+	{
+	return nodes[node]$node_type;
+	}
+	
+
+event remote_connection_handshake_done(p: event_peer)
+	{
+	if ( nodes[p$descr]$node_type == WORKER )
+		++worker_count;
+	}
+event remote_connection_closed(p: event_peer)
+	{
+	if ( nodes[p$descr]$node_type == WORKER )
+		--worker_count;
+	}
+
+event bro_init() &priority=5
 	{
 	# If a node is given, but it's an unknown name we need to fail.
 	if ( node != "" && node !in nodes )
diff --git a/scripts/base/frameworks/cluster/setup-connections.bro b/scripts/base/frameworks/cluster/setup-connections.bro
index 956a6194f4..7928d0c6ec 100644
--- a/scripts/base/frameworks/cluster/setup-connections.bro
+++ b/scripts/base/frameworks/cluster/setup-connections.bro
@@ -60,13 +60,12 @@ event bro_init() &priority=9
 				                                   $connect=T, $retry=1mins, 
 				                                   $class=node];
 			}
-		
 		else if ( me$node_type == WORKER )
 			{
 			if ( n$node_type == MANAGER && me$manager == i )
 				Communication::nodes["manager"] = [$host=nodes[i]$ip, $p=nodes[i]$p,
 				                                   $connect=T, $retry=1mins, 
-				                                   $class=node];
+				                                   $class=node, $events=manager_events];
 			
 			if ( n$node_type == PROXY && me$proxy == i )
 				Communication::nodes["proxy"] = [$host=nodes[i]$ip, $p=nodes[i]$p,
diff --git a/scripts/base/frameworks/metrics/__load__.bro b/scripts/base/frameworks/metrics/__load__.bro
index a10fe855df..35f6b30fb5 100644
--- a/scripts/base/frameworks/metrics/__load__.bro
+++ b/scripts/base/frameworks/metrics/__load__.bro
@@ -1 +1,11 @@
 @load ./main
+
+# The cluster framework must be loaded first.
+@load base/frameworks/cluster
+
+# Load either the cluster support script or the non-cluster support script.
+@if ( Cluster::is_enabled() )
+@load ./cluster
+@else
+@load ./non-cluster
+@endif
\ No newline at end of file
diff --git a/scripts/base/frameworks/metrics/cluster.bro b/scripts/base/frameworks/metrics/cluster.bro
new file mode 100644
index 0000000000..94281eb883
--- /dev/null
+++ b/scripts/base/frameworks/metrics/cluster.bro
@@ -0,0 +1,146 @@
+##! This implements transparent cluster support for the metrics framework.
+##! Do not load this file directly.  It's only meant to be loaded automatically
+##! and will be depending on if the cluster framework has been enabled.
+##! The goal of this script is to make metric calculation completely and
+##! transparently automated when running on a cluster.
+
+@load base/frameworks/cluster
+
+module Metrics;
+
+export {
+	## This event is sent by the manager in a cluster to initiate the 3
+	## collection of metrics values 
+	global cluster_collect: event(uid: string, id: ID, filter_name: string);
+	
+	## This event is sent by nodes that are collecting metrics after receiving
+	## a request for the metric filter from the manager.
+	global cluster_results: event(uid: string, id: ID, filter_name: string, data: MetricTable, done: bool);
+	
+	## This event is used internally by workers to send result chunks.
+	global send_data: event(uid: string, id: ID, filter_name: string, data: MetricTable);
+	
+	## This value allows a user to decide how large of result groups the 
+	## workers should transmit values.
+	const cluster_send_in_groups_of = 50 &redef;
+}
+
+# This is maintained by managers so they can know what data they requested and
+# when they requested it.
+global requested_results: table[string] of time = table() &create_expire=5mins;
+
+# TODO: Both of the next variables make the assumption that a value never 
+#       takes longer than 5 minutes to transmit from workers to manager.  This needs to 
+#       be tunable or self-tuning.  These should also be restructured to be
+#       maintained within a single variable.
+# This variable is maintained by manager nodes as they collect and aggregate 
+# results.
+global collecting_results: table[string, ID, string] of MetricTable &create_expire=5mins;
+
+# This variable is maintained by manager nodes to track how many "dones" they
+# collected per collection unique id.  Once the number of results for a uid 
+# matches the number of peer nodes that results should be coming from, the 
+# result is written out and deleted from here.
+# TODO: add an &expire_func in case not all results are received.
+global done_with: table[string] of count &create_expire=5mins &default=0;
+
+# Add events to the cluster framework to make this work.
+redef Cluster::manager_events += /Metrics::cluster_collect/;
+redef Cluster::worker_events += /Metrics::cluster_results/;
+
+# The metrics collection process can only be done by a manager.
+@if ( Cluster::local_node_type() == Cluster::MANAGER )
+event Metrics::log_it(filter: Filter)
+	{
+	local uid = unique_id("");
+	
+	# Set some tracking variables.
+	requested_results[uid] = network_time();
+	collecting_results[uid, filter$id, filter$name] = table();
+	
+	# Request data from peers.
+	event Metrics::cluster_collect(uid, filter$id, filter$name);
+	# Schedule the log_it event for the next break period.
+	schedule filter$break_interval { Metrics::log_it(filter) };
+	}
+@endif
+
+@if ( Cluster::local_node_type() == Cluster::WORKER )
+
+event Metrics::send_data(uid: string, id: ID, filter_name: string, data: MetricTable)
+	{
+	#print fmt("WORKER %s: sending data for uid %s...", Cluster::node, uid);
+	
+	local local_data: MetricTable;
+	local num_added = 0;
+	for ( index in data )
+		{
+		local_data[index] = data[index];
+		delete data[index];
+		
+		# Only send cluster_send_in_groups_of at a time.  Queue another
+		# event to send the next group.
+		if ( cluster_send_in_groups_of == ++num_added )
+			break;
+		}
+	
+	local done = F;
+	# If data is empty, this metric is done.
+	if ( |data| == 0 )
+		done = T;
+	
+	event Metrics::cluster_results(uid, id, filter_name, local_data, done);
+	if ( ! done )
+		event Metrics::send_data(uid, id, filter_name, data);
+	}
+
+event Metrics::cluster_collect(uid: string, id: ID, filter_name: string)
+	{
+	#print fmt("WORKER %s: received the cluster_collect event.", Cluster::node);
+	
+	event Metrics::send_data(uid, id, filter_name, store[id, filter_name]);
+		
+	# Lookup the actual filter and reset it, the reference to the data
+	# currently stored will be maintained interally by the send_data event.
+	reset(filter_store[id, filter_name]);
+	}
+@endif
+
+
+@if ( Cluster::local_node_type() == Cluster::MANAGER )
+
+event Metrics::cluster_results(uid: string, id: ID, filter_name: string, data: MetricTable, done: bool)
+	{
+	#print fmt("MANAGER: receiving results from %s", get_event_peer()$descr);
+	
+	local local_data = collecting_results[uid, id, filter_name];
+	for ( index in data )
+		{
+		if ( index !in local_data )
+			local_data[index] = 0;
+		local_data[index] += data[index];
+		}
+	
+	# Mark another worker as being "done" for this uid.
+	if ( done )
+		++done_with[uid];
+	
+	# If the data has been collected from all peers, we are done and ready to log.
+	if ( Cluster::worker_count == done_with[uid] )
+		{
+		local ts = network_time();
+		# Log the time this was initially requested if it's available.
+		if ( uid in requested_results )
+			ts = requested_results[uid];
+			
+		write_log(ts, filter_store[id, filter_name], local_data);
+		if ( [uid, id, filter_name] in collecting_results )
+			delete collecting_results[uid, id, filter_name];
+		if ( uid in done_with )
+			delete done_with[uid];
+		if ( uid in requested_results )
+			delete requested_results[uid];
+		}
+	}
+
+@endif
\ No newline at end of file
diff --git a/scripts/base/frameworks/metrics/main.bro b/scripts/base/frameworks/metrics/main.bro
index 29f18ab824..38ed17b36f 100644
--- a/scripts/base/frameworks/metrics/main.bro
+++ b/scripts/base/frameworks/metrics/main.bro
@@ -1,5 +1,7 @@
 ##! This is the implementation of the metrics framework.
 
+@load base/frameworks/notice
+
 module Metrics;
 
 export {
@@ -13,16 +15,7 @@ export {
 	## current value to the logging stream.
 	const default_break_interval = 15mins &redef;
 	
-	type Info: record {
-		ts:           time   &log;
-		metric_id:    ID     &log;
-		filter_name:  string &log;
-		agg_subnet:   string &log &optional;
-		index:        string &log &optional;
-		value:        count  &log;
-	};
-	
-	type Entry: record {
+	type Index: record {
 		## Host is the value to which this metric applies.
 		host:         addr &optional;
 		
@@ -33,11 +26,19 @@ export {
 		## value in a Host header.  This is an example of a non-host based
 		## metric since multiple IP addresses could respond for the same Host
 		## header value.
-		index:        string &default="";
+		str:        string &optional;
 		
-		## The value by which the counter should be increased in each filter
-		## where this entry is accepted.
-		increment:    count &default=1;
+		## The CIDR block that this metric applies to.  This is typically
+		## only used internally for host based aggregation.
+		network:      subnet &optional;
+	} &log;
+	
+	type Info: record {
+		ts:           time   &log;
+		metric_id:    ID     &log;
+		filter_name:  string &log;
+		index:        Index  &log;
+		value:        count  &log;
 	};
 	
 	# TODO: configure a metrics filter logging stream to log the current
@@ -52,11 +53,11 @@ export {
 		name:              string                  &default="default";
 		## A predicate so that you can decide per index if you would like
 		## to accept the data being inserted.
-		pred:              function(entry: Entry): bool &optional;
+		pred:              function(index: Index): bool &optional;
 		## Global mask by which you'd like to aggregate traffic.
 		aggregation_mask:  count                   &optional;
 		## This is essentially applying names to various subnets.
-		aggregation_table: table[subnet] of string &optional;
+		aggregation_table: table[subnet] of subnet &optional;
 		## The interval at which the metric should be "broken" and written
 		## to the logging stream.
 		break_interval:    interval                &default=default_break_interval;
@@ -68,6 +69,7 @@ export {
 		## A straight threshold for generating a notice.
 		notice_threshold:  count                   &optional;
 		## A series of thresholds at which to generate notices.
+		## TODO: This is not implemented yet!
 		notice_thresholds: vector of count         &optional;
 		## If this and a $notice_threshold value are set, this notice type
 		## will be generated by the metrics framework.
@@ -75,15 +77,23 @@ export {
 	};
 	
 	global add_filter: function(id: ID, filter: Filter);
-	global add_data: function(id: ID, entry: Entry);
+	global add_data: function(id: ID, index: Index, increment: count);
+	
+	# This is the event that is used to "finish" metrics and adapt the metrics
+	# framework for clustered or non-clustered usage.
+	global log_it: event(filter: Filter);	
 	
 	global log_metrics: event(rec: Info);
 }
 
-global metric_filters: table[ID] of vector of Filter = table();
+redef record Notice::Info += {
+	metric_index: Index &log &optional;
+};
 
-type MetricIndex: table[string] of count &default=0;
-type MetricTable: table[string] of MetricIndex;
+global metric_filters: table[ID] of vector of Filter = table();
+global filter_store: table[ID, string] of Filter = table();
+
+type MetricTable: table[Index] of count &default=0;
 # This is indexed by metric ID and stream filter name.
 global store: table[ID, string] of MetricTable = table();
 
@@ -95,63 +105,45 @@ event bro_init() &priority=5
 	{
 	Log::create_stream(METRICS, [$columns=Info, $ev=log_metrics]);
 	}
-	
+
+function write_log(ts: time, filter: Filter, data: MetricTable)
+	{
+	for ( index in data )
+		{
+		local val = data[index];
+		local m: Info = [$ts=ts, 
+		                 $metric_id=filter$id,
+		                 $filter_name=filter$name,
+		                 $index=index,
+		                 $value=val];
+
+		if ( m$index?$host &&
+		     filter?$notice_threshold &&
+		     m$value >= filter$notice_threshold )
+			{
+			NOTICE([$note=filter$note,
+			        $msg=fmt("Metrics threshold crossed by %s %d/%d", index$host, m$value, filter$notice_threshold),
+			        $src=m$index$host, $n=m$value,
+			        $metric_index=index]);
+			}
+		
+		else if ( filter?$notice_thresholds &&
+		          m$value >= filter$notice_thresholds[thresholds[cat(filter$id,filter$name)]] )
+			{
+			# TODO: implement this
+			}
+		
+		if ( filter$log )
+			Log::write(METRICS, m);
+		}
+	}
+
+
 function reset(filter: Filter)
 	{
 	store[filter$id, filter$name] = table();
 	}
 
-event log_it(filter: Filter)
-	{
-	# If this node is the manager in a cluster, this needs to request values
-	# for this metric from all of the workers.
-	
-	local id = filter$id;
-	local name = filter$name;
-	for ( agg_subnet in store[id, name] )
-		{
-		local metric_values = store[id, name][agg_subnet];
-		for ( index in metric_values )
-			{
-			local val = metric_values[index];
-			local m: Info = [$ts=network_time(), 
-			                 $metric_id=id,
-			                 $filter_name=name,
-			                 $agg_subnet=fmt("%s", agg_subnet),
-			                 $index=index,
-			                 $value=val];
-			
-			if ( filter?$notice_threshold &&
-			     m$value >= filter$notice_threshold )
-				{
-				print m;
-				NOTICE([$note=filter$note, 
-				        $msg=fmt("Metrics threshold crossed by %s %d/%d", m$agg_subnet, m$value, filter$notice_threshold),
-				        $n=m$value]);
-				}
-			
-			else if ( filter?$notice_thresholds &&
-			          m$value >= filter$notice_thresholds[thresholds[cat(id,name)]] )
-				{
-				# TODO: implement this
-				}
-			
-			# If there wasn't an index, remove the field.
-			if ( index == "" )
-				delete m$index;
-			
-			# If there wasn't an aggregation subnet, remove the field.
-			if ( agg_subnet == "" )
-				delete m$agg_subnet;
-				
-			Log::write(METRICS, m);
-			}
-		}
-	reset(filter);
-	
-	schedule filter$break_interval { log_it(filter) };
-	}
-
 function add_filter(id: ID, filter: Filter)
 	{
 	if ( filter?$aggregation_table && filter?$aggregation_mask )
@@ -177,13 +169,13 @@ function add_filter(id: ID, filter: Filter)
 		metric_filters[id] = vector();
 	metric_filters[id][|metric_filters[id]|] = filter;
 
+	filter_store[id, filter$name] = filter;
 	store[id, filter$name] = table();
 	
-	# Only do this on the manager if in a cluster.
-	schedule filter$break_interval { log_it(filter) };
+	schedule filter$break_interval { Metrics::log_it(filter) };
 	}
 	
-function add_data(id: ID, entry: Entry)
+function add_data(id: ID, index: Index, increment: count)
 	{
 	if ( id !in metric_filters )
 		return;
@@ -196,38 +188,28 @@ function add_data(id: ID, entry: Entry)
 		local filter = filters[filter_id];
 		
 		# If this filter has a predicate, run the predicate and skip this
-		# entry if the predicate return false.
+		# index if the predicate return false.
 		if ( filter?$pred &&
-		     ! filter$pred(entry) )
+		     ! filter$pred(index) )
 			next;
 		
-		local agg_subnet = "";
 		local filt_store = store[id, filter$name];
-		if ( entry?$host )
+		if ( index?$host )
 			{
 			if ( filter?$aggregation_mask )
 				{
-				local agg_mask = filter$aggregation_mask;
-				agg_subnet = fmt("%s", mask_addr(entry$host, agg_mask));
+				index$network = mask_addr(index$host, filter$aggregation_mask);
+				delete index$host;
 				}
 			else if ( filter?$aggregation_table )
 				{
-				agg_subnet = fmt("%s", filter$aggregation_table[entry$host]);
-				# if an aggregation table is being used and the value isn't 
-				# in the table, that means we aren't interested in it.
-				if ( agg_subnet == "" )
-					next;
+				index$network = filter$aggregation_table[index$host];
+				delete index$host;
 				}
-			else
-				agg_subnet = fmt("%s", entry$host);
 			}
 
-		if ( agg_subnet !in filt_store )
-			filt_store[agg_subnet] = table();
-
-		local fs = filt_store[agg_subnet];
-		if ( entry$index !in fs )
-			fs[entry$index] = 0;
-		fs[entry$index] = fs[entry$index] + entry$increment;
+		if ( index !in filt_store )
+			filt_store[index] = 0;
+		filt_store[index] += increment;
 		}
 	}
diff --git a/scripts/base/frameworks/metrics/non-cluster.bro b/scripts/base/frameworks/metrics/non-cluster.bro
new file mode 100644
index 0000000000..a96210649e
--- /dev/null
+++ b/scripts/base/frameworks/metrics/non-cluster.bro
@@ -0,0 +1,17 @@
+
+module Metrics;
+
+export {
+
+}
+
+event Metrics::log_it(filter: Filter)
+	{
+	local id = filter$id;
+	local name = filter$name;
+	
+	write_log(network_time(), filter, store[id, name]);
+	reset(filter);
+	
+	schedule filter$break_interval { Metrics::log_it(filter) };
+	}
diff --git a/scripts/base/init-default.bro b/scripts/base/init-default.bro
index 32201fddc0..1cf125c3ab 100644
--- a/scripts/base/init-default.bro
+++ b/scripts/base/init-default.bro
@@ -23,11 +23,11 @@
 @load base/frameworks/signatures
 @load base/frameworks/packet-filter
 @load base/frameworks/software
-@load base/frameworks/intel
-@load base/frameworks/metrics
 @load base/frameworks/communication
 @load base/frameworks/control
 @load base/frameworks/cluster
+@load base/frameworks/metrics
+@load base/frameworks/intel
 @load base/frameworks/reporter
 
 @load base/protocols/conn
diff --git a/scripts/base/utils/site.bro b/scripts/base/utils/site.bro
index 96fc46ec56..536c891572 100644
--- a/scripts/base/utils/site.bro
+++ b/scripts/base/utils/site.bro
@@ -17,9 +17,11 @@ export {
 	## Networks that are considered "local".
 	const local_nets: set[subnet] &redef;
 	
-	## This is used for mapping between local networks and string
-	## values for the CIDRs represented.
-	global local_nets_table: table[subnet] of string = {};
+	## This is used for retrieving the subnet when you multiple 
+	## :bro:id:`local_nets`.  A membership query can be done with an 
+	## :bro:type:`addr` and the table will yield the subnet it was found 
+	## within.
+	global local_nets_table: table[subnet] of subnet = {};
 
 	## Networks that are considered "neighbors".
 	const neighbor_nets: set[subnet] &redef;
@@ -145,6 +147,6 @@ event bro_init() &priority=10
 
 	# Create the local_nets mapping table.
 	for ( cidr in Site::local_nets )
-		local_nets_table[cidr] = fmt("%s", cidr);
+		local_nets_table[cidr] = cidr;
 
 	}
diff --git a/scripts/policy/frameworks/metrics/conn-example.bro b/scripts/policy/frameworks/metrics/conn-example.bro
index e67117a7e0..61360496cf 100644
--- a/scripts/policy/frameworks/metrics/conn-example.bro
+++ b/scripts/policy/frameworks/metrics/conn-example.bro
@@ -14,7 +14,7 @@ event bro_init()
 
 event connection_established(c: connection)
 	{
-	Metrics::add_data(CONNS_ORIGINATED, [$host=c$id$orig_h]);
-	Metrics::add_data(CONNS_RESPONDED,  [$host=c$id$resp_h]);
+	Metrics::add_data(CONNS_ORIGINATED, [$host=c$id$orig_h], 1);
+	Metrics::add_data(CONNS_RESPONDED,  [$host=c$id$resp_h], 1);
 	}
 	
\ No newline at end of file
diff --git a/scripts/policy/frameworks/metrics/http-example.bro b/scripts/policy/frameworks/metrics/http-example.bro
index 904ec9a227..94592a852f 100644
--- a/scripts/policy/frameworks/metrics/http-example.bro
+++ b/scripts/policy/frameworks/metrics/http-example.bro
@@ -1,5 +1,4 @@
 
-
 redef enum Metrics::ID += {
 	HTTP_REQUESTS_BY_STATUS_CODE,
 	HTTP_REQUESTS_BY_HOST_HEADER,
@@ -7,16 +6,21 @@ redef enum Metrics::ID += {
 
 event bro_init()
 	{
-	Metrics::add_filter(HTTP_REQUESTS_BY_HOST_HEADER, [$break_interval=5mins]);
-	
-	# Site::local_nets must be defined in order for this to actually do anything.
-	Metrics::add_filter(HTTP_REQUESTS_BY_STATUS_CODE, [$aggregation_table=Site::local_nets_table, $break_interval=5mins]);
+	# TODO: these are waiting on a fix with table vals + records before they will work.
+	#Metrics::add_filter(HTTP_REQUESTS_BY_HOST_HEADER, 
+	#                    [$pred(index: Index) = { return Site:is_local_addr(index$host) },
+	#                     $aggregation_mask=24,
+	#                     $break_interval=5mins]);
+	#
+	## Site::local_nets must be defined in order for this to actually do anything.
+	#Metrics::add_filter(HTTP_REQUESTS_BY_STATUS_CODE, [$aggregation_table=Site::local_nets_table,
+	#                                                   $break_interval=5mins]);
 	}
 
 event HTTP::log_http(rec: HTTP::Info)
 	{
 	if ( rec?$host )
-		Metrics::add_data(HTTP_REQUESTS_BY_HOST_HEADER, [$index=rec$host]);
+		Metrics::add_data(HTTP_REQUESTS_BY_HOST_HEADER, [$str=rec$host]);
 	if ( rec?$status_code )
-		Metrics::add_data(HTTP_REQUESTS_BY_STATUS_CODE, [$host=rec$id$orig_h, $index=fmt("%d", rec$status_code)]);
+		Metrics::add_data(HTTP_REQUESTS_BY_STATUS_CODE, [$host=rec$id$orig_h, $str=fmt("%d", rec$status_code)]);
 	}
\ No newline at end of file
diff --git a/scripts/policy/frameworks/metrics/ssl-example.bro b/scripts/policy/frameworks/metrics/ssl-example.bro
index e043690feb..f3c5b8b902 100644
--- a/scripts/policy/frameworks/metrics/ssl-example.bro
+++ b/scripts/policy/frameworks/metrics/ssl-example.bro
@@ -8,8 +8,8 @@ event bro_init()
 	{
 	Metrics::add_filter(SSL_SERVERNAME, 
 		[$name="no-google-ssl-servers",
-		 $pred(entry: Metrics::Entry) = { 
-		    return (/google\.com$/ !in entry$index); 
+		 $pred(index: Metrics::Index) = { 
+		    return (/google\.com$/ !in index$str); 
 		 },
 		 $break_interval=10secs
 		]);
@@ -18,5 +18,5 @@ event bro_init()
 event SSL::log_ssl(rec: SSL::Info)
 	{
 	if ( rec?$server_name )
-		Metrics::add_data(SSL_SERVERNAME, [$index=rec$server_name]);
+		Metrics::add_data(SSL_SERVERNAME, [$str=rec$server_name], 1);
 	}
\ No newline at end of file
diff --git a/testing/btest/Baseline/policy.frameworks.metrics.basic-cluster/manager-1.metrics.log b/testing/btest/Baseline/policy.frameworks.metrics.basic-cluster/manager-1.metrics.log
new file mode 100644
index 0000000000..ff692027b2
--- /dev/null
+++ b/testing/btest/Baseline/policy.frameworks.metrics.basic-cluster/manager-1.metrics.log
@@ -0,0 +1,4 @@
+# ts	metric_id	filter_name	index.host	index.str	index.network	value
+1313429477.091485	TEST_METRIC	foo-bar	6.5.4.3	-	-	4
+1313429477.091485	TEST_METRIC	foo-bar	1.2.3.4	-	-	6
+1313429477.091485	TEST_METRIC	foo-bar	7.2.1.5	-	-	2
diff --git a/testing/btest/Baseline/policy.frameworks.metrics.basic/metrics.log b/testing/btest/Baseline/policy.frameworks.metrics.basic/metrics.log
new file mode 100644
index 0000000000..fb4a2c4528
--- /dev/null
+++ b/testing/btest/Baseline/policy.frameworks.metrics.basic/metrics.log
@@ -0,0 +1,4 @@
+# ts	metric_id	filter_name	index.host	index.str	index.network	value
+1313430544.678529	TEST_METRIC	foo-bar	6.5.4.3	-	-	2
+1313430544.678529	TEST_METRIC	foo-bar	1.2.3.4	-	-	3
+1313430544.678529	TEST_METRIC	foo-bar	7.2.1.5	-	-	1
diff --git a/testing/btest/Baseline/policy.frameworks.metrics.notice/notice.log b/testing/btest/Baseline/policy.frameworks.metrics.notice/notice.log
new file mode 100644
index 0000000000..112fe69e4b
--- /dev/null
+++ b/testing/btest/Baseline/policy.frameworks.metrics.notice/notice.log
@@ -0,0 +1,4 @@
+# ts	uid	id.orig_h	id.orig_p	id.resp_h	id.resp_p	note	msg	sub	src	dst	p	n	peer_descr	actions	policy_items	dropped	remote_location.country_code	remote_location.region	remote_location.city	remote_location.latitude	remote_location.longitude	metric_index.host	metric_index.str	metric_index.network
+1313432466.662314	-	-	-	-	-	Test_Notice	Metrics threshold crossed by 6.5.4.3 2/1	-	6.5.4.3	-	-	2	bro	Notice::ACTION_LOG	4	-	-	-	-	-	-	6.5.4.3	-	-
+1313432466.662314	-	-	-	-	-	Test_Notice	Metrics threshold crossed by 1.2.3.4 3/1	-	1.2.3.4	-	-	3	bro	Notice::ACTION_LOG	4	-	-	-	-	-	-	1.2.3.4	-	-
+1313432466.662314	-	-	-	-	-	Test_Notice	Metrics threshold crossed by 7.2.1.5 1/1	-	7.2.1.5	-	-	1	bro	Notice::ACTION_LOG	4	-	-	-	-	-	-	7.2.1.5	-	-
diff --git a/testing/btest/policy/frameworks/metrics/basic-cluster.bro b/testing/btest/policy/frameworks/metrics/basic-cluster.bro
new file mode 100644
index 0000000000..eda41c3759
--- /dev/null
+++ b/testing/btest/policy/frameworks/metrics/basic-cluster.bro
@@ -0,0 +1,38 @@
+# @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: sleep 1
+# @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 6
+# @TEST-EXEC: btest-diff manager-1/metrics.log
+
+@TEST-START-FILE cluster-layout.bro
+redef Cluster::nodes = {
+	["manager-1"] = [$node_type=Cluster::MANAGER, $ip=127.0.0.1, $p=37757/tcp, $workers=set("worker-1")],
+	["proxy-1"] = [$node_type=Cluster::PROXY,     $ip=127.0.0.1, $p=37758/tcp, $manager="manager-1", $workers=set("worker-1")],
+	["worker-1"] = [$node_type=Cluster::WORKER,   $ip=127.0.0.1, $p=37760/tcp, $manager="manager-1", $proxy="proxy-1", $interface="eth0"],
+	["worker-2"] = [$node_type=Cluster::WORKER,   $ip=127.0.0.1, $p=37761/tcp, $manager="manager-1", $proxy="proxy-1", $interface="eth1"],
+};
+@TEST-END-FILE
+
+redef enum Metrics::ID += {
+	TEST_METRIC,
+};
+
+event bro_init() &priority=5
+	{
+	Metrics::add_filter(TEST_METRIC, 
+		[$name="foo-bar",
+		 $break_interval=3secs]);
+	}
+
+@if ( Cluster::local_node_type() == Cluster::WORKER )
+
+event bro_init()
+	{
+	Metrics::add_data(TEST_METRIC, [$host=1.2.3.4], 3);
+	Metrics::add_data(TEST_METRIC, [$host=6.5.4.3], 2);
+	Metrics::add_data(TEST_METRIC, [$host=7.2.1.5], 1);
+	}
+
+@endif
\ No newline at end of file
diff --git a/testing/btest/policy/frameworks/metrics/basic.bro b/testing/btest/policy/frameworks/metrics/basic.bro
new file mode 100644
index 0000000000..43e7ac28ef
--- /dev/null
+++ b/testing/btest/policy/frameworks/metrics/basic.bro
@@ -0,0 +1,16 @@
+# @TEST-EXEC: bro %INPUT
+# @TEST-EXEC: btest-diff metrics.log
+
+redef enum Metrics::ID += {
+	TEST_METRIC,
+};
+
+event bro_init() &priority=5
+	{
+	Metrics::add_filter(TEST_METRIC, 
+		[$name="foo-bar",
+		 $break_interval=3secs]);
+	Metrics::add_data(TEST_METRIC, [$host=1.2.3.4], 3);
+	Metrics::add_data(TEST_METRIC, [$host=6.5.4.3], 2);
+	Metrics::add_data(TEST_METRIC, [$host=7.2.1.5], 1);
+	}
diff --git a/testing/btest/policy/frameworks/metrics/notice.bro b/testing/btest/policy/frameworks/metrics/notice.bro
new file mode 100644
index 0000000000..3451af18f4
--- /dev/null
+++ b/testing/btest/policy/frameworks/metrics/notice.bro
@@ -0,0 +1,23 @@
+# @TEST-EXEC: bro %INPUT
+# @TEST-EXEC: btest-diff notice.log
+
+redef enum Notice::Type += {
+	Test_Notice,
+};
+
+redef enum Metrics::ID += {
+	TEST_METRIC,
+};
+
+event bro_init() &priority=5
+	{
+	Metrics::add_filter(TEST_METRIC, 
+		[$name="foo-bar",
+		 $break_interval=3secs,
+		 $note=Test_Notice,
+		 $notice_threshold=1,
+		 $log=F]);
+	Metrics::add_data(TEST_METRIC, [$host=1.2.3.4], 3);
+	Metrics::add_data(TEST_METRIC, [$host=6.5.4.3], 2);
+	Metrics::add_data(TEST_METRIC, [$host=7.2.1.5], 1);
+	}