Checkpoint

scripts/base/frameworks/measurement/__load__.bro

@@ -0,0 +1,13 @@
+@load ./main
+
+@load ./plugins
+
+# 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

scripts/base/frameworks/measurement/cluster.bro

@@ -0,0 +1,308 @@
+##! 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
+@load ./main
+
+module Measurement;
+
+export {
+	## Allows a user to decide how large of result groups the 
+	## workers should transmit values for cluster metric aggregation.
+	const cluster_send_in_groups_of = 50 &redef;
+	
+	## The percent of the full threshold value that needs to be met 
+	## on a single worker for that worker to send the value to its manager in
+	## order for it to request a global view for that value.  There is no
+	## requirement that the manager requests a global view for the index
+	## since it may opt not to if it requested a global view for the index
+	## recently.
+	const cluster_request_global_view_percent = 0.2 &redef;
+
+	## This is to deal with intermediate update overload.  A manager will only allow
+	## this many intermediate update requests to the workers to be inflight at 
+	## any given time.  Requested intermediate updates are currently thrown out
+	## and not performed.  In practice this should hopefully have a minimal effect.
+	const max_outstanding_global_views = 10 &redef;
+
+	## Intermediate updates can cause overload situations on very large clusters.
+	## This option may help reduce load and correct intermittent problems.
+	## The goal for this option is also meant to be temporary.
+	const enable_intermediate_updates = T &redef;
+
+	# Event sent by the manager in a cluster to initiate the 
+	# collection of metrics values for a filter.
+	global cluster_filter_request: event(uid: string, id: string, filter_name: string);
+
+	# Event sent by nodes that are collecting metrics after receiving
+	# a request for the metric filter from the manager.
+	global cluster_filter_response: event(uid: string, id: string, filter_name: string, data: MetricTable, done: bool);
+
+	# This event is sent by the manager in a cluster to initiate the
+	# collection of a single index value from a filter.  It's typically
+	# used to get intermediate updates before the break interval triggers
+	# to speed detection of a value crossing a threshold.
+	global cluster_index_request: event(uid: string, id: string, filter_name: string, index: Index);
+
+	# This event is sent by nodes in response to a 
+	# :bro:id:`Measurement::cluster_index_request` event.
+	global cluster_index_response: event(uid: string, id: string, filter_name: string, index: Index, val: ResultVal);
+
+	# This is sent by workers to indicate that they crossed the percent of the 
+	# current threshold by the percentage defined globally in 
+	# :bro:id:`Measurement::cluster_request_global_view_percent`
+	global cluster_index_intermediate_response: event(id: string, filter_name: string, index: Measurement::Index);
+
+	# This event is scheduled internally on workers to send result chunks.
+	global send_data: event(uid: string, id: string, filter_name: string, data: MetricTable);
+}
+
+
+# Add events to the cluster framework to make this work.
+redef Cluster::manager2worker_events += /Measurement::cluster_(filter_request|index_request)/;
+redef Cluster::worker2manager_events += /Measurement::cluster_(filter_response|index_response|index_intermediate_response)/;
+
+@if ( Cluster::local_node_type() != Cluster::MANAGER )
+# This variable is maintained to know what indexes they have recently sent as 
+# intermediate updates so they don't overwhelm their manager. The count that is
+# yielded is the number of times the percentage threshold has been crossed and
+# an intermediate result has been received.
+global recent_global_view_indexes: table[string, string, Index] of count &create_expire=1min &default=0;
+
+# This is done on all non-manager node types in the event that a metric is 
+# being collected somewhere other than a worker.
+function data_added(filter: Filter, index: Index, val: ResultVal)
+	{
+	# If an intermediate update for this value was sent recently, don't send
+	# it again.
+	if ( [filter$id, filter$name, index] in recent_global_view_indexes )
+		return;
+
+	# If val is 5 and global view % is 0.1 (10%), pct_val will be 50.  If that
+	# crosses the full threshold then it's a candidate to send as an 
+	# intermediate update.
+	if ( enable_intermediate_updates && 
+	     check_thresholds(filter, index, val, cluster_request_global_view_percent) )
+		{
+		# kick off intermediate update
+		event Measurement::cluster_index_intermediate_response(filter$id, filter$name, index);
+		++recent_global_view_indexes[filter$id, filter$name, index];
+		}
+	}
+
+event Measurement::send_data(uid: string, id: string, 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 Measurement::cluster_filter_response(uid, id, filter_name, local_data, done);
+	if ( ! done )
+		event Measurement::send_data(uid, id, filter_name, data);
+	}
+
+event Measurement::cluster_filter_request(uid: string, id: string, filter_name: string)
+	{
+	#print fmt("WORKER %s: received the cluster_filter_request event for %s.", Cluster::node, id);
+	
+	# Initiate sending all of the data for the requested filter.
+	event Measurement::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 internally by the send_data event.
+	reset(filter_store[id, filter_name]);
+	}
+	
+event Measurement::cluster_index_request(uid: string, id: string, filter_name: string, index: Index)
+	{
+	if ( [id, filter_name] in store && index in store[id, filter_name] )
+		{
+		#print fmt("WORKER %s: received the cluster_index_request event for %s=%s.", Cluster::node, index2str(index), data);
+		event Measurement::cluster_index_response(uid, id, filter_name, index, store[id, filter_name][index]);
+		}
+	else
+		{
+		# We need to send an empty response if we don't have the data so that the manager
+		# can know that it heard back from all of the workers.
+		event Measurement::cluster_index_response(uid, id, filter_name, index, [$begin=network_time(), $end=network_time()]);
+		}
+	}
+
+@endif
+
+
+@if ( Cluster::local_node_type() == Cluster::MANAGER )
+
+# This variable is maintained by manager nodes as they collect and aggregate 
+# results.
+global filter_results: table[string, string, string] of MetricTable &read_expire=1min;
+
+# 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;
+
+# 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 &read_expire=1min &default=0;
+
+# This variable is maintained by managers to track intermediate responses as 
+# they are getting a global view for a certain index.
+global index_requests: table[string, string, string, Index] of ResultVal &read_expire=1min;
+
+# This variable is maintained by managers to prevent overwhelming communication due
+# to too many intermediate updates.  Each metric filter is tracked separately so that 
+# one metric won't overwhelm and degrade other quieter metrics.
+global outstanding_global_views: table[string, string] of count &default=0;
+
+# Managers handle logging.
+event Measurement::finish_period(filter: Filter)
+	{
+	#print fmt("%.6f MANAGER: breaking %s filter for %s metric", network_time(), filter$name, filter$id);
+	local uid = unique_id("");
+	
+	# Set some tracking variables.
+	requested_results[uid] = network_time();
+	if ( [uid, filter$id, filter$name] in filter_results )
+		delete filter_results[uid, filter$id, filter$name];
+	filter_results[uid, filter$id, filter$name] = table();
+	
+	# Request data from peers.
+	event Measurement::cluster_filter_request(uid, filter$id, filter$name);
+	# Schedule the next finish_period event.
+	schedule filter$every { Measurement::finish_period(filter) };
+	}
+
+# This is unlikely to be called often, but it's here in case there are metrics
+# being collected by managers.
+function data_added(filter: Filter, index: Index, val: ResultVal)
+	{
+	if ( check_thresholds(filter, index, val, 1.0) )
+		threshold_crossed(filter, index, val);
+	}
+	
+event Measurement::cluster_index_response(uid: string, id: string, filter_name: string, index: Index, val: ResultVal)
+	{
+	#print fmt("%0.6f MANAGER: receiving index data from %s - %s=%s", network_time(), get_event_peer()$descr, index2str(index), val);
+
+	# We only want to try and do a value merge if there are actually measured datapoints
+	# in the ResultVal.
+	if ( val$num > 0 && [uid, id, filter_name, index] in index_requests )
+		index_requests[uid, id, filter_name, index] = merge_result_vals(index_requests[uid, id, filter_name, index], val);
+	else
+		index_requests[uid, id, filter_name, index] = val;
+
+	# Mark that this worker is done.
+	++done_with[uid];
+
+	#print fmt("worker_count:%d :: done_with:%d", Cluster::worker_count, done_with[uid]);
+	if ( Cluster::worker_count == done_with[uid] )
+		{
+		local ir = index_requests[uid, id, filter_name, index];
+		if ( check_thresholds(filter_store[id, filter_name], index, ir, 1.0) )
+			{
+			threshold_crossed(filter_store[id, filter_name], index, ir);
+			}
+		delete done_with[uid];
+		delete index_requests[uid, id, filter_name, index];
+		# Check that there is an outstanding view before subtracting.
+		if ( outstanding_global_views[id, filter_name] > 0 )
+			--outstanding_global_views[id, filter_name];
+		}
+	}
+
+# Managers handle intermediate updates here.
+event Measurement::cluster_index_intermediate_response(id: string, filter_name: string, index: Index)
+	{
+	#print fmt("MANAGER: receiving intermediate index data from %s", get_event_peer()$descr);
+	#print fmt("MANAGER: requesting index data for %s", index2str(index));
+
+	if ( [id, filter_name] in outstanding_global_views &&
+	     |outstanding_global_views[id, filter_name]| > max_outstanding_global_views )
+		{
+		# Don't do this intermediate update.  Perhaps at some point in the future 
+		# we will queue and randomly select from these ignored intermediate
+		# update requests.
+		return;
+		}
+
+	++outstanding_global_views[id, filter_name];
+
+	local uid = unique_id("");
+	event Measurement::cluster_index_request(uid, id, filter_name, index);
+	}
+
+event Measurement::cluster_filter_response(uid: string, id: string, filter_name: string, data: MetricTable, done: bool)
+	{
+	#print fmt("MANAGER: receiving results from %s", get_event_peer()$descr);
+	
+	# Mark another worker as being "done" for this uid.
+	if ( done )
+		++done_with[uid];
+
+	local local_data = filter_results[uid, id, filter_name];
+	local filter = filter_store[id, filter_name];
+
+	for ( index in data )
+		{
+		if ( index in local_data )
+			local_data[index] = merge_result_vals(local_data[index], data[index]);
+		else
+			local_data[index] = data[index];
+
+		# If a filter is done being collected, thresholds for each index
+		# need to be checked so we're doing it here to avoid doubly iterating 
+		# over each index.
+		if ( Cluster::worker_count == done_with[uid] )
+			{
+			if ( check_thresholds(filter, index, local_data[index], 1.0) )
+				{
+				threshold_crossed(filter, index, local_data[index]);
+				}
+			}
+		}
+	
+	# If the data has been collected from all peers, we are done and ready to finish.
+	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];
+			delete requested_results[uid];
+			}
+		
+		if ( filter?$period_finished )
+			filter$period_finished(ts, filter$id, filter$name, local_data);
+
+		# Clean up
+		delete filter_results[uid, id, filter_name];
+		delete done_with[uid];
+		# Not sure I need to reset the filter on the manager.
+		reset(filter);
+		}
+	}
+
+@endif

