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.
This commit is contained in:
Seth Hall 2011-08-15 15:57:48 -04:00
parent 2af9d9bc20
commit 3919a35b9b
18 changed files with 407 additions and 117 deletions

View file

@ -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

View file

@ -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 )

View file

@ -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,

View file

@ -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

View file

@ -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

View file

@ -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();
@ -96,62 +106,44 @@ 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;
}
}

View file

@ -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) };
}

View file

@ -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

View file

@ -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;
}

View file

@ -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);
}

View file

@ -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)]);
}

View file

@ -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);
}

View file

@ -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

View file

@ -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

View file

@ -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 - -

View file

@ -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

View file

@ -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);
}

View file

@ -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);
}