zeek-dev October 2017

zeek-dev@lists.zeek.org

9 discussions

Re: [Bro-Dev] [Bro-Commits] [git/bro] topic/actor-system: First-pass broker-enabled Cluster scripting API + misc. (07ad06b)

by Robin Sommer

This is coming together quite nicely. Not sure if it's stable yet, but I'll just go ahead with some feedback I noticed looking over the new cluster API: - One thing I can't quite tell is if this is still aiming to maintain compatibility with the old communication system, like by keeping the proxies and also the *_events patterns. Looking at setup-connections, it seems so. I'd say just go ahead and remove all legacy pieces. Maintain two schemes in parallel is cumbersome, and I think it's fine to just force everything over to Broker. - Is the idea for the "*_topic" constants that one just picks the apppropiate one when sending events? Like if I want to publish something to all workers, I'd publish to Cluster::worker_topic? I think that's fine, though I'm wondering if we could compress the API there somehow so that Cluster doesn't need to export all those constants indvidiually. One idea would be a function that returns a topic based on node type? - I like the Pools! How about moving Pool with its functions out of the main.bro, just for clarity. - Looks like the hello/bye events are broadcasted by all nodes. Is that on purpose, or should that be limited to just one, like just the master sending them out? Or does it not matter and this provides for more redundancy? - create_store() vs "stores": Is the idea that I'd normally use create_store() and that populates the table, but I could also redef it myself instead of using create_store() to create more custom entries? If so, maybe make that a bit more explicit in the comments that there're two ways to configure that table. Robin On Fri, Oct 27, 2017 at 12:44 -0500, Jonathan Siwek wrote: > Repository : ssh://git@bro-ids.icir.org/bro > On branch : topic/actor-system > Link : https://github.com/bro/bro/commit/07ad06b083d16f9cf1c86041cf7287335a74ebbb > > >--------------------------------------------------------------- > > commit 07ad06b083d16f9cf1c86041cf7287335a74ebbb > Author: Jon Siwek <jsiwek(a)illinois.edu> > Date: Fri Oct 27 12:44:54 2017 -0500 > > First-pass broker-enabled Cluster scripting API + misc. > > - Remove Broker::Options, Broker::configure(). This was only > half implemented (e.g. the "routable" flag wasn't used), and using > a function to set these options was awkward: the only way to > override defaults was via calling configure() in a bro_init with > *lower* priority such that the last call "wins". Also doesn't > really make sense for it to be a function since the underlying > broker library can't adapt to changes in these configuration > values dynamically at runtime, so instead there's just now > two options you can redef: "Broker::forward_messages" and > "Broker::log_topic". > > - Add Broker::node_id() to get a unique identifier for the Bro instance's > broker endpoint. This is used by the Cluster API to map node name > (e.g. "manager") to broker endpoint so that one can track which nodes > are still alive. > > - Fix how broker-based communication interacts with --pseudo-realtime > and reading pcaps: bro now terminates at end of reading pcap when > broker is active (this should now be equivalent to how RemoteSerializer > worked). > > - New broker-enabled Cluster framework API > - Still uses Cluster::nodes as the means of setting up cluster network > - See Cluster::stores, Cluster::StoreInfo, and Cluster::create_store > for how broker data stores are integrated into cluster operation > > - Update several unit tests to new Cluster API. Failing tests at > the moment are mostly from scripts/frameworks that haven't been > ported to the new Cluster API. > > > >--------------------------------------------------------------- > > 07ad06b083d16f9cf1c86041cf7287335a74ebbb > aux/broker | 2 +- > scripts/base/frameworks/broker/main.bro | 47 ++-- > scripts/base/frameworks/broker/store.bro | 14 +- > scripts/base/frameworks/cluster/__load__.bro | 7 - > scripts/base/frameworks/cluster/main.bro | 279 ++++++++++++++++++++- > .../base/frameworks/cluster/setup-connections.bro | 150 +++++++++++ > scripts/base/frameworks/control/main.bro | 1 + > src/broker/Manager.cc | 20 +- > src/broker/Manager.h | 15 +- > src/broker/comm.bif | 21 +- > src/iosource/PktSrc.cc | 20 +- > testing/btest/Baseline/plugins.hooks/output | 18 +- > .../manager-1..stdout | 4 + > .../manager-1.reporter.log | 11 +- > testing/btest/broker/remote_log_types.bro | 2 +- > .../base/frameworks/cluster/start-it-up-logger.bro | 47 +++- > .../base/frameworks/cluster/start-it-up.bro | 48 +++- > .../frameworks/control/configuration_update.bro | 2 +- > .../scripts/base/frameworks/control/id_value.bro | 2 +- > .../scripts/base/frameworks/control/shutdown.bro | 2 +- > .../logging/field-extension-cluster-error.bro | 7 +- > .../frameworks/logging/field-extension-cluster.bro | 7 +- > 22 files changed, 601 insertions(+), 125 deletions(-) > > diff --git a/aux/broker b/aux/broker > index 76375d0..e1d637c 160000 > --- a/aux/broker > +++ b/aux/broker > @@ -1 +1 @@ > -Subproject commit 76375d07f5bf1ffc9711e064644bf865eda7a828 > +Subproject commit e1d637c816955a451079b419f438307960109346 > diff --git a/scripts/base/frameworks/broker/main.bro b/scripts/base/frameworks/broker/main.bro > index 47533a4..3ac25ed 100644 > --- a/scripts/base/frameworks/broker/main.bro > +++ b/scripts/base/frameworks/broker/main.bro > @@ -20,6 +20,11 @@ export { > ## use already. > const default_listen_retry = 30sec &redef; > > + ## Default address on which to listen. > + ## > + ## .. bro:see:: Broker::listen > + const default_listen_address = "" &redef; > + > ## Default interval to retry connecting to a peer if it cannot be made to work > ## initially, or if it ever becomes disconnected. > const default_connect_retry = 30sec &redef; > @@ -55,14 +60,12 @@ export { > ## all peers. > const ssl_keyfile = "" &redef; > > - ## The available configuration options when enabling Broker. > - type Options: record { > - ## Whether this Broker instance relays messages not destined to itself. > - ## By default, routing is disabled. > - routable: bool &default = F; > - ## The topic prefix where to publish logs. > - log_topic: string &default = "bro/logs/"; > - }; > + ## Forward all received messages to subscribing peers. > + const forward_messages = F &redef; > + > + ## The topic prefix where logs will be published. The log's stream id > + ## is appended when writing to a particular stream. > + const log_topic = "bro/logs/" &redef; > > type ErrorCode: enum { > ## The unspecified default error code. > @@ -153,13 +156,6 @@ export { > val: Broker::Data; > }; > > - ## Configures the local endpoint. > - ## > - ## options: Configures the local Broker endpoint behavior. > - ## > - ## Returns: true if configuration was successfully performed.. > - global configure: function(options: Options &default = Options()): bool; > - > ## Listen for remote connections. > ## > ## a: an address string on which to accept connections, e.g. > @@ -174,7 +170,8 @@ export { > ## Returns: the bound port or 0/? on failure. > ## > ## .. bro:see:: Broker::status > - global listen: function(a: string &default = "", p: port &default=default_port, > + global listen: function(a: string &default = default_listen_address, > + p: port &default = default_port, > retry: interval &default = default_listen_retry): port; > ## Initiate a remote connection. > ## > @@ -213,6 +210,9 @@ export { > ## Returns: a list of all peer connections. > global peers: function(): vector of PeerInfo; > > + ## Returns: a unique identifier for the local broker endpoint. > + global node_id: function(): string; > + > ## Publishes an event at a given topic. > ## > ## topic: a topic associated with the event message. > @@ -278,16 +278,6 @@ export { > > module Broker; > > -event bro_init() &priority=-10 > - { > - configure(); # Configure with defaults. > - } > - > -function configure(options: Options &default = Options()): bool > - { > - return __configure(options); > - } > - > event retry_listen(a: string, p: port, retry: interval) > { > listen(a, p, retry); > @@ -318,6 +308,11 @@ function peers(): vector of PeerInfo > return __peers(); > } > > +function node_id(): string > + { > + return __node_id(); > + } > + > function publish(topic: string, ev: Event): bool > { > return __publish(topic, ev); > diff --git a/scripts/base/frameworks/broker/store.bro b/scripts/base/frameworks/broker/store.bro > index ed735e0..6b22f5d 100644 > --- a/scripts/base/frameworks/broker/store.bro > +++ b/scripts/base/frameworks/broker/store.bro > @@ -61,7 +61,7 @@ export { > options: BackendOptions &default = BackendOptions()): opaque of Broker::Store; > > ## Create a clone of a master data store which may live with a remote peer. > - ## A clone automatically synchronizes to the master by automatically > + ## A clone automatically synchronizes to the master by > ## receiving modifications and applying them locally. Direct modifications > ## are not possible, they must be sent through the master store, which then > ## automatically broadcasts the changes out to clones. But queries may be > @@ -70,18 +70,6 @@ export { > ## > ## name: the unique name which identifies the master data store. > ## > - ## b: the storage backend to use. > - ## > - ## options: tunes how some storage backends operate. > - ## > - ## resync: the interval at which to re-attempt synchronizing with the master > - ## store should the connection be lost. If the clone has not yet > - ## synchronized for the first time, updates and queries queue up > - ## until the synchronization completes. After, if the connection > - ## to the master store is lost, queries continue to use the clone's > - ## version, but updates will be lost until the master is once again > - ## available. > - ## > ## Returns: a handle to the data store. > global create_clone: function(name: string): opaque of Broker::Store; > > diff --git a/scripts/base/frameworks/cluster/__load__.bro b/scripts/base/frameworks/cluster/__load__.bro > index 1717b83..4f193c0 100644 > --- a/scripts/base/frameworks/cluster/__load__.bro > +++ b/scripts/base/frameworks/cluster/__load__.bro > @@ -19,13 +19,6 @@ redef peer_description = Cluster::node; > > @load ./setup-connections > > -# 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 > - > -## Set the port that this node is supposed to listen on. > -redef Communication::listen_port = Cluster::nodes[Cluster::node]$p; > - > @if ( Cluster::local_node_type() == Cluster::MANAGER ) > @load ./nodes/manager > # If no logger is defined, then the manager receives logs. > diff --git a/scripts/base/frameworks/cluster/main.bro b/scripts/base/frameworks/cluster/main.bro > index 261f3f1..c94ed28 100644 > --- a/scripts/base/frameworks/cluster/main.bro > +++ b/scripts/base/frameworks/cluster/main.bro > @@ -7,10 +7,96 @@ > ##! ``@load base/frameworks/cluster``. > > @load base/frameworks/control > +@load base/frameworks/broker > > module Cluster; > > export { > + ## Whether the cluster framework uses broker to perform remote communication. > + const use_broker = T &redef; > + > + ## The topic name used for exchanging general messages that are relevant to > + ## any node in a cluster. Used with broker-enabled cluster communication. > + const broadcast_topic = "bro/cluster/broadcast" &redef; > + > + ## The topic name used for exchanging messages that are relevant to > + ## logger nodes in a cluster. Used with broker-enabled cluster communication. > + const logger_topic = "bro/cluster/logger" &redef; > + > + ## The topic name used for exchanging messages that are relevant to > + ## manager nodes in a cluster. Used with broker-enabled cluster communication. > + const manager_topic = "bro/cluster/manager" &redef; > + > + ## The topic name used for exchanging messages that are relevant to > + ## proxy nodes in a cluster. Used with broker-enabled cluster communication. > + const proxy_topic = "bro/cluster/proxy" &redef; > + > + ## The topic name used for exchanging messages that are relevant to > + ## worker nodes in a cluster. Used with broker-enabled cluster communication. > + const worker_topic = "bro/cluster/worker" &redef; > + > + ## The topic name used for exchanging messages that are relevant to > + ## time machine nodes in a cluster. Used with broker-enabled cluster communication. > + const time_machine_topic = "bro/cluster/time_machine" &redef; > + > + ## The topic prefix used for exchanging messages that are relevant to > + ## a named node in a cluster. Used with broker-enabled cluster communication. > + const node_topic_prefix = "bro/cluster/node/" &redef; > + > + ## Name of the node on which master data stores will be created if no other > + ## has already been specified by the user in :bro:see:`Cluster::stores`. > + const default_master_node = "manager" &redef; > + > + ## The type of data store backend that will be used for all data stores if > + ## no other has already been specified by the user in :bro:see:`Cluster::stores`. > + const default_backend = Broker::MEMORY &redef; > + > + ## The type of persistent data store backend that will be used for all data > + ## stores if no other has already been specified by the user in > + ## :bro:see:`Cluster::stores`. This will be used when script authors call > + ## :bro:see:`Cluster::create_store` with the *persistent* argument set true. > + const default_persistent_backend = Broker::SQLITE &redef; > + > + ## Setting a default dir will, for persistent backends that have not > + ## been given an explicit file path via :bro:see:`Cluster::stores`, > + ## automatically create a path within this dir that is based on the name of > + ## the data store. > + const default_store_dir = "" &redef; > + > + ## Information regarding a cluster-enabled data store. > + type StoreInfo: record { > + ## The name of the data store. > + name: string &optional; > + ## The store handle. > + store: opaque of Broker::Store &optional; > + ## The name of the cluster node on which the master version of the data > + ## store resides. > + master_node: string &default=default_master_node; > + ## Whether the data store is the master version or a clone. > + master: bool &default=F; > + ## The type of backend used for storing data. > + backend: Broker::BackendType &default=default_backend; > + ## Parameters used for configuring the backend. > + options: Broker::BackendOptions &default=Broker::BackendOptions(); > + }; > + > + ## A table of cluster-enabled data stores that have been created, indexed > + ## by their name. To customize a particular data store, you may redef this, > + ## defining the :bro:see:`StoreInfo` to associate with the store's name. > + global stores: table[string] of StoreInfo &default=StoreInfo() &redef; > + > + ## Sets up a cluster-enabled data store. They will also still properly > + ## function for uses that are not operating a cluster. > + ## > + ## name: the name of the data store to create. > + ## > + ## persistent: whether the data store must be persistent. > + ## > + ## Returns: the store's information. For master stores, the store will be > + ## ready to use immediately. For clones, the store field will not > + ## be set until the node containing the master store has connected. > + global create_store: function(name: string, persistent: bool &default=F): StoreInfo; > + > ## The cluster logging stream identifier. > redef enum Log::ID += { LOG }; > > @@ -18,6 +104,8 @@ export { > type Info: record { > ## The time at which a cluster message was generated. > ts: time; > + ## The name of the node that is creating the log record. > + node: string; > ## A message indicating information about the cluster's operation. > message: string; > } &log; > @@ -92,8 +180,7 @@ export { > ## If the *ip* field is a non-global IPv6 address, this field > ## can specify a particular :rfc:`4007` ``zone_id``. > zone_id: string &default=""; > - ## The port to which this local node can connect when > - ## establishing communication. > + ## The port that this node will listen on for peer connections. > p: port; > ## Identifier for the interface a worker is sniffing. > interface: string &optional; > @@ -108,6 +195,8 @@ export { > workers: set[string] &optional; > ## Name of a time machine node with which this node connects. > time_machine: string &optional; > + ## A unique identifier assigned to the node by the broker framework. > + id: string &optional; > }; > > ## This function can be called at any time to determine if the cluster > @@ -134,6 +223,8 @@ export { > ## named cluster-layout.bro somewhere in the BROPATH. It will be > ## automatically loaded if the CLUSTER_NODE environment variable is set. > ## Note that BroControl handles all of this automatically. > + ## The table is typically indexed by node names/labels (e.g. "manager" > + ## or "worker-1"). > const nodes: table[string] of Node = {} &redef; > > ## Indicates whether or not the manager will act as the logger and receive > @@ -148,6 +239,15 @@ export { > > ## Interval for retrying failed connections between cluster nodes. > const retry_interval = 1min &redef; > + > + ## When using broker-enabled cluster framework, nodes use this event to > + ## exchange their user-defined name along with a string that uniquely > + ## identifies it for the duration of its lifetime (this string may change if > + ## the node dies and has to reconnect later). > + global hello: event(name: string, id: string); > + > + ## Write a message to the cluster logging stream. > + global log: function(msg: string); > } > > function is_enabled(): bool > @@ -163,13 +263,112 @@ function local_node_type(): NodeType > event remote_connection_handshake_done(p: event_peer) &priority=5 > { > if ( p$descr in nodes && nodes[p$descr]$node_type == WORKER ) > - ++worker_count; > + { > + if ( use_broker ) > + Reporter::error(fmt("broker-enabled cluster using old comms: '%s' ", node)); > + else > + ++worker_count; > + } > } > > event remote_connection_closed(p: event_peer) &priority=5 > { > if ( p$descr in nodes && nodes[p$descr]$node_type == WORKER ) > - --worker_count; > + { > + if ( use_broker ) > + Reporter::error(fmt("broker-enabled cluster using old comms: '%s' ", node)); > + else > + --worker_count; > + } > + } > + > +event Cluster::hello(name: string, id: string) &priority=10 > + { > + if ( name !in nodes ) > + { > + Reporter::error(fmt("Got Cluster::hello msg from unexpected node: %s", name)); > + return; > + } > + > + local n = nodes[name]; > + > + if ( n?$id && n$id != id ) > + Reporter::error(fmt("Got Cluster::hello msg from duplicate node: %s", name)); > + > + n$id = id; > + Cluster::log(fmt("got hello from %s (%s)", name, id)); > + > + if ( n$node_type == WORKER ) > + ++worker_count; > + > + for ( store_name in stores ) > + { > + local info = stores[store_name]; > + > + if ( info?$store ) > + next; > + > + if ( info$master ) > + next; > + > + if ( info$master_node == name ) > + { > + info$store = Broker::create_clone(info$name); > + Cluster::log(fmt("created clone store: %s", info$name)); > + } > + } > + } > + > +event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string) &priority=10 > + { > + if ( ! use_broker ) > + return; > + > + if ( ! Cluster::is_enabled() ) > + return; > + > + local e = Broker::make_event(Cluster::hello, node, Broker::node_id()); > + Broker::publish(Cluster::broadcast_topic, e); > + } > + > +event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string) &priority=10 > + { > + if ( ! use_broker ) > + return; > + > + for ( node_name in nodes ) > + { > + local n = nodes[node_name]; > + > + if ( n?$id && n$id == endpoint$id ) > + { > + Cluster::log(fmt("node down: %s", node_name)); > + delete n$id; > + > + if ( n$node_type == WORKER ) > + --worker_count; > + > + for ( store_name in stores ) > + { > + local info = stores[store_name]; > + > + if ( ! info?$store ) > + next; > + > + if ( info$master ) > + next; > + > + if ( info$master_node == node_name ) > + { > + Broker::close(info$store); > + delete info$store; > + Cluster::log(fmt("clone store closed: %s", info$name)); > + } > + } > + > + break; > + } > + } > } > > event bro_init() &priority=5 > @@ -183,3 +382,75 @@ event bro_init() &priority=5 > > Log::create_stream(Cluster::LOG, [$columns=Info, $path="cluster"]); > } > + > +function create_store(name: string, persistent: bool &default=F): Cluster::StoreInfo > + { > + local info = stores[name]; > + info$name = name; > + > + if ( Cluster::default_store_dir != "" ) > + { > + local default_options = Broker::BackendOptions(); > + local path = Cluster::default_store_dir + "/" + name; > + > + if ( info$options$sqlite$path == default_options$sqlite$path ) > + info$options$sqlite$path = path + ".sqlite"; > + > + if ( info$options$rocksdb$path == default_options$rocksdb$path ) > + info$options$rocksdb$path = path + ".rocksdb"; > + } > + > + if ( persistent ) > + { > + switch ( info$backend ) { > + case Broker::MEMORY: > + info$backend = Cluster::default_persistent_backend; > + break; > + case Broker::SQLITE: > + fallthrough; > + case Broker::ROCKSDB: > + # no-op: user already asked for a specific persistent backend. > + break; > + default: > + Reporter::error(fmt("unhandled data store type: %s", info$backend)); > + break; > + } > + } > + > + if ( ! Cluster::is_enabled() ) > + { > + if ( info?$store ) > + { > + Reporter::warning(fmt("duplicate cluster store creation for %s", name)); > + return info; > + } > + > + info$store = Broker::create_master(name, info$backend, info$options); > + info$master = T; > + stores[name] = info; > + Cluster::log(fmt("created master store: %s", name)); > + return info; > + } > + > + if ( info$master_node !in Cluster::nodes ) > + Reporter::fatal(fmt("master node '%s' for cluster store '%s' does not exist", > + info$master_node, name)); > + > + if ( Cluster::node == info$master_node ) > + { > + info$store = Broker::create_master(name, info$backend, info$options); > + info$master = T; > + stores[name] = info; > + return info; > + } > + > + info$master = F; > + stores[name] = info; > + Cluster::log(fmt("pending clone store creation: %s", name)); > + return info; > + } > + > +function log(msg: string) > + { > + Log::write(Cluster::LOG, [$ts = network_time(), $node = node, $message = msg]); > + } > diff --git a/scripts/base/frameworks/cluster/setup-connections.bro b/scripts/base/frameworks/cluster/setup-connections.bro > index 971a55d..2e775da 100644 > --- a/scripts/base/frameworks/cluster/setup-connections.bro > +++ b/scripts/base/frameworks/cluster/setup-connections.bro > @@ -3,13 +3,163 @@ > > @load ./main > @load base/frameworks/communication > +@load base/frameworks/broker > > @if ( Cluster::node in Cluster::nodes ) > > module Cluster; > > +type NamedNode: record { > + name: string; > + node: Node; > +}; > + > +function nodes_with_type(node_type: NodeType): vector of NamedNode > + { > + local rval: vector of NamedNode = vector(); > + > + for ( name in Cluster::nodes ) > + { > + local n = Cluster::nodes[name]; > + > + if ( n$node_type != node_type ) > + next; > + > + rval[|rval|] = NamedNode($name=name, $node=n); > + } > + > + return rval; > + } > + > +function connect_peer(node_type: NodeType, node_name: string): bool > + { > + local nn = nodes_with_type(node_type); > + > + for ( i in nn ) > + { > + local n = nn[i]; > + > + if ( n$name != node_name ) > + next; > + > + Cluster::log(fmt("initiate peering with %s:%s, retry=%s", > + n$node$ip, n$node$p, Cluster::retry_interval)); > + return Broker::peer(cat(n$node$ip), n$node$p, Cluster::retry_interval); > + } > + > + return F; > + } > + > event bro_init() &priority=9 > { > + if ( ! use_broker ) > + return; > + > + local self = nodes[node]; > + > + switch ( self$node_type ) { > + case NONE: > + return; > + case CONTROL: > + break; > + case LOGGER: > + Broker::subscribe(Cluster::logger_topic); > + Broker::subscribe(Broker::log_topic); > + break; > + case MANAGER: > + Broker::subscribe(Cluster::manager_topic); > + > + if ( Cluster::manager_is_logger ) > + Broker::subscribe(Broker::log_topic); > + > + break; > + case PROXY: > + Broker::subscribe(Cluster::proxy_topic); > + break; > + case WORKER: > + Broker::subscribe(Cluster::worker_topic); > + break; > + case TIME_MACHINE: > + Broker::subscribe(Cluster::time_machine_topic); > + break; > + default: > + Reporter::error(fmt("Unhandled cluster node type: %s", self$node_type)); > + return; > + } > + > + Broker::subscribe(Cluster::broadcast_topic); > + Broker::subscribe(Cluster::node_topic_prefix + node); > + > + Broker::listen(Broker::default_listen_address, > + self$p, > + Broker::default_listen_retry); > + > + Cluster::log(fmt("listening on %s:%s", Broker::default_listen_address, self$p)); > + > + switch ( self$node_type ) { > + case MANAGER: > + if ( self?$logger ) > + connect_peer(LOGGER, self$logger); > + > + if ( self?$time_machine ) > + connect_peer(TIME_MACHINE, self$time_machine); > + > + break; > + case PROXY: > + if ( self?$logger ) > + connect_peer(LOGGER, self$logger); > + > + if ( self?$manager ) > + connect_peer(MANAGER, self$manager); > + > + local proxies = nodes_with_type(PROXY); > + > + for ( i in proxies ) > + { > + local proxy = proxies[i]; > + > + if ( proxy$node?$proxy ) > + Broker::peer(cat(proxy$node$ip), proxy$node$p, Cluster::retry_interval); > + } > + > + break; > + case WORKER: > + if ( self?$logger ) > + connect_peer(LOGGER, self$logger); > + > + if ( self?$manager ) > + connect_peer(MANAGER, self$manager); > + > + if ( self?$proxy ) > + connect_peer(PROXY, self$proxy); > + > + if ( self?$time_machine ) > + connect_peer(TIME_MACHINE, self$time_machine); > + > + break; > + } > + } > + > +event bro_init() &priority=-10 > + { > + if ( use_broker ) > + return; > + > + local lp = Cluster::nodes[Cluster::node]$p; > + enable_communication(); > + listen(Communication::listen_interface, > + lp, > + Communication::listen_ssl, > + Communication::listen_ipv6, > + Communication::listen_ipv6_zone_id, > + Communication::listen_retry); > + } > + > +event bro_init() &priority=9 > + { > + if ( use_broker ) > + return; > + > local me = nodes[node]; > > for ( i in Cluster::nodes ) > diff --git a/scripts/base/frameworks/control/main.bro b/scripts/base/frameworks/control/main.bro > index 5c68c47..e3b58ef 100644 > --- a/scripts/base/frameworks/control/main.bro > +++ b/scripts/base/frameworks/control/main.bro > @@ -5,6 +5,7 @@ > module Control; > > export { > + ## Whether the control framework uses broker to perform remote communication. > const use_broker = T &redef; > > ## The address of the host that will be controlled. > diff --git a/src/broker/Manager.cc b/src/broker/Manager.cc > index e9d6280..b2cbe87 100644 > --- a/src/broker/Manager.cc > +++ b/src/broker/Manager.cc > @@ -113,7 +113,6 @@ Manager::BrokerState::BrokerState(broker::broker_options options) > > Manager::Manager() > { > - routable = false; > bound_port = 0; > > next_timestamp = 1; > @@ -128,6 +127,7 @@ void Manager::InitPostScript() > { > DBG_LOG(DBG_BROKER, "Initializing"); > > + log_topic = get_option("Broker::log_topic")->AsString()->CheckString(); > log_id_type = internal_type("Log::ID")->AsEnumType(); > writer_id_type = internal_type("Log::Writer")->AsEnumType(); > > @@ -144,6 +144,7 @@ void Manager::InitPostScript() > > broker::broker_options options; > options.disable_ssl = get_option("Broker::disable_ssl")->AsBool(); > + options.forward = get_option("Broker::forward_messages")->AsBool(); > > bstate = std::make_shared<BrokerState>(options); > } > @@ -176,18 +177,6 @@ bool Manager::Active() > return bound_port > 0 || bstate->endpoint.peers().size(); > } > > -bool Manager::Configure(bool arg_routable, std::string arg_log_topic) > - { > - DBG_LOG(DBG_BROKER, "Configuring endpoint: routable=%s log_topic=%s", > - (routable ? "yes" : "no"), arg_log_topic.c_str()); > - > - routable = arg_routable; > - log_topic = arg_log_topic; > - > - // TODO: process routable flag > - return true; > - } > - > uint16_t Manager::Listen(const string& addr, uint16_t port) > { > bound_port = bstate->endpoint.listen(addr, port); > @@ -233,6 +222,11 @@ std::vector<broker::peer_info> Manager::Peers() const > return bstate->endpoint.peers(); > } > > +std::string Manager::NodeID() const > + { > + return to_string(bstate->endpoint.node_id()); > + } > + > bool Manager::PublishEvent(string topic, std::string name, broker::vector args) > { > if ( ! bstate->endpoint.peers().size() ) > diff --git a/src/broker/Manager.h b/src/broker/Manager.h > index a430c57..5af23e4 100644 > --- a/src/broker/Manager.h > +++ b/src/broker/Manager.h > @@ -74,15 +74,6 @@ public: > bool Active(); > > /** > - * Configure the local Broker endpoint. > - * @param routable Whether the context of this endpoint routes messages not > - * @param log_topic The topic prefix for logs we this endpoint published. > - * destined to itself. By default endpoints do not route. > - * @return true if configuration was successful. > - */ > - bool Configure(bool routable = false, std::string log_topic=""); > - > - /** > * Listen for remote connections. > * @param port the TCP port to listen on. > * @param addr an address string on which to accept connections, e.g. > @@ -115,6 +106,11 @@ public: > std::vector<broker::peer_info> Peers() const; > > /** > + * @return a unique identifier for this broker endpoint. > + */ > + std::string NodeID() const; > + > + /** > * Send an event to any interested peers. > * @param topic a topic string associated with the message. > * Peers advertise interest by registering a subscription to some prefix > @@ -296,7 +292,6 @@ private: > broker::endpoint& Endpoint() > { assert(bstate); return bstate->endpoint; } > > - bool routable; > std::string log_topic; > uint16_t bound_port; > > diff --git a/src/broker/comm.bif b/src/broker/comm.bif > index 411e3d4..c7c94d4 100644 > --- a/src/broker/comm.bif > +++ b/src/broker/comm.bif > @@ -7,8 +7,6 @@ > > module Broker; > > -type Broker::Options: record; > - > ## Generated when something changes in the Broker sub-system. > event Broker::status%(endpoint: EndpointInfo, msg: string%); > > @@ -51,20 +49,6 @@ enum PeerStatus %{ > RECONNECTING, > %} > > -function Broker::__configure%(options: Broker::Options%): bool > - %{ > - auto routable = false; > - auto log_topic = ""; > - > - if ( auto routable_val = options->AsRecordVal()->Lookup(0) ) > - routable = routable_val->AsBool(); > - > - if ( auto log_topic_val = options->AsRecordVal()->Lookup(1) ) > - log_topic = log_topic_val->AsString()->CheckString(); > - > - return new Val(broker_mgr->Configure(routable, log_topic), TYPE_BOOL); > - %} > - > function Broker::__listen%(a: string, p: port%): port > %{ > if ( ! p->IsTCP() ) > @@ -140,3 +124,8 @@ function Broker::__peers%(%): PeerInfos > > return rval; > %} > + > +function Broker::__node_id%(%): string > + %{ > + return new StringVal(broker_mgr->NodeID()); > + %} > diff --git a/src/iosource/PktSrc.cc b/src/iosource/PktSrc.cc > index a9362a0..343801a 100644 > --- a/src/iosource/PktSrc.cc > +++ b/src/iosource/PktSrc.cc > @@ -10,6 +10,8 @@ > #include "Hash.h" > #include "Net.h" > #include "Sessions.h" > +#include "broker/Manager.h" > +#include "iosource/Manager.h" > > #include "pcap/pcap.bif.h" > > @@ -304,13 +306,19 @@ bool PktSrc::ExtractNextPacketInternal() > return 1; > } > > - if ( pseudo_realtime && using_communication && ! IsOpen() ) > + if ( pseudo_realtime && ! IsOpen() ) > { > - // Source has gone dry, we're done. > - if ( remote_trace_sync_interval ) > - remote_serializer->SendFinalSyncPoint(); > - else > - remote_serializer->Terminate(); > + if ( using_communication ) > + { > + // Source has gone dry, we're done. > + if ( remote_trace_sync_interval ) > + remote_serializer->SendFinalSyncPoint(); > + else > + remote_serializer->Terminate(); > + } > + > + if ( broker_mgr->Active() ) > + iosource_mgr->Terminate(); > } > > SetIdle(true); > diff --git a/testing/btest/Baseline/plugins.hooks/output b/testing/btest/Baseline/plugins.hooks/output > index f7c61c0..5eefc64 100644 > --- a/testing/btest/Baseline/plugins.hooks/output > +++ b/testing/btest/Baseline/plugins.hooks/output > @@ -148,8 +148,6 @@ > 0.000000 MetaHookPost CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_SYSLOG, {514/udp})) -> <no result> > 0.000000 MetaHookPost CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_TEREDO, {3544/udp})) -> <no result> > 0.000000 MetaHookPost CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_XMPP, {5222<...>/tcp})) -> <no result> > -0.000000 MetaHookPost CallFunction(Broker::__configure, <frame>, ([routable=F, log_topic=bro<...>/])) -> <no result> > -0.000000 MetaHookPost CallFunction(Broker::configure, <frame>, ([routable=F, log_topic=bro<...>/])) -> <no result> > 0.000000 MetaHookPost CallFunction(Cluster::is_enabled, <frame>, ()) -> <no result> > 0.000000 MetaHookPost CallFunction(Cluster::is_enabled, <null>, ()) -> <no result> > 0.000000 MetaHookPost CallFunction(Files::register_analyzer_add_callback, <frame>, (Files::ANALYZER_EXTRACT, FileExtract::on_add{ if (!FileExtract::args?$extract_filename) FileExtract::args$extract_filename = cat(extract-, FileExtract::f$last_active, -, FileExtract::f$source, -, FileExtract::f$id)FileExtract::f$info$extracted = FileExtract::args$extract_filenameFileExtract::args$extract_filename = build_path_compressed(FileExtract::prefix, FileExtract::args$extract_filename)FileExtract::f$info$extracted_cutoff = Fmkdir(FileExtract::prefix)})) -> <no result> > @@ -251,7 +249,7 @@ > 0.000000 MetaHookPost CallFunction(Log::__create_stream, <frame>, (Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird])) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::__create_stream, <frame>, (X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509])) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::__create_stream, <frame>, (mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql])) -> <no result> > -0.000000 MetaHookPost CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T])) -> <no result> > +0.000000 MetaHookPost CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T])) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::add_default_filter, <frame>, (Broker::LOG)) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::add_default_filter, <frame>, (Cluster::LOG)) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::add_default_filter, <frame>, (Communication::LOG)) -> <no result> > @@ -384,7 +382,7 @@ > 0.000000 MetaHookPost CallFunction(Log::create_stream, <frame>, (Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird])) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::create_stream, <frame>, (X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509])) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::create_stream, <frame>, (mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql])) -> <no result> > -0.000000 MetaHookPost CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T])) -> <no result> > +0.000000 MetaHookPost CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T])) -> <no result> > 0.000000 MetaHookPost CallFunction(NetControl::check_plugins, <frame>, ()) -> <no result> > 0.000000 MetaHookPost CallFunction(NetControl::init, <null>, ()) -> <no result> > 0.000000 MetaHookPost CallFunction(Notice::want_pp, <frame>, ()) -> <no result> > @@ -872,8 +870,6 @@ > 0.000000 MetaHookPre CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_SYSLOG, {514/udp})) > 0.000000 MetaHookPre CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_TEREDO, {3544/udp})) > 0.000000 MetaHookPre CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_XMPP, {5222<...>/tcp})) > -0.000000 MetaHookPre CallFunction(Broker::__configure, <frame>, ([routable=F, log_topic=bro<...>/])) > -0.000000 MetaHookPre CallFunction(Broker::configure, <frame>, ([routable=F, log_topic=bro<...>/])) > 0.000000 MetaHookPre CallFunction(Cluster::is_enabled, <frame>, ()) > 0.000000 MetaHookPre CallFunction(Cluster::is_enabled, <null>, ()) > 0.000000 MetaHookPre CallFunction(Files::register_analyzer_add_callback, <frame>, (Files::ANALYZER_EXTRACT, FileExtract::on_add{ if (!FileExtract::args?$extract_filename) FileExtract::args$extract_filename = cat(extract-, FileExtract::f$last_active, -, FileExtract::f$source, -, FileExtract::f$id)FileExtract::f$info$extracted = FileExtract::args$extract_filenameFileExtract::args$extract_filename = build_path_compressed(FileExtract::prefix, FileExtract::args$extract_filename)FileExtract::f$info$extracted_cutoff = Fmkdir(FileExtract::prefix)})) > @@ -975,7 +971,7 @@ > 0.000000 MetaHookPre CallFunction(Log::__create_stream, <frame>, (Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird])) > 0.000000 MetaHookPre CallFunction(Log::__create_stream, <frame>, (X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509])) > 0.000000 MetaHookPre CallFunction(Log::__create_stream, <frame>, (mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql])) > -0.000000 MetaHookPre CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T])) > +0.000000 MetaHookPre CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T])) > 0.000000 MetaHookPre CallFunction(Log::add_default_filter, <frame>, (Broker::LOG)) > 0.000000 MetaHookPre CallFunction(Log::add_default_filter, <frame>, (Cluster::LOG)) > 0.000000 MetaHookPre CallFunction(Log::add_default_filter, <frame>, (Communication::LOG)) > @@ -1108,7 +1104,7 @@ > 0.000000 MetaHookPre CallFunction(Log::create_stream, <frame>, (Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird])) > 0.000000 MetaHookPre CallFunction(Log::create_stream, <frame>, (X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509])) > 0.000000 MetaHookPre CallFunction(Log::create_stream, <frame>, (mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql])) > -0.000000 MetaHookPre CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T])) > +0.000000 MetaHookPre CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T])) > 0.000000 MetaHookPre CallFunction(NetControl::check_plugins, <frame>, ()) > 0.000000 MetaHookPre CallFunction(NetControl::init, <null>, ()) > 0.000000 MetaHookPre CallFunction(Notice::want_pp, <frame>, ()) > @@ -1596,8 +1592,6 @@ > 0.000000 | HookCallFunction Analyzer::register_for_ports(Analyzer::ANALYZER_SYSLOG, {514/udp}) > 0.000000 | HookCallFunction Analyzer::register_for_ports(Analyzer::ANALYZER_TEREDO, {3544/udp}) > 0.000000 | HookCallFunction Analyzer::register_for_ports(Analyzer::ANALYZER_XMPP, {5222<...>/tcp}) > -0.000000 | HookCallFunction Broker::__configure([routable=F, log_topic=bro<...>/]) > -0.000000 | HookCallFunction Broker::configure([routable=F, log_topic=bro<...>/]) > 0.000000 | HookCallFunction Cluster::is_enabled() > 0.000000 | HookCallFunction Files::register_analyzer_add_callback(Files::ANALYZER_EXTRACT, FileExtract::on_add{ if (!FileExtract::args?$extract_filename) FileExtract::args$extract_filename = cat(extract-, FileExtract::f$last_active, -, FileExtract::f$source, -, FileExtract::f$id)FileExtract::f$info$extracted = FileExtract::args$extract_filenameFileExtract::args$extract_filename = build_path_compressed(FileExtract::prefix, FileExtract::args$extract_filename)FileExtract::f$info$extracted_cutoff = Fmkdir(FileExtract::prefix)}) > 0.000000 | HookCallFunction Files::register_for_mime_type(Files::ANALYZER_PE, application/x-dosexec) > @@ -1698,7 +1692,7 @@ > 0.000000 | HookCallFunction Log::__create_stream(Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird]) > 0.000000 | HookCallFunction Log::__create_stream(X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509]) > 0.000000 | HookCallFunction Log::__create_stream(mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql]) > -0.000000 | HookCallFunction Log::__write(PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T]) > +0.000000 | HookCallFunction Log::__write(PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T]) > 0.000000 | HookCallFunction Log::add_default_filter(Broker::LOG) > 0.000000 | HookCallFunction Log::add_default_filter(Cluster::LOG) > 0.000000 | HookCallFunction Log::add_default_filter(Communication::LOG) > @@ -1831,7 +1825,7 @@ > 0.000000 | HookCallFunction Log::create_stream(Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird]) > 0.000000 | HookCallFunction Log::create_stream(X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509]) > 0.000000 | HookCallFunction Log::create_stream(mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql]) > -0.000000 | HookCallFunction Log::write(PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T]) > +0.000000 | HookCallFunction Log::write(PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T]) > 0.000000 | HookCallFunction NetControl::check_plugins() > 0.000000 | HookCallFunction NetControl::init() > 0.000000 | HookCallFunction Notice::want_pp() > diff --git a/testing/btest/Baseline/scripts.base.frameworks.cluster.start-it-up/manager-1..stdout b/testing/btest/Baseline/scripts.base.frameworks.cluster.start-it-up/manager-1..stdout > index 7c8eb5e..5b10602 100644 > --- a/testing/btest/Baseline/scripts.base.frameworks.cluster.start-it-up/manager-1..stdout > +++ b/testing/btest/Baseline/scripts.base.frameworks.cluster.start-it-up/manager-1..stdout > @@ -1,4 +1,8 @@ > Connected to a peer > Connected to a peer > Connected to a peer > +Got fully_connected event > +Got fully_connected event > Connected to a peer > +Got fully_connected event > +Got fully_connected event > diff --git a/testing/btest/Baseline/scripts.base.frameworks.logging.field-extension-cluster-error/manager-1.reporter.log b/testing/btest/Baseline/scripts.base.frameworks.logging.field-extension-cluster-error/manager-1.reporter.log > index b7d8c11..e7972c2 100644 > --- a/testing/btest/Baseline/scripts.base.frameworks.logging.field-extension-cluster-error/manager-1.reporter.log > +++ b/testing/btest/Baseline/scripts.base.frameworks.logging.field-extension-cluster-error/manager-1.reporter.log > @@ -3,11 +3,10 @@ > #empty_field (empty) > #unset_field - > #path reporter > -#open 2016-09-22-23-31-34 > +#open 2017-10-26-19-18-59 > #fields _write_ts _stream _system_name ts level message location > #types time string string time enum string string > -1474587094.261799 reporter manager-1 0.000000 Reporter::WARNING WriterFrontend communication/Log::WRITER_ASCII expected 11 fields in write, got 8. Skipping line. (empty) > -1474587094.261799 reporter manager-1 0.000000 Reporter::WARNING WriterFrontend communication/Log::WRITER_ASCII expected 11 fields in write, got 8. Skipping line. (empty) > -1474587094.261799 reporter manager-1 0.000000 Reporter::WARNING WriterFrontend communication/Log::WRITER_ASCII expected 11 fields in write, got 8. Skipping line. (empty) > -1474587099.984660 reporter manager-1 0.000000 Reporter::INFO received termination signal (empty) > -#close 2016-09-22-23-31-40 > +1509045539.693078 reporter manager-1 0.000000 Reporter::WARNING Write using filter 'default' on path 'broker' changed to use new path 'broker-2' to avoid conflict with filter '' (empty) > +1509045539.699623 reporter manager-1 0.000000 Reporter::WARNING WriterFrontend cluster/Log::WRITER_ASCII expected 6 fields in write, got 3. Skipping line. (empty) > +1509045547.196521 reporter manager-1 0.000000 Reporter::INFO received termination signal (empty) > +#close 2017-10-26-19-19-07 > diff --git a/testing/btest/broker/remote_log_types.bro b/testing/btest/broker/remote_log_types.bro > index aeaf1b9..56d63eb 100644 > --- a/testing/btest/broker/remote_log_types.bro > +++ b/testing/btest/broker/remote_log_types.bro > @@ -1,4 +1,4 @@ > - @TEST-SERIALIZE: brokercomm > +# @TEST-SERIALIZE: brokercomm > > # @TEST-EXEC: btest-bg-run recv "bro -b ../recv.bro >recv.out" > # @TEST-EXEC: btest-bg-run send "bro -b ../send.bro >send.out" > diff --git a/testing/btest/scripts/base/frameworks/cluster/start-it-up-logger.bro b/testing/btest/scripts/base/frameworks/cluster/start-it-up-logger.bro > index 97f3698..72251c0 100644 > --- a/testing/btest/scripts/base/frameworks/cluster/start-it-up-logger.bro > +++ b/testing/btest/scripts/base/frameworks/cluster/start-it-up-logger.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run logger-1 CLUSTER_NODE=logger-1 BROPATH=$BROPATH:.. bro %INPUT > # @TEST-EXEC: sleep 1 > @@ -38,10 +38,16 @@ global fully_connected_nodes = 0; > event fully_connected() > { > ++fully_connected_nodes; > + > if ( Cluster::node == "logger-1" ) > { > if ( peer_count == 5 && fully_connected_nodes == 5 ) > - terminate_communication(); > + { > + if ( Cluster::use_broker ) > + terminate(); > + else > + terminate_communication(); > + } > } > } > > @@ -49,6 +55,43 @@ redef Cluster::worker2logger_events += /fully_connected/; > redef Cluster::proxy2logger_events += /fully_connected/; > redef Cluster::manager2logger_events += /fully_connected/; > > +event bro_init() > + { > + Broker::auto_publish(Cluster::logger_topic, fully_connected); > + } > + > +event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string) > + { > + print "Connected to a peer"; > + ++peer_count; > + > + if ( Cluster::node == "logger-1" ) > + { > + if ( peer_count == 5 && fully_connected_nodes == 5 ) > + { > + if ( Cluster::use_broker ) > + terminate(); > + else > + terminate_communication(); > + } > + } > + else if ( Cluster::node == "manager-1" ) > + { > + if ( peer_count == 5 ) > + event fully_connected(); > + } > + else > + { > + if ( peer_count == 3 ) > + event fully_connected(); > + } > + } > + > +event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string) > + { > + terminate(); > + } > + > event remote_connection_handshake_done(p: event_peer) > { > print "Connected to a peer"; > diff --git a/testing/btest/scripts/base/frameworks/cluster/start-it-up.bro b/testing/btest/scripts/base/frameworks/cluster/start-it-up.bro > index acb9c36..b0fcc69 100644 > --- a/testing/btest/scripts/base/frameworks/cluster/start-it-up.bro > +++ b/testing/btest/scripts/base/frameworks/cluster/start-it-up.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run manager-1 BROPATH=$BROPATH:.. CLUSTER_NODE=manager-1 bro %INPUT > # @TEST-EXEC: sleep 1 > @@ -8,7 +8,7 @@ > # @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 30 > -# @TEST-EXEC: btest-diff manager-1/.stdout > +# @TEST-EXEC: TEST_DIFF_CANONIFIER=$SCRIPTS/diff-sort btest-diff manager-1/.stdout > # @TEST-EXEC: btest-diff proxy-1/.stdout > # @TEST-EXEC: btest-diff proxy-2/.stdout > # @TEST-EXEC: btest-diff worker-1/.stdout > @@ -32,17 +32,59 @@ global fully_connected_nodes = 0; > > event fully_connected() > { > + if ( ! is_remote_event() ) > + return; > + > + print "Got fully_connected event"; > fully_connected_nodes = fully_connected_nodes + 1; > + > if ( Cluster::node == "manager-1" ) > { > if ( peer_count == 4 && fully_connected_nodes == 4 ) > - terminate_communication(); > + { > + if ( Cluster::use_broker ) > + terminate(); > + else > + terminate_communication(); > + } > } > } > > redef Cluster::worker2manager_events += /fully_connected/; > redef Cluster::proxy2manager_events += /fully_connected/; > > +event bro_init() > + { > + Broker::auto_publish(Cluster::manager_topic, fully_connected); > + } > + > +event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string) > + { > + print "Connected to a peer"; > + peer_count = peer_count + 1; > + > + if ( Cluster::node == "manager-1" ) > + { > + if ( peer_count == 4 && fully_connected_nodes == 4 ) > + { > + if ( Cluster::use_broker ) > + terminate(); > + else > + terminate_communication(); > + } > + } > + else > + { > + if ( peer_count == 2 ) > + event fully_connected(); > + } > + } > + > +event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string) > + { > + terminate(); > + } > + > event remote_connection_handshake_done(p: event_peer) > { > print "Connected to a peer"; > diff --git a/testing/btest/scripts/base/frameworks/control/configuration_update.bro b/testing/btest/scripts/base/frameworks/control/configuration_update.bro > index 535a357..8d8fb0a 100644 > --- a/testing/btest/scripts/base/frameworks/control/configuration_update.bro > +++ b/testing/btest/scripts/base/frameworks/control/configuration_update.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run controllee BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controllee Communication::listen_port=65531/tcp Broker::default_port=65531/tcp > # @TEST-EXEC: sleep 5 > diff --git a/testing/btest/scripts/base/frameworks/control/id_value.bro b/testing/btest/scripts/base/frameworks/control/id_value.bro > index bee37a2..1b5e354 100644 > --- a/testing/btest/scripts/base/frameworks/control/id_value.bro > +++ b/testing/btest/scripts/base/frameworks/control/id_value.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run controllee BROPATH=$BROPATH:.. bro %INPUT only-for-controllee frameworks/control/controllee Communication::listen_port=65532/tcp Broker::default_port=65532/tcp > # @TEST-EXEC: btest-bg-run controller BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controller Control::host=127.0.0.1 Control::host_port=65532/tcp Control::cmd=id_value Control::arg=test_var > diff --git a/testing/btest/scripts/base/frameworks/control/shutdown.bro b/testing/btest/scripts/base/frameworks/control/shutdown.bro > index 9c0c104..2869c3f 100644 > --- a/testing/btest/scripts/base/frameworks/control/shutdown.bro > +++ b/testing/btest/scripts/base/frameworks/control/shutdown.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run controllee BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controllee Communication::listen_port=65530/tcp Broker::default_port=65530/tcp > # @TEST-EXEC: btest-bg-run controller BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controller Control::host=127.0.0.1 Control::host_port=65530/tcp Control::cmd=shutdown > diff --git a/testing/btest/scripts/base/frameworks/logging/field-extension-cluster-error.bro b/testing/btest/scripts/base/frameworks/logging/field-extension-cluster-error.bro > index 6ac7a5e..c0ab10b 100644 > --- a/testing/btest/scripts/base/frameworks/logging/field-extension-cluster-error.bro > +++ b/testing/btest/scripts/base/frameworks/logging/field-extension-cluster-error.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run manager-1 "cp ../cluster-layout.bro . && CLUSTER_NODE=manager-1 bro %INPUT" > # @TEST-EXEC: sleep 1 > @@ -43,6 +43,11 @@ event terminate_me() { > terminate(); > } > > +event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string) > + { > + schedule 1sec { terminate_me() }; > + } > + > event remote_connection_closed(p: event_peer) { > schedule 1sec { terminate_me() }; > } > diff --git a/testing/btest/scripts/base/frameworks/logging/field-extension-cluster.bro b/testing/btest/scripts/base/frameworks/logging/field-extension-cluster.bro > index fb51251..6740743 100644 > --- a/testing/btest/scripts/base/frameworks/logging/field-extension-cluster.bro > +++ b/testing/btest/scripts/base/frameworks/logging/field-extension-cluster.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run manager-1 "cp ../cluster-layout.bro . && CLUSTER_NODE=manager-1 bro %INPUT" > # @TEST-EXEC: sleep 1 > @@ -39,6 +39,11 @@ event terminate_me() { > terminate(); > } > > +event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string) > + { > + schedule 1sec { terminate_me() }; > + } > + > event remote_connection_closed(p: event_peer) { > schedule 1sec { terminate_me() }; > } > > > > _______________________________________________ > bro-commits mailing list > bro-commits(a)bro.org > http://mailman.icsi.berkeley.edu/mailman/listinfo/bro-commits > -- Robin Sommer * ICSI/LBNL * robin(a)icir.org * www.icir.org/robin

