Merge pull request #293 from timescale/feat/remove-pipeline-inline-sources

feat!: remove pipeline-embedded sources: blocks

Author: mostafa

.github/workflows/ci.yml

@@ -251,18 +251,20 @@ jobs:
             --pipeline tests/fixtures/dynamic-pipelines/pipelines/multi_format.yml \
             --pipeline tests/fixtures/dynamic-pipelines/pipelines/extract_languages.yml \
             --pipeline tests/fixtures/dynamic-pipelines/pipelines/include_expansion.yml \
+            --source tests/fixtures/dynamic-pipelines/source-files/ \
             --resolve-sources --verbose
       - name: Golden file comparison for rsigma pipeline resolve
         run: |
           failed=0
           for pipeline in tests/fixtures/dynamic-pipelines/pipelines/*.yml; do
             name=$(basename "$pipeline" .yml)
             golden="tests/fixtures/dynamic-pipelines/golden/${name}.json"
+            source_file="tests/fixtures/dynamic-pipelines/source-files/${name}.yml"
             if [ ! -f "$golden" ]; then
               echo "SKIP: no golden file for ${name}"
               continue
             fi
-            actual=$(./target/release/rsigma pipeline resolve --pipeline "$pipeline" --pretty)
+            actual=$(./target/release/rsigma pipeline resolve --pipeline "$pipeline" --source-file "$source_file" --pretty)
             if ! diff -u "$golden" <(echo "$actual") > /tmp/diff_${name}.txt 2>&1; then
               echo "FAIL: ${name} resolve output differs from golden file"
               cat /tmp/diff_${name}.txt

CHANGELOG.md

@@ -4,6 +4,15 @@ All notable changes to RSigma are documented in this file. Each entry correspond
 
 ## [Unreleased]
 
+### Removed pipeline-embedded `sources:` blocks (#293)
+
+Dynamic source declarations no longer live inside pipeline files. A pipeline that still declares an inline `sources:` block is now rejected with a hard parse error that points at `rsigma rule migrate-sources`; source declarations come exclusively from standalone `--source` files, and a pipeline only references them with `${source.<id>}`. This completes the deprecation cycle started in v0.12.0 (#135, visible-deprecated) and continued in v0.13.0 (#136, hidden from docs).
+
+- **Library API.** `rsigma_eval::Pipeline` drops its `sources` field; `Pipeline::is_dynamic()` is now driven purely by `${source.*}` references, and `validate_source_refs` no longer takes a pipeline-local declaration set. `parse_sources` is now exported for tooling that reads a raw `sources:` block. The runtime `RuntimeEngine` gains `set_external_sources`, resolving and expanding references against the external declarations (carried across hot-reload), and `expand_includes` takes the external sources for its remote-include check.
+- **Reference detection fix.** List-valued pipeline `vars` (the common `value_placeholders` shape, e.g. `malicious_commands: ["${source.cmd_list}"]`) are now correctly recognized as dynamic source references; previously only scalar var values were scanned, which the removed inline `sources:` block had masked.
+- **`rule migrate-sources`** reads the inline `sources:` block directly (rather than through the now-rejecting pipeline parser) so it keeps working as the migration path.
+- **Docs and tests** move to the external-only model throughout; the runtime `pipeline_deprecation` module and its stderr warning are gone.
+
 ### Removed the deprecated flat CLI aliases (#292)
 
 The twelve flat top-level subcommands (`eval`, `daemon`, `parse`, `validate`, `lint`, `fields`, `condition`, `stdin`, `convert`, `list-targets`, `list-formats`, `resolve`) are removed. They shipped as visible-deprecated forwarders in v0.12.0 (#124), were hidden from `rsigma --help` in v0.13.0 (#125), and reach end-of-life here. Invoking a removed alias now fails with clap's `unrecognized subcommand` error and lists the available command groups. Use the noun-led groups instead: `engine eval`, `engine daemon`, `rule parse`, `rule validate`, `rule lint`, `rule fields`, `rule condition`, `rule stdin`, `backend convert`, `backend targets`, `backend formats`, and `pipeline resolve`. The per-alias forwarding dispatch and the stderr deprecation warning are gone; the group enums remain the single source of truth for every argument.

crates/rsigma-cli/README.md

@@ -1041,7 +1041,7 @@ cat rule.yml | rsigma rule stdin
 
 ### `rule migrate-sources`: Extract pipeline sources into standalone files
 
-Extract pipeline-embedded `sources:` blocks into standalone source files. Pipeline-embedded sources are deprecated; this tool automates the migration to the `--source` flag.
+Extract pipeline-embedded `sources:` blocks into standalone source files. Pipeline-embedded sources were removed in v1.0; this tool automates the migration to the `--source` flag for any pipeline that still declares them.
 
 | Argument | Type | Default | Description |
 |----------|------|---------|-------------|
@@ -1197,7 +1197,7 @@ rsigma engine daemon -r rules/ -p pipelines/ --source sources.d/   # loads all *
 
 External source files decouple source configuration from pipeline logic, so pipelines stay reusable across environments. Source IDs must be unique across every `--source` file. The flag is repeatable, so multiple files can be combined (each with its own per-team or per-data-source ownership).
 
-> **Deprecated.** Declaring `sources:` inline in a pipeline file is deprecated and will be removed in v1.0 (tracked in [#137](https://github.com/timescale/rsigma/issues/137)). The parser still accepts it but prints a `warning:` line on stderr at every load. Migrate with `rsigma rule migrate-sources -p <dir-or-file> -o sources.yml` and load the result via `--source sources.yml`.
+> **Removed in v1.0.** Declaring `sources:` inline in a pipeline file is no longer accepted (tracked in [#137](https://github.com/timescale/rsigma/issues/137)); the parser rejects such a pipeline with a hard error pointing at the migration tool. Migrate with `rsigma rule migrate-sources -p <dir-or-file> -o sources.yml` and load the result via `--source sources.yml`.
 
 ### Source types
 

crates/rsigma-cli/src/commands/daemon.rs

@@ -1484,28 +1484,15 @@ fn run_daemon(
             process::exit(exit_code::CONFIG_ERROR);
         });
 
-    let pipeline_sources: Vec<_> = pipelines
-        .iter()
-        .flat_map(|p| {
-            p.sources
-                .iter()
-                .map(|s| (s.clone(), p.name.clone()))
-                .collect::<Vec<_>>()
-        })
-        .collect();
-
-    // The pipeline-embedded `sources:` deprecation warning is emitted from
-    // `load_pipelines` (called above), which de-duplicates across hot-reloads
-    // and covers every CLI entry point that loads a pipeline file.
-
-    let source_registry = rsigma_runtime::sources::registry::DaemonSourceRegistry::new(
-        external_sources,
-        pipeline_sources,
-    )
-    .unwrap_or_else(|e| {
-        eprintln!("Source ID collision: {e}");
-        process::exit(exit_code::CONFIG_ERROR);
-    });
+    // Source declarations come exclusively from external `--source` files;
+    // pipelines only reference them. The registry therefore holds only the
+    // external sources.
+    let source_registry =
+        rsigma_runtime::sources::registry::DaemonSourceRegistry::from_external(external_sources)
+            .unwrap_or_else(|e| {
+                eprintln!("Source ID collision: {e}");
+                process::exit(exit_code::CONFIG_ERROR);
+            });
 
     // `value_parser` (and the config schema) restrict this to off/summary/full.
     let match_detail = match_detail

crates/rsigma-cli/src/commands/migrate_sources.rs

@@ -5,7 +5,7 @@ use std::collections::HashMap;
 use std::path::PathBuf;
 
 use clap::Args;
-use rsigma_eval::parse_pipeline_file;
+use rsigma_eval::parse_sources;
 
 /// Arguments for `rsigma rule migrate-sources`.
 #[derive(Args, Debug)]
@@ -79,23 +79,26 @@ pub(crate) fn cmd_migrate_sources(args: MigrateSourcesArgs) {
             }
         };
 
-        let pipeline = match parse_pipeline_file(path) {
-            Ok(p) => p,
+        // The pipeline parser rejects inline `sources:` blocks, so read the
+        // block straight out of the raw YAML instead. A pipeline that no
+        // longer declares sources is simply skipped.
+        let sources = match extract_inline_sources(&content) {
+            Ok(sources) => sources,
             Err(e) => {
-                eprintln!("Error parsing pipeline {}: {e}", path.display());
+                eprintln!("Error parsing sources in {}: {e}", path.display());
                 std::process::exit(crate::exit_code::RULE_ERROR);
             }
         };
 
-        if pipeline.sources.is_empty() {
+        if sources.is_empty() {
             pipelines_without_sources += 1;
             continue;
         }
 
         pipelines_with_sources += 1;
 
         let mut extracted = Vec::new();
-        for source in &pipeline.sources {
+        for source in &sources {
             if let Some(prev_pipeline) = seen_ids.get(&source.id) {
                 eprintln!(
                     "Error: source ID '{}' declared in both '{}' and '{}'. \
@@ -211,6 +214,24 @@ struct ExtractedSource {
     raw_yaml: String,
 }
 
+/// Parse a pipeline file's raw YAML and return its inline `sources:`
+/// declarations, or an empty vector when the file declares none. Errors only
+/// on a malformed `sources:` block (an unknown source type, a missing `id`,
+/// etc.), so a clean, already-migrated pipeline is a no-op.
+fn extract_inline_sources(
+    content: &str,
+) -> Result<Vec<rsigma_eval::pipeline::sources::DynamicSource>, rsigma_eval::EvalError> {
+    let value: yaml_serde::Value = yaml_serde::from_str(content)
+        .map_err(|e| rsigma_eval::EvalError::InvalidModifiers(format!("pipeline YAML: {e}")))?;
+    let Some(node) = value
+        .as_mapping()
+        .and_then(|m| m.get(yaml_serde::Value::String("sources".to_string())))
+    else {
+        return Ok(Vec::new());
+    };
+    parse_sources(node)
+}
+
 /// Extract the raw YAML text for a single source entry from a pipeline file.
 /// Falls back to a simple serialization if the source can't be found by ID.
 fn extract_source_yaml(content: &str, source_id: &str) -> String {

crates/rsigma-cli/src/commands/resolve.rs

@@ -83,6 +83,11 @@ async fn resolve_async(
         }
     }
 
+    // Pipelines only reference sources now; the declarations come from the
+    // `--source-file` flags loaded above. Parse each pipeline so a stale
+    // inline `sources:` block still surfaces its migration error, and note
+    // any pipeline that references no sources (the command's output is
+    // driven entirely by the loaded source declarations either way).
     for path in &pipeline_paths {
         let pipeline = match parse_pipeline_file(path) {
             Ok(p) => p,
@@ -92,25 +97,11 @@ async fn resolve_async(
             }
         };
 
-        if !pipeline.sources.is_empty() {
-            rsigma_runtime::warn_pipeline_inline_sources(path, &pipeline.name);
-        }
-
-        if !pipeline.is_dynamic() && source_files.is_empty() {
+        if !pipeline.is_dynamic() {
             eprintln!(
-                "Pipeline '{}' has no dynamic sources, skipping.",
+                "Pipeline '{}' references no dynamic sources.",
                 pipeline.name
             );
-            continue;
-        }
-
-        for source in &pipeline.sources {
-            if let Some(ref filter) = source_filter
-                && source.id != *filter
-            {
-                continue;
-            }
-            all_sources.push((pipeline.name.clone(), source.clone()));
         }
     }
 

crates/rsigma-cli/src/commands/validate.rs

@@ -171,35 +171,34 @@ fn resolve_validate_sources(
             let mut resolved_pipelines = Vec::with_capacity(pipelines.len());
             let mut source_errors: Vec<String> = Vec::new();
 
-            // Resolve external sources first so they populate the cache
-            if !external_sources.is_empty()
-                && let Err(e) = rt.block_on(rsigma_runtime::sources::resolve_all(
-                    &resolver,
-                    &external_sources,
-                ))
-            {
-                source_errors.push(format!("external sources: {e}"));
-            }
+            // Resolve the external sources once into a shared data map that
+            // every dynamic pipeline's `${source.*}` references expand against.
+            let resolved_data = match rt.block_on(rsigma_runtime::sources::resolve_all(
+                &resolver,
+                &external_sources,
+            )) {
+                Ok(data) => data,
+                Err(e) => {
+                    source_errors.push(format!("external sources: {e}"));
+                    std::collections::HashMap::new()
+                }
+            };
 
             for pipeline in &pipelines {
                 if pipeline.is_dynamic() {
-                    match rt.block_on(rsigma_runtime::sources::resolve_all(
-                        &resolver,
-                        &pipeline.sources,
-                    )) {
-                        Ok(resolved_data) => {
-                            let expanded =
-                                rsigma_runtime::sources::template::TemplateExpander::expand(
-                                    pipeline,
-                                    &resolved_data,
-                                );
-                            resolved_pipelines.push(expanded);
-                        }
-                        Err(e) => {
-                            source_errors.push(format!("pipeline '{}': {e}", pipeline.name));
-                            resolved_pipelines.push(pipeline.clone());
-                        }
+                    let mut expanded = rsigma_runtime::sources::template::TemplateExpander::expand(
+                        pipeline,
+                        &resolved_data,
+                    );
+                    if let Err(e) = rsigma_runtime::sources::include::expand_includes(
+                        &mut expanded,
+                        &resolved_data,
+                        &external_sources,
+                        false,
+                    ) {
+                        source_errors.push(format!("pipeline '{}': {e}", pipeline.name));
                     }
+                    resolved_pipelines.push(expanded);
                 } else {
                     resolved_pipelines.push(pipeline.clone());
                 }

crates/rsigma-cli/src/daemon/server.rs

@@ -384,6 +384,17 @@ pub async fn run_daemon(config: DaemonConfig) {
         let resolver: Arc<dyn rsigma_runtime::sources::SourceResolver> = instrumented;
         engine.set_source_resolver(resolver.clone());
 
+        // Feed the engine the external source declarations so it can resolve
+        // and expand `${source.*}` references in the pipelines.
+        engine.set_external_sources(
+            config
+                .source_registry
+                .sources()
+                .into_iter()
+                .cloned()
+                .collect(),
+        );
+
         // Resolve dynamic sources at startup (blocks on required sources)
         if let Err(e) = engine.resolve_dynamic_pipelines().await {
             tracing::error!(error = %e, "Failed to resolve required dynamic sources at startup");

crates/rsigma-cli/src/main.rs

@@ -568,16 +568,14 @@ pub(crate) fn load_pipelines(paths: &[PathBuf]) -> Vec<Pipeline> {
                 Ok(p) => {
                     eprintln!("Loaded pipeline: {} (priority {})", p.name, p.priority);
                     if p.is_dynamic() {
-                        let source_ids: Vec<&str> =
-                            p.sources.iter().map(|s| s.id.as_str()).collect();
-                        eprintln!("  dynamic source(s): {}", source_ids.join(", "));
-                    }
-                    // The inline-sources deprecation warning lives in
-                    // rsigma-runtime, which is only linked with the `daemon`
-                    // feature. Builds without it cannot resolve sources anyway.
-                    #[cfg(feature = "daemon")]
-                    if !p.sources.is_empty() {
-                        rsigma_runtime::warn_pipeline_inline_sources(path, &p.name);
+                        let source_ids: Vec<&str> = p
+                            .dynamic_references()
+                            .iter()
+                            .map(|r| r.source_id.as_str())
+                            .collect::<std::collections::BTreeSet<&str>>()
+                            .into_iter()
+                            .collect();
+                        eprintln!("  dynamic source ref(s): {}", source_ids.join(", "));
                     }
                     pipelines.push(p);
                 }