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]