SuperAPI AI Directory
Partner with us中文AI newsFavorites
Skillsstreamling-plugin-basics

streamling-plugin-basics

Coding & development

View repository

Updated 10-16 05:49

About this skill

streamling plugin basics — Agent Skill

A streamling plugin is an independently compiled cdylib loaded at runtime via a stable ABI (abi_stable). A plugin crate can register any mix of six component kinds: source, transform, sink, preprocessor, UDF, and side output. This skill covers the crate skeleton, registration, the lifecycle every component shares, and the cross-cutting utilities (options, metrics, state, async runtime, errors). For the per-kind trait patterns, see the sibling skills.

Read this first, then jump to the specific skill:

Crate setup

A plugin is a Cargo crate compiled to cdylib (+ rlib so tests work) that depends on streamling-plugin.

# Cargo.toml
[package]
name = "my_plugins"
version = "0.1.0"
edition = "2024"

[lib]
name = "my_plugins"
crate-type = ["cdylib", "rlib"]

[dependencies]
streamling-plugin = "0.2.0"          # or path = "../crates/streamling-plugin"
abi_stable = "0.11.3"                # required by the init macro
async-trait = "0.1.83"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
tracing = "0.1"
arrow = "55.2.0"
arrow-schema = "55.2.0"
serde = { version = "1", features = ["derive"] }

crate-type = ["cdylib", "rlib"] is mandatory — cdylib is what streamling loads; rlib lets unit tests link against your code.

Registration (lib.rs)

Register every component with a macro, then invoke exactly one init macro at the bottom:

mod sink;
mod source;

use crate::sink::MySink;
use crate::source::MySource;
use streamling_plugin::{
    init_plugin_with_async_runtime, register_plugin_sink, register_plugin_source,
};

// id can be "namespace.name" (two string args) or just "name" (one string arg).
register_plugin_source!("my_plugin", "rest_source", MySource);
register_plugin_sink!("my_sink", MySink);

init_plugin_with_async_runtime!();

Macro reference

Macro Form Notes
register_plugin_source! ("ns", "name", T) or ("name", T) T::new(rt, state, metrics, opts)
register_plugin_transform! ("ns", "name", T) or ("name", T) T::new(schema, rt, state, metrics, opts)
register_plugin_sink! ("ns", "name", T) or ("name", T) T::new(schema, rt, state, metrics, opts)
register_plugin_preprocessor! ("ns", "name", T) or ("name", T) T::new(opts)
register_plugin_udf! (T) T: ScalarUDFImpl + Default
register_plugin_udf_fn! (create_fn) fn() -> ScalarUDF
register_plugin_side_output! ("id", T) or (T) T: SideOutputPlugin
set_plugin_output_buffer! ("plugin.id", capacity) backpressure cap on the output channel
set_plugin_input_buffer! ("plugin.id", capacity) cap on the input channel
init_plugin_with_async_runtime! () required, exactly once, at the bottom

The plugin.id string is namespace.name when you pass two args, else name. The buffer-setter macros take that exact full id (e.g. set_plugin_output_buffer!("my_plugin.rest_source", 1)).

Two init macros exist. Use init_plugin_with_async_runtime!() — it embeds a real Tokio runtime (DirectTokioProxy), so your plugin can call tokio::time::sleep, tokio::task::spawn_blocking, etc. directly. The bare init_plugin!() variant expects the host to supply the runtime and is rarely what a standalone plugin wants.

Constructor contracts (exact)

The registration macros generate a closure that calls your new() with a fixed argument order. Match it exactly or the macro will fail to compile:

Kind new(...) arguments Return
Source rt, state_backend_factory, metrics_recorder, options Result<Self, PluginInitializationError> or Self
Transform schema, rt, state_backend_factory, metrics_recorder, options Result<Self, _> or Self
Sink schema, rt, state_backend_factory, metrics_recorder, options Result<Self, _> or Self
Preprocessor options Result<Self, _>

Where:

  • rt: PluginAsyncRuntimeObj
  • state_backend_factory: PluginStateBackendFactory
  • metrics_recorder: PluginMetricsRecorder
  • schema: arrow_schema::SchemaRef (the upstream input schema — not passed to sources, which define their own)
  • options: HashMap<String, String> (parsed from the pipeline YAML options: map)

The Result<Self, PluginInitializationError> form is preferred — it lets you reject bad config at create time with a clear Configuration(..) message instead of panicking. (The generator already catches panics and converts them to Configuration errors, but an explicit Result is cleaner.)

Lifecycle

Every source/transform/sink runs the same protocol, driven by streamling over message channels:

