EAI-5958: Add kubeletstats collector to stamp workload_id on pod CPU/mem metrics

sources/otel-lgtm-stack/v1.0.7/templates/collectors-logs-metrics-k8s.yaml

@@ -722,4 +722,133 @@ spec:
         metrics:
           receivers: [prometheus]
           processors: [memory_limiter, batch, attributes]
+          exporters: [otlp]
+---
+# Per-pod CPU / memory metrics enriched with workload_id.
+# cAdvisor metrics scraped by otel-collector-metrics-k8s are node-scoped, so their
+# pod/namespace arrive as datapoint labels and cannot be associated to a pod by the
+# k8sattributes processor (which works at the resource level). This daemonset uses the
+# kubeletstats receiver, which emits pod/container-scoped metrics with k8s.pod.* as
+# resource attributes, so k8sattributes can look up the pod and stamp its
+# airm.silogen.ai/workload-id label. A transform then copies the identity resource
+# attributes down to datapoint attributes so they survive OTLP ingestion as series
+# labels (workload_id, project_id, pod, namespace, container) — matching how GPU and
+# vLLM metrics already carry workload_id for per-workload observability.
+apiVersion: opentelemetry.io/v1beta1
+kind: OpenTelemetryCollector
+metadata:
+  name: otel-collector-metrics-kubelet
+  namespace: {{ .Release.Namespace }}
+spec:
+  mode: daemonset
+  serviceAccount: otel-collector
+  image: "ghcr.io/open-telemetry/opentelemetry-collector-releases/opentelemetry-collector-contrib:0.113.0"
+  env:
+    # Address the kubelet by node IP, not node name: cluster DNS (CoreDNS) does not
+    # resolve node hostnames here, so ${env:K8S_NODE_NAME}:10250 fails with "no such host".
+    # status.hostIP is always routable.
+    - name: K8S_NODE_IP
+      valueFrom:
+        fieldRef:
+          fieldPath: status.hostIP
+    # Node name is used only as the local k8sattributes watch filter (not for DNS), so it
+    # is safe here despite node hostnames being unresolvable via cluster DNS.
+    - name: K8S_NODE_NAME
+      valueFrom:
+        fieldRef:
+          fieldPath: spec.nodeName
+  podAnnotations:
+    prometheus.io/scrape: "true"
+    prometheus.io/port: "8888"
+  resources:
+    limits:
+      cpu: {{ .Values.collectors.resources.logs.limits.cpu | quote }}
+      memory: {{ .Values.collectors.resources.logs.limits.memory }}
+    requests:
+      cpu: {{ .Values.collectors.resources.logs.requests.cpu | quote }}
+      memory: {{ .Values.collectors.resources.logs.requests.memory }}
+  tolerations:
+    # Run on every node, including tainted GPU/inference nodes, so per-workload CPU/memory
+    # is collected fleet-wide (matches the node-exporter daemonset's scheduling).
+    - operator: Exists
+  config:
+    receivers:
+      kubeletstats:
+        collection_interval: 30s
+        auth_type: serviceAccount
+        endpoint: "https://${env:K8S_NODE_IP}:10250"
+        insecure_skip_verify: true
+        metric_groups:
+          # pod-level only: consumers (AIWB) query the k8s_pod_* series. The container
+          # group would roughly double per-container cardinality (overlapping the existing
+          # cAdvisor scrape) for series nothing reads.
+          - pod
+    processors:
+      memory_limiter:
+        check_interval: 5s
+        limit_percentage: 80
+        spike_limit_percentage: 25
+      batch:
+        send_batch_size: 2000
+        timeout: 10s
+      k8sattributes:
+        auth_type: serviceAccount
+        # Only watch pods on this collector's own node. kubeletstats already scrapes just
+        # the local node's kubelet, so local pod metadata is all the enrichment needs — this
+        # keeps each daemonset pod from holding a cluster-wide pod cache (cost scales with
+        # pods-per-node instead of pods x nodes).
+        filter:
+          node_from_env_var: K8S_NODE_NAME
+        extract:
+          labels:
+          - from: pod
+            key: airm.silogen.ai/workload-id
+            tag_name: workload.id
+          - from: pod
+            key: airm.silogen.ai/project-id
+            tag_name: project.id
+          metadata:
+          - k8s.namespace.name
+          - k8s.pod.name
+          - k8s.pod.uid
+          - k8s.node.name
+          - k8s.container.name
+        passthrough: false
+        pod_association:
+        - sources:
+          - from: resource_attribute
+            name: k8s.pod.uid
+        - sources:
+          - from: resource_attribute
+            name: k8s.pod.name
+          - from: resource_attribute
+            name: k8s.namespace.name
+      # Copy identity resource attributes down to datapoint attributes so they are
+      # emitted as Prometheus series labels (resource attributes otherwise land in
+      # target_info, not on the series itself).
+      transform/workload-id-labels:
+        error_mode: ignore
+        metric_statements:
+          - context: datapoint
+            statements:
+              - set(attributes["workload_id"], resource.attributes["workload.id"]) where resource.attributes["workload.id"] != nil
+              - set(attributes["project_id"], resource.attributes["project.id"]) where resource.attributes["project.id"] != nil
+              - set(attributes["namespace"], resource.attributes["k8s.namespace.name"]) where resource.attributes["k8s.namespace.name"] != nil
+              - set(attributes["pod"], resource.attributes["k8s.pod.name"]) where resource.attributes["k8s.pod.name"] != nil
+              - set(attributes["container"], resource.attributes["k8s.container.name"]) where resource.attributes["k8s.container.name"] != nil
+      attributes:
+        actions:
+          - key: k8s_cluster_name
+            action: insert
+            value: {{ .Values.cluster.name | quote }}
+    exporters:
+      otlp:
+        endpoint: http://lgtm-stack.otel-lgtm-stack.svc.cluster.local:4317
+        tls:
+          insecure: true
+    service:
+      pipelines:
+        metrics:
+          receivers: [kubeletstats]
+          processors: [memory_limiter, batch, k8sattributes, transform/workload-id-labels, attributes]
           exporters: [otlp]