3 years, 7 months

[Bro-Dev] File Analysis Inconsistencies

by Aaron Eppert

I crafted a custom file analysis plugin that attaches to specific MIME types via file_sniff and fires an appropriate event once processing has been completed. I had to jump through a few hoops to make a file analysis plugin, first, but those were cleared and everything runs and loads appropriately there (bro -NN verified.) My test regime is very straight forward, I have several PCAPs cooked up that contain simple HTTP file GETs (that extract otherwise properly and do not exhibit missing_bytes) and I am running them via `bro -C -r <>.pcap`. My issue comes with utter and complete inconsistency with execution - it is, effectively, a coin flip, with zero changes. When I have dumped the buffers being processed, as my file analysis plugin has a secondary verification to make sure the data passed is appropriate - which is confusing, as the mime type fires correct, which seems to indicate a bug somewhere in the data path - the correct execution, clearly has the proper data in it. The invalid executions, again changing nothing other than a subsequent execution, shows a buffer of what appears to be completely random data. I currently cannot supply the file analysis plugin for inspection, but would very much appreciate insight in how to find the root cause. It very much seems to be upstream. If I run the analysis portion of the plugin as a free standing executable outside of Bro against the data transferred via HTTP, everything works perfect and the structures are filled accordingly. I saw BIT-1832, and there could be similar root causes in there, but I have not had time to investigate otherwise. The issues I am raising, again, are command line replay via command line, not even “live” network traffic or tcpreplay over a NIC/dummy interface. Aaron

