fix(loki): serialize the structured metadata to JSON (#21461)

* feat(loki): serialize the structured metadata

* feat(loki): serialize structured_metadata in json

* feat(loki): properly serialize for the API

* docs(loki): add structured metadata json serialization to changelog

* style(cargo-fmt): run cargo fmt

---------

Co-authored-by: Max Boone <max.boone@maxem.energy>

changelog.d/21443-serialize-loki-structured-metadata-to-json.fix.md

@@ -0,0 +1,3 @@
+Serializes structured metadata to JSON for the Grafana Loki sink
+
+authors: maxboone

src/sinks/loki/event.rs

@@ -137,7 +137,11 @@ pub struct LokiEvent {
 
 impl ByteSizeOf for LokiEvent {
     fn allocated_bytes(&self) -> usize {
-        self.timestamp.allocated_bytes() + self.event.allocated_bytes()
+        self.timestamp.allocated_bytes()
+            + self.event.allocated_bytes()
+            + self.structured_metadata.iter().fold(0, |res, item| {
+                res + item.0.allocated_bytes() + item.1.allocated_bytes()
+            })
     }
 }
 
@@ -146,10 +150,18 @@ impl Serialize for LokiEvent {
     where
         S: serde::Serializer,
     {
-        let mut seq = serializer.serialize_seq(Some(2))?;
+        let mut seq = serializer.serialize_seq(Some(3))?;
         seq.serialize_element(&self.timestamp.to_string())?;
         let event = String::from_utf8_lossy(&self.event);
         seq.serialize_element(&event)?;
+        // Convert structured_metadata into a map structure
+        seq.serialize_element(
+            &self
+                .structured_metadata
+                .iter()
+                .cloned()
+                .collect::<HashMap<String, String>>(),
+        )?;
         seq.end()
     }
 }

src/sinks/loki/tests.rs

@@ -175,3 +175,28 @@ async fn timestamp_out_of_range() {
 
     assert!(sink.encoder.encode_event(e1).is_none());
 }
+
+#[tokio::test]
+async fn structured_metadata_as_json() {
+    let (config, cx) = load_sink::<LokiConfig>(
+        r#"
+        endpoint = "http://localhost:3100"
+        labels = {test = "structured_metadata"}
+        structured_metadata.bar = "{{ foo }}"
+        encoding.codec = "json"
+        encoding.except_fields = ["foo"]
+        "#,
+    )
+    .unwrap();
+    let client = config.build_client(cx).unwrap();
+    let mut sink = LokiSink::new(config, client).unwrap();
+
+    let mut e1 = Event::Log(LogEvent::from("hello world"));
+    e1.as_mut_log().insert("foo", "bar");
+
+    let event = sink.encoder.encode_event(e1).unwrap();
+    let body = serde_json::json!(event.event);
+    let expected_metadata = serde_json::json!({"bar": "bar"});
+
+    assert_eq!(body[2], expected_metadata);
+}