In-Cluster Event Broker¶

continuum-event-broker is the in-cluster endpoint of the event bus tunnel. A router running outside the GPU cluster cannot reach worker Pod IPs directly, so it cannot subscribe to the Dynamo ZMQ publishers that emit kv_metrics load signals and kv-events. The broker runs inside the cluster, initiates an outbound connection to the router's TLS event bus, and relays publisher bytes to the router.

  router (bus server)                     cluster
  ┌─────────────────────┐   HTTP/2+TLS   ┌──────────────────────┐   ZMQ/TCP   ┌───────────┐
  │ discovery, ZMTP     │◀──────────────│ continuum-event-      │────────────▶│ publisher │
  │ client, decoding    │ session +      │ broker (bus client)   │             │ (worker)  │
  └─────────────────────┘ data streams   └──────────────────────┘             └───────────┘

The router owns worker discovery and decides every subscription. The broker acts strictly as a byte tunnel rather than a re-encoding proxy: for each subscribe command, it opens a TCP connection to the publisher and relays raw ZMTP bytes bidirectionally over a single HTTP/2 stream without parsing them. The router's bounded ZMTP client handles the handshake, topic subscription, and frame decoding through the tunnel, so Dynamo protocol upgrades do not require broker changes.

The wire protocol lives in the standalone crates/continuum-event-bus crate, the router-side listener and broker subscriber live in src/event_bus/, and the deployment manifests live under deploy/ (see Deployment).

The same event bus also supports a second client role, kv_listener: continuum-kv-listener dials out to the router over the same transport to push per-backend KV event streams when the router cannot reach the listener directly. See KV Cache Optimization and Dynamo Worker Discovery for details. The broker binary itself never announces sources and rejects any announced_source target with unsupported_target.

What the broker does¶

1. Metadata sync: Lists and watches DynamoWorkerMetadata resources in its configured Kubernetes namespaces to build the set of advertised publisher endpoints. The broker waits until every watched namespace completes its initial list before dialing the bus.
2. Session handshake: Sends POST /bus/v1/session over HTTP/2 and TLS with its bearer credential, followed by a hello frame declaring the broker role, its client ID, its Kubernetes namespace, and the Dynamo namespaces it covers. The router replies with welcome containing the negotiated protocol version, session ID, and effective resource limits.
3. Heartbeat: Sends heartbeat frames at the negotiated interval and terminates the session if the router remains silent past the idle timeout.
4. Stream setup: Validates each subscribe target against the Security model, connects to the publisher, sends POST /bus/v1/streams/{id}, and reports stream_opened. If validation or connection fails, it reports stream_refused without opening a stream.
5. Byte relay: Relays bytes until either side closes. Streams that terminate on their own report stream_closed (upstream_closed or buffer_overflow), whereas streams closed by an unsubscribe command close silently.
6. Reconnection and drain handling: Reconnects with jittered exponential backoff (reconnect_backoff, 1s to 60s by default) when a session drops, closing any active streams with the session. On a router drain, the broker stops accepting new streams, lets open streams finish until the router's deadline, and then reconnects.
7. Rebalance throttling: A rebalance drain reconnects after a short jittered delay (250 to 500 ms), unless another rebalance occurred within the last minute, in which case it falls back to normal exponential backoff so repeated rebalances cannot trigger a tight reconnect loop.
8. Graceful shutdown: On SIGTERM or SIGINT, sends draining, rejects new subscriptions with draining, lets existing streams finish until drain_deadline, and exits with status 0. A second signal forces an immediate exit.

A session and all of its data streams share a single TLS connection, so a dropped connection terminates the session and all streams together. The broker uses hyper's HTTP/2 client over tokio-rustls (aws-lc-rs provider) because each data stream requires concurrent streaming request and response bodies for its entire lifetime.

Security model¶

The broker dials a target only after every validation check passes, preventing a compromised router from using the broker to probe arbitrary cluster addresses.

The address filter evaluates resolved socket addresses right before dialing, so DNS records cannot bypass it. Loopback, link-local (including the cloud metadata IP 169.254.169.254), unspecified, multicast, and broadcast addresses are always rejected unless explicitly covered by an allowed_cidrs entry that falls entirely within that special range (0.0.0.0/0 does not qualify). When allowed_cidrs is non-empty, any resolved address outside the configured CIDRs is also rejected.