initialize()                          # once, before any data
  └─ process_batch(...)               # repeated: source→generate_batch, transform/sink→process_batch
       process_checkpoint_marker(E)   # a checkpoint epoch is starting
       process_checkpoint_finalizer(E)# that epoch is durably committed — safe to flush/advance
  terminate()                         # graceful shutdown

Checkpoint semantics (load-bearing):

  • process_checkpoint_marker(epoch) — a checkpoint beginning. Returning Ok propagates the marker downstream. For sinks this means "an ack should be sent upstream."
  • process_checkpoint_finalizer(epoch) — the epoch is durably committed. This is where you persist durable progress: advance a cursor, flush a buffer, save state. If you maintain resumable progress, do it here, not in process_batch.
  • process_batch returning Err fails the pipeline. Empty batches are a normal "no data right now" signal — see the source skill.

SupportsGracefulShutdown is a required supertrait for source/transform/sink:

fn is_running(&self) -> bool;                 // false → engine stops polling
async fn terminate(&self) -> Result<(), PluginError>;  // release resources, set running=false

The standard idiom is an Arc<AtomicBool> flipped to false in terminate().

Async runtime

rt: PluginAsyncRuntimeObj exposes FFI-safe async primitives:

rt.spawn(fut);                       // spawn a detached task
rt.sleep(RDuration::from_millis(100)).await;   // non-blocking sleep
rt.timeout(RDuration::from_secs(5), fut).await;
rt.block_on(fut);                    // rare; block on a future
rt.yield_now().await;

RDuration comes from abi_stable::std_types. Because init_plugin_with_async_runtime!() installs a real Tokio runtime, you can also use tokio::time::sleep, tokio::task::spawn_blocking, and tokio::sync types directly inside your methods.

Options & secrets

options: HashMap<String, String> arrives flat from the pipeline YAML. For anything non-trivial, centralize parsing in a small helper struct with typed accessors. Secrets must come from environment variables, not YAML — adopt this PluginOptions helper (it checks ENV_PREFIX__KEY before the YAML value and warns on plaintext secrets):

use std::collections::HashMap;
use streamling_plugin::PluginError;
use tracing::warn;

pub struct PluginOptions {
    options: HashMap<String, String>,
    env_prefix: String,
    plugin_name: String,
}

impl PluginOptions {
    pub fn new(options: HashMap<String, String>, plugin_name: &str, env_prefix: &str) -> Self {
        Self { options, env_prefix: env_prefix.to_string(), plugin_name: plugin_name.to_string() }
    }

    /// Required value: env var wins, then YAML, else error.
    pub fn get(&self, key: &str) -> Result<String, PluginError> {
        let env_key = format!("{}__{}", self.env_prefix, key.to_uppercase());
        if let Ok(v) = std::env::var(&env_key) { return Ok(v); }
        self.options.get(key).cloned().ok_or_else(|| {
            PluginError::Internal(format!("{}: required option '{}' is not set", self.plugin_name, key))
        })
    }

    /// Optional value with default.
    pub fn get_or(&self, key: &str, default: &str) -> String {
        let env_key = format!("{}__{}", self.env_prefix, key.to_uppercase());
        std::env::var(&env_key).ok().or_else(|| self.options.get(key).cloned())
            .unwrap_or_else(|| default.to_string())
    }

    /// Secret: env var only; warn if found in plaintext YAML.
    pub fn get_secret(&self, key: &str) -> Option<String> {
        let env_key = format!("{}__{}", self.env_prefix, key.to_uppercase());
        std::env::var(&env_key).ok().or_else(|| {
            self.options.get(key).map(|v| {
                warn!("{key} set in plaintext YAML — prefer env var {env_key}.");
                v.clone()
            })
        })
    }

    pub fn get_usize(&self, key: &str, default: usize) -> usize {
        self.options.get(key).and_then(|s| s.parse().ok()).unwrap_or(default)
    }
}

Convention: prefix is STREAMLING__PLUGIN__<UPPER_NAME> (e.g. STREAMLING__PLUGIN__MY_SINK). Parse numeric/bool options with .parse() and a sensible default — never unwrap() user input.

Metrics & identity labels

metrics_recorder: PluginMetricsRecorder emits custom Prometheus series:

Method Series type
record_count(name, u64) / record_count_w_tags(name, u64, vec![("k","v")]) counter
record_latency(name, Duration) / record_latency_w_tags(name, dur, tags) timing (ms)
record_gauge(name, u64) / record_gauge_w_tags(name, u64, tags) gauge

