fix(docker_logs source): ensure that docker labels are flattened

Fixes: #5716

Previously labels on docker containers were inserted in such a way that
dots in their names would end up creating a nested structure in the log
event due to the semantics of `LogEvent.insert`.

For example:

```json
{
  "container_created_at": "2021-04-16T18:53:19.946155600Z",
  "container_id": "d6bd69d4bc64bef20b4e992dcc23113741067e8268762694f92899504ae14319",
  "container_name": "docker_echo_1",
  "host": "COMP-C02DV25MML87",
  "image": "hashicorp/http-echo:latest",
  "label": {
    "com": {
      "docker": {
        "compose": {
          "config-hash": "e7e5ba19811180f27a7af36667652d0cd686599e6184cb023d9b71d791ff6a1e",
          "container-number": "1",
          "oneoff": "False",
          "project": {
            "config_files": "docker-compose.yml",
            "working_dir": "/private/tmp/docker"
          },
          "service": "echo",
          "version": "1.27.4"
        }
      }
    }
  },
  "message": "2021/04/16 19:14:10 localhost:5678 172.29.0.1:61824 \"GET / HTTP/1.1\" 200 6 \"curl/7.64.1\" 35.6µs",
  "source_type": "docker",
  "stream": "stdout",
  "timestamp": "2021-04-16T19:14:10.400790400Z"
}
```

This change ensures that labels are inserted as-is as keys:

```json
{
  "container_created_at": "2021-04-16T18:53:19.946155600Z",
  "container_id": "d6bd69d4bc64bef20b4e992dcc23113741067e8268762694f92899504ae14319",
  "container_name": "docker_echo_1",
  "host": "COMP-C02DV25MML87",
  "image": "hashicorp/http-echo:latest",
  "label": {
    "com.docker.compose.config-hash": "e7e5ba19811180f27a7af36667652d0cd686599e6184cb023d9b71d791ff6a1e",
    "com.docker.compose.container-number": "1",
    "com.docker.compose.oneoff": "False",
    "com.docker.compose.project": "docker",
    "com.docker.compose.project.config_files": "docker-compose.yml",
    "com.docker.compose.project.working_dir": "/private/tmp/docker",
    "com.docker.compose.service": "echo",
    "com.docker.compose.version": "1.27.4"
  },
  "message": "2021/04/16 19:12:01 localhost:5678 172.29.0.1:61820 \"GET / HTTP/1.1\" 200 6 \"curl/7.64.1\" 18.1µs",
  "source_type": "docker",
  "stream": "stdout",
  "timestamp": "2021-04-16T19:12:01.622769500Z"
}
```

Signed-off-by: Jesse Szwedko <[email protected]>

src/event/value.rs

@@ -271,6 +271,13 @@ impl Value {
         self.as_bytes()
     }
 
+    pub fn as_map(&self) -> Option<&BTreeMap<String, Value>> {
+        match &self {
+            Value::Map(map) => Some(map),
+            _ => None,
+        }
+    }
+
     pub fn as_timestamp(&self) -> Option<&DateTime<Utc>> {
         match &self {
             Value::Timestamp(ts) => Some(ts),

src/sources/docker_logs.rs

@@ -2,7 +2,7 @@ use super::util::MultilineConfig;
 use crate::{
     config::{log_schema, DataType, SourceConfig, SourceContext, SourceDescription},
     event::merge_state::LogEventMergeState,
-    event::{self, Event, LogEvent, Value},
+    event::{self, Event, LogEvent, PathComponent, PathIter, Value},
     internal_events::{
         DockerLogsCommunicationError, DockerLogsContainerEventReceived,
         DockerLogsContainerMetadataFetchFailed, DockerLogsContainerUnwatch,
@@ -885,8 +885,13 @@ impl ContainerLogInfo {
             log_event.insert(CONTAINER, self.id.0.clone());
 
             // Labels.
-            for (key, value) in self.metadata.labels.iter() {
-                log_event.insert(key.clone(), value.clone());
+            if !self.metadata.labels.is_empty() {
+                let prefix_path = PathIter::new("label").collect::<Vec<_>>();
+                for (key, value) in self.metadata.labels.iter() {
+                    let mut path = prefix_path.clone();
+                    path.push(PathComponent::Key(key.clone()));
+                    log_event.insert_path(path, value.clone());
+                }
             }
 
             // Container name.
@@ -961,7 +966,7 @@ impl ContainerLogInfo {
 
 struct ContainerMetadata {
     /// label.key -> String
-    labels: Vec<(String, Value)>,
+    labels: HashMap<String, String>,
     /// name -> String
     name: Value,
     /// name
@@ -978,17 +983,7 @@ impl ContainerMetadata {
         let name = details.name.unwrap();
         let created = details.created.unwrap();
 
-        let labels = config
-            .labels
-            .as_ref()
-            .map(|map| {
-                map.iter()
-                    .map(|(key, value)| {
-                        (("label.".to_owned() + key), Value::from(value.to_owned()))
-                    })
-                    .collect()
-            })
-            .unwrap_or_default();
+        let labels = config.labels.unwrap_or_default();
 
         Ok(ContainerMetadata {
             labels,
@@ -1656,6 +1651,41 @@ mod integration_tests {
         assert!(is_empty(exclude_out));
     }
 
+    #[tokio::test]
+    async fn flat_labels() {
+        trace_init();
+
+        let message = "18";
+        let name = "vector_test_flat_labels";
+        let label = "vector.test.label.flat.labels";
+
+        let docker = docker(None, None).unwrap();
+        let id = running_container(name, Some(label), message, &docker).await;
+        let out = source_with(&[name], None);
+
+        let events = collect_n(out, 1).await;
+        let _ = container_kill(&id, &docker).await;
+        container_remove(&id, &docker).await;
+
+        let log = events[0].as_log();
+        assert_eq!(log[log_schema().message_key()], message.into());
+        assert_eq!(log[&*super::CONTAINER], id.into());
+        assert!(log.get(&*super::CREATED_AT).is_some());
+        assert_eq!(log[&*super::IMAGE], "busybox".into());
+        assert!(log
+            .get("label")
+            .unwrap()
+            .as_map()
+            .unwrap()
+            .get(label)
+            .is_some());
+        assert_eq!(events[0].as_log()[&super::NAME], name.into());
+        assert_eq!(
+            events[0].as_log()[log_schema().source_type_key()],
+            "docker".into()
+        );
+    }
+
     #[tokio::test]
     async fn log_longer_than_16kb() {
         trace_init();