Every refusal returns stream_refused to the router and increments continuum_event_broker_refusals_total{code}. Warning logs are rate-limited per error code to prevent log flooding: the broker logs the first refusal in each 60-second window, counts subsequent occurrences silently, and logs the next window's first refusal along with a suppressed count.

Before the initial list completes, no endpoints are advertised and the broker does not dial the bus. Even if a subscribe command were evaluated in that state, the target check rejects it with target_not_advertised so the router can retry later.

The broker reads only the event_channels map from DynamoWorkerMetadata. It ignores the endpoints map, which holds request-plane addresses (<ip>:<port>/<id>/<endpoint>) rather than ZMQ publishers. Because any principal with permission to create DynamoWorkerMetadata in a watched namespace can advertise an endpoint, Kubernetes RBAC is the primary trust boundary and allowed_cidrs provides a second independent boundary.

The bus_url must use https://; plaintext URLs are rejected as configuration errors. The bearer credential is loaded from a mounted file or environment variable, sent exclusively in a sensitive Authorization header, and never written to logs.

Set allowed_cidrs to the Pod CIDR¶

We strongly recommend setting allowed_cidrs to the cluster's Pod CIDR (for example, 10.244.0.0/16). While the DynamoWorkerMetadata check restricts targets to advertised endpoints, allowed_cidrs bounds the blast radius of a forged or misconfigured metadata resource.

When allowed_cidrs is empty, the broker permits any advertised non-special IP address, which could allow a forged resource to point at a Service IP, node address, another tenant's Pod, or an internal database. Special-purpose ranges (loopback, link-local including 169.254.169.254, unspecified, multicast, and broadcast) remain blocked regardless.

EndpointSlice readiness¶

The broker does not filter targets by EndpointSlice readiness. Its security check exists to verify that a target is a genuine Dynamo publisher, and an unready worker's publisher is still a valid publisher endpoint.

The router decides when to subscribe based on its own readiness-aware discovery. Omitting EndpointSlice checks keeps the broker's RBAC footprint limited to a single custom resource. Because each worker's DynamoWorkerMetadata is owned by its Pod, Kubernetes garbage-collects the metadata resource when the Pod terminates, removing its endpoints from the broker's view.

The view across watch restarts¶

After the initial list completes, the broker retains its last known metadata snapshot across watch restarts and replaces it atomically when the next list finishes. Dropping the snapshot during a temporary Kubernetes API server outage would reject all new subscriptions; retaining it means an endpoint deleted during the outage remains dialable until the relist completes, while still subject to allowed_cidrs.

Bounded buffering¶

Each data stream buffers at most max_stream_buffer_bytes (up to 1 MiB, or less if negotiated in welcome) toward the router. Across all streams in a session, total buffered data is capped at 64 MiB so a router that opens many streams without granting HTTP/2 window capacity cannot force the broker to allocate max_streams * max_stream_buffer_bytes.

Every queued chunk charges its byte length plus a fixed 64-byte queue-item overhead against both limits, including at the minimum 16 KiB stream budget, so a publisher sending tiny frames is accounted at its true memory cost.

The broker reads from the publisher socket continuously without applying TCP backpressure, avoiding silent drops at the publisher's ZMQ high-water mark. If a chunk would exceed either buffer limit, the broker resets the HTTP/2 stream, closes the publisher socket, and sends stream_closed with buffer_overflow, prompting the router to resubscribe. In the reverse direction (router to publisher), the 64 KiB HTTP/2 stream window bounds how much data the router can send ahead of the socket.

Opening a stream is bounded¶

The router expires pending subscribe commands after 30 seconds (STREAM_OPEN_TIMEOUT) and rejects late data streams with unknown_stream. To finish within that window, the broker enforces strict per-stage deadlines before opening a stream:

• DNS resolution: up to 5 seconds
• Upstream TCP connect: up to 10 seconds total across all resolved addresses, with each individual attempt also capped by upstream_connect_timeout
• Bus HTTP/2 slot acquisition: up to 5 seconds to obtain a free stream slot on the bus connection
• Router response head: up to 10 seconds for the router to acknowledge the stream request headers