scripts/base/frameworks/measurement/main.bro

@@ -0,0 +1,367 @@
+##! The metrics framework provides a way to count and measure data.  
+
+@load base/utils/queue
+
+module Measurement;
+
+export {
+	## The metrics logging stream identifier.
+	redef enum Log::ID += { LOG };
+	
+	## This is the interval for how often threshold based notices will happen 
+	## after they have already fired.
+	const threshold_crossed_restart_interval = 1hr &redef;
+	
+	## The various calculations are all defined as plugins.
+	type Calculation: enum {
+		PLACEHOLDER
+	};
+
+	## Represents a thing which is having metrics collected for it.  An instance
+	## of this record type and an id together represent a single measurement.
+	type Index: record {
+		## A non-address related metric or a sub-key for an address based metric.
+		## An example might be successful SSH connections by client IP address
+		## where the client string would be the index value.
+		## Another example might be number of HTTP requests to a particular
+		## 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.
+		str:          string &optional;
+	
+		## Host is the value to which this metric applies.
+		host:         addr &optional;
+	} &log;
+	
+	## Represents data being added for a single metric data point.
+	## Only supply a single value here at a time.
+	type DataPoint: record {
+		## Count value.
+		num:       count       &optional;
+		## Double value.
+		dbl:       double      &optional;
+		## String value.
+		str:       string      &optional;
+	};
+
+	## Value supplied when a metric is finished.  It contains all
+	## of the measurements collected for the metric.  Most of the
+	## fields are added by calculation plugins.
+	type ResultVal: record {
+		## The time when this result was first started.
+		begin:    time          &log;
+
+		## The time when the last value was added to this result.
+		end:      time          &log;
+
+		## The number of measurements received.
+		num:      count         &log &default=0;
+
+		## A sample of something being measured.  This is helpful in 
+		## some cases for collecting information to do further detection
+		## or better logging for forensic purposes.
+		samples:  vector of string   &optional;
+	};
+	
+	type Measurement: record {
+		## The calculations to perform on the data.
+		apply:          set[Calculation];
+		
+		## A predicate so that you can decide per index if you would like
+		## to accept the data being inserted.
+		pred:           function(index: Measurement::Index, data: Measurement::DataPoint): bool &optional;
+		
+		## A function to normalize the index.  This can be used to aggregate or
+		## normalize the entire index.
+		normalize_func: function(index: Measurement::Index): Index &optional;
+
+		## A number of sample DataPoints to collect.
+		samples:        count &optional;
+	};
+
+
+	type Results: record {
+		begin: time;
+		end:   time;
+		result
+	};
+
+	## Type to store a table of metrics result values.
+	type ResultTable: table[Index] of Results;
+
+	## Filters define how the data from a metric is aggregated and handled.  
+	## Filters can be used to set how often the measurements are cut 
+	## and logged or how the data within them is aggregated.
+	type Filter: record {
+		## A name for the filter in case multiple filters are being
+		## applied to the same metric.  In most cases the default 
+		## filter name is fine and this field does not need to be set.
+		id:              string;
+
+		## The interval at which this filter should be "broken" and written
+		## to the logging stream.  The counters are also reset to zero at 
+		## this time so any threshold based detection needs to be set to a 
+		## number that should be expected to happen within this period.
+		every:              interval;
+
+		## Optionally provide a function to calculate a value from the ResultVal 
+		## structure which will be used for thresholding.  If no function is 
+		## provided, then in the following order of preference either the 
+		## $unique or the $sum fields will be used.
+		threshold_val_func: function(val: Measurement::ResultVal): count  &optional;
+
+		## The threshold value for calling the $threshold_crossed callback.
+		threshold:          count                    &optional;
+		
+		## A series of thresholds for calling the $threshold_crossed callback.
+		threshold_series:   vector of count          &optional;
+		
+		## A callback with the full collection of ResultVals for this filter.
+		## It's best to not access any global state outside of the variables
+		## given to the callback because there is no assurance provided as to
+		## where the callback will be executed on clusters.
+		period_finished:    function(data: Measurement::ResultTable) &optional;
+
+		## A callback that is called when a threshold is crossed.
+		threshold_crossed:  function(index: Measurement::Index, val: Measurement::ResultVal) &optional;
+	};
+	
+	## Function to associate a metric filter with a metric ID.
+	## 
+	## id: The metric ID that the filter should be associated with.
+	##
+	## filter: The record representing the filter configuration.
+	global add_filter: function(id: string, filter: Measurement::Filter);
+	
+	## Add data into a metric.  This should be called when
+	## a script has measured some point value and is ready to increment the
+	## counters.
+	##
+	## id: The metric identifier that the data represents.
+	##
+	## index: The metric index that the value is to be added to.
+	##
+	## increment: How much to increment the counter by.
+	global add_data: function(id: string, index: Measurement::Index, data: Measurement::DataPoint);
+
+	## Helper function to represent a :bro:type:`Measurement::Index` value as 
+	## a simple string.
+	## 
+	## index: The metric index that is to be converted into a string.
+	##
+	## Returns: A string reprentation of the metric index.
+	global index2str: function(index: Measurement::Index): string;
+	
+	## Event to access metrics records as they are passed to the logging framework.
+	global log_metrics: event(rec: Measurement::Info);
+	
+}
+
+redef record Filter += {
+	# Internal use only.  The metric that this filter applies to.  The value is automatically set.
+	id: string &optional;
+};
+
+redef record ResultVal += {
+	# Internal use only.  This is the queue where samples
+	# are maintained since the queue is self managing for
+	# the number of samples requested.
+	sample_queue: Queue::Queue &optional;
+
+	# Internal use only.  Indicates if a simple threshold was already crossed.
+	is_threshold_crossed: bool &default=F;
+
+	# Internal use only.  Current index for threshold series.
+	threshold_series_index: count &default=0;
+};
+
+# Store the filters indexed on the metric identifier and filter name.
+global filter_store: table[string, string] of Filter = table();
+
+# This is indexed by metric id and filter name.
+global store: table[string, string] of ResultTable = table();
+
+# This is a hook for watching thresholds being crossed.  It is called whenever
+# index values are updated and the new val is given as the `val` argument.
+# It's only prototyped here because cluster and non-cluster have separate 
+# implementations.
+global data_added: function(filter: Filter, index: Index, val: ResultVal);
+
+# Prototype the hook point for plugins to do calculations.
+global add_to_calculation: hook(filter: Filter, val: double, data: DataPoint, result: ResultVal);
+# Prototype the hook point for plugins to merge Measurements.
+global plugin_merge_measurements: hook(result: ResultVal, rv1: ResultVal, rv2: ResultVal);
+
+# Event that is used to "finish" metrics and adapt the metrics
+# framework for clustered or non-clustered usage.
+global finish_period: event(filter: Measurement::Filter);
+
+event bro_init() &priority=5
+	{
+	Log::create_stream(Measurement::LOG, [$columns=Info, $ev=log_metrics]);
+	}
+
+function index2str(index: Index): string
+	{
+	local out = "";
+	if ( index?$host )
+		out = fmt("%shost=%s", out, index$host);
+	if ( index?$str )
+		out = fmt("%s%sstr=%s", out, |out|==0 ? "" : ", ", index$str);
+	return fmt("metric_index(%s)", out);
+	}
+
+function merge_result_vals(rv1: ResultVal, rv2: ResultVal): ResultVal
+	{
+	local result: ResultVal;
+
+	# Merge $begin (take the earliest one)
+	result$begin = (rv1$begin < rv2$begin) ? rv1$begin : rv2$begin;
+
+	# Merge $end (take the latest one)
+	result$end = (rv1$end > rv2$end) ? rv1$end : rv2$end;
+
+	# Merge $num
+	result$num = rv1$num + rv2$num;
+
+	hook plugin_merge_measurements(result, rv1, rv2);
+	
+	# Merge $sample_queue
+	if ( rv1?$sample_queue && rv2?$sample_queue )
+		result$sample_queue = Queue::merge(rv1$sample_queue, rv2$sample_queue);
+	else if ( rv1?$sample_queue )
+		result$sample_queue = rv1$sample_queue;
+	else if ( rv2?$sample_queue )
+		result$sample_queue = rv2$sample_queue;
+	
+	# Merge $threshold_series_index
+	result$threshold_series_index = (rv1$threshold_series_index > rv2$threshold_series_index) ? rv1$threshold_series_index : rv2$threshold_series_index;
+
+	# Merge $is_threshold_crossed
+	if ( rv1$is_threshold_crossed || rv2$is_threshold_crossed )
+		result$is_threshold_crossed = T;
+
+	return result;
+	}
+	
+function reset(filter: Filter)
+	{
+	if ( [filter$id, filter$name] in store )
+		delete store[filter$id, filter$name];
+
+	store[filter$id, filter$name] = table();
+	}
+
+function add_filter(id: string, filter: Filter)
+	{
+	if ( [id, filter$name] in store )
+		{
+		Reporter::warning(fmt("invalid Metric filter (%s): Filter with same name already exists.", filter$name));
+		return;
+		}
+
+	if ( ! filter?$id )
+		filter$id = id;
+	
+	filter_store[id, filter$name] = filter;
+	store[id, filter$name] = table();
+
+	schedule filter$every { Measurement::finish_period(filter) };
+	}
+
+function add_data(id: string, index: Index, data: DataPoint)
+	{
+	# Try to add the data to all of the defined filters for the metric.
+	for ( [metric_id, filter_id] in filter_store )
+		{
+		local filter = filter_store[metric_id, filter_id];
+		
+		# If this filter has a predicate, run the predicate and skip this
+		# index if the predicate return false.
+		if ( filter?$pred && ! filter$pred(index, data) )
+			next;
+		
+		#if ( filter?$normalize_func )
+		#	index = filter$normalize_func(copy(index));
+		
+		local metric_tbl = store[id, filter$name];
+		if ( index !in metric_tbl )
+			metric_tbl[index] = [$begin=network_time(), $end=network_time()];
+
+		local result = metric_tbl[index];
+
+		# If a string was given, fall back to 1.0 as the value.
+		local val = 1.0;
+		if ( data?$num || data?$dbl )
+			val = data?$dbl ? data$dbl : data$num;
+
+		++result$num;
+		# Continually update the $end field.
+		result$end=network_time();
+
+		#if ( filter?$samples && filter$samples > 0 && data?$str )
+		#	{
+		#	if ( ! result?$sample_queue )
+		#		result$sample_queue = Queue::init([$max_len=filter$samples]);
+		#	Queue::push(result$sample_queue, data$str);
+		#	}
+
+		hook add_to_calculation(filter, val, data, result);
+		data_added(filter, index, result);
+		}
+	}
+
+# This function checks if a threshold has been crossed.  It is also used as a method to implement 
+# mid-break-interval threshold crossing detection for cluster deployments.
+function check_thresholds(filter: Filter, index: Index, val: ResultVal, modify_pct: double): bool
+	{
+	if ( ! (filter?$threshold || filter?$threshold_series) )
+		return;
+
+	local watch = 0.0;
+	if ( val?$unique )
+		watch = val$unique;
+	else if ( val?$sum )
+		watch = val$sum;
+
+	if ( filter?$threshold_val_func )
+		watch = filter$threshold_val_func(val);
+
+	if ( modify_pct < 1.0 && modify_pct > 0.0 )
+		watch = watch/modify_pct;
+
+	if ( ! val$is_threshold_crossed &&
+	     filter?$threshold && watch >= filter$threshold )
+		{
+		# A default threshold was given and the value crossed it.
+		return T;
+		}
+
+	if ( filter?$threshold_series &&
+	     |filter$threshold_series| >= val$threshold_series_index &&
+	     watch >= filter$threshold_series[val$threshold_series_index] )
+		{
+		# A threshold series was given and the value crossed the next 
+		# value in the series.
+		return T;
+		}
+
+	return F;
+	}
+
+function threshold_crossed(filter: Filter, index: Index, val: ResultVal)
+	{
+	if ( ! filter?$threshold_crossed )
+		return;
+
+	if ( val?$sample_queue )
+		val$samples = Queue::get_str_vector(val$sample_queue);
+
+	filter$threshold_crossed(index, val);
+	val$is_threshold_crossed = T;
+
+	# Bump up to the next threshold series index if a threshold series is being used.
+	if ( filter?$threshold_series )
+		++val$threshold_series_index;
+	}
+

