From a63a9ae71462ebf51bec143e9e0e83a1535fadb3 Mon Sep 17 00:00:00 2001 From: "matteo.gazzetta" Date: Fri, 31 Jul 2026 10:24:21 +0200 Subject: [PATCH] feat: support OpenTelemetry metrics alongside kube-state-metrics KRR hardcodes kube-state-metrics / cAdvisor / node-exporter metric names. Clusters whose Kubernetes state metrics come from an OpenTelemetry Collector (k8s_cluster + hostmetrics receivers) expose the same information under different names, so KRR finds nothing there. Introduce a metric dialect for Kubernetes *state* metrics, selected with `--prometheus-metrics-dialect {auto,kube-state-metrics,otel}`: | KRR needs | kube-state-metrics | otel | |-----------------|--------------------------------------------------|------------------------------------| | node memory | kube_node_status_capacity{resource="memory"} | system_memory_limit_bytes | | node CPU | kube_node_status_capacity{resource="cpu"} | system_cpu_logical_count | | memory requests | kube_pod_container_resource_requests | k8s_container_memory_request_bytes | | CPU requests | kube_pod_container_resource_requests | k8s_container_cpu_request | | memory limits | kube_pod_container_resource_limits | k8s_container_memory_limit_bytes | | pod phase | kube_pod_status_phase{phase="Running"} == 1 | k8s_pod_phase == 2 (phase enum) | | OOMKills | kube_pod_container_status_last_terminated_reason | k8s_container_status_reason | Container usage metrics (container_cpu_usage_seconds_total, container_memory_working_set_bytes) are deliberately left out of the dialect: they keep their cAdvisor names in every setup, so an OpenTelemetry-only deployment still has to scrape cAdvisor (e.g. via the Collector's prometheus receiver). This keeps the change to state metrics only. The otel dialect requires the Collector to map the Kubernetes resource attributes onto the namespace / pod / container labels, because that is how the state metrics are joined with the cAdvisor usage metrics; a pipeline exporting them verbatim as k8s_namespace_name etc. cannot work. Dialect detection probes for the mapped labels, not only for the metric name, and logs which mapping is missing instead of accepting such a pipeline silently. The README carries the transform processor that does the mapping. The node capacity metrics also change on the kube-state-metrics side, which fixes #420: machine_memory_bytes and machine_cpu_cores are deprecated and are missing on Azure Managed Prometheus, so the cluster summary now reads kube_node_status_capacity, grouped by `node` instead of by `instance`. Pod owner resolution is a separate axis, because the k8s_cluster receiver has no equivalent of the kube_*_owner metrics. Added `--prometheus-owner-resolution {auto,kube-state-metrics,recording-rule}`: - kube-state-metrics: the existing kube_replicaset_owner / kube_replicationcontroller_owner / kube_job_owner / kube_pod_owner chain, including the per-kind handling of Rollout, DeploymentConfig, CronJob and GroupedJob and the KRR_OWNER_BATCH_SIZE batching. - recording-rule: one query against a pod owner recording rule (namespace_workload_pod:kube_pod_owner:relabel by default, configurable via --prometheus-workload-recording-rule). Saves a query per workload on large clusters. It only covers Deployment, DaemonSet, StatefulSet, Job and ReplicaSet, so other kinds fall back to the owner chain instead of silently returning no pods. Both `auto` modes probe kube-state-metrics first and are cached per service, so an existing kube-state-metrics installation keeps sending the queries it sent before, at the cost of one extra probe per cluster. Probes look back one hour (KRR_DIALECT_PROBE_WINDOW) rather than reading an instant vector, so a scrape gap cannot silently switch the dialect for a whole scan. When no dialect is detected, KRR warns and falls back to kube-state-metrics. Also stop reading the metric loader in the no-data warning of gather_data: the loader is unbound when its constructor raised, which turned that path into a NameError. The README now lists every metric KRR queries and which component exposes it, which covers #354, plus a table mapping each one to its OpenTelemetry name. tests/test_metric_dialects.py pins the generated PromQL for both dialects, using the pre-existing kube-state-metrics queries as golden values, and covers detection precedence, the unmapped-label rejection, probe windowing, caching, and the owner resolution fallbacks. Apart from the node capacity metrics, the only change to a kube-state-metrics query is the quote style in the kube-system request queries ('kube-system' -> "kube-system"). Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 54 ++- .../integrations/prometheus/metrics/base.py | 6 + .../integrations/prometheus/metrics/memory.py | 17 +- .../prometheus_metrics_service.py | 245 ++++++++-- robusta_krr/core/models/config.py | 13 + robusta_krr/core/models/metric_dialects.py | 176 +++++++ robusta_krr/main.py | 28 ++ tests/test_metric_dialects.py | 433 ++++++++++++++++++ 8 files changed, 926 insertions(+), 46 deletions(-) create mode 100644 robusta_krr/core/models/metric_dialects.py create mode 100644 tests/test_metric_dialects.py diff --git a/README.md b/README.md index c5f8d7c1..470f74de 100644 --- a/README.md +++ b/README.md @@ -124,7 +124,7 @@ Read more about [how KRR works](#how-krr-works) ### Requirements -KRR requires Prometheus 2.26+, [kube-state-metrics](https://github.com/kubernetes/kube-state-metrics) & [cAdvisor](https://github.com/google/cadvisor). +KRR requires Prometheus 2.26+, [kube-state-metrics](https://github.com/kubernetes/kube-state-metrics) & [cAdvisor](https://github.com/google/cadvisor). cAdvisor is always required; kube-state-metrics can be replaced by an OpenTelemetry Collector, see [Using OpenTelemetry instead of kube-state-metrics](#requirements).
Which metrics does KRR need? @@ -132,13 +132,53 @@ No setup is required if you use kube-prometheus-stack or diff --git a/robusta_krr/core/integrations/prometheus/metrics/base.py b/robusta_krr/core/integrations/prometheus/metrics/base.py index 347e6b93..409ee008 100644 --- a/robusta_krr/core/integrations/prometheus/metrics/base.py +++ b/robusta_krr/core/integrations/prometheus/metrics/base.py @@ -16,6 +16,7 @@ from robusta_krr.core.abstract.metrics import BaseMetric from robusta_krr.core.abstract.strategies import PodsTimeData from robusta_krr.core.models.config import settings +from robusta_krr.core.models.metric_dialects import DEFAULT_DIALECT, MetricDialect from robusta_krr.core.models.objects import K8sObjectData @@ -55,6 +56,9 @@ class PrometheusMetric(BaseMetric): You can override this method to change the way the results are combined. This parameter specifies the maximum number of pods per query. Set to None to disable batching + + `dialect`: the metric naming dialect to use for Kubernetes state metrics. + Defaults to kube-state-metrics naming. """ query_type: QueryType = QueryType.Query @@ -67,9 +71,11 @@ def __init__( prometheus: CustomPrometheusConnect, service_name: str, executor: Optional[ThreadPoolExecutor] = None, + dialect: MetricDialect = DEFAULT_DIALECT, ) -> None: self.prometheus = prometheus self.service_name = service_name + self.dialect = dialect self.executor = executor diff --git a/robusta_krr/core/integrations/prometheus/metrics/memory.py b/robusta_krr/core/integrations/prometheus/metrics/memory.py index 0cbd7e40..40d14f12 100644 --- a/robusta_krr/core/integrations/prometheus/metrics/memory.py +++ b/robusta_krr/core/integrations/prometheus/metrics/memory.py @@ -82,28 +82,29 @@ class MaxOOMKilledMemoryLoader(PrometheusMetric): def get_query(self, object: K8sObjectData, duration: str, step: str) -> str: pods_selector = "|".join(pod.name for pod in object.pods) cluster_label = self.get_prometheus_cluster_label() + dialect = self.dialect + reason_label = dialect.oom_reason_label return f""" max_over_time( max( max( - kube_pod_container_resource_limits{{ - resource="memory", - namespace="{object.namespace}", + {dialect.container_memory_limit}{{ + {dialect.memory_resource_selector}namespace="{object.namespace}", pod=~"{pods_selector}", container="{object.container}" {cluster_label} - }} + }} ) by (pod, container, job) - * on(pod, container, job) group_left(reason) + * on(pod, container, job) group_left({reason_label}) max( - kube_pod_container_status_last_terminated_reason{{ - reason="OOMKilled", + {dialect.oom_reason_metric}{{ + {reason_label}="OOMKilled", namespace="{object.namespace}", pod=~"{pods_selector}", container="{object.container}" {cluster_label} }} - ) by (pod, container, job, reason) + ) by (pod, container, job, {reason_label}) ) by (container, pod, job) [{duration}:{step}] ) diff --git a/robusta_krr/core/integrations/prometheus/metrics_service/prometheus_metrics_service.py b/robusta_krr/core/integrations/prometheus/metrics_service/prometheus_metrics_service.py index bc19dcab..95899c02 100644 --- a/robusta_krr/core/integrations/prometheus/metrics_service/prometheus_metrics_service.py +++ b/robusta_krr/core/integrations/prometheus/metrics_service/prometheus_metrics_service.py @@ -3,7 +3,7 @@ import os from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta -from typing import Iterable, List, Optional, Dict, Any +from typing import Any, Dict, Iterable, List, Optional from kubernetes.client import ApiClient from prometheus_api_client import PrometheusApiClientException @@ -13,6 +13,14 @@ from robusta_krr.core.abstract.strategies import PodsTimeData from robusta_krr.core.integrations import openshift from robusta_krr.core.models.config import settings +from robusta_krr.core.models.metric_dialects import ( + DEFAULT_DIALECT, + DIALECTS, + MetricDialect, + MetricDialectName, + OwnerResolutionName, + get_dialect, +) from robusta_krr.core.models.objects import K8sObjectData, PodData from robusta_krr.utils.batched import batched from robusta_krr.utils.service_discovery import MetricsServiceDiscovery @@ -23,6 +31,13 @@ PROM_REFRESH_CREDS_SEC = int(os.environ.get("PROM_REFRESH_CREDS_SEC", "600")) # 10 minutes +# Workload types covered by the `namespace_workload_pod:kube_pod_owner:relabel` +# recording rule. Any other kind falls back to owner metrics. +RECORDING_RULE_WORKLOAD_TYPES = frozenset({"deployment", "daemonset", "statefulset", "job", "replicaset"}) + +# Lookback used by dialect and owner resolution detection. +DIALECT_PROBE_WINDOW = os.environ.get("KRR_DIALECT_PROBE_WINDOW", "1h") + logger = logging.getLogger("krr") @@ -110,6 +125,10 @@ def __init__( self.prom_config = None self.prometheus = None self._last_init_at = None + self._dialect: Optional[MetricDialect] = None + self._dialect_lock = asyncio.Lock() + self._owner_resolution: Optional[OwnerResolutionName] = None + self._owner_resolution_lock = asyncio.Lock() self.get_prometheus() def get_prometheus(self): @@ -199,6 +218,116 @@ async def get_history_range(self, history_duration: timedelta) -> tuple[datetime logger.debug(f"Returned from get_history_range: {result}") raise ValueError("Error while getting history range") from e + def _single_cluster_label(self) -> str: + # use this for queries with no labels. turn ', cluster="xxx"' to 'cluster="xxx"' + return self.get_prometheus_cluster_label().replace(",", "") + + @staticmethod + def _join_matchers(*matchers: str) -> str: + """Joins PromQL label matchers, skipping the empty ones.""" + + return ", ".join(matcher for matcher in matchers if matcher) + + async def _metric_has_data(self, metric: str, selector: str = "") -> bool: + """ + Checks whether a metric exists in this Prometheus within the probe window. + Used to auto-detect which metric naming dialect this cluster exposes. + + The window matters: an instant vector only sees the Prometheus lookback + delta, so a scrape gap or a collector restart would report the metric as + absent and silently switch the whole scan to another dialect. + """ + + matchers = self._join_matchers(selector, self._single_cluster_label()) + query = f"count(last_over_time({metric}{{ {matchers} }}[{DIALECT_PROBE_WINDOW}]))" + try: + return bool(await self.query(query)) + except Exception as e: + logger.debug(f"Probe query for {metric} failed, assuming it is absent: {e}") + return False + + async def get_dialect(self) -> MetricDialect: + """ + Returns the metric naming dialect to use, detecting it once per service. + """ + + if self._dialect is not None: + return self._dialect + + async with self._dialect_lock: + if self._dialect is None: + self._dialect = await self._detect_dialect() + logger.info(f"Using the '{self._dialect.name.value}' metrics dialect for cluster {self.cluster}") + return self._dialect + + async def _detect_dialect(self) -> MetricDialect: + configured = settings.prometheus_metrics_dialect + if configured != MetricDialectName.AUTO: + return get_dialect(configured) + + for dialect in DIALECTS: + if await self._metric_has_data(dialect.probe_metric, dialect.probe_selector): + return dialect + + # A dialect whose metric is there but whose labels are not is a + # configuration mistake worth naming, rather than a missing metrics source. + for dialect in DIALECTS: + if dialect.probe_selector and await self._metric_has_data(dialect.probe_metric): + logger.warning( + "%s exists but has no %s, so the '%s' dialect cannot be used. Map the Kubernetes resource " + "attributes onto the namespace / pod / container labels in your metrics pipeline, see " + "https://github.com/robusta-dev/krr#requirements", + dialect.probe_metric, + dialect.probe_selector, + dialect.name.value, + ) + + logger.warning( + "Could not detect a metrics dialect (none of %s returned data in the last %s), falling back to '%s'. " + "Set --prometheus-metrics-dialect to select one explicitly.", + ", ".join(dialect.probe_metric for dialect in DIALECTS), + DIALECT_PROBE_WINDOW, + DEFAULT_DIALECT.name.value, + ) + return DEFAULT_DIALECT + + async def get_owner_resolution(self) -> OwnerResolutionName: + """ + Returns the strategy used to map a workload to its pods, detecting it once per service. + """ + + if self._owner_resolution is not None: + return self._owner_resolution + + async with self._owner_resolution_lock: + if self._owner_resolution is None: + self._owner_resolution = await self._detect_owner_resolution() + logger.info( + f"Using the '{self._owner_resolution.value}' pod owner resolution for cluster {self.cluster}" + ) + return self._owner_resolution + + async def _detect_owner_resolution(self) -> OwnerResolutionName: + configured = settings.prometheus_owner_resolution + if configured != OwnerResolutionName.AUTO: + return configured + + # kube-state-metrics owner metrics are the historical behaviour, so they win when present. + if await self._metric_has_data("kube_pod_owner"): + return OwnerResolutionName.KUBE_STATE_METRICS + if await self._metric_has_data(settings.prometheus_workload_recording_rule): + return OwnerResolutionName.RECORDING_RULE + + logger.warning( + "Neither kube_pod_owner nor %s returned data in the last %s, so no workload can be mapped to its " + "historic pods and recommendations will only cover running pods. Falling back to '%s'. Set " + "--prometheus-owner-resolution and --prometheus-workload-recording-rule explicitly to override.", + settings.prometheus_workload_recording_rule, + DIALECT_PROBE_WINDOW, + OwnerResolutionName.KUBE_STATE_METRICS.value, + ) + return OwnerResolutionName.KUBE_STATE_METRICS + async def gather_data( self, object: K8sObjectData, @@ -210,8 +339,9 @@ async def gather_data( ResourceHistoryData: The gathered resource history data. """ logger.debug(f"Gathering {LoaderClass.__name__} metric for {object}") + dialect = await self.get_dialect() try: - metric_loader = LoaderClass(self.get_prometheus(), self.name(), self.executor) + metric_loader = LoaderClass(self.get_prometheus(), self.name(), self.executor, dialect) data = await metric_loader.load_data(object, period, step) except Exception: logger.exception("Failed to gather resource history data for %s", object) @@ -224,9 +354,8 @@ async def gather_data( object.add_warning("NoPrometheusMemoryMetrics") if LoaderClass.warning_on_no_data: - logger.warning( - f"{metric_loader.service_name} returned no {metric_loader.__class__.__name__} metrics for {object}" - ) + # NOTE: not read off the loader, which is unbound when its constructor raised. + logger.warning(f"{self.name()} returned no {LoaderClass.__name__} metrics for {object}") return data @@ -253,20 +382,23 @@ async def query_and_validate(self, prom_query) -> Any: async def get_cluster_summary(self) -> Dict[str, Any]: cluster_label = self.get_prometheus_cluster_label() + single_cluster_label = self._single_cluster_label() + dialect = await self.get_dialect() + + node_memory_matchers = self._join_matchers(dialect.node_memory_selector, single_cluster_label) + node_cpu_matchers = self._join_matchers(dialect.node_cpu_selector, single_cluster_label) - # use this for queries with no labels. turn ', cluster="xxx"' to 'cluster="xxx"' - single_cluster_label = cluster_label.replace(",", "") memory_query = f""" - sum(max by (instance) (machine_memory_bytes{{ {single_cluster_label} }})) + sum(max by ({dialect.node_group_by_label}) ({dialect.node_memory_total}{{ {node_memory_matchers} }})) """ cpu_query = f""" - sum(max by (instance) (machine_cpu_cores{{ {single_cluster_label} }})) + sum(max by ({dialect.node_group_by_label}) ({dialect.node_cpu_total}{{ {node_cpu_matchers} }})) """ kube_system_requests_mem = f""" - sum(max(kube_pod_container_resource_requests{{ namespace='kube-system', resource='memory' {cluster_label} }}) by (job, pod, container) ) + sum(max({dialect.container_memory_request}{{ {dialect.memory_resource_selector}namespace="kube-system" {cluster_label} }}) by (job, pod, container) ) """ kube_system_requests_cpu = f""" - sum(max(kube_pod_container_resource_requests{{ namespace='kube-system', resource='cpu' {cluster_label} }}) by (job, pod, container) ) + sum(max({dialect.container_cpu_request}{{ {dialect.cpu_resource_selector}namespace="kube-system" {cluster_label} }}) by (job, pod, container) ) """ try: cluster_memory_result = await self.query_and_validate(memory_query) @@ -284,26 +416,16 @@ async def get_cluster_summary(self) -> Dict[str, Any]: logger.error(f"Exception occurred while getting cluster summary: {e}") return {} - async def load_pods(self, object: K8sObjectData, period: timedelta) -> list[PodData]: + async def _load_related_pods_by_owner_metrics( + self, object: K8sObjectData, period_literal: str, cluster_label: str + ) -> list[dict]: """ - List pods related to the object and add them to the object's pods list. - Args: - object (K8sObjectData): The Kubernetes object. - period (timedelta): The time period for which to gather data. + Resolves the pods of a workload by walking the kube-state-metrics owner chain. """ - logger.debug(f"Adding historic pods for {object}") - - period_seconds = period.total_seconds() - if period_seconds <= 86400: # one day - hours_literal = min(int(period.total_seconds()) // 3600, 32) - period_literal = f"{hours_literal}h" - else: - days_literal = min(int(period.total_seconds()) // 3600 // 24, 32) - period_literal = f"{days_literal}d" pod_owners: Iterable[str] pod_owner_kind: str - cluster_label = self.get_prometheus_cluster_label() + if object.kind in ["Deployment", "Rollout"]: replicasets = await self.query(f""" kube_replicaset_owner{{ @@ -358,7 +480,7 @@ async def load_pods(self, object: K8sObjectData, period: timedelta) -> list[PodD pod_owners = [object.name] pod_owner_kind = object.kind - related_pods_result = [] + related_pods_result: list[dict] = [] batch_size = int(os.environ.get("KRR_OWNER_BATCH_SIZE", 100)) for owner_group in batched(pod_owners, batch_size): owners_regex = "|".join(owner_group) @@ -373,6 +495,68 @@ async def load_pods(self, object: K8sObjectData, period: timedelta) -> list[PodD ) """) related_pods_result.extend(related_pods_result_item) + + return related_pods_result + + async def _load_related_pods_by_recording_rule( + self, object: K8sObjectData, period_literal: str, cluster_label: str + ) -> list[dict]: + """ + Resolves the pods of a workload with a single query against a pod owner + recording rule (`namespace_workload_pod:kube_pod_owner:relabel` by default). + + This replaces the owner chain with one query, which matters on large + clusters, and it is the only option when raw owner metrics are not + available - for example when Kubernetes state metrics come from the + OpenTelemetry `k8s_cluster` receiver instead of kube-state-metrics. + """ + + return await self.query(f""" + last_over_time( + {settings.prometheus_workload_recording_rule}{{ + workload="{object.name}", + workload_type="{object.kind.lower()}", + namespace="{object.namespace}" + {cluster_label} + }}[{period_literal}] + ) + """) + + async def load_pods(self, object: K8sObjectData, period: timedelta) -> list[PodData]: + """ + List pods related to the object and add them to the object's pods list. + Args: + object (K8sObjectData): The Kubernetes object. + period (timedelta): The time period for which to gather data. + """ + + logger.debug(f"Adding historic pods for {object}") + + period_seconds = int(period.total_seconds()) + if period_seconds <= 86400: # one day + hours_literal = min(period_seconds // 3600, 32) + period_literal = f"{hours_literal}h" + else: + days_literal = min(period_seconds // 3600 // 24, 32) + period_literal = f"{days_literal}d" + + cluster_label = self.get_prometheus_cluster_label() + dialect = await self.get_dialect() + owner_resolution = await self.get_owner_resolution() + + use_recording_rule = owner_resolution == OwnerResolutionName.RECORDING_RULE + if use_recording_rule and object.kind.lower() not in RECORDING_RULE_WORKLOAD_TYPES: + logger.debug( + f"{object.kind} is not covered by {settings.prometheus_workload_recording_rule}, " + "resolving its pods with owner metrics instead" + ) + use_recording_rule = False + + if use_recording_rule: + related_pods_result = await self._load_related_pods_by_recording_rule(object, period_literal, cluster_label) + else: + related_pods_result = await self._load_related_pods_by_owner_metrics(object, period_literal, cluster_label) + if related_pods_result == []: return [] @@ -385,12 +569,11 @@ async def load_pods(self, object: K8sObjectData, period: timedelta) -> list[PodD for pod_group in batched(related_pods, 100): group_regex = "|".join(pod_group) pods_status_result = await self.query(f""" - kube_pod_status_phase{{ - phase="Running", - {related_pod_label}=~"{group_regex}", + {dialect.pod_phase_metric}{{ + {dialect.pod_phase_running_selector}{related_pod_label}=~"{group_regex}", namespace="{object.namespace}" {cluster_label} - }} == 1 + }} == {dialect.pod_phase_running_value} """) current_pods_set |= {pod["metric"][related_pod_label] for pod in pods_status_result} del pods_status_result diff --git a/robusta_krr/core/models/config.py b/robusta_krr/core/models/config.py index e423c6c7..0ad637d6 100644 --- a/robusta_krr/core/models/config.py +++ b/robusta_krr/core/models/config.py @@ -14,6 +14,7 @@ from robusta_krr.core.abstract import formatters from robusta_krr.core.abstract.strategies import AnyStrategy, BaseStrategy +from robusta_krr.core.models.metric_dialects import MetricDialectName, OwnerResolutionName from robusta_krr.core.models.objects import KindLiteral logger = logging.getLogger("krr") @@ -42,6 +43,18 @@ class Config(pd.BaseSettings): prometheus_ssl_enabled: bool = pd.Field(False) prometheus_cluster_label: Optional[str] = pd.Field(None) prometheus_label: Optional[str] = pd.Field(None) + prometheus_metrics_dialect: MetricDialectName = pd.Field( + MetricDialectName.AUTO, + description="Naming dialect of the Kubernetes state metrics in Prometheus", + ) + prometheus_owner_resolution: OwnerResolutionName = pd.Field( + OwnerResolutionName.AUTO, + description="How to map a workload to its pods", + ) + prometheus_workload_recording_rule: str = pd.Field( + "namespace_workload_pod:kube_pod_owner:relabel", + description="Recording rule used by the 'recording-rule' pod owner resolution", + ) eks_managed_prom: bool = pd.Field(False) eks_managed_prom_profile_name: Optional[str] = pd.Field(None) eks_access_key: Optional[str] = pd.Field(None) diff --git a/robusta_krr/core/models/metric_dialects.py b/robusta_krr/core/models/metric_dialects.py new file mode 100644 index 00000000..a3eee5da --- /dev/null +++ b/robusta_krr/core/models/metric_dialects.py @@ -0,0 +1,176 @@ +""" +Metric naming dialects for Kubernetes *state* metrics. + +KRR needs two kinds of metrics: + +1. Container **usage** metrics (``container_cpu_usage_seconds_total``, + ``container_memory_working_set_bytes``). These come from cAdvisor and are + assumed to be present under their cAdvisor names in every supported setup. + An OpenTelemetry Collector can provide them by scraping cAdvisor through the + ``prometheus`` receiver. + +2. Kubernetes **state** metrics (pod phase, pod owners, resource requests and + limits, node capacity, OOMKill reasons). These are named differently + depending on who produced them: + + * ``kube-state-metrics`` + ``cAdvisor``/``node-exporter`` (the default), or + * the OpenTelemetry Collector ``k8s_cluster`` and ``hostmetrics`` receivers. + +Only group 2 is described by a :class:`MetricDialect`, so switching dialects +never changes how usage metrics are queried. + +Every dialect must expose its series under the Prometheus-conventional +``namespace`` / ``pod`` / ``container`` labels, because that is what the cAdvisor +usage metrics use and the two are joined on those labels. A Collector exporting +resource attributes verbatim produces ``k8s_namespace_name`` / ``k8s_pod_name`` / +``k8s_container_name`` instead, which does not work; map them first, see the +OpenTelemetry section of the README for the transform to do it. Dialect +auto-detection checks for the mapped labels, not only for the metric name, so a +Collector without that mapping is reported instead of being silently accepted. +""" + +from __future__ import annotations + +import enum + +import pydantic as pd + + +class MetricDialectName(str, enum.Enum): + AUTO = "auto" + KUBE_STATE_METRICS = "kube-state-metrics" + OTEL = "otel" + + +class OwnerResolutionName(str, enum.Enum): + AUTO = "auto" + KUBE_STATE_METRICS = "kube-state-metrics" + RECORDING_RULE = "recording-rule" + + +class MetricDialect(pd.BaseModel): + """ + Metric and label names of a Kubernetes state metrics source. + + Every field is a metric name, a label name, or a PromQL label-selector + fragment that is interpolated into a query. Selector fragments are used as a + *prefix* inside the label braces, so they must end with ", " when they are + not empty. The node selectors are the exception, see below. + """ + + name: MetricDialectName + + # Node capacity, used for the cluster summary. The node selectors carry no + # trailing comma, they are joined with the cluster label by the caller. + node_memory_total: str + node_cpu_total: str + node_memory_selector: str + node_cpu_selector: str + node_group_by_label: str + + # Container resource requests and limits. + container_memory_request: str + container_cpu_request: str + container_memory_limit: str + # kube-state-metrics encodes the resource in a label, OTel in the metric name. + memory_resource_selector: str + cpu_resource_selector: str + + # Pod phase. + pod_phase_metric: str + pod_phase_running_selector: str + # kube-state-metrics exposes one gauge per phase (1 = in this phase), OTel a + # single gauge holding the phase enum (2 = Running). + pod_phase_running_value: str + + # OOMKill detection. + oom_reason_metric: str + oom_reason_label: str + + # Metric used to detect this dialect when `--prometheus-metrics-dialect auto`, + # plus the label matcher that tells this dialect apart from a source that + # exposes the same metric under unmapped labels. No trailing comma. + probe_metric: str + probe_selector: str + + class Config: + allow_mutation = False + + +KSM_DIALECT = MetricDialect( + name=MetricDialectName.KUBE_STATE_METRICS, + # NOTE: machine_memory_bytes / machine_cpu_cores are deprecated and missing + # in some hosted Prometheus offerings (Azure Managed Prometheus), so the + # kube-state-metrics capacity metric is used instead. + node_memory_total="kube_node_status_capacity", + node_cpu_total="kube_node_status_capacity", + node_memory_selector='resource="memory"', + node_cpu_selector='resource="cpu"', + node_group_by_label="node", + container_memory_request="kube_pod_container_resource_requests", + container_cpu_request="kube_pod_container_resource_requests", + container_memory_limit="kube_pod_container_resource_limits", + memory_resource_selector='resource="memory", ', + cpu_resource_selector='resource="cpu", ', + pod_phase_metric="kube_pod_status_phase", + pod_phase_running_selector='phase="Running", ', + pod_phase_running_value="1", + oom_reason_metric="kube_pod_container_status_last_terminated_reason", + oom_reason_label="reason", + probe_metric="kube_pod_status_phase", + probe_selector="", +) + +# Names produced by the OpenTelemetry Collector: +# node capacity -> hostmetrics receiver (cpu + memory scrapers) +# container resources -> k8s_cluster receiver +# pod phase -> k8s_cluster receiver (k8s.pod.phase) +# OOMKill reason -> k8s_cluster receiver (k8s.container.status.reason, +# disabled by default, must be enabled explicitly) +# k8s.container.status.reason and system.memory.limit are both opt-in, and the +# k8s.* resource attributes have to be mapped onto the namespace / pod / +# container labels. See the README for the Collector configuration. +OTEL_DIALECT = MetricDialect( + name=MetricDialectName.OTEL, + node_memory_total="system_memory_limit_bytes", + node_cpu_total="system_cpu_logical_count", + node_memory_selector="", + node_cpu_selector="", + node_group_by_label="host_name", + container_memory_request="k8s_container_memory_request_bytes", + container_cpu_request="k8s_container_cpu_request", + container_memory_limit="k8s_container_memory_limit_bytes", + memory_resource_selector="", + cpu_resource_selector="", + pod_phase_metric="k8s_pod_phase", + pod_phase_running_selector="", + pod_phase_running_value="2", + oom_reason_metric="k8s_container_status_reason", + oom_reason_label="k8s_container_status_reason", + probe_metric="k8s_pod_phase", + # `namespace` only exists once the Collector maps k8s.namespace.name onto it. + probe_selector='namespace!=""', +) + +# Order matters: auto-detection probes dialects in this order and keeps the +# first one that returns data, so an existing kube-state-metrics setup always +# resolves to the same queries it used before dialects existed. +DIALECTS: tuple[MetricDialect, ...] = (KSM_DIALECT, OTEL_DIALECT) + +DEFAULT_DIALECT = KSM_DIALECT + +_BY_NAME = {dialect.name: dialect for dialect in DIALECTS} + + +def get_dialect(name: MetricDialectName) -> MetricDialect: + """ + Returns the dialect for an explicit dialect name. + + Raises: + ValueError: if called with MetricDialectName.AUTO, which has to be + resolved by probing Prometheus instead. + """ + + if name == MetricDialectName.AUTO: + raise ValueError("MetricDialectName.AUTO must be resolved by probing, not by name") + return _BY_NAME[name] diff --git a/robusta_krr/main.py b/robusta_krr/main.py index c9da3009..c1862126 100644 --- a/robusta_krr/main.py +++ b/robusta_krr/main.py @@ -17,6 +17,7 @@ from robusta_krr.core.abstract import formatters from robusta_krr.core.abstract.strategies import BaseStrategy from robusta_krr.core.models.config import Config +from robusta_krr.core.models.metric_dialects import MetricDialectName, OwnerResolutionName from robusta_krr.core.runner import Runner, publish_input_error from robusta_krr.utils.version import get_version @@ -147,6 +148,30 @@ def run_strategy( help="The label in prometheus used to differentiate clusters. (Only relevant for centralized prometheus)", rich_help_panel="Prometheus Settings", ), + prometheus_metrics_dialect: MetricDialectName = typer.Option( + MetricDialectName.AUTO.value, + "--prometheus-metrics-dialect", + help="Naming of the Kubernetes state metrics: 'kube-state-metrics' (kube_* metrics, cAdvisor and " + "node-exporter), 'otel' (OpenTelemetry Collector k8s_cluster and hostmetrics receivers), or " + "'auto' to detect it. Container usage metrics always use cAdvisor names.", + rich_help_panel="Prometheus Settings", + ), + prometheus_owner_resolution: OwnerResolutionName = typer.Option( + OwnerResolutionName.AUTO.value, + "--prometheus-owner-resolution", + help="How to find the pods of a workload: 'kube-state-metrics' walks the kube_*_owner chain, " + "'recording-rule' uses a single pod owner recording rule, 'auto' prefers the owner chain and " + "falls back to the recording rule.", + rich_help_panel="Prometheus Settings", + ), + prometheus_workload_recording_rule: str = typer.Option( + "namespace_workload_pod:kube_pod_owner:relabel", + "--prometheus-workload-recording-rule", + help="Recording rule used by the 'recording-rule' pod owner resolution. Must expose the " + "'workload', 'workload_type', 'pod' and 'namespace' labels, plus the label passed to " + "--prometheus-cluster-label when that flag is used.", + rich_help_panel="Prometheus Settings", + ), eks_managed_prom: bool = typer.Option( False, "--eks-managed-prom", @@ -371,6 +396,9 @@ def run_strategy( prometheus_ssl_enabled=prometheus_ssl_enabled, prometheus_cluster_label=prometheus_cluster_label, prometheus_label=prometheus_label, + prometheus_metrics_dialect=prometheus_metrics_dialect, + prometheus_owner_resolution=prometheus_owner_resolution, + prometheus_workload_recording_rule=prometheus_workload_recording_rule, eks_managed_prom=eks_managed_prom, eks_managed_prom_region=eks_managed_prom_region, eks_assume_role=eks_assume_role, diff --git a/tests/test_metric_dialects.py b/tests/test_metric_dialects.py new file mode 100644 index 00000000..d515839d --- /dev/null +++ b/tests/test_metric_dialects.py @@ -0,0 +1,433 @@ +""" +Tests for the Kubernetes state metrics dialects and for pod owner resolution. + +The kube-state-metrics expectations are golden copies of the queries KRR sent +before dialects existed, so any accidental change to the default behaviour +fails here. +""" + +import asyncio +import re +from datetime import timedelta + +import pytest + +from robusta_krr.core.integrations.prometheus.metrics import MaxOOMKilledMemoryLoader +from robusta_krr.core.integrations.prometheus.metrics_service.prometheus_metrics_service import PrometheusMetricsService +from robusta_krr.core.models.allocations import ResourceAllocations +from robusta_krr.core.models.config import Config +from robusta_krr.core.models.metric_dialects import ( + KSM_DIALECT, + OTEL_DIALECT, + MetricDialectName, + OwnerResolutionName, + get_dialect, +) +from robusta_krr.core.models.objects import K8sObjectData, PodData + +RECORDING_RULE = "namespace_workload_pod:kube_pod_owner:relabel" +OTEL_PROBE = 'k8s_pod_phase{namespace!=""}' + + +def normalized(query: str) -> str: + return re.sub(r"\s+", " ", query).strip() + + +def make_object(kind: str = "Deployment") -> K8sObjectData: + return K8sObjectData( + cluster="mock-cluster", + name="mock-object-1", + container="mock-container-1", + pods=[PodData(name="mock-pod-1", deleted=False), PodData(name="mock-pod-2", deleted=False)], + namespace="default", + kind=kind, + allocations=ResourceAllocations( + requests={"cpu": 1, "memory": 1}, # type: ignore + limits={"cpu": 2, "memory": 2}, # type: ignore + ), + ) + + +@pytest.fixture +def configure(): + def _configure(**kwargs) -> None: + Config.set_config( + Config( + format="table", + show_cluster_name=False, + strategy="simple", + log_to_stderr=False, + other_args={}, + # NOTE: Config validates its defaults, and the "*" default of + # `resources` is only accepted when it comes from the CLI. + resources=["Deployment"], + **kwargs, + ) + ) + + _configure() + return _configure + + +PROBE_RE = re.compile(r"count\(last_over_time\((?P[^{]+)\{(?P.*)\}\[\w+\]\)\)") + + +class FakePrometheusMetricsService(PrometheusMetricsService): + """ + A PrometheusMetricsService that records queries instead of sending them. + + `existing_metrics` holds what this Prometheus would answer a detection probe + with, spelled exactly as the probe asks for it: `"kube_pod_status_phase"` for + a probe without a selector and `'k8s_pod_phase{namespace!=""}'` for one with. + A metric that exists under unmapped labels is therefore expressed by listing + it without the selector. + """ + + def __init__(self, *, existing_metrics: tuple[str, ...] = (), responses: list | None = None) -> None: + self.cluster = "mock-cluster" + self.api_client = None + self.executor = None + self._dialect = None + self._dialect_lock = asyncio.Lock() + self._owner_resolution = None + self._owner_resolution_lock = asyncio.Lock() + + self.prometheus = None + self.existing_metrics = set(existing_metrics) + self.responses = responses if responses is not None else [] + self.queries: list[str] = [] + + def get_prometheus(self): + return self.prometheus + + async def query(self, query: str) -> list: + self.queries.append(query) + + probe = PROBE_RE.fullmatch(normalized(query)) + if probe: + return [{"metric": {}, "value": [0, "1"]}] if self._probe_key(probe) in self.existing_metrics else [] + + if self.responses: + return self.responses.pop(0) + return [{"metric": {}, "value": [0, "1"]}] + + def _probe_key(self, probe: re.Match) -> str: + metric = probe.group("metric") + # The cluster label is orthogonal to what a probe is asking about. + selector = probe.group("matchers").replace(normalized(self._single_cluster_label()), "").strip(" ,") + return f"{metric}{{{selector}}}" if selector else metric + + @property + def data_queries(self) -> list[str]: + """Queries that are not dialect detection probes.""" + + return [query for query in self.queries if not normalized(query).startswith("count(")] + + +# --------------------- Query building --------------------- # + + +def test_oom_killed_query_with_kube_state_metrics(configure): + query = MaxOOMKilledMemoryLoader(prometheus=None, service_name="Prometheus", dialect=KSM_DIALECT).get_query( + make_object(), "1d", "30m" + ) + + assert normalized(query) == normalized(""" + max_over_time( + max( + max( + kube_pod_container_resource_limits{ + resource="memory", namespace="default", + pod=~"mock-pod-1|mock-pod-2", + container="mock-container-1" + } + ) by (pod, container, job) + * on(pod, container, job) group_left(reason) + max( + kube_pod_container_status_last_terminated_reason{ + reason="OOMKilled", + namespace="default", + pod=~"mock-pod-1|mock-pod-2", + container="mock-container-1" + } + ) by (pod, container, job, reason) + ) by (container, pod, job) + [1d:30m] + ) + """) + + +def test_oom_killed_query_with_otel(configure): + query = MaxOOMKilledMemoryLoader(prometheus=None, service_name="Prometheus", dialect=OTEL_DIALECT).get_query( + make_object(), "1d", "30m" + ) + + assert normalized(query) == normalized(""" + max_over_time( + max( + max( + k8s_container_memory_limit_bytes{ + namespace="default", + pod=~"mock-pod-1|mock-pod-2", + container="mock-container-1" + } + ) by (pod, container, job) + * on(pod, container, job) group_left(k8s_container_status_reason) + max( + k8s_container_status_reason{ + k8s_container_status_reason="OOMKilled", + namespace="default", + pod=~"mock-pod-1|mock-pod-2", + container="mock-container-1" + } + ) by (pod, container, job, k8s_container_status_reason) + ) by (container, pod, job) + [1d:30m] + ) + """) + + +def test_dialect_defaults_to_kube_state_metrics(configure): + loader = MaxOOMKilledMemoryLoader(prometheus=None, service_name="Prometheus") + + assert loader.dialect == KSM_DIALECT + + +@pytest.mark.asyncio +async def test_gather_data_passes_the_detected_dialect_to_the_loader(configure): + service = FakePrometheusMetricsService(existing_metrics=(OTEL_PROBE,)) + seen = {} + + class RecordingLoader(MaxOOMKilledMemoryLoader): + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + seen["dialect"] = self.dialect + + async def load_data(self, object, period, step): + return {} + + await service.gather_data(make_object(), RecordingLoader, timedelta(days=1)) + + assert seen["dialect"] == OTEL_DIALECT + + +@pytest.mark.asyncio +async def test_cluster_summary_queries_with_kube_state_metrics(configure): + service = FakePrometheusMetricsService(existing_metrics=("kube_pod_status_phase",)) + + await service.get_cluster_summary() + + assert [normalized(query) for query in service.data_queries] == [ + 'sum(max by (node) (kube_node_status_capacity{ resource="memory" }))', + 'sum(max by (node) (kube_node_status_capacity{ resource="cpu" }))', + 'sum(max(kube_pod_container_resource_requests{ resource="memory", namespace="kube-system" }) ' + "by (job, pod, container) )", + 'sum(max(kube_pod_container_resource_requests{ resource="cpu", namespace="kube-system" }) ' + "by (job, pod, container) )", + ] + + +@pytest.mark.asyncio +async def test_cluster_summary_queries_with_otel(configure): + service = FakePrometheusMetricsService(existing_metrics=(OTEL_PROBE,)) + + await service.get_cluster_summary() + + assert [normalized(query) for query in service.data_queries] == [ + "sum(max by (host_name) (system_memory_limit_bytes{ }))", + "sum(max by (host_name) (system_cpu_logical_count{ }))", + 'sum(max(k8s_container_memory_request_bytes{ namespace="kube-system" }) by (job, pod, container) )', + 'sum(max(k8s_container_cpu_request{ namespace="kube-system" }) by (job, pod, container) )', + ] + + +@pytest.mark.asyncio +async def test_cluster_summary_keeps_the_cluster_label(configure): + configure( + prometheus_metrics_dialect=MetricDialectName.KUBE_STATE_METRICS, + prometheus_label="cluster", + prometheus_cluster_label="prod", + ) + service = FakePrometheusMetricsService() + + await service.get_cluster_summary() + + queries = [normalized(query) for query in service.data_queries] + assert queries[0] == 'sum(max by (node) (kube_node_status_capacity{ resource="memory", cluster="prod" }))' + assert queries[2] == ( + 'sum(max(kube_pod_container_resource_requests{ resource="memory", namespace="kube-system" , ' + 'cluster="prod" }) by (job, pod, container) )' + ) + + +# --------------------- Dialect detection --------------------- # + + +@pytest.mark.asyncio +async def test_detection_prefers_kube_state_metrics(configure): + service = FakePrometheusMetricsService(existing_metrics=("kube_pod_status_phase", OTEL_PROBE)) + + assert await service.get_dialect() == KSM_DIALECT + + +@pytest.mark.asyncio +async def test_detection_falls_back_to_otel(configure): + service = FakePrometheusMetricsService(existing_metrics=(OTEL_PROBE,)) + + assert await service.get_dialect() == OTEL_DIALECT + + +@pytest.mark.asyncio +async def test_detection_defaults_to_kube_state_metrics_when_nothing_matches(configure): + service = FakePrometheusMetricsService() + + assert await service.get_dialect() == KSM_DIALECT + + +@pytest.mark.asyncio +async def test_otel_is_rejected_when_the_labels_are_not_mapped(configure, caplog): + # k8s_pod_phase exists, but the Collector left the Kubernetes resource + # attributes as k8s_namespace_name etc. instead of mapping them to namespace. + service = FakePrometheusMetricsService(existing_metrics=("k8s_pod_phase",)) + + assert await service.get_dialect() == KSM_DIALECT + assert "k8s_pod_phase exists but has no" in caplog.text + + +@pytest.mark.asyncio +async def test_probes_use_a_lookback_window(configure): + # An instant vector would report a metric as absent after a scrape gap, and + # detection would silently switch dialects for the whole scan. + service = FakePrometheusMetricsService(existing_metrics=(OTEL_PROBE,)) + + await service.get_dialect() + + assert all("last_over_time" in query for query in service.queries) + assert service.queries != [] + + +@pytest.mark.asyncio +async def test_detection_is_cached(configure): + service = FakePrometheusMetricsService(existing_metrics=(OTEL_PROBE,)) + + await service.get_dialect() + probe_count = len(service.queries) + await service.get_dialect() + + assert len(service.queries) == probe_count + + +@pytest.mark.asyncio +async def test_explicit_dialect_is_not_probed(configure): + configure(prometheus_metrics_dialect=MetricDialectName.OTEL) + service = FakePrometheusMetricsService(existing_metrics=("kube_pod_status_phase",)) + + assert await service.get_dialect() == OTEL_DIALECT + assert service.queries == [] + + +def test_get_dialect_rejects_auto(): + with pytest.raises(ValueError): + get_dialect(MetricDialectName.AUTO) + + +# --------------------- Owner resolution --------------------- # + + +@pytest.mark.asyncio +async def test_owner_resolution_prefers_owner_metrics(configure): + service = FakePrometheusMetricsService(existing_metrics=("kube_pod_owner", RECORDING_RULE)) + + assert await service.get_owner_resolution() == OwnerResolutionName.KUBE_STATE_METRICS + + +@pytest.mark.asyncio +async def test_owner_resolution_falls_back_to_recording_rule(configure): + service = FakePrometheusMetricsService(existing_metrics=(RECORDING_RULE,)) + + assert await service.get_owner_resolution() == OwnerResolutionName.RECORDING_RULE + + +@pytest.mark.asyncio +async def test_owner_resolution_defaults_to_owner_metrics(configure, caplog): + service = FakePrometheusMetricsService() + + assert await service.get_owner_resolution() == OwnerResolutionName.KUBE_STATE_METRICS + # Neither source has data, so pod lists will be empty. Say so. + assert "Neither kube_pod_owner nor" in caplog.text + + +@pytest.mark.asyncio +async def test_load_pods_with_owner_metrics(configure): + configure( + prometheus_metrics_dialect=MetricDialectName.KUBE_STATE_METRICS, + prometheus_owner_resolution=OwnerResolutionName.KUBE_STATE_METRICS, + ) + service = FakePrometheusMetricsService( + responses=[ + [{"metric": {"replicaset": "mock-object-1-abc"}}], + [{"metric": {"pod": "mock-pod-1"}}, {"metric": {"pod": "mock-pod-2"}}], + [{"metric": {"pod": "mock-pod-1"}}], + ] + ) + + pods = await service.load_pods(make_object(), timedelta(days=1)) + + assert sorted((pod.name, pod.deleted) for pod in pods) == [("mock-pod-1", False), ("mock-pod-2", True)] + queries = [normalized(query) for query in service.data_queries] + assert queries[0] == ( + 'kube_replicaset_owner{ owner_name="mock-object-1", owner_kind="Deployment", namespace="default" }[24h]' + ) + assert queries[1] == ( + 'last_over_time( kube_pod_owner{ owner_name=~"mock-object-1-abc", owner_kind="ReplicaSet", ' + 'namespace="default" }[24h] )' + ) + assert queries[2] == ( + 'kube_pod_status_phase{ phase="Running", pod=~"mock-pod-1|mock-pod-2", namespace="default" } == 1' + ) + + +@pytest.mark.asyncio +async def test_load_pods_with_recording_rule(configure): + configure( + prometheus_metrics_dialect=MetricDialectName.OTEL, + prometheus_owner_resolution=OwnerResolutionName.RECORDING_RULE, + ) + service = FakePrometheusMetricsService( + responses=[ + [{"metric": {"pod": "mock-pod-1"}}, {"metric": {"pod": "mock-pod-2"}}], + [{"metric": {"pod": "mock-pod-1"}}, {"metric": {"pod": "mock-pod-2"}}], + ] + ) + + pods = await service.load_pods(make_object(), timedelta(days=1)) + + assert sorted((pod.name, pod.deleted) for pod in pods) == [("mock-pod-1", False), ("mock-pod-2", False)] + queries = [normalized(query) for query in service.data_queries] + assert queries[0] == ( + f'last_over_time( {RECORDING_RULE}{{ workload="mock-object-1", workload_type="deployment", ' + 'namespace="default" }[24h] )' + ) + assert queries[1] == 'k8s_pod_phase{ pod=~"mock-pod-1|mock-pod-2", namespace="default" } == 2' + + +@pytest.mark.asyncio +async def test_load_pods_falls_back_for_kinds_missing_from_recording_rule(configure): + configure( + prometheus_metrics_dialect=MetricDialectName.KUBE_STATE_METRICS, + prometheus_owner_resolution=OwnerResolutionName.RECORDING_RULE, + ) + service = FakePrometheusMetricsService(responses=[[], []]) + + assert await service.load_pods(make_object(kind="Rollout"), timedelta(days=1)) == [] + assert normalized(service.data_queries[0]).startswith("kube_replicaset_owner{") + + +@pytest.mark.asyncio +async def test_load_pods_uses_hours_for_sub_day_periods(configure): + configure(prometheus_owner_resolution=OwnerResolutionName.KUBE_STATE_METRICS) + service = FakePrometheusMetricsService(responses=[[], []]) + + await service.load_pods(make_object(kind="StatefulSet"), timedelta(hours=6)) + + assert "[6h]" in service.data_queries[0]