fix: aws_ec2_metadata transform when using log namespacing

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 {