When any deadline expires, the broker closes the upstream socket, releases the stream slot, and reports the failure:

Configuration¶

See config.event-broker.yaml.example.

Unknown fields are rejected at startup. Running continuum-event-broker --check-config validates the configuration file, bearer credential, and CA bundle and exits.

RBAC¶

The broker requires only get, list, and watch permissions on dynamoworkermetadatas.nvidia.com in each watched namespace:

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: continuum-event-broker
  namespace: <watched namespace>
rules:
  - apiGroups: ["nvidia.com"]
    resources: ["dynamoworkermetadatas"]
    verbs: ["get", "list", "watch"]

The broker never creates, updates, patches, or deletes Kubernetes resources. Both the Helm chart and the Kustomize bundle render this Role in each watched namespace (see Deployment).

Bounds on the metadata view¶

The broker implements its own lightweight DynamoWorkerMetadata parser without depending on the main router crate, enforcing the same safety bounds:

• Resource and publisher caps: at most 16,384 resources per watched namespace, 1,024 channel entries inspected and 256 publishers retained per resource, and 65,536 total publisher references (including duplicate advertisements across resources).
• Field size caps: channel keys up to 1 KiB, channel values up to 64 KiB, and endpoint strings up to 256 bytes.
• Paged list streaming: list queries fetch 32 resources per page and parse each page before requesting the next, holding at most one raw page in memory at a time. Opening a watch response is bounded to 30 seconds.
• Cap enforcement on updates: list pagination enforces the 65,536 publisher reference cap before committing the snapshot. If a watch update would exceed the cap, the previous advertisement for that resource is removed first so stale addresses are immediately deauthorized.
• Selective parsing and fault tolerance: deserialization extracts only spec.data.event_channels and skips model cards and request-plane endpoints. Malformed entries or invalid resources are skipped and counted in continuum_event_broker_metadata_skipped_total without failing the enclosing list page.

Metrics¶

All metric label sets have bounded cardinality: error and close codes come from the protocol enum (with unrecognized values mapped to unknown), and dynamic identifiers such as stream IDs, hosts, and namespaces are never used as labels. The /readyz endpoint returns 200 OK once the initial metadata sync completes.

Packaging¶

The release workflow builds continuum-event-broker for all four Linux targets and packages it alongside its man page (continuum-event-broker(1)) in the Linux release archives and Debian package. The docker-event-broker CI job publishes a distroless static container image to ghcr.io/lablup/continuum-event-broker using crates/continuum-event-broker/Dockerfile.

Because the broker crate is standalone (with its own Cargo.lock outside the router workspace), it adds no dependencies to the router feature graph. CI runs formatting, linting, and unit tests for the crate on every Rust change.

Deployment¶

You can deploy the broker inside the GPU cluster using either the standalone Helm chart (deploy/helm/continuum-event-broker) or the Kustomize bundle (deploy/kubernetes/event-broker/). On the router side, values-dynamo-event-bus.yaml in the router Helm chart exposes the event bus listener.

Both deployment methods run a single-Pod Deployment per broker identity (client_id). Because the router allows only one active session per client_id and a new connection with the same ID supersedes the existing one, high availability requires deploying two distinct broker identities rather than scaling a single Deployment to two replicas.

When two broker identities cover the same Dynamo namespace, the router distributes subscriptions across both and fails over subscriptions within one reconnect cycle if either broker disconnects (see Several brokers per namespace). For step-by-step setup instructions, see In-cluster event broker in the deployment guide.

Testing¶

The crate's test suite runs entirely in-process without a Kubernetes cluster. It uses an in-process TLS HTTP/2 bus with an ephemeral certificate and local TCP publishers to verify bidirectional byte relay, refusal codes, request-plane address filtering, non-blocking buffer overflow, stream-open deadlines, rebalance backoff, reconnect handling, router drain, terminal errors, idle timeouts, and SIGTERM draining.

The metadata parser is tested against the router's DynamoWorkerMetadata fixtures, and full end-to-end tunnel tests run against the router listener in tests/event_bus/.