Skip to main content

Module config

Module config 

Source
Expand description

Pipeline configuration: typed framework sections plus opaque per-component passthrough, loaded from YAML with ${VAR:-default} environment interpolation.

The framework owns the typed sections (pipeline, checkpoint, backpressure, metrics) and validates them strictly (deny_unknown_fields at every level). The source, deserializer, and sink sections are single-key mappings selecting a component type; their bodies are opaque ComponentConfigs handed to the component’s factory, which deserializes its own typed config. This keeps connectors fully decoupled from the framework schema.

pipeline: { name: orders, threads: 4, io_threads: 2 }
checkpoint: { interval: 5s, max_pending_batches: 1024 }
backpressure: { max_inflight_bytes: 256MiB }
source:
  kafka:                                   # KafkaSourceConfig
    brokers: ${KAFKA_BROKERS:-localhost:9092}
    topic: orders
    group_id: orders-etl                   # required (no default)
deserializer:
  avro:                                    # AvroSettings (confluent mode)
    registry:
      url: ${SCHEMA_REGISTRY_URL:?schema registry required}
sink:
  clickhouse:                              # ClickHouseSinkConfig
    table: orders_local
    columns: [id, amount, ts]              # required; order is the wire contract
    shards:
      - { replicas: ["http://ch-0-0:8123", "http://ch-0-1:8123"] }
metrics: { exporter: prometheus, listen: 0.0.0.0:9090 }

Environment interpolation runs on the raw text before parsing — see interpolate for the exact semantics of ${VAR}, ${VAR:-default}, ${VAR:?message}, and $$.

§Example

use etl_core::config::PipelineConfig;

let cfg = PipelineConfig::from_str(r#"
pipeline: { name: demo }
source: { memory: {} }
sink: { memory: {} }
"#).unwrap();

assert_eq!(cfg.pipeline.name, "demo");
assert_eq!(cfg.pipeline.io_threads, 2);                    // default
assert_eq!(cfg.checkpoint.max_pending_batches, 1024);      // default
assert_eq!(cfg.source.type_tag(), "memory");

Structs§

BackpressureSection
backpressure: — in-flight budget and hysteresis.
CheckpointSection
checkpoint: — watermark commit policy.
ComponentConfig
An opaque component section: { <type_tag>: { ...connector config... } }.
MetricsSection
metrics: — exporter selection and observability knobs.
PipelineConfig
Root of a pipeline’s configuration file.
PipelineSection
pipeline: — identity and thread budget.

Enums§

ConfigError
Why a pipeline configuration could not be loaded or is invalid.
E2eBasis
Which timestamp anchors end-to-end latency.
MetricsExporter
Metrics exporter selection.
PinningMode
How pipeline threads are pinned to cores.
YamlValue
Re-export of serde_yaml::Value, the opaque body type carried by a ComponentConfig. serde_yaml is a 0.x dependency, so exposing its Value directly in etl-core’s public API would tie our semver to theirs; this alias is the documented exemption (mirroring the bytes and AvroValue re-export pattern — see docs/DESIGN.md § Dependency policy). A major bump of the YAML crate becomes a breaking change here, and only here.