Add Vector internal telemetry pipeline

config/agent/sinks/vector_aggregator.yml

@@ -4,6 +4,7 @@ inputs:
   - tag_stream_kubernetes_container_logs
   - tag_stream_kubernetes_flow_logs
   - tag_stream_kubernetes_host_metrics
+  - tag_stream_vector_agent_metrics
 address: vector-aggregator.vector.svc:8585
 tls:
   enabled: true
@@ -14,3 +15,14 @@ tls:
   verify_hostname: true
 healthcheck:
   enabled: true
+keepalive:
+  interval_secs: 60
+  timeout_secs: 20
+batch:
+  max_bytes: 2097152 # ~2MB
+  max_events: 1000
+  timeout_secs: 1
+buffer:
+  type: disk
+  max_size: 1073741824 # ~1GB
+  when_full: block

config/agent/sources/vector_agent_metrics.yml

@@ -0,0 +1,2 @@
+type: internal_metrics
+scrape_interval_secs: 1

config/agent/transforms/tag_stream_vector_agent_metrics.yml

@@ -0,0 +1,8 @@
+type: remap
+inputs:
+  - vector_agent_metrics
+source: |
+  .tags."@vector.stream" = "metrics.vector.agent"
+  .tags."kubernetes.cluster" = "$${KUBERNETES_CLUSTER_NAME}"
+  .tags.host = "$${VECTOR_SELF_NODE_NAME}"
+  .tags.role = "agent"

config/aggregator/sinks/elasticsearch_logs.yml

@@ -18,7 +18,8 @@ data_stream:
   auto_routing: true
 bulk:
   action: create
+compression: gzip
 batch:
-  max_bytes: 10000000 # ~10MB
+  max_bytes: 10485760 # ~10MB
   max_events: 10000
   timeout_secs: 5

config/aggregator/sinks/elasticsearch_metrics.yml

@@ -2,6 +2,8 @@ type: elasticsearch
 api_version: "v8"
 inputs:
   - tag_elastic_data_stream_kubernetes_host_metrics
+  - tag_elastic_data_stream_vector_agent_metrics
+  - tag_elastic_data_stream_vector_aggregator_metrics
 endpoints:
   - https://elasticsearch-es-http.elastic-stack.svc:9200
 tls:
@@ -18,7 +20,8 @@ data_stream:
   namespace: '{{ .tags."data_stream.namespace" }}'
 bulk:
   action: create
+compression: gzip
 batch:
-  max_bytes: 10000000 # ~10MB
+  max_bytes: 10485760 # ~10MB
   max_events: 20000
-  timeout_secs: 2
+  timeout_secs: 1

config/aggregator/sources/vector_aggregator_metrics.yml

@@ -0,0 +1,2 @@
+type: internal_metrics
+scrape_interval_secs: 1

config/aggregator/transforms/parse_json.yml

@@ -4,6 +4,8 @@ inputs:
   - route_agent_streams.kubernetes_container_logs
   - route_agent_streams.kubernetes_flow_logs
   - route_agent_streams.kubernetes_host_metrics
+  - route_agent_streams.vector_agent_metrics
+  - tag_stream_vector_aggregator_metrics
 source: |
   parsed, err = parse_json(._raw)
   if err == null {

config/aggregator/transforms/route_agent_streams.yml

@@ -14,3 +14,6 @@ route:
   kubernetes_host_metrics:
     type: vrl
     source: .tags."@vector.stream" == "metrics.kubernetes.host"
+  vector_agent_metrics:
+    type: vrl
+    source: .tags."@vector.stream" == "metrics.vector.agent"

config/aggregator/transforms/route_parsed_streams.yml

@@ -18,3 +18,9 @@ route:
   unifi_logs:
     type: vrl
     source: .@vector.stream == "logs.unifi"
+  vector_agent_metrics:
+    type: vrl
+    source: .tags."@vector.stream" == "metrics.vector.agent"
+  vector_aggregator_metrics:
+    type: vrl
+    source: .tags."@vector.stream" == "metrics.vector.aggregator"

config/aggregator/transforms/tag_elastic_data_stream_kubernetes_container_logs.yml

@@ -10,4 +10,4 @@ source: |
     namespace = "default"
   }
 
-  .data_stream.namespace = namespace
+  .data_stream.namespace = replace!(namespace, "-", "_")

config/aggregator/transforms/tag_elastic_data_stream_vector_agent_metrics.yml

@@ -0,0 +1,7 @@
+type: remap
+inputs:
+  - route_parsed_streams.vector_agent_metrics
+source: |
+  .tags."data_stream.type" = "metrics"
+  .tags."data_stream.dataset" = "vector.agent"
+  .tags."data_stream.namespace" = "default"