3 years, 7 months

[Bro-Dev] BinPac - Many repeated messages in the same packet

by Aaron Eppert

I am running into an implementation issue with BinPac and would hope to find a few pointers. I have a protocol that loads a given TCP packet with as many publish messages as possible in a worst case scenario - often it just has a single message, but it is not guaranteed. When a publish message contains more than one subsequent message, there is not an indicator that another message follows. The packet looks, generally like this: +-------------------------------------+ | Message 0 | +-------------------------------------+ | Message 1 | +-------------------------------------+ | Message 2 | +-------------------------------------+ | ... | +-------------------------------------+ | Message N-2 | +-------------------------------------+ | Message N-1 | +-------------------------------------+ | Message N | +-------------------------------------+ The protocol definition code I have written as follows: type SPROTO_messages = record { thdr : uint8; hdrlen : uint8; variable_header : case msg_type of { SPROTO_CONNECT -> connect_packet : SPROTO_connect(hdrlen); SPROTO_SUBSCRIBE -> subscribe_packet : SPROTO_subscribe(hdrlen); SPROTO_SUBACK -> suback_packet : SPROTO_suback(hdrlen); SPROTO_PUBLISH -> publish_packet : SPROTO_publish(hdrlen); SPROTO_UNSUBSCRIBE -> unsubscribe_packet : SPROTO_unsubscribe(hdrlen); default -> none : empty; }; } &let { msg_type : uint8 = (thdr >> 4); }; type SPROTO_PDU(is_orig: bool) = record { sproto_messages : SPROTO_messages[]; } &byteorder=bigendian; — I can tell via Wireshark that I am definitely missing messages. Any advice on a better way to implement the above would be greatly appreciated. Aaron

