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;
15pub mod cli;
16pub mod commands;
17pub mod compose;
18pub mod config;
19pub mod env_config;
20pub mod env_loader;
21pub mod error;
22pub mod executor;
23pub mod expand;
24pub mod init_template;
25pub mod interpolate;
26#[cfg(feature = "lineage")]
27pub mod lineage_glue;
28pub mod merge;
29pub mod obs;
30pub mod registry;
31pub mod replication;
32#[cfg(feature = "schedule")]
33pub mod schedule;
34pub mod secrets;
35#[cfg(feature = "serve")]
36pub mod serve;
37pub mod state;
38pub mod transforms;
39
40pub use error::{CliError, CliResult};
41
42/// Convenience entry point for integration tests and custom hosts: parse a
43/// YAML config string, expand the matrix, and run all rows.
44///
45/// This skips the `install_observability` call that [`commands::run::run`]
46/// performs — callers can wire their own `metrics` recorder / tracing
47/// subscriber before calling this function (or not at all).
48pub async fn run_from_yaml_str(yaml: &str) -> CliResult<executor::RunSummary> {
49    // Parse first, then resolve ${env}/${file}/${secret} INTO the parsed tree
50    // (post-parse) so a resolved value can never alter the document's structure
51    // (F43) — mirroring the binary's `from_path` path.
52    let mut value: serde_json::Value =
53        serde_yaml::from_str(yaml).map_err(|e| CliError::ParseConfig {
54            path: std::path::PathBuf::from("<yaml-string>"),
55            message: e.to_string(),
56        })?;
57    interpolate::interpolate_value(&mut value)?;
58    let interpolated = serde_yaml::to_string(&value).map_err(|e| CliError::ParseConfig {
59        path: std::path::PathBuf::from("<yaml-string>"),
60        message: e.to_string(),
61    })?;
62    let mut cfg: config::PipelineConfig =
63        serde_yaml::from_str(&interpolated).map_err(|e| CliError::ParseConfig {
64            path: std::path::PathBuf::from("<yaml-string>"),
65            message: e.to_string(),
66        })?;
67    if cfg.version != 1 {
68        return Err(CliError::ParseConfig {
69            path: std::path::PathBuf::from("<yaml-string>"),
70            message: format!(
71                "unsupported pipeline version {}, only version 1 is recognised",
72                cfg.version
73            ),
74        });
75    }
76    crate::secrets::resolve_secrets(&mut cfg).await?;
77    let pipeline_name = cfg.name.clone().unwrap_or_else(|| "unnamed".to_string());
78    let auth = auth_catalog::build_auth_catalog(cfg.auth.as_ref())?;
79    let resilience = match &cfg.resilience {
80        Some(spec) => Some(spec.to_policy()?),
81        None => None,
82    };
83    let nodes = expand::expand(&cfg)?;
84    executor::run_expanded(
85        nodes,
86        executor::ExecuteOptions {
87            pipeline_name,
88            execution: cfg.execution.clone(),
89            dry_run: false,
90            limit: None,
91            state_path_override: None,
92            shard: None,
93            auth,
94            clock: chrono::Utc::now().fixed_offset(),
95            cancel: None,
96            resilience,
97            #[cfg(feature = "lineage")]
98            lineage: None,
99            #[cfg(feature = "lineage")]
100            lineage_cfg: None,
101        },
102    )
103    .await
104}
faucet_cli/lib.rs

faucet_cli/
lib.rs
