From 6b3b02b072f14911369813fe9c8c87d50527fcf7 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Fri, 9 Oct 2026 13:26:07 +0200 Subject: [PATCH] orchestrator-kubernetes: Fix replica metrics fetches failing on idle connections The Kubernetes client's 60 s read timeout fires on pooled connections that idle for close to 60 s, because hyper-timeout runs the read timer while a connection sits idle and does not reset it when a request is written. The per-minute replica metrics fetch reuses such connections for processes 1+ and fails with `client error (SendRequest)`, leaving their metrics NULL. Raise the read timeout to 120 s, above the pool's 90 s idle expiry. Count failed per-process metrics fetches in `mz_orchestrator_kubernetes_process_metrics_fetch_failures_total`, labeled by fetch step and error kind, and log the full error chain. Closes: CPU-306 Co-Authored-By: Claude Opus 5.5 --- doc/user/data/metrics.yml | 7 ++ src/environmentd/src/environmentd/main.rs | 73 +++++++------ src/orchestrator-kubernetes/Cargo.toml | 3 + src/orchestrator-kubernetes/src/lib.rs | 47 +++++--- src/orchestrator-kubernetes/src/metrics.rs | 101 ++++++++++++++++++ .../src/metrics/tests.rs | 52 +++++++++ src/orchestrator-kubernetes/src/util.rs | 13 ++- 7 files changed, 248 insertions(+), 48 deletions(-) create mode 100644 src/orchestrator-kubernetes/src/metrics.rs create mode 100644 src/orchestrator-kubernetes/src/metrics/tests.rs diff --git a/doc/user/data/metrics.yml b/doc/user/data/metrics.yml index 130383902f05a..e0fa94e3478b8 100644 --- a/doc/user/data/metrics.yml +++ b/doc/user/data/metrics.yml @@ -1687,6 +1687,13 @@ metrics: - case source: src/compute-types/src/plan.rs visibility: internal +- name: mz_orchestrator_kubernetes_process_metrics_fetch_failures_total + help: The number of failed per-process metrics fetches, by fetch step and error kind. + labels: + - kind + - step + source: src/orchestrator-kubernetes/src/metrics.rs + visibility: internal - name: mz_otel_on_close help: count of on_close events sent to otel source: src/ore/src/tracing.rs diff --git a/src/environmentd/src/environmentd/main.rs b/src/environmentd/src/environmentd/main.rs index 13e31c2f095dd..fcc47236e0fb8 100644 --- a/src/environmentd/src/environmentd/main.rs +++ b/src/environmentd/src/environmentd/main.rs @@ -824,41 +824,44 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> { let orchestrator = Arc::new( runtime - .block_on(KubernetesOrchestrator::new(KubernetesOrchestratorConfig { - context: args.orchestrator_kubernetes_context.clone(), - scheduler_name: args.orchestrator_kubernetes_scheduler_name, - priority_class_name: args.orchestrator_kubernetes_priority_class_name, - service_annotations: args - .orchestrator_kubernetes_service_annotation - .into_iter() - .map(|l| (l.key, l.value)) - .collect(), - service_labels: args - .orchestrator_kubernetes_service_label - .into_iter() - .map(|l| (l.key, l.value)) - .collect(), - service_node_selector: args - .orchestrator_kubernetes_service_node_selector - .into_iter() - .map(|l| (l.key, l.value)) - .collect(), - service_affinity: args.orchestrator_kubernetes_service_affinity, - service_tolerations: args.orchestrator_kubernetes_service_tolerations, - service_account: args.orchestrator_kubernetes_service_account, - image_pull_policy: args.orchestrator_kubernetes_image_pull_policy, - aws_external_id_prefix: args.aws_external_id_prefix.clone(), - coverage: args.orchestrator_kubernetes_coverage, - ephemeral_volume_storage_class: args - .orchestrator_kubernetes_ephemeral_volume_class - .clone(), - service_fs_group: args.orchestrator_kubernetes_service_fs_group.clone(), - name_prefix: args.orchestrator_kubernetes_name_prefix.clone(), - collect_pod_metrics: !args - .orchestrator_kubernetes_disable_pod_metrics_collection, - enable_prometheus_scrape_annotations: args - .orchestrator_kubernetes_enable_prometheus_scrape_annotations, - })) + .block_on(KubernetesOrchestrator::new( + KubernetesOrchestratorConfig { + context: args.orchestrator_kubernetes_context.clone(), + scheduler_name: args.orchestrator_kubernetes_scheduler_name, + priority_class_name: args.orchestrator_kubernetes_priority_class_name, + service_annotations: args + .orchestrator_kubernetes_service_annotation + .into_iter() + .map(|l| (l.key, l.value)) + .collect(), + service_labels: args + .orchestrator_kubernetes_service_label + .into_iter() + .map(|l| (l.key, l.value)) + .collect(), + service_node_selector: args + .orchestrator_kubernetes_service_node_selector + .into_iter() + .map(|l| (l.key, l.value)) + .collect(), + service_affinity: args.orchestrator_kubernetes_service_affinity, + service_tolerations: args.orchestrator_kubernetes_service_tolerations, + service_account: args.orchestrator_kubernetes_service_account, + image_pull_policy: args.orchestrator_kubernetes_image_pull_policy, + aws_external_id_prefix: args.aws_external_id_prefix.clone(), + coverage: args.orchestrator_kubernetes_coverage, + ephemeral_volume_storage_class: args + .orchestrator_kubernetes_ephemeral_volume_class + .clone(), + service_fs_group: args.orchestrator_kubernetes_service_fs_group.clone(), + name_prefix: args.orchestrator_kubernetes_name_prefix.clone(), + collect_pod_metrics: !args + .orchestrator_kubernetes_disable_pod_metrics_collection, + enable_prometheus_scrape_annotations: args + .orchestrator_kubernetes_enable_prometheus_scrape_annotations, + }, + &metrics_registry, + )) .context("creating kubernetes orchestrator")?, ); let secrets_controller: Arc = match args.secrets_controller { diff --git a/src/orchestrator-kubernetes/Cargo.toml b/src/orchestrator-kubernetes/Cargo.toml index f7555f04432ff..5a21a9c983115 100644 --- a/src/orchestrator-kubernetes/Cargo.toml +++ b/src/orchestrator-kubernetes/Cargo.toml @@ -32,5 +32,8 @@ sha2.workspace = true tokio.workspace = true tracing.workspace = true +[dev-dependencies] +mz-ore = { path = "../ore", default-features = false, features = ["test"] } + [features] default = [] diff --git a/src/orchestrator-kubernetes/src/lib.rs b/src/orchestrator-kubernetes/src/lib.rs index a4c5efc750338..c40cd24fa8496 100644 --- a/src/orchestrator-kubernetes/src/lib.rs +++ b/src/orchestrator-kubernetes/src/lib.rs @@ -52,6 +52,8 @@ use mz_orchestrator::{ ServiceProcessMetrics, ServiceStatus, recommended_k8s_labels, scheduling_config::*, }; use mz_ore::cast::CastInto; +use mz_ore::error::ErrorExt; +use mz_ore::metrics::MetricsRegistry; use mz_ore::retry::Retry; use mz_ore::task::AbortOnDropHandle; use serde::Deserialize; @@ -59,7 +61,10 @@ use sha2::{Digest, Sha256}; use tokio::sync::{mpsc, oneshot}; use tracing::{error, info, warn}; +use crate::metrics::{FetchStep, OrchestratorMetrics}; + pub mod cloud_resource_controller; +mod metrics; pub mod secrets; pub mod util; @@ -164,6 +169,7 @@ pub struct KubernetesOrchestrator { vpc_endpoint_api: Api, namespaces: Mutex>>, resource_reader: Arc, + metrics: OrchestratorMetrics, } impl fmt::Debug for KubernetesOrchestrator { @@ -176,6 +182,7 @@ impl KubernetesOrchestrator { /// Creates a new Kubernetes orchestrator from the provided configuration. pub async fn new( config: KubernetesOrchestratorConfig, + metrics_registry: &MetricsRegistry, ) -> Result { let (client, kubernetes_namespace) = util::create_client(config.context.clone()).await?; let resource_reader = @@ -188,6 +195,7 @@ impl KubernetesOrchestrator { vpc_endpoint_api: Api::default_namespaced(client), namespaces: Mutex::new(BTreeMap::new()), resource_reader, + metrics: OrchestratorMetrics::register_into(metrics_registry), }) } } @@ -206,6 +214,7 @@ impl Orchestrator for KubernetesOrchestrator { command_rx, name_prefix: self.config.name_prefix.clone().unwrap_or_default(), collect_pod_metrics: self.config.collect_pod_metrics, + metrics: self.metrics.clone(), } .spawn(format!("kubernetes-orchestrator-worker:{namespace}")); @@ -306,6 +315,7 @@ struct OrchestratorWorker { command_rx: mpsc::UnboundedReceiver, name_prefix: String, collect_pod_metrics: bool, + metrics: OrchestratorMetrics, } #[derive(Deserialize, Clone, Debug)] @@ -1565,18 +1575,28 @@ impl OrchestratorWorker { let clusterd_usage_fut = get_clusterd_usage(self_, service_name, i); let (metrics, clusterd_usage) = - match futures::future::join(self_.metrics_api.get(&name), clusterd_usage_fut).await - { - (Ok(metrics), Ok(clusterd_usage)) => (metrics, Some(clusterd_usage)), - (Ok(metrics), Err(e)) => { - warn!("Failed to fetch clusterd usage for {name}: {e}"); - (metrics, None) - } - (Err(e), _) => { - warn!("Failed to get metrics for {name}: {e}"); - return ServiceProcessMetrics::default(); - } - }; + futures::future::join(self_.metrics_api.get(&name), clusterd_usage_fut).await; + let clusterd_usage = match clusterd_usage { + Ok(clusterd_usage) => Some(clusterd_usage), + Err(e) => { + self_ + .metrics + .record_fetch_error(FetchStep::ClusterdUsage, e.as_ref()); + warn!("Failed to fetch clusterd usage for {name}: {e:#}"); + None + } + }; + let metrics = match metrics { + Ok(metrics) => metrics, + Err(e) => { + self_.metrics.record_fetch_error(FetchStep::PodMetrics, &e); + warn!( + "Failed to get metrics for {name}: {}", + e.display_with_causes() + ); + return ServiceProcessMetrics::default(); + } + }; let Some(PodMetricsContainer { usage: PodMetricsContainerUsage { @@ -1586,6 +1606,9 @@ impl OrchestratorWorker { .. }) = metrics.containers.get(0) else { + self_ + .metrics + .record_fetch_failure(FetchStep::PodMetrics, "no_containers"); warn!("metrics result contained no containers for {name}"); return ServiceProcessMetrics::default(); }; diff --git a/src/orchestrator-kubernetes/src/metrics.rs b/src/orchestrator-kubernetes/src/metrics.rs new file mode 100644 index 0000000000000..f4f5d828e1f9f --- /dev/null +++ b/src/orchestrator-kubernetes/src/metrics.rs @@ -0,0 +1,101 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +//! Prometheus metrics for the Kubernetes orchestrator. + +use std::error::Error; +use std::io; + +use kube::error::Error as K8sError; +use mz_ore::metric; +use mz_ore::metrics::MetricsRegistry; +use mz_ore::metrics::raw::IntCounterVec; + +/// Metrics for a [`KubernetesOrchestrator`](crate::KubernetesOrchestrator). +#[derive(Debug, Clone)] +pub(crate) struct OrchestratorMetrics { + process_metrics_fetch_failures: IntCounterVec, +} + +/// The step of a per-process metrics fetch that failed. +#[derive(Debug, Clone, Copy)] +pub(crate) enum FetchStep { + /// Reading the pod's `PodMetrics` from the Kubernetes metrics API. + PodMetrics, + /// Reading usage metrics from the clusterd process. + ClusterdUsage, +} + +impl FetchStep { + fn as_str(&self) -> &'static str { + match self { + FetchStep::PodMetrics => "pod_metrics", + FetchStep::ClusterdUsage => "clusterd_usage", + } + } +} + +impl OrchestratorMetrics { + pub(crate) fn register_into(registry: &MetricsRegistry) -> Self { + Self { + process_metrics_fetch_failures: registry.register(metric!( + name: "mz_orchestrator_kubernetes_process_metrics_fetch_failures_total", + help: "The number of failed per-process metrics fetches, by fetch step and error kind.", + var_labels: ["step", "kind"], + )), + } + } + + /// Records a failed fetch whose error has the given `kind`. + pub(crate) fn record_fetch_failure(&self, step: FetchStep, kind: &str) { + self.process_metrics_fetch_failures + .with_label_values(&[step.as_str(), kind]) + .inc(); + } + + /// Records a failed fetch, classifying `error` by [`error_kind`]. + pub(crate) fn record_fetch_error(&self, step: FetchStep, error: &(dyn Error + 'static)) { + self.record_fetch_failure(step, error_kind(error)); + } +} + +/// Classifies an error by walking its source chain. +/// +/// Returns `timeout` if any error in the chain is a timeout. Otherwise returns the kind of the +/// outermost error that is a Kubernetes client or HTTP client error: `api` for an error status +/// from the API server, `decode` for a response that failed to deserialize, `transport` for a +/// connection or protocol error. Returns `other` for anything else. +pub(crate) fn error_kind(error: &(dyn Error + 'static)) -> &'static str { + let mut kind = None; + let mut next = Some(error); + while let Some(err) = next { + next = err.source(); + if let Some(e) = err.downcast_ref::() { + if e.kind() == io::ErrorKind::TimedOut { + return "timeout"; + } + } else if let Some(e) = err.downcast_ref::() { + if e.is_timeout() { + return "timeout"; + } + kind.get_or_insert_with(|| if e.is_decode() { "decode" } else { "transport" }); + } else if let Some(e) = err.downcast_ref::() { + kind.get_or_insert(match e { + K8sError::Api(_) => "api", + K8sError::SerdeError(_) => "decode", + K8sError::HyperError(_) | K8sError::Service(_) => "transport", + _ => "other", + }); + } + } + kind.unwrap_or("other") +} + +#[cfg(test)] +mod tests; diff --git a/src/orchestrator-kubernetes/src/metrics/tests.rs b/src/orchestrator-kubernetes/src/metrics/tests.rs new file mode 100644 index 0000000000000..cfa8b6131c071 --- /dev/null +++ b/src/orchestrator-kubernetes/src/metrics/tests.rs @@ -0,0 +1,52 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +use std::io; + +use anyhow::Context; +use kube::error::Error as K8sError; + +use super::error_kind; + +fn service_error(kind: io::ErrorKind) -> K8sError { + K8sError::Service(Box::new(io::Error::from(kind))) +} + +#[mz_ore::test] +fn error_kind_finds_nested_timeout() { + assert_eq!( + error_kind(&service_error(io::ErrorKind::TimedOut)), + "timeout" + ); + + let wrapped: anyhow::Result<()> = + Err(service_error(io::ErrorKind::TimedOut)).context("failed to get service"); + assert_eq!(error_kind(wrapped.unwrap_err().as_ref()), "timeout"); +} + +#[mz_ore::test] +fn error_kind_classifies_kube_errors() { + assert_eq!( + error_kind(&service_error(io::ErrorKind::ConnectionReset)), + "transport" + ); + + let serde_err = serde_json::from_str::("x").unwrap_err(); + assert_eq!(error_kind(&K8sError::SerdeError(serde_err)), "decode"); + + let wrapped: anyhow::Result<()> = + Err(service_error(io::ErrorKind::ConnectionReset)).context("failed to get service"); + assert_eq!(error_kind(wrapped.unwrap_err().as_ref()), "transport"); +} + +#[mz_ore::test] +fn error_kind_defaults_to_other() { + let err = anyhow::anyhow!("internal-http port missing in service spec"); + assert_eq!(error_kind(err.as_ref()), "other"); +} diff --git a/src/orchestrator-kubernetes/src/util.rs b/src/orchestrator-kubernetes/src/util.rs index 10478bee6a16c..621181a0135fc 100644 --- a/src/orchestrator-kubernetes/src/util.rs +++ b/src/orchestrator-kubernetes/src/util.rs @@ -38,7 +38,18 @@ pub async fn create_client(context: String) -> Result<(Client, String), anyhow:: }; kubeconfig.connect_timeout = Some(Duration::from_secs(10)); - kubeconfig.read_timeout = Some(Duration::from_secs(60)); + // The read timeout bounds how long a hung Kubernetes call can block an orchestrator worker, + // which handles one command at a time. + // + // NOTE: The read timeout must exceed the connection pool's 90 s idle expiry. hyper-timeout + // starts the read timer while a pooled connection sits idle and does not reset it when a + // request is written. A request on a connection that idled for close to `read_timeout` + // therefore times out before the response arrives and fails with + // `client error (SendRequest)`. + // + // TODO(CPU-306): Return to 60 s once kube-client enables hyper-timeout's + // `reset_reader_on_write`. + kubeconfig.read_timeout = Some(Duration::from_secs(120)); kubeconfig.write_timeout = Some(Duration::from_secs(60)); let namespace = kubeconfig.default_namespace.clone();