3 years, 7 months

[Bro-Dev] design summary: porting Bro scripts to use Broker

by Siwek, Jon

I want to check if there’s any feedback on the approach I’m planning to take when porting over Bro’s scripts to use Broker. There’s two major areas to consider: (1) how users specify network topology e.g. either for traditional cluster configuration or manually connecting Bro instances and (2) replacing &synchronized with Broker’s distributed data storage features. Broker-Based Topology ===================== It’s again useful to decompose topology specification it into two main use-cases: Creating Clusters, e.g w/ BroControl ------------------------------------ This use-case should look familiar once ported to use Broker: the existing “cluster” framework will be used for specifying the topology of the cluster and for automatically setting up the connections between nodes. The one thing that will differ is the event subscription mechanism, which needs to change since Broker itself handles that differently, but I think the general idea can remain similar. The current mechanism for handling event subscription/publication: const Cluster::manager2worker_events = /Drop::.*/ &redef; # similar patterns follow for all node combinations... And a script author that is making their script usable on clusters writes: redef Cluster::manager2worker_events += /^Intel::(cluster_new_item|purge_item)$/; The new mechanism: # contains topic prefixes const Cluster::manager_subscriptions: set[string] &redef; # contains (topic string, event name) pairs const Cluster::manager_publications: set[string, string] &redef; # similar sets follow for all node types… And a script author writes: # topic naming convention relates to file hierarchy/organization of scripts redef Cluster::manager_subscriptions += { "bro/event/framework/control", "bro/event/framework/intel", }; # not sure how to get around referencing events via strings: can't use 'any' # to stick event values directly into the set, maybe that’s ok since we can # at least detect lookup failures at bro_init time and emit errors/abort. redef Cluster::manager_publications += { ["bro/event/framework/control/configuration_update_request", "Control::configuration_update_request"], ["bro/event/framework/intel/cluster_new_item", "Intel::cluster_new_item"], }; Then subscriptions and auto-publications still get automatically set up by the cluster framework in bro_init(). Other Manual/Custom Topologies ------------------------------ I don’t see anything to do here as the Broker API already has enough to set up peerings and subscriptions in arbitrary ways. The old “communication” framework scripts can just go away as most of its functions have direct corollaries in the new “broker” framework. The one thing that is missing is the “Communication::nodes” table which acts as both a state-tracking structure and an API that users may use to have the comm. framework automatically set up connections between the nodes in the table. I find this redundant — there’s two APIs to accomplish the same thing, with the table being an additional layer of indirection to the actual connect/listen functions a user can just as easily use themselves. I also think it’s not useful for state-tracking as a user operating at the level of this use-case is can easily track nodes themselves or has some other notion of the state structures they need to track that is more intuitive for the particular problem they're solving. Unless there’s arguments or I find it’s actually needed, I don’t plan to port this to Broker. Broker-Based Data Distribution ============================== Replacing &synchronized requires completely new APIs that script authors can easily use to work for both cluster and non-cluster use-cases and independently of a user’s choice of persistent storage backend. Broker Framework API -------------------- const Broker::default_master_node = "manager" &redef; const Broker::default_backend = MEMORY &redef; # Setting a default dir will, for persistent backends that have not # been given an explicit file path, automatically create a path within this # dir that is based on the name of the data store. const Broker::default_store_dir = "" &redef; type Broker::StoreInfo: record { name: string &optional; store: opaque of Broker::Store &optional; master_node: string &default=Broker::default_master_node; master: bool &default=F; backend: Broker::BackendType &default=default_backend; options: Broker::BackendOptions &default=Broker::BackendOptions(); }; # Primarily used by users to set up master store location and backend # configuration, but also possible to lookup an existing/open store by name. global Broker::stores: table[string] of StoreInfo &default=StoreInfo() &redef; # Set up data stores to properly function regardless of whether user is # operating a cluster. This also automatically sets up the store to # be a clone or a master as is appropriate for the the local node type. # It does this by inspecting the state of the “Broker::stores” table, # which a user configures in advance via redef. # (I have pseudo-code written, let me know if you want to see it all). global Broker::InitStore: function(name: string): opaque of Broker::Store; Script-Author Example Usage --------------------------- # Script author that wants to utilize data stores doesn't have to be aware of # whether user is running a cluster or if they want to use persistent storage # backends. const Software::tracked_store_name = "bro/framework/software/tracked" &redef; global Software::tracked_store: opaque of Broker::Store; event bro_init() &priority = +10 { Software::tracked_store = Broker::InitStore(Software::tracked_store_name); } Bro-User Example Usage ---------------------- # User needs to be able to choose data store backends and which cluster node the # the master store lives on. They can either do this manually, or BroControl # will autogenerate the following in cluster-layout.bro: # Explicitly configure an individual store. redef Broker::stores += { ["bro/framework/software/tracked"] = [$master_node = "some_node", $backend=Broker::SQLITE, $options=Broker::BackendOptions( $sqlite=Broker::SQLiteOptions( $path="/home/jon/tracked_software.sqlite"))]; }; # Or set new default configurations for stores. redef Broker::default_master_node = "manager"; redef Broker::default_backend = Broker::MEMORY; redef Broker::default_store_dir = "/home/jon/stores"; # Then Broker::InitStore() will end up creating the right type of store. BroControl Example Usage ------------------------ BroControl users will have a new “datastore.cfg" file they may customize: # The default file will contain a just a basic [default] section # and would set up all data stores on the manager node, using the default # backend (in-memory). If a user wants to globally change to persistent # storage and also give a canonical storage node, they can do that here. [default] master = manager backend = MEMORY # When using persistent backends as default, need to specify a directory to # store databases in. Files will be auto-named based on the store's name. dir = /home/jon/stores # If a user has special needs regarding persistence/residence, they can # further customize individual stores: [bro/framework/software/tracked] master = some_node backend = SQLITE path = /home/jon/tracked_software.sqlite