sources/otel-lgtm-stack/v1.0.8/templates/collectors-logs-metrics-k8s.yaml

@@ -724,4 +724,133 @@ spec:
         metrics:
           receivers: [prometheus]
           processors: [memory_limiter, batch, attributes]
-          exporters: [otlp]
+          exporters: [otlp]
+---
+# Per-pod CPU / memory metrics enriched with workload_id.
+# cAdvisor metrics scraped by otel-collector-metrics-k8s are node-scoped, so their
+# pod/namespace arrive as datapoint labels and cannot be associated to a pod by the
+# k8sattributes processor (which works at the resource level). This daemonset uses the
+# kubeletstats receiver, which emits pod/container-scoped metrics with k8s.pod.* as
+# resource attributes, so k8sattributes can look up the pod and stamp its
+# airm.silogen.ai/workload-id label. A transform then copies the identity resource
+# attributes down to datapoint attributes so they survive OTLP ingestion as series
+# labels (workload_id, project_id, pod, namespace, container) — matching how GPU and
+# vLLM metrics already carry workload_id for per-workload observability.
+apiVersion: opentelemetry.io/v1beta1
+kind: OpenTelemetryCollector
+metadata:
+  name: otel-collector-metrics-kubelet
+  namespace: {{ .Release.Namespace }}
+spec:
+  mode: daemonset
+  serviceAccount: otel-collector
+  image: "ghcr.io/open-telemetry/opentelemetry-collector-releases/opentelemetry-collector-contrib:0.113.0"
+  env:
+    # Address the kubelet by node IP, not node name: cluster DNS (CoreDNS) does not
+    # resolve node hostnames here, so ${env:K8S_NODE_NAME}:10250 fails with "no such host".
+    # status.hostIP is always routable.
+    - name: K8S_NODE_IP
+      valueFrom:
+        fieldRef:
+          fieldPath: status.hostIP
+    # Node name is used only as the local k8sattributes watch filter (not for DNS), so it
+    # is safe here despite node hostnames being unresolvable via cluster DNS.
+    - name: K8S_NODE_NAME
+      valueFrom:
+        fieldRef:
+          fieldPath: spec.nodeName
+  podAnnotations:
+    prometheus.io/scrape: "true"
+    prometheus.io/port: "8888"
+  resources:
+    limits:
+      cpu: {{ .Values.collectors.resources.logs.limits.cpu | quote }}
+      memory: {{ .Values.collectors.resources.logs.limits.memory }}
+    requests:
+      cpu: {{ .Values.collectors.resources.logs.requests.cpu | quote }}
+      memory: {{ .Values.collectors.resources.logs.requests.memory }}
+  tolerations:
+    # Run on every node, including tainted GPU/inference nodes, so per-workload CPU/memory
+    # is collected fleet-wide (matches the node-exporter daemonset's scheduling).
+    - operator: Exists
+  config:
+    receivers:
+      kubeletstats:
+        collection_interval: {{ .Values.collectors.scrapeInterval }}
+        auth_type: serviceAccount
+        endpoint: "https://${env:K8S_NODE_IP}:10250"
+        insecure_skip_verify: true
+        metric_groups:
+          # pod-level only: consumers (AIWB) query the k8s_pod_* series. The container
+          # group would roughly double per-container cardinality (overlapping the existing
+          # cAdvisor scrape) for series nothing reads.
+          - pod
+    processors:
+      memory_limiter:
+        check_interval: {{ .Values.collectors.memoryLimiter.checkInterval }}
+        limit_percentage: {{ .Values.collectors.memoryLimiter.limitPercentage }}
+        spike_limit_percentage: {{ .Values.collectors.memoryLimiter.spikeLimitPercentage }}
+      batch:
+        send_batch_size: 2000
+        timeout: 10s
+      k8sattributes:
+        auth_type: serviceAccount
+        # Only watch pods on this collector's own node. kubeletstats already scrapes just
+        # the local node's kubelet, so local pod metadata is all the enrichment needs — this
+        # keeps each daemonset pod from holding a cluster-wide pod cache (cost scales with
+        # pods-per-node instead of pods x nodes).
+        filter:
+          node_from_env_var: K8S_NODE_NAME
+        extract:
+          labels:
+          - from: pod
+            key: airm.silogen.ai/workload-id
+            tag_name: workload.id
+          - from: pod
+            key: airm.silogen.ai/project-id
+            tag_name: project.id
+          metadata:
+          - k8s.namespace.name
+          - k8s.pod.name
+          - k8s.pod.uid
+          - k8s.node.name
+          - k8s.container.name
+        passthrough: false
+        pod_association:
+        - sources:
+          - from: resource_attribute
+            name: k8s.pod.uid
+        - sources:
+          - from: resource_attribute
+            name: k8s.pod.name
+          - from: resource_attribute
+            name: k8s.namespace.name
+      # Copy identity resource attributes down to datapoint attributes so they are
+      # emitted as Prometheus series labels (resource attributes otherwise land in
+      # target_info, not on the series itself).
+      transform/workload-id-labels:
+        error_mode: ignore
+        metric_statements:
+          - context: datapoint
+            statements:
+              - set(attributes["workload_id"], resource.attributes["workload.id"]) where resource.attributes["workload.id"] != nil
+              - set(attributes["project_id"], resource.attributes["project.id"]) where resource.attributes["project.id"] != nil
+              - set(attributes["namespace"], resource.attributes["k8s.namespace.name"]) where resource.attributes["k8s.namespace.name"] != nil
+              - set(attributes["pod"], resource.attributes["k8s.pod.name"]) where resource.attributes["k8s.pod.name"] != nil
+              - set(attributes["container"], resource.attributes["k8s.container.name"]) where resource.attributes["k8s.container.name"] != nil
+      attributes:
+        actions:
+          - key: k8s_cluster_name
+            action: insert
+            value: {{ .Values.cluster.name | quote }}
+    exporters:
+      otlp:
+        endpoint: http://lgtm-stack.otel-lgtm-stack.svc.cluster.local:4317
+        tls:
+          insecure: true
+    service:
+      pipelines:
+        metrics:
+          receivers: [kubeletstats]
+          processors: [memory_limiter, batch, k8sattributes, transform/workload-id-labels, attributes]
+          exporters: [otlp]