fix(cli): validate transform VRL and conditions with --no-environment

vector validate --no-environment now catches VRL compilation errors and
invalid condition expressions in transforms before any environment
resources are available.

Design:
- Add validate(&TransformContext) to TransformConfig; remap and filter
  implement it to compile VRL programs and conditions respectively
- route, exclusive_route, and sample also validate their condition fields
- Introduce validate_transforms(), called in both --no-environment and
  normal validate paths, which builds stub enrichment tables from the
  config so VRL can resolve table name references without loading data
- Remove the validate() call from check_outputs: that function is for
  graph/wiring checks only; transform compilation belongs in
  validate_transforms()
- Short-circuit environment checks when transform validation fails to
  avoid running health checks against external services for known-invalid
  configs
- Add transforms-filter and transforms-aws_ec2_metadata to cli-tests
  feature set so the new integration tests compile correctly
- Convert all CLI integration test configs to YAML; add create_yaml_file
  helper so test files carry the .yaml extension Vector uses for format
  detection

Fixes #15037

Author: pront

changelog.d/15037_validate_no_environment_vrl.fix.md

@@ -0,0 +1,3 @@
+Fixed `vector validate --no-environment` so it reports VRL and condition compilation errors for transforms without requiring full environment-dependent component initialization.
+
+authors: pront

src/conditions/mod.rs

@@ -210,6 +210,14 @@ impl AnyCondition {
             AnyCondition::Map(m) => m.build(enrichment_tables, metrics_storage),
         }
     }
