- Rust 73.6%
- TypeScript 16.8%
- JavaScript 3.7%
- Shell 3.6%
- HTML 1.9%
- Other 0.4%
Erik Stephens
1721174aa4
Some checks failed
/ test (push) Failing after 3s
Adds a `meta` transform stage as the one place a pipeline operator may touch out-of-band event metadata. Its function is called as `f(e, meta)` and returns a patch merged into the event's metadata, so read-modify-write happens in one pass:
- module: meta
function: 'function(e, meta) { return { ingress_ms: e.sched_ms }; }'
`e` is a copy of the payload — only the returned patch is applied, so the stage cannot alter payload fields (use rename/drop for that). Writable keys come from a typed `WRITABLE` whitelist, currently just `ingress_ms`; exposing another field is one row plus, if it needs new validation, one `MetaKind` variant. Values are normalised as well as validated: JS numbers arrive as f64, and storing `1000.0` would make the `as_u64()` reads in latency.rs return None — a silent "missing data" failure rather than a loud one. An unknown key, a bad value, a non-object return, or a thrown exception fails the event through the error channel. Non-Data events pass through without invoking the function, since they carry no meta at all.
This supersedes an earlier approach that hung `this.getIngressStamp()` / `this.setIngressStamp(v)` off the shared JS sandbox. That put per-event mutable state on an object whose other properties (hostname, util, config) are ambient and set-once, which is why it needed a reset guard to scrub state between calls; it also did not survive a second metadata field, since the backing AtomicI64 only works for one non-negative integer.
Separately, JS exceptions were surfacing as the useless "Exception generated by QuickJS" because the real message lives on the pending exception and must be fetched via Ctx::catch(). `describe_js_error` now covers every call helper and is public for bespoke call sites. In fail2ban this was worse than bad wording: `ban()` used `.ok()?`, so a throwing ban was indistinguishable from a deliberate null return. It still means "not banned, retry later", but is now reported.
`tap` built its own runtime rather than going through `create_sandbox`, so it had neither the interrupt handler nor the Deadline userdata — a runaway tap function would wedge the worker thread, and adding `guard_call` alone would have compiled and silently done nothing. It now shares the standard sandbox setup, and gains its first tests. `run_sync_pub` existed for plugins building their own contexts; tap was the last, so it is now test-only.
Both runaway-function tests are wrapped in a timeout so a regression fails instead of hanging the suite. The equivalent test in js_sandbox.rs still has the hang-on-regression shape — it is a sync test, so a timeout there needs a spawned thread.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
|
||
|---|---|---|
| .agents/commands/fjx | ||
| .claude | ||
| .forgejo/workflows | ||
| .vscode | ||
| doc | ||
| e2e | ||
| logbus-plugin-sdk | ||
| openspec | ||
| src | ||
| tests | ||
| ts-port | ||
| wiki | ||
| .eslintrc.yml | ||
| .gitignore | ||
| .gitlab-ci.yml | ||
| .pre-commit-config.yaml | ||
| .prettierrc | ||
| AGENTS.md | ||
| Cargo.lock | ||
| Cargo.toml | ||
| CLAUDE.md | ||
| fjx.json | ||
| HANDOFF.md | ||
| justfile | ||
| LICENSE | ||
| README.md | ||
| test.just | ||
LogBus
A "Swiss Army" knife for handling events with a focus on processing system & application logs. See the docs for more information.
Vision
LogBus operates at Layer 1: Signal Conditioning — the digital layer that cleans, normalizes, and shapes raw event streams before they can be understood. Think of it as DSP for log data: filtering noise, normalizing formats, aggregating samples, applying deterministic transforms.
The conditioned signal it produces is the foundation for Layer 2: Signal Intelligence — pattern detection, anomaly identification, cross-source correlation, meaning extraction. Whether L2 lives inside LogBus or in a downstream system is an open question, but L1 has to be right first.
See docs/vision.md for more.
Hacking
See just help for list of helpful tasks.
Unit Testing
just test::units will test plugins. Plugins should have their spec files colocated beside them. Plugins should strive to be simple functions, but here are some ways they are not:
-
If a plugin needs to manage state, then that can persist in its closure and managed however the plugin sees fit. For example, the write-opensearch output plugin batches events into an array until it is ready to perform a bulk insert.
-
Some plugins may require special start & stop handling.
End-to-End Testing
just test::e2e will test more real-world like scenarios and exercise the main engine bits, but limited to simpler inputs & outputs (files).
How Events Flow Between Stages (Rust port)
Each stage in the pipeline is an async task. Stages are connected by bounded MPSC channels (capacity 4096). The engine wires them up based on the inputs: declarations in the YAML config.
[Source] --[4096]--> [fan_out] --[4096]--> [Transform] --[4096]--> [fan_out] --[4096]--> [Sink]
Fan-in (multiple upstream stages → one stage) is handled natively by MPSC: each upstream fan-out task holds a clone of the downstream stage's sender. The downstream's receiver closes only when all upstream senders are dropped — i.e., when every upstream stage has finished.
Fan-out (one stage → multiple downstream stages) is handled by a lightweight fan_out task spawned per stage. It reads from the stage's output channel and clones each event (cheap — events are Arc<Value>) to each downstream's input channel.
Backpressure
Backpressure is automatic and requires no plugin code. Because mpsc::Sender::send().await suspends when the channel buffer is full, a slow stage naturally stalls everything upstream:
- The sink's input buffer fills up (disk is slow)
- The upstream
fan_out'ssend().awaitsuspends - The transform's output buffer fills up
- The transform's
send().awaitsuspends - The transform stops reading its input buffer
- The source's
send().awaitsuspends - The source stops reading from I/O (file, stdin, Kafka)
The maximum event accumulation before the source stalls is capacity × number of channels in the chain (default capacity: 4096, tunable via --event-buffer). No events are dropped; the source simply pauses until the sink catches up. Tokio parks suspended tasks with zero CPU overhead.
The UI stats interceptor is the one deliberate exception: it receives events via try_send (non-blocking) so a slow browser connection can never cause backpressure into the data pipeline.
Known Issues, Limitations, Warts
-
There should be an example demonstrating how to "commit" events. For example, when consuming from a kafka topic, should only commit offsets once the final pipeline stage has confirmed that the event was received successfully. Not yet sure if there needs to be 1st class support for such a mechanism or if can achieve by wiring up the latter stage's output to the earlier stage's input.
-
Would be nice if plugins could communicate the type (and maybe shape as well) of the events they process. Then, a user could be warned if a pipeline configuration contains stages wired up erroneously.
-
Some plugins require pipeline-defined functions. For simple functions, inlined in yaml is not cumbersome. For more complicated functions, would be nice to define those in a proper package for a better dev & test story.