3 years, 7 months

[Bro-Dev] Performance Enhancements

by Jim Mellander

Hi all: One item of particular interest to me from Brocon was this tidbit from Packetsled's lightning talk: "Optimizing core loops (like net_run() ) with preprocessor branch prediction macros likely() and unlikely() for ~3% speedup. We optimize for maximum load." After conversing with Leo Linsky of Packetsled, I wanted to initiate a conversation about easy performance improvements that may be within fairly easy reach: 1. Obviously, branch prediction, as mentioned above. 3% speedup for (almost) free is nothing to sneeze at. 2. Profiling bro to identify other hot spots that could benefit from optimization. 3. Best practices for compiling Bro (compiler options, etc.) 4. Data structure revisit (hash functions, perhaps?) etc. Perhaps the Bro core team is working on some, all, or a lot more in this area. It might be nice to get the Bro community involved too. Is anyone else interested? Jim Mellander ESNet

3 years, 8 months

Re: [Bro-Dev] Performance Enhancements

by Jim Mellander

Well, I found pathological behavior with BaseList::remove Instrumenting it with a printf of num_entries & i after the for loop, running against a test pcap then summarizing with awk gives: Count, num_entries, i 1 3583 3536 1 3584 3537 1 3623 3542 1 3624 3543 1 3628 3620 1 3629 3621 1 3636 3562 1 3636 3625 1 3637 3563 1 3637 3626 1 3644 3576 1 3644 3641 1 3645 3577 1 3645 3642 1 3647 3641 1 3648 3642 1 3650 3629 1 3651 3630 1 3658 3647 1 3659 3648 1 3663 3655 1 3664 3656 1 3673 3629 1 3674 3630 1 3697 3686 1 3698 3687 1 3981 3595 1 3982 3596 1 4372 3978 1 4373 3979 1 4374 3656 1 4374 4371 1 4375 3657 1 4375 4372 1 4554 4371 1 4555 4372 1 4571 4367 1 4571 4551 1 4572 4368 1 4572 4552 1 4968 4566 1 4969 4567 1 5058 4566 1 5059 4567 1 5160 4963 1 5161 4964 1 5258 5157 1 5259 5158 1 5342 4566 1 5343 4567 1 5353 5253 1 5354 5254 1 5356 3638 1 5356 5337 1 5356 5350 1 5356 5351 1 5356 5353 1 5357 3639 1 5357 5338 1 5357 5351 1 5357 5352 1 5357 5354 1 5367 5351 1 5368 5352 1 5369 4556 1 5370 4557 1 5374 3675 1 5374 5366 1 5375 3676 1 5375 5367 1 5379 3664 1 5379 5045 1 5380 3665 1 5380 5046 1 5384 3601 1 5385 3602 1 5386 5354 1 5387 5355 1 5392 5370 1 5393 5371 1 5404 5363 1 5404 5381 1 5405 5364 1 5405 5382 1 5408 5341 1 5408 5368 1 5408 5399 1 5409 5342 1 5409 5369 1 5409 5400 1 5413 5401 1 5413 5403 1 5414 5402 1 5414 5404 1 5416 5408 1 5417 5409 1 5429 5395 1 5430 5396 1 5439 5381 1 5439 5406 1 5440 5382 1 5440 5407 1 5460 5436 1 5461 5437 1 5463 5407 1 5464 5408 1 5465 5397 1 5465 5460 1 5466 5398 1 5466 5461 1 5474 5359 1 5474 5451 1 5474 5456 1 5474 5471 1 5475 5360 1 5475 5452 1 5475 5457 1 5475 5472 1 5479 5456 1 5479 5476 1 5480 5457 1 5480 5477 1 5481 5416 1 5482 5417 1 5493 5426 1 5493 5474 1 5493 5488 1 5494 5427 1 5494 5475 1 5494 5489 1 5497 5357 1 5497 5367 1 5497 5461 1 5497 5462 1 5497 5480 1 5497 5488 1 5498 5358 1 5498 5368 1 5498 5462 1 5498 5463 1 5498 5481 1 5498 5489 1 5499 3682 1 5499 5460 1 5499 5476 1 5499 5478 1 5499 5480 1 5500 3683 1 5500 5461 1 5500 5477 1 5500 5479 1 5500 5481 2 3612 3609 2 3613 3610 2 3689 3686 2 3690 3687 2 3697 3694 2 3698 3695 2 5374 5371 2 5375 5372 2 5384 5381 2 5385 5382 2 5463 5460 2 5464 5461 2 5493 5465 2 5494 5466 2 5497 5484 2 5498 5485 2 5499 5482 2 5499 5488 2 5499 5490 2 5499 5492 2 5499 5494 2 5500 5483 2 5500 5489 2 5500 5491 2 5500 5493 2 5500 5495 3 4571 4568 3 4572 4569 3 5493 5490 3 5494 5491 3 5497 5490 3 5497 5492 3 5498 5491 3 5498 5493 3 5499 5486 3 5500 5487 4 3647 3644 4 3648 3645 5 5379 5376 5 5380 5377 7 5499 5496 7 5500 5497 10 5497 5494 10 5498 5495 26 3 2 5861 3 1 13714 4 4 14130 4 1 34914 4 0 74299 3 3 1518194 2 1 2648755 2 2 8166358 3 0 13019625 2 0 62953139 0 0 71512938 1 1 104294506 1 0 there are 286 instances where the list has over 3000 entries, and the desired entry is near the end... That linear search has got to be killing performance, even though its uncommon :-( The case of num_entries being 0 can be optimized a bit, but is relatively minor. Now, I'll see if I can identify the offending List Jim On Mon, Oct 9, 2017 at 1:03 PM, Clark, Gilbert <gc355804(a)ohio.edu> wrote: > If you look in one of the subdirectories or another, in ages past there > was a little shell script to incrementally execute bro against a specific > trace, loading one script at a time to see the effect each of them had on > the overall runtime. I can't remember what it was called offhand, but it > was useful for quick and dirty testing. > > > And yes, function calls in bro script land are pretty horrific in terms of > overhead. Line per line, bro script in general isn't terribly efficient > just because it doesn't do any of the things a modern interpreter might > (e.g. just-in-time compilation). That's not a criticism, it's only a note > - most folks rely on horizontal scaling to deal with faster ingress, which > I think makes a whole lot of sense. > > > Just in the event it's useful, I've attached some profiling I did on the > script function overhead with the crap I wrote: these are some graphs on > which script functions call which other script functions, how many times > that happens, and how much time is spent in each function. > > > avggraph is the average time spent per-call, and resgraph is the aggregate > time spent in each function across the entire run. The numbers' formatting > needed some fixing, but never made it that far ... > > > I know Robin et. al. were working on different approaches for > next-generation scripting kind of stuff, but haven't kept up well enough to > really know where those are. > > > One thing I played around with was using plugin hooks to integrate other > scripting languages into the bro fast path (luajit was my weapon of choice) > and seeing if conversion from bro script to one of those other languages > might improve the run time. Other languages would still be less efficient > than C, and anything garbage collected would need to be *really* carefully > used, but ... it struck me as an idea that might be worth a look :) > > > And yeah, generally speaking, most of the toolkits I've played with for > software-based packet processing absolutely do use memory pools for the > fast path. They also use burst fetch tricks (to amortize the cost of > fetching packets over X packets, rather than fetching one packet at a > time), and I've also seen quite a bit of prefetch / SIMD to try to keep > things moving quickly once the packets make it to the CPU. > > > Things start to get pretty crazy as packet rates increase, though: once > you hit about 10 - 15 Gbps, even a *cache miss* on a modern system is > enough to force a drop ... > > > For what it's worth ... > > > -Gilbert > > > ------------------------------ > *From:* Azoff, Justin S <jazoff(a)illinois.edu> > *Sent:* Monday, October 9, 2017 10:19:26 AM > *To:* Jim Mellander > *Cc:* Clark, Gilbert; bro-dev(a)bro.org > *Subject:* Re: [Bro-Dev] Performance Enhancements > > > > On Oct 6, 2017, at 5:59 PM, Jim Mellander <jmellander(a)lbl.gov> wrote: > > > > I particularly like the idea of an allocation pool that per-packet > information can be stored, and reused by the next packet. > > Turns out bro does this most of the time.. unless you use the next_packet > event. Normal connections use the sessions cache which holds connection > objects, but new_packet has its own code path that creates the ip header > from scratch for each packet. I tried to pre-allocate PortVal objects, but > I think I was screwing something up with 'Ref' and bro would just segfault > on the 2nd connection. > > > > There also are probably some optimizations of frequent operations now > that we're in a 64-bit world that could prove useful - the one's complement > checksum calculation in net_util.cc is one that comes to mind, especially > since it works effectively a byte at a time (and works with even byte > counts only). Seeing as this is done per-packet on all tcp payload, > optimizing this seems reasonable. Here's a discussion of do the checksum > calc in 64-bit arithmetic: https://locklessinc.com/articles/tcp_checksum/ > - this website also has an x64 allocator that is claimed to be faster than > tcmalloc, see: https://locklessinc.com/benchmarks_allocator.shtml (note: > I haven't tried anything from this source, but find it interesting). > The TCP/IP Checksum - Lockless Inc > <https://locklessinc.com/articles/tcp_checksum/> > locklessinc.com > The above obvious algorithm handles 16-bits at a time. Usually, if you > process more data at a time, then performance is better. Since we have a > 64-bit machine, we ... > > Benchmarks of the Lockless Memory Allocator > <https://locklessinc.com/benchmarks_allocator.shtml> > locklessinc.com > The speed of various memory allocators is compared. > > > > I couldn't get this code to return the right checksums inside bro (some > casting issue?), but if it is faster it should increase performance by a > small percentage. Comparing 'bro -b' runs on a pcap with 'bro -b -C' runs > (which should show what kind of performance increase we would get if that > function took 0s to run) shows a decent chunk of time taken computing > checksums. > > > I'm guessing there are a number of such "small" optimizations that could > provide significant performance gains. > > I've been trying to figure out the best way to profile bro. So far > attempting to use linux perf, or google perftools hasn't been able to shed > much light on anything. I think the approach I was using to benchmark > certain operations in the bro language is the better approach. > > Instead of running bro and trying to profile it to figure out what is > causing the most load, simply compare the execution of two bro runs with > slightly different scripts/settings. I think this will end up being the > better approach because it answers real questions like "If I load this > script or change this setting what is the performance impact on the bro > process". When I did this last I used this method to compare the > performance from one bro commit to the next, but I never tried comparing > bro with one set of scripts loaded to bro with a different set of scripts > loaded. > > For example, the simplest and most dramatic test I came up with so far: > > $ time bro -r 2009-M57-day11-18.trace -b > real 0m2.434s > user 0m2.236s > sys 0m0.200s > > $ cat np.bro > event new_packet(c: connection, p: pkt_hdr) > { > > } > > $ time bro -r 2009-M57-day11-18.trace -b np.bro > real 0m10.588s > user 0m10.392s > sys 0m0.204s > > We've been saying for a while that adding that event is expensive, but I > don't know if it's even been quantified. > > The main thing I still need to figure out is how to do this type of test > in a cluster environment while replaying a long pcap. > > > > Somewhat related, came across this presentation yesterday: > > https://www.youtube.com/watch?v=NH1Tta7purM&feature=youtu.be > <https://www.youtube.com/watch?v=NH1Tta7purM&feature=youtu.be> > CppCon 2017: Carl Cook “When a Microsecond Is an Eternity: High > Performance Trading Systems in C++” > <https://www.youtube.com/watch?v=NH1Tta7purM&feature=youtu.be> > www.youtube.com > http://CppCon.org — Presentation Slides, PDFs, Source Code and other > presenter materials are available at: https://github.com/CppCon/CppCon2017 > — Automated t... > > > > CppCon 2017: Carl Cook “When a Microsecond Is an Eternity: High > Performance Trading Systems in C++” > > Among other things, he mentions using a memory pool for objects instead of > creating/deleting them. > > > > — > Justin Azoff > >

