diff --git a/src/transforms/aws_ec2_metadata.rs b/src/transforms/aws_ec2_metadata.rs index f78373b7863af..b79fbafccf7a4 100644 --- a/src/transforms/aws_ec2_metadata.rs +++ b/src/transforms/aws_ec2_metadata.rs @@ -16,6 +16,7 @@ use tokio::time::{sleep, Duration, Instant}; use tracing::Instrument; use vector_config::configurable_component; use vector_core::config::LogNamespace; +use vrl::value::kind::Collection; use vrl::value::Kind; use crate::config::OutputId; @@ -274,6 +275,11 @@ impl TransformConfig for Ec2Metadata { .map(|(output, definition)| { let mut schema_definition = definition.clone(); + // If the event is not an object, it will be converted to an object in this transform + if !schema_definition.event_kind().contains_object() { + *schema_definition.event_kind_mut() = Kind::object(Collection::empty()); + } + for path in paths { schema_definition = schema_definition.with_field(path, Kind::bytes().or_undefined(), None); @@ -708,6 +714,38 @@ enum Ec2MetadataError { }, } +#[cfg(test)] +mod test { + use crate::config::schema::Definition; + use crate::config::{LogNamespace, OutputId, TransformConfig}; + use crate::transforms::aws_ec2_metadata::Ec2Metadata; + use enrichment::TableRegistry; + use lookup::OwnedTargetPath; + use vrl::owned_value_path; + use vrl::value::Kind; + + #[tokio::test] + async fn schema_def_with_string_input() { + let transform_config = Ec2Metadata { + namespace: Some(OwnedTargetPath::event(owned_value_path!("ec2", "metadata")).into()), + ..Default::default() + }; + + let input_definition = + Definition::new(Kind::bytes(), Kind::any_object(), [LogNamespace::Vector]); + + let mut outputs = transform_config.outputs( + TableRegistry::default(), + &[(OutputId::dummy(), input_definition)], + LogNamespace::Vector, + ); + assert_eq!(outputs.len(), 1); + let output = outputs.pop().unwrap(); + let actual_schema_def = output.schema_definitions(true)[&OutputId::dummy()].clone(); + assert!(actual_schema_def.event_kind().is_object()); + } +} + #[cfg(feature = "aws-ec2-metadata-integration-tests")] #[cfg(test)] mod integration_tests {