faucet_cli/
lib.rs

1#![cfg_attr(docsrs, feature(doc_cfg))]
2
3//! # faucet-cli
4//!
5//! A config-driven runner for [`faucet-stream`](https://docs.rs/faucet-stream)
6//! pipelines.  Define a source, optional transforms, a sink, and (optionally)
7//! a state store in a YAML or JSON file, then run it with the `faucet` binary —
8//! no Rust code required.
9//!
10//! The library half of this crate exposes the same building blocks the binary
11//! uses (config parsing, env interpolation, the connector registry) so that
12//! integrations and tests can reuse them.
13
14pub mod auth_catalog;
15#[cfg(feature = "catalog")]
16pub mod catalog;
17pub mod cli;
18pub mod commands;
19pub mod compose;
20pub mod config;
21pub mod dlq_replay;
22pub mod env_config;
23pub mod env_loader;
24pub mod error;
25pub mod executor;
26pub mod expand;
27pub mod init_template;
28pub mod interpolate;
29#[cfg(feature = "lineage")]
30pub mod lineage_glue;
31pub mod merge;
32#[cfg(feature = "notify")]
33pub mod notify;
34pub mod obs;
35pub mod pipeline_test;
36pub mod registry;
37pub mod replication;
38#[cfg(feature = "schedule")]
39pub mod schedule;
40pub mod secrets;
41#[cfg(feature = "serve")]
42pub mod serve;
43pub mod sla;
44pub mod state;
45pub mod transforms;
46
47pub use error::{CliError, CliResult};
48
49/// Convenience entry point for integration tests and custom hosts: parse a
50/// YAML config string, expand the matrix, and run all rows.
51///
52/// This skips the `install_observability` call that [`commands::run::run`]
53/// performs — callers can wire their own `metrics` recorder / tracing
54/// subscriber before calling this function (or not at all).
55pub async fn run_from_yaml_str(yaml: &str) -> CliResult<executor::RunSummary> {
56    // Parse first, then resolve ${env}/${file}/${secret} INTO the parsed tree
57    // (post-parse) so a resolved value can never alter the document's structure
58    // (F43) — mirroring the binary's `from_path` path.
59    let mut value: serde_json::Value =
60        serde_yaml::from_str(yaml).map_err(|e| CliError::ParseConfig {
61            path: std::path::PathBuf::from("<yaml-string>"),
62            message: e.to_string(),
63        })?;
64    interpolate::interpolate_value(&mut value)?;
65    let interpolated = serde_yaml::to_string(&value).map_err(|e| CliError::ParseConfig {
66        path: std::path::PathBuf::from("<yaml-string>"),
67        message: e.to_string(),
68    })?;
69    let mut cfg: config::PipelineConfig =
70        serde_yaml::from_str(&interpolated).map_err(|e| CliError::ParseConfig {
71            path: std::path::PathBuf::from("<yaml-string>"),
72            message: e.to_string(),
73        })?;
74    if cfg.version != 1 {
75        return Err(CliError::ParseConfig {
76            path: std::path::PathBuf::from("<yaml-string>"),
77            message: format!(
78                "unsupported pipeline version {}, only version 1 is recognised",
79                cfg.version
80            ),
81        });
82    }
83    crate::secrets::resolve_secrets(&mut cfg).await?;
84    let pipeline_name = cfg.name.clone().unwrap_or_else(|| "unnamed".to_string());
85    let auth = auth_catalog::build_auth_catalog(cfg.auth.as_ref())?;
86    let resilience = match &cfg.resilience {
87        Some(spec) => Some(spec.to_policy()?),
88        None => None,
89    };
90    #[cfg(feature = "catalog")]
91    let catalog = match cfg.catalog.as_ref() {
92        Some(spec) => Some(catalog::connect_from_spec(spec).await?),
93        None => None,
94    };
95    let nodes = expand::expand(&cfg)?;
96    executor::run_expanded(
97        nodes,
98        executor::ExecuteOptions {
99            pipeline_name,
100            execution: cfg.execution.clone(),
101            dry_run: false,
102            limit: None,
103            state_path_override: None,
104            shard: None,
105            auth,
106            clock: chrono::Utc::now().fixed_offset(),
107            cancel: None,
108            resilience,
109            sla: cfg.sla.clone(),
110            #[cfg(feature = "lineage")]
111            lineage: None,
112            #[cfg(feature = "lineage")]
113            lineage_cfg: None,
114            #[cfg(feature = "notify")]
115            notifier: None,
116            #[cfg(feature = "catalog")]
117            catalog,
118        },
119    )
120    .await
121}
faucet_cli/lib.rs

faucet_cli/
lib.rs