3 years, 8 months

Re: [Bro-Dev] Performance Enhancements

by Azoff, Justin S

> On Oct 9, 2017, at 4:03 PM, Clark, Gilbert <gc355804(a)ohio.edu> wrote: > > If you look in one of the subdirectories or another, in ages past there was a little shell script to incrementally execute bro against a specific trace, loading one script at a time to see the effect each of them had on the overall runtime. I can't remember what it was called offhand, but it was useful for quick and dirty testing. > > And yes, function calls in bro script land are pretty horrific in terms of overhead. Line per line, bro script in general isn't terribly efficient just because it doesn't do any of the things a modern interpreter might (e.g. just-in-time compilation). That's not a criticism, it's only a note - most folks rely on horizontal scaling to deal with faster ingress, which I think makes a whole lot of sense. >From what my little microbenchmarks have been showing me, bro scripts are mostly fast, but there are some operations that may be slower than they should be.. like for loops or set/table lookups on small or empty tables. > Just in the event it's useful, I've attached some profiling I did on the script function overhead with the crap I wrote: these are some graphs on which script functions call which other script functions, how many times that happens, and how much time is spent in each function. > > avggraph is the average time spent per-call, and resgraph is the aggregate time spent in each function across the entire run. The numbers' formatting needed some fixing, but never made it that far ... Something like this from an hours worth of bro runtime would be neat. One potential issue is that if a function is called a lot, it could mean it's either something that needs to be optimized so it is faster, or it could mean it's something that needs to be refactored so it's not called so much. > I know Robin et. al. were working on different approaches for next-generation scripting kind of stuff, but haven't kept up well enough to really know where those are. http://www.icir.org/hilti/ I believe. > One thing I played around with was using plugin hooks to integrate other scripting languages into the bro fast path (luajit was my weapon of choice) and seeing if conversion from bro script to one of those other languages might improve the run time. Other languages would still be less efficient than C, and anything garbage collected would need to be *really* carefully used, but ... it struck me as an idea that might be worth a look :) I'm not sure how hard it would be to write a transpiler for bro scripts and convert them completely to something like lua. Other than maybe ip address and subnets as data types, I think they overlap fairly well. > And yeah, generally speaking, most of the toolkits I've played with for software-based packet processing absolutely do use memory pools for the fast path. They also use burst fetch tricks (to amortize the cost of fetching packets over X packets, rather than fetching one packet at a time), and I've also seen quite a bit of prefetch / SIMD to try to keep things moving quickly once the packets make it to the CPU. > > Things start to get pretty crazy as packet rates increase, though: once you hit about 10 - 15 Gbps, even a *cache miss* on a modern system is enough to force a drop ... Data rate is just one part of it.. the number of packets per second and new sessions per second has a huge impact as well. Handling 10Gbps of a few concurrent connections is easy, but 10Gbps of DNS will not work so well. DNS is one analyzer/script that I think could benefit from being ported to C++. The script doesn't do much and there aren't many scripts out there that want the lower level dns events.. Having the analyzer merge the request/reply directly and bypass the scripts entirely could boost performance by quite a bit. — Justin Azoff

