A dynamic plugin hub for HAProxy SPOE agents. Loads Rust plugins as shared libraries at runtime, dispatches SPOE messages (running independent plugins in parallel), and returns aggregated variables to HAProxy.
HAProxy's Stream Processing Offload Engine (SPOE) allows offloading request processing to external agents via the SPOP protocol. This hub acts as a single SPOE agent that multiplexes work across dynamically loaded plugins, so you can add, update, or remove processing logic without modifying or recompiling the hub itself.
Key features:
- Parallel plugin execution — independent plugins run concurrently, with topological sorting for dependency chains
- Bounded plugin execution — per-plugin concurrency and queue limits reject overload explicitly; timed-out calls retain capacity until their blocking work actually returns
- Variable namespacing — plugin output variables are auto-prefixed (e.g.,
geoip.country_code) to prevent conflicts - Pre-parsed data — plugins receive typed, structured arguments instead of raw SPOP bytes
- Hot reload — SIGHUP or a config-file change reloads configuration and swaps plugins without dropping connections; plugins whose
libraryandparamsare unchanged keep running, and in-flight dispatches finish on the replaced plugins before they are drained and torn down (swap → quiesce → drain → shutdown) - Graceful shutdown — SIGTERM drains in-flight messages before exiting
- Observability — structured JSON logging and built-in Prometheus metrics endpoint
- Rust toolchain (stable, >= 1.86)
- HAProxy >= 2.0 with SPOE support
git clone https://github.com/haproxy-haptic/haproxy-spoa-hub.git
cd haproxy-spoa-hub
cargo build --releaseThe hub binary is at target/release/haproxy-spoa-hub. The build also produces example plugins: libnoop_plugin.so (sets a static variable) and libecho_plugin.so (echoes all message arguments back as variables).
Create a configuration file (e.g., config.toml) with both plugins registered for the same SPOE message — the hub runs them in parallel automatically:
plugin_dir = "/opt/spoa-plugins"
default_timeout_ms = 100
log_level = "info"
metrics_addr = "0.0.0.0:9090"
[[listeners]]
type = "unix"
address = "/var/run/haproxy-spoa-hub.sock"
[[plugins]]
name = "noop"
library = "libnoop_plugin.so"
messages = ["check-request"]
[[plugins]]
name = "echo"
library = "libecho_plugin.so"
messages = ["check-request"]Copy both plugins and start the hub:
mkdir -p /opt/spoa-plugins
cp target/release/libnoop_plugin.so target/release/libecho_plugin.so /opt/spoa-plugins/
haproxy-spoa-hub --config config.tomlBoth plugins handle the same check-request message. Since neither depends on the other, the hub executes them concurrently — total latency is the slower of the two, not the sum.
Add an SPOE filter to your frontend. First, the SPOE agent definition (spoe-hub.conf):
[spoa-hub]
spoe-agent spoa-hub-agent
messages check-request
option var-prefix hub
timeout hello 2s
timeout idle 120s
timeout processing 500ms
use-backend spoa-hub-backend
spoe-message check-request
args src=src method=method path=path
event on-frontend-http-request
Then reference it in haproxy.cfg. Variables from each plugin are automatically namespaced with the plugin name:
frontend http-in
bind *:8080
filter spoe engine spoa-hub config /etc/haproxy/spoe-hub.conf
# noop plugin sets "result", available as hub.noop.result
http-request set-header X-Noop %[var(sess.hub.noop.result)]
# echo plugin mirrors args back, e.g. hub.echo.src
http-request set-header X-Client-Src %[var(txn.hub.echo.src)]
default_backend app# Validate configuration without starting
haproxy-spoa-hub --config config.toml --check
# Send a request through HAProxy and check the hub logs
curl http://localhost:8080/The hub logs structured JSON to stderr. Both plugins load at startup and execute in parallel on each request:
{"timestamp":"...","level":"INFO","message":"plugin loaded","plugin":"noop","version":"0.1.0"}
{"timestamp":"...","level":"INFO","message":"plugin loaded","plugin":"echo","version":"0.1.0"}
{"timestamp":"...","level":"INFO","message":"listening","transport":"unix","address":"/var/run/haproxy-spoa-hub.sock"}
{"timestamp":"...","level":"DEBUG","message":"plugin completed","plugin":"noop","latency_us":55,"variables":1}
{"timestamp":"...","level":"DEBUG","message":"plugin completed","plugin":"echo","latency_us":69,"variables":1}
{"timestamp":"...","level":"DEBUG","message":"message processed","message_name":"check-request","latency_us":330}To add a plugin that depends on another's output, use depends_on:
[[plugins]]
name = "bot-detect"
library = "libbot_detection.so"
messages = ["check-request"]
depends_on = ["ja3"] # runs after ja3, receives ja3.hash in message argsA dependent executes only after every declared dependency returns ok. If an
upstream plugin errors, times out, is overloaded, or is skipped, all transitive
dependents are skipped without entering the blocking pool. Every plugin in the
message's execution plan still publishes a host-owned status: ok, error,
timeout, overloaded, or skipped.
| Field | Type | Required | Default | Description |
|---|---|---|---|---|
plugin_dir |
string | yes | — | Path to directory containing plugin .so files |
default_timeout_ms |
integer | no | 100 | Default per-plugin execution timeout |
log_level |
string | no | "info" |
Log level (RUST_LOG syntax) |
metrics_addr |
string | no | disabled | Prometheus endpoint bind address (e.g., "0.0.0.0:9090") |
worker_threads |
integer | no | num CPUs | Tokio runtime worker threads |
max_connections |
integer | no | unlimited | Maximum concurrent SPOP connections |
blocking_thread_keep_alive_secs |
integer | no | 300 | How long idle blocking pool threads stay alive before being reaped. Higher values avoid thread re-creation overhead (~10-50µs) on bursty traffic. Tokio's built-in default is 10 seconds. |
max_blocking_threads |
integer | no | sum of plugin concurrency limits | Process-wide blocking-pool cap. When set, it must be at least the sum of resolved plugin concurrency limits so admitted work cannot enter Tokio's unmetered internal queue. |
| Field | Type | Required | Description |
|---|---|---|---|
type |
string | yes | "unix" or "tcp" |
address |
string | yes | Socket path or host:port |
| Field | Type | Required | Default | Description |
|---|---|---|---|---|
name |
string | yes | — | Unique identifier (used as variable namespace prefix) |
library |
string | yes | — | Filename of .so in plugin_dir |
messages |
array | yes | — | SPOE message names this plugin handles |
depends_on |
array | no | [] |
Plugin names that must complete before this one |
timeout_ms |
integer | no | default_timeout_ms |
Per-plugin timeout override |
max_concurrency |
integer | no | available CPUs | Maximum simultaneously executing calls for this plugin. Timed-out calls keep occupying a slot until they actually return. |
max_queue |
integer | no | 0 |
Maximum calls waiting for a concurrency slot. Zero rejects excess work immediately. |
queue_timeout_ms |
integer | no | remaining timeout_ms |
Maximum queue wait when max_queue > 0; the total call still cannot exceed timeout_ms. |
params |
table | no | {} |
Typed parameters passed to init() (supports nested tables, arrays, and all TOML types) |
On reload, a plugin whose library and params are unchanged keeps its
running instance — init() isn't called again, so connections and caches
it holds survive — and the other fields take effect on that instance. A
plugin whose library or params changed, or that was added, is created and
initialised anew; one that was removed is drained and shut down.
The defaults favor bounded latency under overload: each plugin is limited to
the host's available parallelism and has no queue. CPU-heavy plugins such as a
WAF should normally set max_concurrency to the CPU capacity reserved for the
hub. Add a small queue only when absorbing a short burst is preferable to an
immediate overload verdict; never use it as a substitute for horizontal
scaling.
Overload, queue-timeout, timeout, and plugin-error outcomes are exported as bounded-cardinality Prometheus metrics and host-owned status variables. Their per-request log entries are debug-level so hostile traffic cannot turn an already bounded execution path into a production log-amplification attack.
!!! note
Top-level keys (plugin_dir, default_timeout_ms, etc.) must appear before any [[listeners]] or [[plugins]] table arrays in the TOML file.
Plugins are standalone Rust crates compiled to shared libraries. You only need the haproxy-spoa-hub-plugin-api crate — no access to the hub source code required.
The hub keeps every successfully opened plugin library mapped until process
exit. A config reload creates, initializes, drains, shuts down, and destroys
the plugin instances whose library or params changed, but never dlcloses
arbitrary plugin code: plugins may embed foreign runtimes or retain
process-global callbacks that cannot be unloaded safely. Replace a plugin .so
by rolling or restarting the hub process; do not expect a config reload to
activate new code at the same library path.
cargo new --lib my-spoa-plugin
cd my-spoa-pluginSet up Cargo.toml:
[lib]
crate-type = ["cdylib"]
[dependencies]
haproxy-spoa-hub-plugin-api = "0.2"
abi_stable = "0.11"Define your plugin struct, then register it with define_plugin!. The macro generates the FFI trait implementation and wraps process() in catch_unwind so a panic doesn't crash the hub:
#![allow(non_camel_case_types, non_local_definitions)]
use haproxy_spoa_hub_plugin_api::*;
#[derive(Debug)]
struct MyPlugin {
db_path: String,
}
define_plugin!(MyPlugin, {
fn new() -> Self {
MyPlugin { db_path: String::new() }
}
fn init(
&mut self,
context: &PluginContext,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// Access per-plugin configuration from [plugins.params]
self.db_path = context.get_config("db_path")
.and_then(|v| v.as_str())
.ok_or("missing required config: db_path")?
.to_owned();
Ok(())
}
fn name(&self) -> &str { "my-plugin" }
fn version(&self) -> &str { env!("CARGO_PKG_VERSION") }
fn process(
&self,
message: &SpoeMessage,
) -> Result<ProcessingResult, Box<dyn std::error::Error + Send + Sync>> {
// Typed argument lookup (O(1), no allocation)
let src = message.get_string("src").unwrap_or("unknown");
Ok(ProcessingResult::single(
TxnVariable::session("result", SpoeValue::String("processed".into())),
))
}
});Message arguments and return values use SpoeValue:
| Variant | Rust type | Description |
|---|---|---|
Null |
— | No value |
Bool(bool) |
bool |
Boolean |
Int32(i32) |
i32 |
Signed 32-bit integer |
Uint32(u32) |
u32 |
Unsigned 32-bit integer |
Int64(i64) |
i64 |
Signed 64-bit integer |
Uint64(u64) |
u64 |
Unsigned 64-bit integer |
Ipv4([u8; 4]) |
[u8; 4] |
IPv4 address |
Ipv6([u8; 16]) |
[u8; 16] |
IPv6 address |
String(RString) |
FFI-safe string | UTF-8 string |
Binary(RVec<u8>) |
FFI-safe byte vec | Raw bytes |
SpoeMessage provides typed accessors for extracting arguments without manual pattern matching: get(), get_string(), get_int(), get_bool(), get_ipv4(), get_ipv6(), get_binary(). All return Option and perform O(1) lookup.
Variable scopes control lifetime in HAProxy: Process, Session, Transaction, Request, Response. Use TxnVariable::transaction() or TxnVariable::session() as shorthand constructors.
You set variables with plain names (e.g., "country_code"). The hub automatically prefixes them with your plugin name. A plugin named "geoip" setting "country_code" produces the HAProxy variable geoip.country_code.
Plugins receive per-plugin parameters via the [plugins.params] TOML table. All native TOML types are supported — strings, integers, floats, booleans, arrays, and nested tables:
[[plugins]]
name = "geoip"
library = "libgeoip_lookup.so"
messages = ["check-request"]
[plugins.params]
db_path = "/data/GeoLite2-City.mmdb"
locale = "en"
max_results = 10
trusted_cidrs = ["10.0.0.0/8", "172.16.0.0/12"]
[plugins.params.cache]
enabled = true
ttl_seconds = 300These are passed to init() as a PluginContext with a typed config map. Use get_config() to look up keys and ConfigValue accessor methods to extract typed values:
fn init(&mut self, context: &PluginContext) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let db_path = context.get_config("db_path")
.and_then(|v| v.as_str())
.ok_or("missing required config: db_path")?;
self.db = MaxMindDB::open(db_path)?;
// Integer parameters
self.max_results = context.get_config("max_results")
.and_then(|v| v.as_integer())
.unwrap_or(5) as usize;
// Nested table access
if let Some(cache) = context.get_config("cache")
.and_then(|v| v.as_table())
{
self.cache_enabled = cache.get("enabled")
.and_then(|v| v.as_bool())
.unwrap_or(false);
}
Ok(())
}ConfigValue provides typed accessors: as_str(), as_integer(), as_float(), as_bool(), as_array(), and as_table(). Each returns None if the value is a different type.
Plugins can declare a JSON Schema to validate their configuration at startup. If the user-provided config doesn't match the schema, the hub rejects the plugin with a clear error message before init() is called.
Add an optional config_schema block at the end of define_plugin!:
fn config_schema(&self) -> Option<&str> {
Some(r#"{
"type": "object",
"required": ["db_path"],
"properties": {
"db_path": { "type": "string" },
"max_results": { "type": "integer", "minimum": 1 },
"cache": {
"type": "object",
"properties": {
"enabled": { "type": "boolean" },
"ttl_seconds": { "type": "integer", "minimum": 0 }
}
}
}
}"#)
}If the config is invalid, the hub logs an error like:
{"level":"ERROR","message":"failed to load plugin","error":"plugin 'my-plugin' config validation failed: \"db_path\" is a required property"}Plugins that don't declare a schema skip validation entirely (the default).
If your plugin needs output from another plugin, declare it in the hub config:
[[plugins]]
name = "bot-detect"
library = "libbot_detection.so"
messages = ["check-request"]
depends_on = ["ja3"]
timeout_ms = 80Your plugin receives the upstream plugin's output variables as additional entries in message.args, keyed by their prefixed names (e.g., "ja3.hash").
The hub supports hub.-prefixed control args in SPOE messages. These are consumed by the hub and never passed to plugins. The first control arg is hub.skip_plugins: a comma-separated list of plugin names to skip for the current request.
Configure HAProxy to set it conditionally using ACLs:
spoe-message check-request
args src=src method=method path=path hub.skip_plugins=var(txn.skip_plugins)
event on-frontend-http-request
acl allowlisted src -f /etc/haproxy/allowlist.txt
http-request set-var(txn.skip_plugins) str("rate-limiter") if allowlistedSkipped plugins incur zero processing overhead (no thread pool scheduling, no FFI call) and publish <plugin>.__status=skipped. If a skipped or unsuccessfully completed plugin has dependents (declared via depends_on), those dependents are automatically skipped too. This prevents a dependent from running with missing or invalid upstream arguments.
The hub logs skipped plugins at debug level and records a spoa_plugin_skips_total Prometheus counter with a plugin label.
cargo build --release
cp target/release/libmy_spoa_plugin.so /opt/spoa-plugins/Add a [[plugins]] entry to the hub config and either restart the hub or send SIGHUP to reload.
Plugins log with the standard log crate. Add log = "0.4" to your plugin's dependencies and use the standard macros:
log::info!("request processed: path={}", path);
log::warn!("lookup failed for IP: {}", addr);Don't install a logger: define_plugin! installs one that forwards every record to the hub, where it appears in the structured JSON output with "target":"plugin", plugin (the [[plugins]] name), log_target (your module path), the hub's level, and — for records logged inside process() — the request's span.req_id. The hub's logger is in place before init() runs, so a log::set_logger in init fails — and an unwrapping env_logger::init() fails the plugin load. Records above the hub's log_level are dropped before they are formatted; with a target-scoped RUST_LOG (for example haproxy_spoa_hub=info) add plugin=<level>, or forwarded records are filtered out. Requires plugin-api 0.8 or later (api_version 4); a plugin built against an older release loads on the current hub but its records go nowhere. Two [[plugins]] entries that load the same library share one logger, and their records carry whichever name was installed last.
Every message carries a req_id argument. When HAProxy sends one the hub adopts
it; otherwise the hub generates one. Either way the value is present, so a plugin
can read it without a fallback:
let req_id = message.get_string("req_id").unwrap_or_default();
log::warn!("upstream rejected the token (req_id={req_id})");Have HAProxy send its own id so hub logs and the HAProxy access log name the same
request — the value is whatever unique-id-format produces:
spoe-message check-request
args req_id=unique-id method=method path=path
The hub validates an adopted id against [A-Za-z0-9._:-]{1,128} and replaces
anything else with a generated id, so the value is always safe to put in a log
line. Rejections are counted by spoa_request_id_source_total{source="rejected"};
the rejected value itself is never logged, since echoing it is the injection risk
the validation exists to prevent.
Never use req_id as a key. It arrives over the wire and an operator chooses
its format, so two requests can share one. Using it as a cache key, a map key, or
a dedup token lets one request's state serve another — for a WAF transaction cache
that means one request's inspection state applied to a different request. It is a
correlation label. A plugin that needs an internal identity must mint its own.
The hub writes structured JSON to stderr. Key fields:
| Field | Description |
|---|---|
timestamp |
ISO 8601 timestamp |
level |
TRACE, DEBUG, INFO, WARN, ERROR |
message |
Event description |
plugin |
Plugin name (on plugin-related events, including records a plugin logged itself) |
latency_us |
Execution time in microseconds (per-plugin and per-message) |
message_name |
SPOE message name (on dispatch events) |
target |
Rust module path; plugin for records forwarded from a plugin's log macros, whose own module path is in log_target |
Configure the log level via the log_level config field or the RUST_LOG environment variable.
Set metrics_addr in the config to expose a /metrics endpoint:
metrics_addr = "0.0.0.0:9090"| Metric | Type | Labels | Description |
|---|---|---|---|
spoa_messages_total |
counter | message |
Total SPOP messages processed |
spoa_request_id_source_total |
counter | source |
Where each message's correlation id came from: adopted (HAProxy sent a usable req_id), generated (none arrived), rejected (one arrived and failed validation — alert on this: the hub and the access log then name every request differently) |
spoa_message_duration_seconds |
histogram | message |
Per-message processing latency |
spoa_active_connections |
gauge | — | Current active SPOP connections |
spoa_plugin_errors_total |
counter | plugin, kind |
Plugin processing errors (kind: transient or permanent) |
spoa_plugin_timeouts_total |
counter | plugin |
Plugin timeout events |
spoa_plugin_overloads_total |
counter | plugin |
Calls rejected because the per-plugin queue was full |
spoa_plugin_queue_timeouts_total |
counter | plugin |
Queued calls that did not acquire capacity in time |
spoa_plugin_in_flight |
gauge | plugin |
Plugin calls still executing, including calls whose caller timed out |
spoa_plugin_queued |
gauge | plugin |
Calls waiting for a plugin concurrency slot |
spoa_plugin_skips_total |
counter | plugin |
Plugins skipped via hub.skip_plugins |
All metrics carry a service="haproxy-spoa-hub" global label.
- Rust >= 1.86 (edition 2024)
- Docker and Docker Compose (for integration tests)
cargo build --workspace # Build everything
cargo test --workspace # Unit + integration tests (83 tests)
cargo clippy --workspace # Lint (pedantic)
cargo bench -p haproxy-spoa-hub # Criterion benchmarks
cargo doc --workspace --no-deps # Build rustdoc
cd tests/integration && make test # Docker integration tests (HAProxy + hub)Cargo.toml Workspace root
crates/
hub/ Hub daemon (binary + library)
src/
main.rs Entry point, CLI, signal handling
config.rs TOML config parsing and validation
listener.rs Unix + TCP listener setup
connection.rs Per-connection SPOP handling
dispatch.rs Message -> plugin routing + parallel exec
plugin_loader.rs abi_stable plugin loading + ABI check
registry.rs Plugin registry + execution plan builder
metrics.rs Prometheus metrics
plugin-api/ Published plugin API crate
src/
lib.rs Re-exports, define_plugin! macro
types.rs ConfigValue, SpoeMessage, SpoeValue, ProcessingResult
plugin_trait.rs #[sabi_trait] SpoePlugin + PluginMod
examples/
noop/ Example: sets one variable
delay/ Test: configurable sleep
echo/ Test: echoes all args back
tests/integration/ Docker-based e2e tests
- Protocol: The
spopcrate handles SPOE wire protocol encoding/decoding (HAPROXY-HELLO, NOTIFY/ACK, DISCONNECT) - Plugin loading:
abi_stableprovides type-safe FFI with load-time ABI verification viaload_from_file() - Dispatch: Every plugin has a bounded admission gate before
spawn_blocking. Independent plugins run concurrently viajoin_all; a timeout detaches the caller but deliberately keeps the plugin's concurrency permit until the blocking call returns - Registry:
Arc<ArcSwap<PluginRegistry>>enables lock-free reload — each connection loads the current registry snapshot when processing a NOTIFY frame. A reload builds the new registry against the old one, carrying over every plugin instance whoselibraryandparamsare unchanged, swaps it in atomically, then retires the old one in phases (registry::swap_and_retire): wait for in-flight dispatches still holding the pre-swap snapshot (quiesce), wait for the replaced plugins' fire-and-forget background work (drain), then shut the replaced plugins down - Execution plans: Topological sort of plugin dependency graphs per SPOE message. Independent plugins form parallel groups, dependent plugins run sequentially
Licensed under the Apache License 2.0.