bento-rs

A Rust port of the Bento stream processor, focused (so far) on infrastructure data collection. Streams are declared in YAML with the upstream shape — an input, a chain of processors, and an output — and run on a small tokio engine with retried, backoff-managed delivery.

input:
  redfish:
    urls: [https://idrac-01.example.com]
    username: svc-collector
    password: ${IDRAC_PASSWORD}
    interval: 15m            # all polling inputs take interval; 0s = run once
pipeline:
  processors:
    - select:
        paths: [Model, Name, Status]
    - set_meta:
        key: model
        value: "${data:Model}"
output:
  mongodb:
    url: ${MONGO_URL}
    database: infra
    collection: inventory

Running

cargo build --release
bento-rs run stream.yaml      # run one stream
bento-rs streams ./streams/   # run every *.yaml in a directory as a stream
bento-rs lint stream.yaml     # validate a config without running it
bento-rs list                 # print registered component types

run and streams accept --metrics <host:port> to serve Prometheus counters at /metrics (per stream: bento_input_received, bento_output_sent, bento_output_batches, bento_input_errors, bento_output_errors) and a /ping liveness check.

Streams REST API

streams mode accepts --api <host:port>, which serves a management API and keeps the process running even when every stream has finished:

bento-rs streams ./streams/ --api 127.0.0.1:4196 --metrics 127.0.0.1:4197

curl localhost:4196/streams                          # all streams + status
curl -X POST localhost:4196/streams/foo -d @foo.yaml # add a stream
curl localhost:4196/streams/foo                      # status + config
curl -X PUT localhost:4196/streams/foo -d @foo.yaml  # replace it
curl -X DELETE localhost:4196/streams/foo            # stop and remove it

Each stream has its own shutdown handle, so removing or replacing one drains it (its in-flight batch finishes delivering) without touching the others. GET /streams/{name} returns the config as submitted — before ${VAR} expansion — so resolved secrets are never readable back.

Streams mode gives file-per-job modularity: each YAML file in the directory is an independent stream named by its file stem, polled on its own interval.

Components

Inputs
file	Reads local files as JSON lines (or one JSON document per file)
generate	Emits a static JSON document per poll (testing/heartbeat)
http_client	Polls JSON HTTP/REST endpoints (basic auth, custom headers)
http_server	streaming — receives POSTed JSON; responds 200 only after delivery
mqtt	streaming — subscribes to MQTT topics; QoS 1 PUBACK held until delivery
nats	streaming — subscribes to a core-NATS subject (queue groups supported)
redfish	BMC hardware inventory via Redfish (Dell iDRAC, HPE iLO, ...) with collection expansion
openshift_oc	OpenShift/Kubernetes resources via the oc/kubectl CLI
stdin	streaming — one message per line until EOF (pipe filter)
subprocess	Runs a command per poll, one message per stdout line
vcenter	VMware vCenter inventory via the vSphere Automation REST API (7.0+)

Streaming inputs run continuously instead of on an interval, and the stream ends when they are exhausted (e.g. EOF on stdin). They also accept a batching block that buffers messages into larger output writes — flushed once count messages accumulate and/or period elapses since the first buffered message, whichever comes first:

input:
  http_server:
    address: 0.0.0.0:4195
    batching:
      count: 50
      period: 2s

Acknowledgements respect batching: an http_server client only gets its 200 after the flush containing its messages is written. Polling inputs don't take batching; they already deliver one batch per poll.

Processors
cache	set/get/add/delete against a named cache; add drops duplicates
dedupe	Drop messages whose key at a path was already seen (in-memory, FIFO-capped)
filter	Keep messages where a dot-path exists / equals a value
http	Per-message HTTP request: enrich at result_path or replace the payload
log	Log a line per message (interpolated), leaving the batch untouched
mapping	Execute a Bloblang program on each message (see below)
select	Project messages down to a set of paths
set	Set a field (supports ${meta:key} interpolation)
remove	Remove fields
set_meta	Set a metadata key (supports ${data:path} interpolation)
sleep	Pause per batch — a crude throttle
split	Explode an array field into one message per element

Outputs
broker	Fan out every batch to a list of child outputs
discard	Drop everything (side-effect-only pipelines)
file	Append JSON lines to a file
mqtt	Publish per message (interpolated topic); QoS 1 waits for the broker's ack
nats	Publish per message to a core-NATS subject (interpolated)
stdout	JSON lines
switch	Route each message to the first case whose Bloblang check matches
http_client	POST batches as JSON
mongodb	Insert documents (metadata under _meta)

output:
  switch:
    cases:
      - check: 'this.kind == "vm"'
        output:
          mongodb: {url: "${MONGO_URL}", database: infra, collection: vms}
      - output:            # no check = catch-all
          file: {path: other.jsonl}

Messages matching no switch case are dropped; a failed child write retries the whole batch (at-least-once per child, as with broker).

log and http templates interpolate ${meta:key} and ${data:path} anywhere in the string.

Broker connectors

input:
  mqtt:
    url: tcp://broker.example.com:1883
    topics: ["sensors/#"]
    qos: 1                     # PUBACK sent only after delivery downstream
output:
  nats:
    url: nats://nats.example.com:4222
    subject: "telemetry.${data:host}"

The mqtt connectors use the pure-Rust rumqttc client. With qos: 1 the input holds each message's PUBACK until the engine confirms delivery (unacknowledged messages are redelivered by the broker), and the output's write only succeeds once the broker has acknowledged every publish in the batch — so the engine's retries give end-to-end at-least-once.

The nats connectors speak the core-NATS text protocol directly (no client dependency, plaintext TCP). Core NATS is fire-and-forget: the input carries no upstream acks (pair it with a buffer for durability) and a successful publish means the bytes reached the server. Queue groups are supported for load-balanced consumption; JetStream and TLS are not yet.

Caches & rate limits

Streams declare labelled resources with the upstream shape and reference them from components; a resource is shared by everything in the stream that names it:

cache_resources:
  - label: seen
    memory: {default_ttl: 10m, max_entries: 100000}
rate_limit_resources:
  - label: api
    local: {count: 10, interval: 1s}
input:
  http_client:
    urls: [https://api.example.com/v1/assets]
    rate_limit: api          # shared window across all requests
pipeline:
  processors:
    - cache:
        resource: seen
        operator: add        # drop messages whose key was seen (TTL'd)
        key: "${data:serial}"
output:
  stdout: {}

The memory cache supports default_ttl and FIFO-capped max_entries; the local rate limit allows count acquisitions per interval and delays the rest. rate_limit is accepted by the http_client input (per request), the http processor (per message) and the http_client output (per batch). The cache processor's get writes the cached value (or null) to result_path.

Bloblang (mapping)

The mapping processor runs a subset of upstream's Bloblang mapping language — a real lexer/parser/evaluator, not string templating:

pipeline:
  processors:
    - mapping: |
        # drop unhealthy machines entirely
        root = if this.status.health != "OK" { deleted() } else { this }
        root.model = this.model.uppercase()
        root.rack  = this.host.split("-").index(1)
        root.secret = deleted()
        root.checked_at = now()
        meta model = this.model
        let parts = this.path.split("/")
        root.first = $parts.index(1).or("unknown")

Supported today: root/meta/let assignments; this, bare paths and root reads; literals including arrays and objects; arithmetic, comparison and boolean operators; if / else if / else expressions (an if with no matching branch skips the assignment); deleted() for fields, metadata, whole messages and map_each elements; lambda methods map_each, filter, any, all (this.tags.map_each(t -> t.uppercase()) — parameters shadow outer lets); meta(), env(), now(), timestamp_unix(); and methods including uppercase, lowercase, trim, contains, has_prefix/suffix, replace_all, split, slice, parse_json, length, join, index, sum, sort, unique, flatten, keys, values, get, exists, merge, without, floor/ceil/round/abs, string/number/bool/type, catch/or for error and null fallbacks, regexes (re_match, re_find_all, re_replace_all with $1 capture groups — escape backslashes in patterns: "\\d+"), and timestamps (ts_unix parses RFC 3339 to unix seconds, ts_rfc3339 formats them back). If root is never assigned the payload passes through unchanged. Single Bloblang expressions also power the switch output's check fields.

Not ported yet: arbitrary strftime-style timestamp formats. Parse errors are caught by bento-rs lint with line numbers.

Every message carries meta.stream and meta.input_type plus input-specific metadata. ${VAR} in configs expands from the environment.

Delivery semantics

At-least-once while the process lives: a failed output write is retried with capped exponential backoff (1s → 60s) until it succeeds or the process is asked to shut down. Polling inputs are snapshot-oriented, so a batch lost to a crash is re-collected on the next interval. There is no on-disk buffer yet (see roadmap).

Streaming inputs are ack-driven: each batch travels as a transaction that the engine acknowledges only after the output write succeeds (or negatively acknowledges on processing failure/shutdown). http_server holds its HTTP response on that ack — a client that got a 200 knows its data reached the output, and anything else means "retry".

Durable buffer

Adding a buffer section decouples input from output through an on-disk write-ahead log, making at-least-once hold across restarts:

input:
  http_server: {address: 0.0.0.0:4195}
buffer:
  file:
    directory: /var/lib/bento/buffer/events
    segment_bytes: 16777216   # per-segment rotation size (default 16 MiB)
output:
  mongodb: {url: "${MONGO_URL}", database: infra, collection: events}

The input side appends each batch and fsyncs before acknowledging — with a buffer, http_server's 200 means "durably persisted" rather than "delivered" — while the drain side reads batches back, runs the processors, delivers with the usual retries, and then commits its position (atomic write, fully-consumed segments deleted). A crash between delivery and commit replays that batch on the next run; a partial record from a crash mid-append is truncated on open. A slow or down output no longer stalls the input — the backlog accumulates on disk instead.

Extending

A new component is one module implementing a one-method async trait plus a registration line in src/registry.rs:

// src/inputs/ping.rs
pub fn new(raw: &serde_yaml::Value) -> Result<Box<dyn Input>> { ... }

#[async_trait]
impl Input for PingInput {
    async fn poll(&self) -> Result<Vec<Message>> { ... }
}

r.register_input("ping", crate::inputs::ping::new);

Config parsing is plain serde — declare a struct with defaults. Processors (process) and outputs (write/close) follow the same pattern.

Port status & roadmap

This is an early, honest subset of upstream Bento. What exists today is the engine shape (streams, config format, component registries, retried delivery) and the collection-oriented components above. Major upstream features not yet ported, roughly in intended order:

• Streaming inputs — ack-driven StreamInput trait landed, with stdin and http_server; file tailing and broker consumers to come
• Bloblang — a real subset via the mapping processor (see above), including lambdas, regexes and RFC 3339 timestamps
• Streams REST API — --api serves add/get/update/remove at /streams[/{name}]
• Buffers — durable segmented file buffer between input and output (see above)
• Batching controls — count/period shaping on streaming inputs; byte-size and expression-based flushing to come
• Metrics — Prometheus /metrics + /ping via --metrics
• Broker connectors — NATS (hand-rolled core protocol) and MQTT (rumqttc) inputs/outputs; Kafka/AMQP left out until needed (heavy dependencies)
• Caches & rate limits — labelled memory caches and local rate limits, shared within a stream (see above)
• Output broker — broker fan-out and Bloblang-checked switch routing

Contributions to any box above are welcome — the component registry means most of them land as self-contained modules.

Running on OpenShift

The Dockerfile builds against rustls (no OpenSSL) on a distroless base and runs as a non-root arbitrary UID, compatible with OpenShift's restricted SCC:

oc new-build --binary --strategy=docker --name bento-rs
oc start-build bento-rs --from-dir . --follow
oc apply -f deploy/openshift.yaml

deploy/openshift.yaml mounts a ConfigMap at /etc/bento/streams (one key per stream file) and injects credentials from a Secret via ${VAR} expansion — adding a collection job is a ConfigMap edit plus a rollout restart, no rebuild.