scripts/base/frameworks/measurement/non-cluster.bro

@@ -0,0 +1,21 @@
+@load ./main
+
+module Measurement;
+
+event Measurement::finish_period(filter: Filter)
+	{
+	local data = store[filter$id, filter$name];
+	if ( filter?$period_finished )
+		filter$period_finished(network_time(), filter$id, filter$name, data);
+
+	reset(filter);
+	
+	schedule filter$every { Measurement::finish_period(filter) };
+	}
+	
+	
+function data_added(filter: Filter, index: Index, val: ResultVal)
+	{
+	if ( check_thresholds(filter, index, val, 1.0) )
+		threshold_crossed(filter, index, val);
+	}

scripts/base/frameworks/measurement/plugins/__load__.bro

@@ -0,0 +1,7 @@
+@load ./average
+@load ./max
+@load ./min
+@load ./std-dev
+@load ./sum
+@load ./unique
+@load ./variance

scripts/base/frameworks/measurement/plugins/average.bro

@@ -0,0 +1,35 @@
+
+module Metrics;
+
+export {
+	redef enum Calculation += { 
+		## Calculate the average of the values.
+		AVERAGE
+	};
+
+	redef record ResultVal += {
+		## For numeric data, this calculates the average of all values.
+		average: double &log &optional;
+	};
+}
+
+hook add_to_calculation(filter: Filter, val: double, data: DataPoint, result: ResultVal)
+	{
+	if ( AVERAGE in filter$measure )
+		{
+		if ( ! result?$average ) 
+			result$average = val;
+		else
+			result$average += (val - result$average) / result$num;
+		}
+	}
+
+hook plugin_merge_measurements(result: ResultVal, rv1: ResultVal, rv2: ResultVal)
+	{
+	if ( rv1?$average && rv2?$average )
+		result$average = ((rv1$average*rv1$num) + (rv2$average*rv2$num))/(rv1$num+rv2$num);
+	else if ( rv1?$average )
+		result$average = rv1$average;
+	else if ( rv2?$average )
+		result$average = rv2$average;
+	}