3 years, 8 months

[Bro-Dev] Bro working well on Mac OS High Sierra, just a couple test failures

by Slagell, Adam J

I had no problems after the upgrade to High Sierra on my “production” box, and I had no troubles compiling Bro 2.5.1 on my laptop. I did, however, get a two errors in the test suite. core.truncation ... failed % 'btest-diff output' failed unexpectedly (exit code 1) % cat .diag == File =============================== #separator \x09 #set_separator , #empty_field (empty) #unset_field - #path weird #open 2017-10-04-18-48-40 #fields ts uid id.orig_h id.orig_p id.resp_h id.resp_p name addl notice peer #types time string addr port addr port string string bool string 1334160095.895421 - - - - - truncated_IP bro #close 2017-10-04-18-48-40 #separator \x09 #set_separator , #empty_field (empty) #unset_field - #path weird #open 2017-10-04-18-48-41 #fields ts uid id.orig_h id.orig_p id.resp_h id.resp_p name addl notice peer #types time string addr port addr port string string bool string 1334156241.519125 - - - - - truncated_IP bro #close 2017-10-04-18-48-41 #separator \x09 #set_separator , #empty_field (empty) #unset_field - #path weird #open 2017-10-04-18-48-41 #fields ts uid id.orig_h id.orig_p id.resp_h id.resp_p name addl notice peer #types time string addr port addr port string string bool string 1334094648.590126 - - - - - truncated_IP bro #close 2017-10-04-18-48-41 #separator \x09 #set_separator , #empty_field (empty) #unset_field - #path weird #open 2017-10-04-18-48-43 #fields ts uid id.orig_h id.orig_p id.resp_h id.resp_p name addl notice peer #types time string addr port addr port string string bool string 1338328954.078361 - - - - - internally_truncated_header - F bro #close 2017-10-04-18-48-43 #separator \x09 #set_separator , #empty_field (empty) #unset_field - #path weird #open 2017-10-04-18-48-43 #fields ts uid id.orig_h id.orig_p id.resp_h id.resp_p name addl notice peer #types time string addr port addr port string string bool string 1404148886.981015 - - - - - bad_IP_checksumbro 1404148887.011158 CHhAvVGS1DHFjwGM9 192.168.4.149 51293 72.21.91.29 443 bad_TCP_checksum - F bro #close 2017-10-04-18-48-43 == Diff =============================== --- /tmp/test-diff.62112.output.baseline.tmp 2017-10-04 18:48:43.000000000 +0000 +++ /tmp/test-diff.62112.output.tmp 2017-10-04 18:48:43.000000000 +0000 @@ -46,5 +46,6 @@ #open XXXX-XX-XX-XX-XX-XX #fields ts uid id.orig_h id.orig_p id.resp_h id.resp_p name addl notice peer #types time string addr port addr port string string bool string -0.000000 - - - - - truncated_link_header bro +XXXXXXXXXX.XXXXXX - - - - - bad_IP_checksumbro +XXXXXXXXXX.XXXXXX CHhAvVGS1DHFjwGM9 192.168.4.149 51293 72.21.91.29 443 bad_TCP_checksum - F bro #close XXXX-XX-XX-XX-XX-XX ======================================= % cat .stderr 1404148887.011158 warning in /Users/slagell/Downloads/bro-2.5.1/scripts/base/misc/find-checksum-offloading.bro, line 54: Your trace file likely has invalid IP and TCP checksums, most likely from NIC checksum offloading. By default, packets with invalid checksums are discarded by Bro unless using the -C command-line option or toggling the 'ignore_checksums' variable. Alternatively, disable checksum offloading by the network adapter to ensure Bro analyzes the actual checksums that are transmitted. 1404148887.011158 warning in /Users/slagell/Downloads/bro-2.5.1/scripts/base/misc/find-filtered-trace.bro, line 48: The analyzed trace file was determined to contain only TCP control packets, which may indicate it's been pre-filtered. By default, Bro reports the missing segments for this type of trace, but the 'detect_filtered_trace' option may be toggled if that's not desired. istate.bro-ipv6-socket ... failed % 'btest-bg-wait 20' failed unexpectedly (exit code 1) % cat .stderr The following processes did not terminate: bro -b ../recv.bro bro -b ../send.bro ----------- <<< [72978] bro -b ../recv.bro received termination signal >>> <<< [72998] bro -b ../send.bro received termination signal >>> ------ Adam J. Slagell Director, Cybersecurity & Networking Division Chief Information Security Officer National Center for Supercomputing Applications University of Illinois at Urbana-Champaign www.slagell.info "Under the Illinois Freedom of Information Act (FOIA), any written communication to or from University employees regarding University business is a public record and may be subject to public disclosure."

3 years, 8 months

[Bro-Dev] Renaming the Bro Project: seeking proposed names from the community

by Dopheide, Jeannette M

This year at BroCon we announced that the Bro Project will be changing its name. While “Bro” was originally meant as an Orwellian reminder of the risk that any monitoring fundamentally entails, it has more recently gained a very different, and quite offensive, reputation (“Bro culture”). To avoid facing instant negative impressions with new users that aren’t aware of the history, the Leadership Team has decided to seek a name change. We are accepting proposed names from the community for two months (due Monday December 4th). The Leadership Team will review the list of possible names and narrow it down to 5 finalists. We will announce the finalists and take a second round of feedback from the community before making the final selection. We hope to announce the new name within the next major release. To submit a proposed name, fill out the form here: https://goo.gl/forms/qwR8s6Yd4H0Bu8Ca2 ------ Jeannette Dopheide Sr. Education, Outreach, and Training Coordinator National Center for Supercomputing Applications University of Illinois at Urbana-Champaign

3 years, 8 months

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

zeek-dev October 2017