config/aggregator/transforms/tag_elastic_data_stream_vector_aggregator_metrics.yml

@@ -0,0 +1,7 @@
+type: remap
+inputs:
+  - route_parsed_streams.vector_aggregator_metrics
+source: |
+  .tags."data_stream.type" = "metrics"
+  .tags."data_stream.dataset" = "vector.aggregator"
+  .tags."data_stream.namespace" = "default"

config/aggregator/transforms/tag_stream_vector_aggregator_metrics.yml

@@ -0,0 +1,8 @@
+type: remap
+inputs:
+  - vector_aggregator_metrics
+source: |
+  .tags."@vector.stream" = "metrics.vector.aggregator"
+  .tags."kubernetes.cluster" = "$${KUBERNETES_CLUSTER_NAME}"
+  .tags.host = "$${VECTOR_SELF_NODE_NAME}"
+  .tags.role = "aggregator"

kubernetes/base/agent/values.yml

@@ -40,6 +40,8 @@ args:
 
 # env -- Set environment variables for Vector containers.
 env:
+  - name: VECTOR_LOG
+    value: "info"
   - name: VECTOR_DANGEROUSLY_ALLOW_ENV_VAR_INTERPOLATION
     value: "true"
   - name: KUBERNETES_CLUSTER_TYPE

kubernetes/base/aggregator/values.yml

@@ -12,7 +12,7 @@ role: "Stateless-Aggregator"
 rollWorkload: false
 
 # replicas -- Specify the number of Pods to create. Valid for the "Aggregator" and "Stateless-Aggregator" roles.
-replicas: 1
+replicas: 3
 
 # autoscaling:
 #   enabled: false
@@ -48,6 +48,8 @@ args:
 
 # env -- Set environment variables for Vector containers.
 env:
+  - name: VECTOR_LOG
+    value: "info"
   - name: VECTOR_DANGEROUSLY_ALLOW_ENV_VAR_INTERPOLATION
     value: "true"
   - name: KUBERNETES_CLUSTER_TYPE
@@ -65,7 +67,6 @@ env:
         name: vector-elastic-credentials
         key: password
 
-
 # resources -- Set Vector resource requests and limits.
 resources:
   requests:

tests/agent/test_tag_stream_vector_agent_metrics.yml

@@ -0,0 +1,14 @@
+tests:
+  - name: test_tag_stream_vector_agent_metrics_populates_stream_field
+    inputs:
+      - insert_at: tag_stream_vector_agent_metrics
+        type: log
+        log_fields:
+          message: "{}"
+    outputs:
+      - extract_from: tag_stream_vector_agent_metrics
+        conditions:
+          - type: vrl
+            source: |
+              assert_eq!(.tags."@vector.stream", "metrics.vector.agent")
+              assert_eq!(.tags.role, "agent")

tests/aggregator/test_tag_elastic_data_stream_kubernetes_container_logs.yml

@@ -16,7 +16,7 @@ tests:
             source: |
               assert_eq!(.data_stream.type, "logs")
               assert_eq!(.data_stream.dataset, "kubernetes.container")
-              assert_eq!(.data_stream.namespace, "kube-system")
+              assert_eq!(.data_stream.namespace, "kube_system")
       - extract_from: tag_elastic_data_stream_kubernetes_container_logs
         conditions:
           - type: vrl

tests/aggregator/test_tag_elastic_data_stream_vector_agent_metrics.yml

@@ -0,0 +1,14 @@
+tests:
+  - name: test_tag_elastic_data_stream_vector_agent_metrics_maps_correct_data_stream_fields
+    inputs:
+      - insert_at: tag_elastic_data_stream_vector_agent_metrics
+        type: log
+        log_fields: {}
+    outputs:
+      - extract_from: tag_elastic_data_stream_vector_agent_metrics
+        conditions:
+          - type: vrl
+            source: |
+              assert_eq!(.tags."data_stream.type", "metrics")
+              assert_eq!(.tags."data_stream.dataset", "vector.agent")
+              assert_eq!(.tags."data_stream.namespace", "default")

tests/aggregator/test_tag_elastic_data_stream_vector_aggregator_metrics.yml

@@ -0,0 +1,14 @@
+tests:
+  - name: test_tag_elastic_data_stream_vector_aggregator_metrics_maps_correct_data_stream_fields
+    inputs:
+      - insert_at: tag_elastic_data_stream_vector_aggregator_metrics
+        type: log
+        log_fields: {}
+    outputs:
+      - extract_from: tag_elastic_data_stream_vector_aggregator_metrics
+        conditions:
+          - type: vrl
+            source: |
+              assert_eq!(.tags."data_stream.type", "metrics")
+              assert_eq!(.tags."data_stream.dataset", "vector.aggregator")
+              assert_eq!(.tags."data_stream.namespace", "default")