scripts/base/frameworks/measurement/plugins/max.bro

@@ -0,0 +1,37 @@
+
+module Metrics;
+
+export {
+	redef enum Calculation += { 
+		## Find the maximum value.
+		MAX
+	};
+
+	redef record ResultVal += {
+		## For numeric data, this tracks the maximum value given.
+		max:      double        &log &optional;
+	};
+}
+
+hook add_to_calculation(filter: Filter, val: double, data: DataPoint, result: ResultVal)
+	{
+	if ( MAX in filter$measure )
+		{
+		if ( ! result?$max ) 
+			result$max = val;
+		else if ( val > result$max )
+			result$max = val;
+		}
+	}
+
+hook plugin_merge_measurements(result: ResultVal, rv1: ResultVal, rv2: ResultVal)
+	{
+	if ( rv1?$max && rv2?$max )
+		result$max = (rv1$max > rv2$max) ? rv1$max : rv2$max;
+	else if ( rv1?$max )
+		result$max = rv1$max;
+	else if ( rv2?$max )
+		result$max = rv2$max;
+	}
+
+

scripts/base/frameworks/measurement/plugins/min.bro

@@ -0,0 +1,35 @@
+
+module Metrics;
+
+export {
+	redef enum Calculation += { 
+		## Find the minimum value.
+		MIN
+	};
+
+	redef record ResultVal += {
+		## For numeric data, this tracks the minimum value given.
+		min:      double        &log &optional;
+	};
+}
+
+hook add_to_calculation(filter: Filter, val: double, data: DataPoint, result: ResultVal)
+	{
+	if ( MIN in filter$measure )
+		{
+		if ( ! result?$min ) 
+			result$min = val;
+		else if ( val < result$min )
+			result$min = val;
+		}
+	}
+
+hook plugin_merge_measurements(result: ResultVal, rv1: ResultVal, rv2: ResultVal)
+	{
+	if ( rv1?$min && rv2?$min )
+		result$min = (rv1$min < rv2$min) ? rv1$min : rv2$min;
+	else if ( rv1?$min )
+		result$min = rv1$min;
+	else if ( rv2?$min )
+		result$min = rv2$min;
+	}