+
+    pub fn validate(
+        &self,
+        enrichment_tables: &vector_lib::enrichment::TableRegistry,
+        metrics_storage: &MetricsStorage,
+    ) -> crate::Result<()> {
+        self.build(enrichment_tables, metrics_storage).map(|_| ())
+    }
 }
 
 impl From<ConditionConfig> for AnyCondition {

src/config/transform.rs

@@ -250,7 +250,7 @@ pub trait TransformConfig: DynClone + NamedComponent + core::fmt::Debug + Send +
     ///
     /// If validation does not succeed, an error variant containing a list of all validation errors
     /// is returned.
-    fn validate(&self, _merged_definition: &schema::Definition) -> Result<(), Vec<String>> {
+    fn validate(&self, _context: &TransformContext) -> Result<(), Vec<String>> {
         Ok(())
     }
 

src/config/validation.rs

@@ -1,5 +1,6 @@
 use std::{collections::HashMap, path::PathBuf};
 
+use crate::transforms::route::UNMATCHED_ROUTE;
 use futures_util::{FutureExt, StreamExt, TryFutureExt, TryStreamExt, stream};
 use heim::{disk::Partition, units::information::byte};
 use indexmap::IndexMap;
@@ -9,7 +10,6 @@ use super::{
     ComponentKey, Config, OutputId, Resource, builder::ConfigBuilder,
     transform::get_transform_output_ids,
 };
-use crate::config::schema;
 
 /// Minimum value (exclusive) for EWMA alpha options.
 /// The alpha value must be strictly greater than this value.
@@ -234,23 +234,22 @@ pub fn check_outputs(config: &ConfigBuilder) -> Result<(), Vec<String>> {
     }
 
     for (key, transform) in config.transforms.iter() {
-        // use the most general definition possible, since the real value isn't known yet.
-        let definition = schema::Definition::any();
-
-        if let Err(errs) = transform.inner.validate(&definition) {
-            errors.extend(errs.into_iter().map(|msg| format!("Transform {key} {msg}")));
-        }
-
-        if get_transform_output_ids(
+        let output_ids = get_transform_output_ids(
             transform.inner.as_ref(),
             key.clone(),
             config.schema.log_namespace(),
         )
-        .any(|output| matches!(output.port, Some(output) if output == DEFAULT_OUTPUT))
-        {
-            errors.push(format!(
-                "Transform {key} cannot have a named output with reserved name: `{DEFAULT_OUTPUT}`"
-            ));
+        .collect::<Vec<_>>();
+
+        for reserved in [DEFAULT_OUTPUT, UNMATCHED_ROUTE] {
+            if output_ids
+                .iter()
+                .any(|output| matches!(&output.port, Some(port) if port == reserved))
+            {
+                errors.push(format!(
+                    "Transform {key} cannot have a named output with reserved name: `{reserved}`"
+                ));
+            }
         }
     }
 

src/transforms/exclusive_route/config.rs

@@ -101,7 +101,7 @@ impl TransformConfig for ExclusiveRouteConfig {
         Input::all()
     }
 
-    fn validate(&self, _: &schema::Definition) -> Result<(), Vec<String>> {
+    fn validate(&self, context: &TransformContext) -> Result<(), Vec<String>> {
         let mut errors = Vec::new();
 
         let mut counts = std::collections::HashMap::new();
@@ -127,6 +127,15 @@ impl TransformConfig for ExclusiveRouteConfig {
             errors.push(format!("Using reserved '{UNMATCHED_ROUTE}' name."));
         }
 
+        for route in &self.routes {
+            if let Err(e) = route
+                .condition
+                .validate(&context.enrichment_tables, &context.metrics_storage)
+            {
+                errors.push(format!("route \"{}\": {e}", route.name));
+            }
+        }
+
         if errors.is_empty() {
             Ok(())
         } else {

src/transforms/filter.rs

@@ -50,6 +50,12 @@ impl TransformConfig for FilterConfig {
         )?)))
     }
 
+    fn validate(&self, context: &TransformContext) -> Result<(), Vec<String>> {
+        self.condition
+            .validate(&context.enrichment_tables, &context.metrics_storage)
+            .map_err(|e| vec![e.to_string()])
+    }
+
     fn input(&self) -> Input {
         Input::all()
     }

src/transforms/remap.rs

@@ -277,6 +277,16 @@ impl TransformConfig for RemapConfig {
         Ok(transform)
     }
 
+    fn validate(&self, context: &TransformContext) -> std::result::Result<(), Vec<String>> {
+        self.compile_vrl_program(
+            context.enrichment_tables.clone(),
+            context.metrics_storage.clone(),
+            context.merged_schema_definition.clone(),
+        )
+        .map(|_| ())
+        .map_err(|e| vec![e.to_string()])
+    }
+
     fn input(&self) -> Input {
         Input::all()
     }

src/transforms/route.rs

@@ -131,13 +131,26 @@ impl TransformConfig for RouteConfig {
         Input::all()
     }
 
-    fn validate(&self, _: &schema::Definition) -> Result<(), Vec<String>> {
+    fn validate(&self, context: &TransformContext) -> Result<(), Vec<String>> {
+        let mut errors = Vec::new();
+
         if self.route.contains_key(UNMATCHED_ROUTE) {
-            Err(vec![format!(
+            errors.push(format!(
                 "cannot have a named output with reserved name: `{UNMATCHED_ROUTE}`"
-            )])
-        } else {
+            ));
+        }
+
+        for (name, condition) in &self.route {
+            if let Err(e) = condition.validate(&context.enrichment_tables, &context.metrics_storage)
+            {
+                errors.push(format!("route \"{name}\": {e}"));
+            }
+        }
+
+        if errors.is_empty() {
             Ok(())
+        } else {
+            Err(errors)
         }
     }
 

src/transforms/sample/config.rs

@@ -222,10 +222,26 @@ impl TransformConfig for SampleConfig {
         Input::new(DataType::Log | DataType::Trace)
     }
 
-    fn validate(&self, _: &schema::Definition) -> Result<(), Vec<String>> {
-        self.sample_rate()
-            .map(|_| ())
-            .map_err(|e| vec![e.to_string()])
+    fn validate(&self, context: &TransformContext) -> Result<(), Vec<String>> {
+        let mut errors = self
+            .sample_rate()
+            .err()
+            .map(|e| vec![e.to_string()])
+            .unwrap_or_default();
+
+        if let Some(Err(e)) = self
+            .exclude
+            .as_ref()
+            .map(|c| c.validate(&context.enrichment_tables, &context.metrics_storage))
+        {
+            errors.push(format!("exclude: {e}"));
+        }
+
+        if errors.is_empty() {
+            Ok(())
+        } else {
+            Err(errors)
+        }
     }
 
     fn outputs(
@@ -317,7 +333,11 @@ mod tests {
             exclude: None,
         };
 
-        assert!(config.validate(&crate::schema::Definition::any()).is_ok());
+        assert!(
+            config
+                .validate(&crate::config::TransformContext::default())
+                .is_ok()
+        );
     }
 
     #[test]

src/validate.rs

@@ -1,19 +1,68 @@
 #![allow(missing_docs)]
 
-use std::{fmt, fs::remove_dir_all, path::PathBuf};
+use std::{collections::HashMap, fmt, fs::remove_dir_all, path::PathBuf};
 
 use clap::Parser;
 use colored::*;
 use exitcode::ExitCode;
+use vector_lib::enrichment::{Case, IndexHandle, TableRegistry};
+use vector_vrl_metrics::MetricsStorage;
+use vrl::value::ObjectMap;
 
 use crate::{
-    config::{self, Config, ConfigDiff, loading::ConfigBuilderLoader},
+    config::{self, Config, ConfigDiff, TransformContext, loading::ConfigBuilderLoader},
+    schema::Definition,
     topology::{
         self,
         builder::{TopologyPieces, TopologyPiecesBuilder},
     },
 };
 
+/// Stub enrichment table used during transform validation so VRL can resolve
+/// table name references without loading actual table data.
+#[derive(Clone)]
+struct StubEnrichmentTable;
+
+impl vector_lib::enrichment::Table for StubEnrichmentTable {
+    fn find_table_row<'a>(
+        &self,
+        _: Case,
+        _: &'a [vector_lib::enrichment::Condition<'a>],
+        _: Option<&[String]>,
+        _: Option<&vrl::value::Value>,
+        _: Option<IndexHandle>,
+    ) -> Result<ObjectMap, vector_lib::enrichment::Error> {
+        unreachable!("stub table is compile-time only")
+    }
+
+    fn find_table_rows<'a>(
+        &self,
+        _: Case,
+        _: &'a [vector_lib::enrichment::Condition<'a>],
+        _: Option<&[String]>,
+        _: Option<&vrl::value::Value>,
+        _: Option<IndexHandle>,
+    ) -> Result<Vec<ObjectMap>, vector_lib::enrichment::Error> {
+        unreachable!("stub table is compile-time only")
+    }
+
+    fn add_index(
+        &mut self,
+        _: Case,
+        _: &[&str],
+    ) -> Result<IndexHandle, vector_lib::enrichment::Error> {
+        Ok(IndexHandle(0))
+    }
+
+    fn index_fields(&self) -> Vec<(Case, Vec<String>)> {
+        vec![]
+    }
+
+    fn needs_reload(&self) -> bool {
+        false
+    }
+}
+
 const TEMPORARY_DIRECTORY: &str = "validate_tmp";
 
 #[derive(Parser, Debug)]
@@ -117,6 +166,8 @@ pub async fn validate(opts: &Opts, color: bool) -> ExitCode {
         None => return exitcode::CONFIG,
     };
 
+    validated &= validate_transforms(&config, &mut fmt).await;
+
     if !opts.no_environment {
         if let Some(tmp_directory) = create_tmp_directory(&mut config, &mut fmt) {
             validated &= validate_environment(opts, &config, &mut fmt).await;
@@ -180,6 +231,73 @@ pub fn validate_config(opts: &Opts, fmt: &mut Formatter) -> Option<Config> {
     Some(config)
 }
 
+async fn validate_transforms(config: &Config, fmt: &mut Formatter) -> bool {
+    let enrichment_tables = TableRegistry::default();
+    // Register stub tables so VRL can resolve configured enrichment table names
+    // without loading actual data. This lets us catch real VRL errors (syntax,
+    // type, wrong table name) while deferring data-loading to the environment phase.
+    let stubs: HashMap<String, Box<dyn vector_lib::enrichment::Table + Send + Sync>> = config
+        .enrichment_tables
+        .keys()
+        .map(|key| {
+            (
+                key.to_string(),
+                Box::new(StubEnrichmentTable)
+                    as Box<dyn vector_lib::enrichment::Table + Send + Sync>,
+            )
+        })
+        .collect();
+    if !stubs.is_empty() {
+        enrichment_tables.load(stubs);
+        // Do not call finish_load(): table_ids() and add_index() (used during
+        // VRL compilation) both operate on the loading stage. finish_load()
+        // would move tables to the ArcSwap and make table_ids() return nothing.
+    }
+    let mut definition_cache = HashMap::new();
+    let mut errors = Vec::new();
+
+    for (key, transform) in config.transforms() {
+        let input_definitions = topology::schema::input_definitions(
+            &transform.inputs,
+            config,
+            enrichment_tables.clone(),
+            &mut definition_cache,
+        )
+        .unwrap_or_default();
+
+        let merged_schema_definition = input_definitions
+            .iter()
+            .map(|(_, definition)| definition.clone())
+            .reduce(Definition::merge)
+            .unwrap_or_else(Definition::any);
+
+        let context = TransformContext {
+            key: Some(key.clone()),
+            globals: config.global.clone(),
+            enrichment_tables: enrichment_tables.clone(),
+            metrics_storage: MetricsStorage::default(),
+            merged_schema_definition,
+            schema: config.schema,
+            ..Default::default()
+        };
+
+        if let Err(errs) = transform.inner.validate(&context) {
+            for err in errs {
+                errors.push(format!("Transform \"{key}\": {err}"));
+            }
+        }
+    }
+
+    if errors.is_empty() {
+        fmt.success("Transforms configuration");
+        true
+    } else {
+        fmt.title("Transform errors");
+        fmt.sub_error(errors);
+        false
+    }
+}
+
 async fn validate_environment(opts: &Opts, config: &Config, fmt: &mut Formatter) -> bool {
     let diff = ConfigDiff::initial(config);