fix: aws_ec2_metadata transform when using log namespacing (#17819)

closes: #17193

This keeps the same runtime behavior (if the even is not an object, it
is converted to an object), but fixes the schema definition calculation
to not panic.

The behavior might not be optimal since it may result in data loss.
Users can remap before hand and convert to an object if needed, and
longer term this functionality should be made available in VRL for more
flexibility.

Author: fuchsnj

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 {