scripts/base/frameworks/measurement/plugins/std-dev.bro

@@ -0,0 +1,36 @@
+@load ./sum
+@load ./variance
+
+module Metrics;
+
+export {
+	redef enum Calculation += { 
+		## Find the standard deviation of the values.
+		STD_DEV
+	};
+
+	redef record ResultVal += {
+		## For numeric data, this calculates the standard deviation.
+		std_dev:  double &log &optional;
+	};
+}
+
+# This depends on the variance plugin which uses priority -5
+hook add_to_calculation(filter: Filter, val: double, data: DataPoint, result: ResultVal) &priority=-10
+	{
+	if ( STD_DEV in filter$measure )
+		{
+		if ( result?$variance )
+			result$std_dev = sqrt(result$variance);
+		}
+	}
+
+hook plugin_merge_measurements(result: ResultVal, rv1: ResultVal, rv2: ResultVal) &priority=-10
+	{
+	if ( rv1?$sum || rv2?$sum )
+		{
+		result$sum = rv1?$sum ? rv1$sum : 0;
+		if ( rv2?$sum )
+			result$sum += rv2$sum;
+		}
+	}

scripts/base/frameworks/measurement/plugins/sum.bro

@@ -0,0 +1,35 @@
+
+module Metrics;
+
+export {
+	redef enum Calculation += { 
+		## Sums the values given.  For string values,
+		## this will be the number of strings given.
+		SUM
+	};
+
+	redef record ResultVal += {
+		## For numeric data, this tracks the sum of all values.
+		sum:      double        &log &optional;
+	};
+}
+
+hook add_to_calculation(filter: Filter, val: double, data: DataPoint, result: ResultVal)
+	{
+	if ( SUM in filter$measure )
+		{
+		if ( ! result?$sum ) 
+			result$sum = 0;
+		result$sum += val;
+		}
+	}
+
+hook plugin_merge_measurements(result: ResultVal, rv1: ResultVal, rv2: ResultVal)
+	{
+	if ( rv1?$sum || rv2?$sum )
+		{
+		result$sum = rv1?$sum ? rv1$sum : 0;
+		if ( rv2?$sum )
+			result$sum += rv2$sum;
+		}
+	}