Metric dispatch is best-effort (channel try_send) — a full channel just logs a warning, never fails the batch.

Identity labels (labels() override on source/transform/sink) declare what this instance is and attach to every built-in metric the node emits. Derive them from options at construction:

fn labels(&self) -> Vec<PluginLabel> {
    vec![PluginLabel::new("dataset", self.dataset_name.clone())]
}

PluginLabel::new(key, value). Keep keys Prometheus-legal and avoid the reserved built-ins (topic, table, url, type, id, operator_type, …).

State backend

state_backend_factory.create::<V>() returns an Arc<PluginStateBackend<V>> for checkpointed, resumable state. V: Serialize + Deserialize + Send + Sync + Clone + Debug + 'static.

let state: Arc<PluginStateBackend<MyState>> = state_backend_factory.create();
let resume: Option<MyState> = state.get().await.map_err(PluginError::State)?;
state.put(MyState { cursor: 123 }).await.map_err(PluginError::State)?;
  • get() / put(v) / remove() — single value under the default key ({reference_name}).
  • get_kv(k) / put_kv(k, v) / remove_kv(k) — keyed values under {prefix}:{k}; set_prefix(Some("")) goes global.
  • Persist durable progress in process_checkpoint_finalizer, restore in initialize. See the source skill for the full resume pattern.

Errors

Two error enums:

// Construction-time (returned from new()):
pub enum PluginInitializationError {
    NotImplemented,
    Configuration(RString),   // bad options/config
    Execution(RString),
}

// Runtime (returned from trait methods):
pub enum PluginError {
    ArrowError(ArrowError),
    IoError(io::Error),
    Internal(String),         // config/setup/programming errors
    Execution(String),        // transient data-processing failures
    State(StateBackendError),
}

Rule of thumb: Internal for "this plugin is misconfigured / something is wrong with me," Execution for "this batch failed but the pipeline can retry." Map external errors with .map_err(|e| PluginError::Internal(format!("..: {e}"))).

Minimal end-to-end skeleton (sink)

// src/sink.rs
use arrow::array::RecordBatch;
use arrow_schema::SchemaRef;
use async_trait::async_trait;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use streamling_plugin::api::{PluginStateBackendFactory, SupportsGracefulShutdown};
use streamling_plugin::r#async::PluginAsyncRuntimeObj;
use streamling_plugin::ffi::PluginMetricsRecorder;
use streamling_plugin::{CheckpointEpoch, PluginError, SinkPlugin};

pub struct PrintSink {
    schema: SchemaRef,
    running: Arc<AtomicBool>,
}

impl PrintSink {
    pub fn new(
        schema: SchemaRef,
        _rt: PluginAsyncRuntimeObj,
        _state: PluginStateBackendFactory,
        _metrics: PluginMetricsRecorder,
        _options: HashMap<String, String>,
    ) -> Self {
        Self { schema, running: Arc::new(AtomicBool::new(true)) }
    }
}

#[async_trait]
impl SupportsGracefulShutdown for PrintSink {
    fn is_running(&self) -> bool { self.running.load(Ordering::SeqCst) }
    async fn terminate(&self) -> Result<(), PluginError> {
        self.running.store(false, Ordering::SeqCst);
        Ok(())
    }
}

#[async_trait]
impl SinkPlugin for PrintSink {
    async fn initialize(&self) -> Result<(), PluginError> { Ok(()) }
    async fn process_batch(&self, batch: RecordBatch) -> Result<(), PluginError> {
        if batch.num_rows() == 0 { return Ok(()); }
        tracing::info!(rows = batch.num_rows(), "received batch");
        Ok(())
    }
    async fn process_checkpoint_marker(&self, _e: CheckpointEpoch) -> Result<(), PluginError> { Ok(()) }
    async fn process_checkpoint_finalizer(&self, _e: CheckpointEpoch) -> Result<(), PluginError> { Ok(()) }
}

Common mistakes

  • Wrong new() arity or order. The macros call a fixed signature (see the table). A source that takes schema first, or a sink that omits it, won't compile.
  • Forgetting init_plugin_with_async_runtime!(). Without it the module exports nothing loadable.
  • Panic in process_batch. It crashes the worker thread. Return Err(PluginError::..) instead.
  • Plaintext secrets in YAML. Use the get_secret env-var pattern.
  • unwrap() on user-supplied options. Parse with defaults; reject bad config via PluginInitializationError::Configuration.
  • Doing durable work in process_batch instead of process_checkpoint_finalizer. Breaks exactly-once — the finalizer is the commit point.

Related skills