OpexFlow Documentation

Try it — live REPL

Edit the record and the program, then run. This is an in-browser evaluator of the documented DSL subset operating on the record object — no server round-trip.

// press Run

JavaScript transforms

When a transform needs loops, regex, or logic the DSL can't express, use a JavaScript transform. It runs in an embedded, sandboxed QuickJS runtime — no filesystem, no network, hard memory + stack limits, and a deadline interrupt. A worker pool (one runtime per scripted route, sharded by machine id) gives parallelism with backpressure.

// enrich.js — runs once per record
function transform(record, ctx) {
  if (record.metadata.temp > 95) return null;        // drop
  record.metadata.risk = record.metadata.temp * 0.8
                      + record.metadata.vib * 1.2;
  return record;                         // pass on (mutated)
}

AI transform generator

Describe the transform in plain language; OpexFlow writes it. The generator talks to any OpenAI-compatible /chat/completions endpoint, is steered to emit DSL first (falling back to JS when the rule isn't expressible), and server-validates the output before returning it.

ai:
  enabled: true
  base_url: "https://api.openai.com/v1"   # any OpenAI-compatible base
  model: "gpt-4o-mini"
  api_key: "\${OPEX_AI_API_KEY}"
  max_tokens: 1200

In the UI's AI tab you sketch the input shape, the rule, and the desired output — e.g. "emit only when temp exceeds 80 or vibration crosses 5.0" — and hit Generate. The model maps your fields onto metadata.* and returns a runnable program you paste into the editor and save. Endpoint: POST /api/transform/generate.

Sources

Each source is a protocol adapter configured under adapters (or nats/mqtt for streaming). All normalize to CanonicalMachineData.

Sinks

Each route declares which sinks it fans out to (omit to inherit the default, which is all sinks). Names map case-insensitively.

Observability

Observability is fully native — no sidecar collector required. The connector serves Prometheus metrics and keeps in-process traces + a SQLite history series, all viewable in its own Observability tab.

• Metrics at /metrics — opex_connector_records_total, opex_connector_sink_total, opex_connector_push_duration_seconds, opex_connector_last_publish_ts.
• Traces — one span per record through source → transform → sink, queryable via /api/obs/traces.
• History — sampled metric series in SQLite (/api/obs/series).
• Alerts — sink-stall detection; delivers to a webhook and/or Telegram.
• Grafana — provisioned dashboards ship in the grafana/ directory.

telemetry:
  stall_threshold_secs: 300
  sample_interval_secs: 15
  alerts:
    webhook_url: "https://hooks/ops"
    telegram: { bot_token: "\${TG_TOKEN}", chat_id: "-100…" }

REST API

The same API the canvas UI uses. JSON over HTTP, served on health_check_port (default 3000).

Deployment

systemd

# /etc/systemd/system/opexmx-connector.service
[Unit]
Description=OpexMX Connector
After=network.target

[Service]
Environment=OPEXMX_HMAC_SECRET=change-me
ExecStart=/usr/local/bin/opexmx-connector /etc/opexmx/config.yaml
Restart=on-failure
User=opexmx

[Install]
WantedBy=multi-user.target

Edge (ARM64)

The same static binary cross-compiles to aarch64-unknown-linux-musl. Deploy identical artifacts from Raspberry Pi to rack server.

CLI

The connector takes an optional config path (defaults to config.yaml):

opexmx-connector                 # run with ./config.yaml
opexmx-connector /etc/opexmx/config.yaml
opexmx-connector --help

Validate transforms from the UI (Check) or the API (POST /api/transform/validate) rather than a separate CLI subcommand. Set secrets via environment variables and reference them as ${VAR} in the config.