Bento gear

The bento gear integrates the Bento engine (MIT Fork) into fluxrig. In the current release, it provides data mapping and logic via Bloblang and Standard I/O (Files, Stdout).

NOTE

Connector Status: The Bento ecosystem supports 100+ connectors. The standard fluxrig binary currently includes the Pure Logic and Local I/O sets to maintain a lightweight footprint. Institutional Connectors (Kafka, SQL, AWS, etc.) are currently in the future roadmap. A custom binary build can enable them.

Attribute	Details
Source Code	pkg/gears/native/bento
Pairs With	N/A (Universal Bridging)
Always Emitted Metadata	None (Inherits from Bento config)
Conditionally Emitted Metadata	None
Mandatory Consumed Metadata	None
Optional Consumed Metadata	None
Signals Sent	None
Signals Subscribed	None

Reference

fluxrig gears doc bento generates the identity, ports, and configuration below from the gear's manifest, so they stay in lockstep with the code.

Type	bento
Category	logic
Status	stable
Terminus	opaque

Wraps the Bento stream processor (file, socket, websocket and HTTP I/O, Bloblang mapping) as a gear.

Ports

Port	Direction	Role	Summary
in	input	message	Message into the Bento pipeline (when an input port is declared).
out.<name>	output (dynamic)	message	Message out of a declared Bento output port.

Configuration

Field	Type	Required	Default	Description
bento	object		-	Embedded Bento config (input/pipeline/output), validated by Bento.
log_level	string		-	override the log level for this gear only (e.g. TRACE, DEBUG, INFO, WARN, ERROR).
ports	object		-	Declared input/output port names (bare names, e.g. 'in'/'out'); each maps to a wired flux endpoint. Omit for the implicit in/out pair.

Architecture

Configuration safety

IMPORTANT

Strict Validation: The Rack performs a "Dry Run" build of the Bento stream during startup. If your configuration is invalid (e.g., syntax errors, unknown plugins), the Rack will fail to start. This ensures strictly valid configurations in production.

Description

The Gear automatically bridges fluxrig and Bento based on your ports definition. It attempts to inject the missing side of the pipeline:

1. Ingress (fluxrig -> Bento):

   • If you define ports.inputs, the gear injects a flux_in_<gearname> input source into the Bento config.
   • Note: Auto-wiring currently only supports the primary (first) input. Multiple inputs require manual broker configuration.

2. Egress (Bento -> fluxrig):

   • If you define ports.outputs and provide no Bento output, the gear injects a flux_out_<gearname> output sink.
   • If you define a complex bento.output (e.g., a switch for routing), you must manually target the flux_out_<gearname> plugin for internal ports.

Seen in use

Enriching an authorization with a network signal uses one bento gear for a cache lookup and a policy decision. It makes an outbound HTTP call under a hard deadline. A second gear serves a simulated API from an http_server input. This case shows why a gear bringing its own input gets no injected buffer. A buffer would make a synchronous responder answer before the pipeline produced a body.

Examples

Source (local input)

Ingest from a local File -> Output to fluxrig out.

- name: "file-ingest"
  type: "bento"
  ports:
    outputs: ["out"]
  config:
    bento:
      input:
        file:
          paths: ["/data/ingress/*.json"]

Sink (local output)

Input from fluxrig in -> Egress to local storage.

- name: "local-archive"
  type: "bento"
  ports:
    inputs: ["in"]
  config:
    bento:
      output:
        file:
          path: "/data/archive/${! meta(\"trace_id\") }.json"

Processor (logic)

Transformation. fluxrig in -> Logic -> fluxrig out.

- name: "pii-masker"
  type: "bento"
  ports:
    inputs: ["in"]
    outputs: ["out"]
  config:
    bento:
      pipeline:
        processors:

          - mapping: 'root = this; root.credit_card = "*****"'

Router (traffic control)

Routes to different internal ports. You must configure the switch output.

- name: "priority-router"
  type: "bento"
  ports:
    inputs: ["in"]
    outputs: ["high", "low"]
  config:
    bento:
      # Manually wire the output switch to specific `fluxrig` ports
      output:
        switch:
          cases:

            - check: 'this.priority == "high"'
              output:
                 flux_out_priority_router: { name: "high" }

            - check: 'this.priority == "low"'
              output:
                 flux_out_priority_router: { name: "low" }

Aggregator (fan-in)

Merges multiple fluxrig streams (app_logs, sys_logs) into one output (out).

- name: "concentrator"
  type: "bento"
  ports:
    inputs: ["app_logs", "sys_logs"] # Auto-wire takes primary (app_logs). Manual intervention recommended for multi-input.
    outputs: ["out"]
  config:
    bento:
      # No bento config needed; wrapper handles Input Broker -> default Pipeline -> Output Sink

Rationale & extended info

Native compilation (zero dependency)

The bento gear runs an Isolated Bento Environment compiled natively into the fluxrig binary. It does not rely on a separate sidecar container, external daemon, or network Hop. It bridges data between fluxrig's internal NATS bus (CBOR) and Bento's internal message format using high-performance, in-memory channels.

Data mapping (fluxMsg ↔ Bento part)

When data crosses the boundary between fluxrig and Bento, an automatic translation occurs:

• Ingress (fluxMsg -> Bento): The fluxMsg.RawPayload becomes the Bento message payload (accessible via Bloblang this). The fluxMsg.Metadata map becomes the Bento message metadata (accessible via Bloblang meta("key")).
• Egress (Bento -> fluxMsg): The Bento payload becomes the fluxMsg.RawPayload. The Bento metadata becomes the fluxMsg.Metadata.

Error handling & routing

If a Bento processor (e.g., Bloblang mapping) fails gracefully, it typically uses Bento's standard error handling mechanisms. However, to interface with fluxrig's structured paths, use Bento's switch output or try processors. Route failed messages conditionally back to specific internal fluxrig ports (e.g., ports: { outputs: ["out", "error"] }).

Observability integration

Bento's native metrics and tracing subsystems are bridged to fluxrig's OpenTelemetry pipeline. When the gear starts, it registers a metrics bridge that exports Bento-internal counters and latencies (e.g., connector pressure, processor timing) as native OTel metrics.