Mirror/zeek
1
0
Fork 0
mirror of https://github.com/zeek/zeek.git synced 2025-10-16 13:38:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

cluster/setup-connections: Support global publish subscribe

Browse source 
Relates to #3917
This commit is contained in:
Arne Welzel 2025-03-13 13:37:24 +01:00
parent 8a4041b776
commit 2fde2f18d6
2 changed files with 45 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

12
scripts/base/frameworks/cluster/main.zeek
View file

@ -24,6 +24,18 @@ export {
## Whether to distribute log messages among available logging nodes.
const enable_round_robin_logging = T &redef;
## Enable global publish/subscribe functionality.
##
## With the Broker cluster backend, by default, nodes of the same
## type do not peer with each other, resulting in limited publish
## subscribe functionality. Setting this value to *T* changes the
## behavior to enable global publish subscribe visibility, where all
## nodes observe all messages published by other cluster nodes.
##
## Non-Broker backends should provide global publish subscribe behavior
## by default and can redef this to *T* in their policy scripts.
const enable_global_pub_sub = F &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 = "zeek/cluster/logger" &redef;

33
scripts/base/frameworks/cluster/setup-connections.zeek
View file

@ -59,6 +59,36 @@ function connect_peers_with_type(node_type: NodeType)
}
}
# Connect to all nodes that have the same type.
#
# To limit the number of connections within a cluster, the logic is to
# connect to all nodes of the same type with a name sorted higher than
# this nodes name itself.
#
# With 3 workers, worker-1 connects to worker-2 and worker-3, worker-2
# connects to worker-3 and worker-3 establishes no extra connections.
function connect_peers_same_type(self_name: string, self_type: Cluster::NodeType)
{
# nnodes is already sorted by name.
local nnodes = nodes_with_type(self_type);
local idx = -1;
# nnodes.indexOf(self_name)
for ( i, nn in nnodes )
if ( nn$name == self_name )
idx = i;
assert idx >= 0, fmt("%s not in %s", self_name, nnodes);
# Establish a connection to all nodes at higher indices.
idx += 1;
while ( idx < |nnodes| )
{
connect_peer(self_type, nnodes[idx]$name);
idx += 1;
}
}
event zeek_init() &priority=-10
{
if ( getenv("ZEEKCTL_CHECK_CONFIG") != "" )
@ -153,4 +183,7 @@ event zeek_init() &priority=-10
break;
}
if ( Cluster::enable_global_pub_sub )
connect_peers_same_type(node, self$node_type);
}