scripts/base/frameworks/measurement/plugins/unique.bro

@@ -0,0 +1,51 @@
+
+module Metrics;
+
+export {
+	redef enum Calculation += { 
+		## Calculate the number of unique values.
+		UNIQUE
+	};
+
+	redef record ResultVal += {
+		## If cardinality is being tracked, the number of unique
+		## items is tracked here.
+		unique: count &log &optional;
+	};
+}
+
+redef record ResultVal += {
+	# Internal use only.  This is not meant to be publically available 
+	# because we don't want to trust that we can inspect the values 
+	# since we will like move to a probalistic data structure in the future.
+	# TODO: in the future this will optionally be a hyperloglog structure
+	unique_vals: set[DataPoint] &optional;
+};
+
+hook add_to_calculation(filter: Filter, val: double, data: DataPoint, result: ResultVal)
+	{
+	if ( UNIQUE in filter$measure )
+		{
+		if ( ! result?$unique_vals ) 
+			result$unique_vals=set();
+		add result$unique_vals[data];
+		}
+	}
+
+hook plugin_merge_measurements(result: ResultVal, rv1: ResultVal, rv2: ResultVal)
+	{
+	if ( rv1?$unique_vals || rv2?$unique_vals )
+		{
+		if ( rv1?$unique_vals )
+			result$unique_vals = rv1$unique_vals;
+		
+		if ( rv2?$unique_vals )
+			if ( ! result?$unique_vals )
+				result$unique_vals = rv2$unique_vals;
+			else
+				for ( val2 in rv2$unique_vals )
+					add result$unique_vals[val2];
+
+		result$unique = |result$unique_vals|;
+		}
+	}

