fix kubernetes host metrics pipeline

config/agent/transforms/tag_stream_kubernetes_host_metrics.yml

@@ -2,4 +2,4 @@ type: remap
 inputs:
   - kubernetes_host_metrics
 source: |
-  .@vector.stream = "metrics.kubernetes.host"
+  .tags."@vector.stream" = "metrics.kubernetes.host"

config/aggregator/sinks/elasticsearch_metrics.yml

@@ -11,7 +11,10 @@ auth:
   user: "$${ELASTIC_USERNAME}"
   password: "$${ELASTIC_PASSWORD}"
 mode: data_stream
+dangerously_allow_unconfined_template_resolution: true
 data_stream:
-  auto_routing: true
+  type: '{{ .tags."data_stream.type" }}'
+  dataset: '{{ .tags."data_stream.dataset" }}'
+  namespace: '{{ .tags."data_stream.namespace" }}'
 bulk:
   action: create

config/aggregator/transforms/route_agent_streams.yml

@@ -13,4 +13,4 @@ route:
     source: .@vector.stream == "logs.kubernetes.flow"
   kubernetes_host_metrics:
     type: vrl
-    source: .@vector.stream == "metrics.kubernetes.host"
+    source: .tags."@vector.stream" == "metrics.kubernetes.host"

config/aggregator/transforms/route_parsed_streams.yml

@@ -14,7 +14,7 @@ route:
     source: .@vector.stream == "logs.kubernetes.flow"
   kubernetes_host_metrics:
     type: vrl
-    source: .@vector.stream == "metrics.kubernetes.host"
+    source: .tags."@vector.stream" == "metrics.kubernetes.host"
   unifi_logs:
     type: vrl
     source: .@vector.stream == "logs.unifi"

config/aggregator/transforms/tag_elastic_data_stream_kubernetes_host_metrics.yml

@@ -2,6 +2,6 @@ type: remap
 inputs:
   - route_parsed_streams.kubernetes_host_metrics
 source: |
-  .data_stream.type = "metrics"
-  .data_stream.dataset = "kubernetes.host"
-  .data_stream.namespace = "default"
+  .tags."data_stream.type" = "metrics"
+  .tags."data_stream.dataset" = "kubernetes.host"
+  .tags."data_stream.namespace" = "default"

tests/agent/test_tag_stream_kubernetes_host_metrics.yml

@@ -10,4 +10,4 @@ tests:
         conditions:
           - type: vrl
             source: |
-              assert_eq!(.@vector.stream, "metrics.kubernetes.host")
+              assert_eq!(.tags."@vector.stream", "metrics.kubernetes.host")

tests/aggregator/test_tag_elastic_data_stream_kubernetes_host_metrics.yml

@@ -9,6 +9,6 @@ tests:
         conditions:
           - type: vrl
             source: |
-              assert_eq!(.data_stream.type, "metrics")
-              assert_eq!(.data_stream.dataset, "kubernetes.host")
-              assert_eq!(.data_stream.namespace, "default")
+              assert_eq!(.tags."data_stream.type", "metrics")
+              assert_eq!(.tags."data_stream.dataset", "kubernetes.host")
+              assert_eq!(.tags."data_stream.namespace", "default")