main.zeek

Broker

The Broker-based communication API and its various options.

Namespace: Broker
Imports: base/bif/comm.bif.zeek, base/bif/messaging.bif.zeek

Summary

Runtime Options

`Broker::metrics_export_endpoint_name`: `string` `&redef`	ID for the metrics exporter.
`Broker::metrics_export_interval`: `interval` `&redef`	Frequency for publishing scraped metrics to the target topic.
`Broker::metrics_export_prefixes`: `vector` `&redef`	Selects prefixes from the local metrics.
`Broker::metrics_export_topic`: `string` `&redef`	Target topic for the metrics.
`Broker::metrics_import_topics`: `vector` `&redef`	Topics for the Prometheus exporter for collecting metrics from other peers in the network and including them in the output.
`Broker::peer_counts_as_iosource`: `bool` `&redef`	Whether calling `Broker::peer` will register the Broker system as an I/O source that will block the process from shutting down.

Redefinable Options

`Broker::aggressive_interval`: `count` `&redef`	Frequency of work-stealing polling attempts for Broker/CAF threads in “aggressive” mode.
`Broker::aggressive_polls`: `count` `&redef`	Number of work-stealing polling attempts for Broker/CAF threads in “aggressive” mode.
`Broker::congestion_queue_size`: `count` `&redef`	The number of buffered messages at the Broker/CAF layer after which a subscriber considers themselves congested (i.e.
`Broker::default_connect_retry`: `interval` `&redef`	Default interval to retry connecting to a peer if it cannot be made to work initially, or if it ever becomes disconnected.
`Broker::default_listen_address`: `string` `&redef`	Default address on which to listen.
`Broker::default_listen_address_websocket`: `string` `&redef`	Default address on which to listen for WebSocket connections.
`Broker::default_listen_retry`: `interval` `&redef`	Default interval to retry listening on a port if it’s currently in use already.
`Broker::default_log_topic_prefix`: `string` `&redef`	The default topic prefix where logs will be published.
`Broker::default_port`: `port` `&redef`	Default port for native Broker communication.
`Broker::default_port_websocket`: `port` `&redef`	Default port for Broker WebSocket communication.
`Broker::disable_ssl`: `bool` `&redef`	If true, do not use SSL for network connections.
`Broker::forward_messages`: `bool` `&redef`	Forward all received messages to subscribing peers.
`Broker::log_batch_interval`: `interval` `&redef`	Max time to buffer log messages before sending the current set out as a batch.
`Broker::log_batch_size`: `count` `&redef`	The max number of log entries per log stream to batch together when sending log messages to a remote logger.
`Broker::max_threads`: `count` `&redef`	Max number of threads to use for Broker/CAF functionality.
`Broker::metrics_port`: `port` `&redef`	Port for Broker’s metric exporter.
`Broker::moderate_interval`: `count` `&redef`	Frequency of work-stealing polling attempts for Broker/CAF threads in “moderate” mode.
`Broker::moderate_polls`: `count` `&redef`	Number of work-stealing polling attempts for Broker/CAF threads in “moderate” mode.
`Broker::moderate_sleep`: `interval` `&redef`	Interval of time for under-utilized Broker/CAF threads to sleep when in “moderate” mode.
`Broker::relaxed_interval`: `count` `&redef`	Frequency of work-stealing polling attempts for Broker/CAF threads in “relaxed” mode.
`Broker::relaxed_sleep`: `interval` `&redef`	Interval of time for under-utilized Broker/CAF threads to sleep when in “relaxed” mode.
`Broker::scheduler_policy`: `string` `&redef`	The CAF scheduling policy to use.
`Broker::ssl_cafile`: `string` `&redef`	Path to a file containing concatenated trusted certificates in PEM format.
`Broker::ssl_capath`: `string` `&redef`	Path to an OpenSSL-style directory of trusted certificates.
`Broker::ssl_certificate`: `string` `&redef`	Path to a file containing a X.509 certificate for this node in PEM format.
`Broker::ssl_keyfile`: `string` `&redef`	Path to the file containing the private key for this node’s certificate.
`Broker::ssl_passphrase`: `string` `&redef`	Passphrase to decrypt the private key specified by `Broker::ssl_keyfile`.

Types

`Broker::Data`: `record`	Opaque communication data.
`Broker::DataVector`: `vector`	Opaque communication data sequence.
`Broker::EndpointInfo`: `record`
`Broker::ErrorCode`: `enum`	Enumerates the possible error types.
`Broker::Event`: `record`	Opaque event communication data.
`Broker::NetworkInfo`: `record`
`Broker::PeerInfo`: `record`
`Broker::PeerInfos`: `vector`
`Broker::PeerStatus`: `enum`	The possible states of a peer endpoint.
`Broker::TableItem`: `record`	Opaque communication data used as a convenient way to wrap key-value pairs that comprise table entries.

Functions

`Broker::auto_publish`: `function`	Automatically send an event to any interested peers whenever it is locally dispatched.
`Broker::auto_unpublish`: `function`	Stop automatically sending an event to peers upon local dispatch.
`Broker::default_log_topic`: `function`	The default implementation for `Broker::log_topic`.
`Broker::flush_logs`: `function`	Sends all pending log messages to remote peers.
`Broker::forward`: `function`	Register a topic prefix subscription for events that should only be forwarded to any subscribing peers and not raise any event handlers on the receiving/forwarding node.
`Broker::listen`: `function`	Listen for remote connections using the native Broker protocol.
`Broker::listen_websocket`: `function`	Listen for remote connections using WebSocket.
`Broker::log_topic`: `function` `&redef`	A function that will be called for each log entry to determine what broker topic string will be used for sending it to peers.
`Broker::node_id`: `function`	Get a unique identifier for the local broker endpoint.
`Broker::peer`: `function`	Initiate a remote connection.
`Broker::peers`: `function`	Get a list of all peer connections.
`Broker::publish_id`: `function`	Publishes the value of an identifier to a given topic.
`Broker::subscribe`: `function`	Register interest in all peer event messages that use a certain topic prefix.
`Broker::unpeer`: `function`	Remove a remote connection.
`Broker::unsubscribe`: `function`	Unregister interest in all peer event messages that use a topic prefix.

Detailed Interface

Runtime Options

Broker::metrics_export_endpoint_name 

Type: string
Attributes: &redef
Default: ""

ID for the metrics exporter. When setting a target topic for the exporter, Broker sets this option to the suffix of the new topic unless the ID is a non-empty string. Since setting a topic starts the periodic publishing of events, we recommend setting the ID always first or avoid setting it at all if the topic suffix serves as a good-enough ID. Zeek overrides any value provided in zeek_init or earlier at startup if the environment variable BROKER_METRICS_ENDPOINT_NAME is defined.

Broker::metrics_export_interval 

Type: interval
Attributes: &redef
Default: 1.0 sec

Frequency for publishing scraped metrics to the target topic. Zeek overrides any value provided in zeek_init or earlier at startup if the environment variable BROKER_METRICS_EXPORT_INTERVAL is defined.

Broker::metrics_export_prefixes 

Type

vector of string

Attributes

&redef

Default

[]

Selects prefixes from the local metrics. Only metrics with prefixes listed in this variable are included when publishing local metrics. Setting an empty vector selects all metrics.

Broker::metrics_export_topic 

Type: string
Attributes: &redef
Default: ""

Target topic for the metrics. Setting a non-empty string starts the periodic publishing of local metrics. Zeek overrides any value provided in zeek_init or earlier at startup if the environment variable BROKER_METRICS_EXPORT_TOPIC is defined.

Broker::metrics_import_topics 

Type

vector of string

Attributes

&redef

Default

[]

Topics for the Prometheus exporter for collecting metrics from other peers in the network and including them in the output. Has no effect when not exporting the metrics to Prometheus.

Zeek overrides any value provided in zeek_init or earlier at startup if the environment variable BROKER_METRICS_IMPORT_TOPICS is defined.

Broker::peer_counts_as_iosource 

Type: bool
Attributes: &redef
Default: T

Whether calling Broker::peer will register the Broker system as an I/O source that will block the process from shutting down. For example, set this to false when you are reading pcaps, but also want to initiate a Broker peering and still shutdown after done reading the pcap.

Redefinable Options

Broker::aggressive_interval 

Type: count
Attributes: &redef
Default: 4

Frequency of work-stealing polling attempts for Broker/CAF threads in “aggressive” mode. Only used for the “stealing” scheduler policy.

Broker::aggressive_polls 

Type: count
Attributes: &redef
Default: 5

Number of work-stealing polling attempts for Broker/CAF threads in “aggressive” mode. Only used for the “stealing” scheduler policy.

Broker::congestion_queue_size 

Type: count
Attributes: &redef
Default: 200

The number of buffered messages at the Broker/CAF layer after which a subscriber considers themselves congested (i.e. tune the congestion control mechanisms).

Broker::default_connect_retry 

Type: interval
Attributes: &redef
Default: 30.0 secs

Default interval to retry connecting to a peer if it cannot be made to work initially, or if it ever becomes disconnected. Use of the ZEEK_DEFAULT_CONNECT_RETRY environment variable (set as number of seconds) will override this option and also any values given to Broker::peer.

Broker::default_listen_address 

Type

string

Attributes

&redef

Default

""

Redefinition

from policy/frameworks/management/agent/boot.zeek

=:

127.0.0.1

Default address on which to listen.

See also: Broker::listen_websocket

Broker::default_listen_retry 

Type: interval
Attributes: &redef
Default: 30.0 secs

Default interval to retry listening on a port if it’s currently in use already. Use of the ZEEK_DEFAULT_LISTEN_RETRY environment variable (set as a number of seconds) will override this option and also any values given to Broker::listen.

Broker::default_log_topic_prefix 

Type: string
Attributes: &redef
Default: "zeek/logs/"

The default topic prefix where logs will be published. The log’s stream id is appended when writing to a particular stream.

Broker::default_port 

Type: port
Attributes: &redef
Default: 9999/tcp

Default port for native Broker communication. Where not specified otherwise, this is the port to connect to and listen on.

Broker::default_port_websocket 

Type: port
Attributes: &redef
Default: 9997/tcp

Default port for Broker WebSocket communication. Where not specified otherwise, this is the port to connect to and listen on for WebSocket connections.

See the Broker documentation for a specification of the message format over WebSocket connections.

Broker::disable_ssl 

Type: bool
Attributes: &redef
Default: F

If true, do not use SSL for network connections. By default, SSL will even be used if no certificates / CAs have been configured. In that case (which is the default) the communication will be encrypted, but not authenticated.

Broker::forward_messages 

Type: bool
Attributes: &redef
Default: F

Forward all received messages to subscribing peers.

Broker::log_batch_interval 

Type: interval
Attributes: &redef
Default: 1.0 sec

Max time to buffer log messages before sending the current set out as a batch.

Broker::log_batch_size 

Type: count
Attributes: &redef
Default: 400

The max number of log entries per log stream to batch together when sending log messages to a remote logger.

Broker::max_threads 

Type: count
Attributes: &redef
Default: 1

Max number of threads to use for Broker/CAF functionality. The ZEEK_BROKER_MAX_THREADS environment variable overrides this setting.

Broker::metrics_port 

Type: port
Attributes: &redef
Default: 0/unknown

Port for Broker’s metric exporter. Setting this to a valid TCP port causes Broker to make metrics available to Prometheus scrapers via HTTP. Zeek overrides any value provided in zeek_init or earlier at startup if the environment variable BROKER_METRICS_PORT is defined.

Broker::moderate_interval 

Type: count
Attributes: &redef
Default: 2

Frequency of work-stealing polling attempts for Broker/CAF threads in “moderate” mode. Only used for the “stealing” scheduler policy.

Broker::moderate_polls 

Type: count
Attributes: &redef
Default: 5

Number of work-stealing polling attempts for Broker/CAF threads in “moderate” mode. Only used for the “stealing” scheduler policy.

Broker::moderate_sleep 

Type: interval
Attributes: &redef
Default: 16.0 msecs

Interval of time for under-utilized Broker/CAF threads to sleep when in “moderate” mode. Only used for the “stealing” scheduler policy.

Broker::relaxed_interval 

Type: count
Attributes: &redef
Default: 1

Frequency of work-stealing polling attempts for Broker/CAF threads in “relaxed” mode. Only used for the “stealing” scheduler policy.

Broker::relaxed_sleep 

Type: interval
Attributes: &redef
Default: 64.0 msecs

Interval of time for under-utilized Broker/CAF threads to sleep when in “relaxed” mode. Only used for the “stealing” scheduler policy.

Broker::scheduler_policy 

Type: string
Attributes: &redef
Default: "sharing"

The CAF scheduling policy to use. Available options are “sharing” and “stealing”. The “sharing” policy uses a single, global work queue along with mutex and condition variable used for accessing it, which may be better for cases that don’t require much concurrency or need lower power consumption. The “stealing” policy uses multiple work queues protected by spinlocks, which may be better for use-cases that have more concurrency needs. E.g. may be worth testing the “stealing” policy along with dedicating more threads if a lot of data store processing is required.

Broker::ssl_cafile 

Type: string
Attributes: &redef
Default: ""

Path to a file containing concatenated trusted certificates in PEM format. If set, Zeek will require valid certificates for all peers.

Broker::ssl_capath 

Type: string
Attributes: &redef
Default: ""

Path to an OpenSSL-style directory of trusted certificates. If set, Zeek will require valid certificates for all peers.

Broker::ssl_certificate 

Type: string
Attributes: &redef
Default: ""

Path to a file containing a X.509 certificate for this node in PEM format. If set, Zeek will require valid certificates for all peers.

Broker::ssl_keyfile 

Type: string
Attributes: &redef
Default: ""

Path to the file containing the private key for this node’s certificate. If set, Zeek will require valid certificates for all peers.

Broker::ssl_passphrase 

Type: string
Attributes: &redef
Default: ""

Passphrase to decrypt the private key specified by Broker::ssl_keyfile. If set, Zeek will require valid certificates for all peers.

Types

Broker::Data 

Type

record

data: opaque of Broker::Data &optional

Opaque communication data.

Broker::DataVector 

Type: vector of Broker::Data

Opaque communication data sequence.

Broker::EndpointInfo 

Type

record

id: string: A unique identifier of the node.
network: Broker::NetworkInfo &optional: Network-level information.

Broker::ErrorCode 

Type

enum

Broker::NO_ERROR: (present if base/bif/comm.bif.zeek is loaded)

Broker::UNSPECIFIED: The unspecified default error code.

Broker::PEER_INCOMPATIBLE: Version incompatibility.

Broker::PEER_INVALID: Referenced peer does not exist.

Broker::PEER_UNAVAILABLE: Remote peer not listening.

Broker::PEER_DISCONNECT_DURING_HANDSHAKE: (present if base/bif/comm.bif.zeek is loaded)

Broker::PEER_TIMEOUT: A peering request timed out.

Broker::MASTER_EXISTS: Master with given name already exists.

Broker::NO_SUCH_MASTER: Master with given name does not exist.

Broker::NO_SUCH_KEY: The given data store key does not exist.

Broker::REQUEST_TIMEOUT: The store operation timed out.

Broker::TYPE_CLASH: The operation expected a different type than provided.

Broker::INVALID_DATA: The data value cannot be used to carry out the desired operation.

Broker::BACKEND_FAILURE: The storage backend failed to execute the operation.

Broker::STALE_DATA: The storage backend failed to execute the operation.

Broker::CANNOT_OPEN_FILE: (present if base/bif/comm.bif.zeek is loaded)

Broker::CANNOT_WRITE_FILE: (present if base/bif/comm.bif.zeek is loaded)

Broker::INVALID_TOPIC_KEY: (present if base/bif/comm.bif.zeek is loaded)

Broker::END_OF_FILE: (present if base/bif/comm.bif.zeek is loaded)

Broker::INVALID_TAG: (present if base/bif/comm.bif.zeek is loaded)

Broker::INVALID_STATUS: (present if base/bif/comm.bif.zeek is loaded)

Broker::CAF_ERROR: Catch-all for a CAF-level problem.

Enumerates the possible error types.

Broker::Event 

Type

record

name: string &optional: The name of the event. Not set if invalid event or arguments.
args: Broker::DataVector: The arguments to the event.

Opaque event communication data.

Broker::NetworkInfo 

Type

record

address: string &log: The IP address or hostname where the endpoint listens.
bound_port: port &log: The port where the endpoint is bound to.

Broker::PeerInfo 

Type

record

peer: Broker::EndpointInfo

status: Broker::PeerStatus

Broker::PeerInfos 

Type: vector of Broker::PeerInfo

Broker::PeerStatus 

Type

enum

Broker::INITIALIZING: The peering process is initiated.

Broker::CONNECTING: Connection establishment in process.

Broker::CONNECTED: Connection established, peering pending.

Broker::PEERED: Successfully peered.

Broker::DISCONNECTED: Connection to remote peer lost.

Broker::RECONNECTING: Reconnecting to peer after a lost connection.

The possible states of a peer endpoint.

Broker::TableItem 

Type

record

key: Broker::Data

val: Broker::Data

Opaque communication data used as a convenient way to wrap key-value pairs that comprise table entries.

Functions

Broker::auto_publish 

Type: function (topic: string, ev: any) : bool

Automatically send an event to any interested peers whenever it is locally dispatched. (For example, using “event my_event(…);” in a script.)

Parameters

topic – a topic string associated with the event message. Peers advertise interest by registering a subscription to some prefix of this topic name.
ev – a Zeek event value.

Returns

true if automatic event sending is now enabled.

Broker::auto_unpublish 

Type: function (topic: string, ev: any) : bool

Stop automatically sending an event to peers upon local dispatch.

Parameters

topic – a topic originally given to Broker::auto_publish.
ev – an event originally given to Broker::auto_publish.

Returns

true if automatic events will not occur for the topic/event pair.

Broker::default_log_topic 

Type: function (id: Log::ID, path: string) : string

The default implementation for Broker::log_topic.

Broker::flush_logs 

Type: function () : count

Sends all pending log messages to remote peers. This normally doesn’t need to be used except for test cases that are time-sensitive.

Broker::forward 

Type: function (topic_prefix: string) : bool

Register a topic prefix subscription for events that should only be forwarded to any subscribing peers and not raise any event handlers on the receiving/forwarding node. i.e. it’s the same as Broker::subscribe except matching events are not raised on the receiver, just forwarded. Use Broker::unsubscribe with the same argument to undo this operation.

Parameters: topic_prefix – a prefix to match against remote message topics. e.g. an empty prefix matches everything and “a” matches “alice” and “amy” but not “bob”.
Returns: true if a new event forwarding/subscription is now registered.

Broker::listen 

Type: function (a: string &default = Broker::default_listen_address &optional, p: port &default = Broker::default_port &optional, retry: interval &default = Broker::default_listen_retry &optional) : port

Listen for remote connections using the native Broker protocol.

Parameters

a – an address string on which to accept connections, e.g. “127.0.0.1”. An empty string refers to INADDR_ANY.
p – the TCP port to listen on. The value 0 means that the OS should choose the next available free port.
retry – If non-zero, retries listening in regular intervals if the port cannot be acquired immediately. 0 disables retries. If the ZEEK_DEFAULT_LISTEN_RETRY environment variable is set (as number of seconds), it overrides any value given here.

Returns

the bound port or 0/? on failure.