scripts/base/frameworks/measurement/plugins/variance.bro

@@ -0,0 +1,65 @@
+@load ./average
+
+module Metrics;
+
+export {
+	redef enum Calculation += { 
+		## Find the variance of the values.
+		VARIANCE
+	};
+
+	redef record ResultVal += {
+		## For numeric data, this calculates the variance.
+		variance: double &log &optional;
+	};
+}
+
+redef record ResultVal += {
+	# Internal use only.  Used for incrementally calculating variance.
+	prev_avg: double &optional;
+
+	# Internal use only.  For calculating incremental variance.
+	var_s: double &optional;
+};
+
+hook add_to_calculation(filter: Filter, val: double, data: DataPoint, result: ResultVal) &priority=5
+	{
+	if ( VARIANCE in filter$measure )
+		result$prev_avg = result$average;
+	}
+
+# Reduced priority since this depends on the average
+hook add_to_calculation(filter: Filter, val: double, data: DataPoint, result: ResultVal) &priority=-5
+	{
+	if ( VARIANCE in filter$measure )
+		{
+		if ( ! result?$var_s )
+			result$var_s = 0.0;
+		result$var_s += (val - result$prev_avg) * (val - result$average);
+		result$variance = (val > 0) ? result$var_s/val : 0.0;
+		}
+	}
+
+# Reduced priority since this depends on the average
+hook plugin_merge_measurements(result: ResultVal, rv1: ResultVal, rv2: ResultVal) &priority=-5
+	{
+	if ( rv1?$var_s && rv2?$var_s )
+		{
+		local rv1_avg_sq = (rv1$average - result$average);
+		rv1_avg_sq = rv1_avg_sq*rv1_avg_sq;
+		local rv2_avg_sq = (rv2$average - result$average);
+		rv2_avg_sq = rv2_avg_sq*rv2_avg_sq;
+		result$var_s = rv1$num*(rv1$var_s/rv1$num + rv1_avg_sq) + rv2$num*(rv2$var_s/rv2$num + rv2_avg_sq);
+		}
+	else if ( rv1?$var_s )
+		result$var_s = rv1$var_s;
+	else if ( rv2?$var_s )
+		result$var_s = rv2$var_s;
+
+	if ( rv1?$prev_avg && rv2?$prev_avg )
+		result$prev_avg = ((rv1$prev_avg*rv1$num) + (rv2$prev_avg*rv2$num))/(rv1$num+rv2$num);
+	else if ( rv1?$prev_avg )
+		result$prev_avg = rv1$prev_avg;
+	else if ( rv2?$prev_avg )
+		result$prev_avg = rv2$prev_avg;
+	}

scripts/base/frameworks/measurement/simple.bro

@@ -0,0 +1,6 @@
+
+module Metrics;
+
+export {
+	
+}