mz_orchestrator_kubernetes/
metrics.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Prometheus metrics for the Kubernetes orchestrator.
11
12use std::error::Error;
13use std::io;
14
15use kube::error::Error as K8sError;
16use mz_ore::metric;
17use mz_ore::metrics::MetricsRegistry;
18use mz_ore::metrics::raw::IntCounterVec;
19
20/// Metrics for a [`KubernetesOrchestrator`](crate::KubernetesOrchestrator).
21#[derive(Debug, Clone)]
22pub(crate) struct OrchestratorMetrics {
23    process_metrics_fetch_failures: IntCounterVec,
24}
25
26/// The step of a per-process metrics fetch that failed.
27#[derive(Debug, Clone, Copy)]
28pub(crate) enum FetchStep {
29    /// Reading the pod's `PodMetrics` from the Kubernetes metrics API.
30    PodMetrics,
31    /// Reading usage metrics from the clusterd process.
32    ClusterdUsage,
33}
34
35impl FetchStep {
36    fn as_str(&self) -> &'static str {
37        match self {
38            FetchStep::PodMetrics => "pod_metrics",
39            FetchStep::ClusterdUsage => "clusterd_usage",
40        }
41    }
42}
43
44impl OrchestratorMetrics {
45    pub(crate) fn register_into(registry: &MetricsRegistry) -> Self {
46        Self {
47            process_metrics_fetch_failures: registry.register(metric!(
48                name: "mz_orchestrator_kubernetes_process_metrics_fetch_failures_total",
49                help: "The number of failed per-process metrics fetches, by fetch step and error kind.",
50                var_labels: ["step", "kind"],
51            )),
52        }
53    }
54
55    /// Records a failed fetch whose error has the given `kind`.
56    pub(crate) fn record_fetch_failure(&self, step: FetchStep, kind: &str) {
57        self.process_metrics_fetch_failures
58            .with_label_values(&[step.as_str(), kind])
59            .inc();
60    }
61
62    /// Records a failed fetch, classifying `error` by [`error_kind`].
63    pub(crate) fn record_fetch_error(&self, step: FetchStep, error: &(dyn Error + 'static)) {
64        self.record_fetch_failure(step, error_kind(error));
65    }
66}
67
68/// Classifies an error by walking its source chain.
69///
70/// Returns `timeout` if any error in the chain is a timeout. Otherwise returns the kind of the
71/// outermost error that is a Kubernetes client or HTTP client error: `api` for an error status
72/// from the API server, `decode` for a response that failed to deserialize, `transport` for a
73/// connection or protocol error. Returns `other` for anything else.
74pub(crate) fn error_kind(error: &(dyn Error + 'static)) -> &'static str {
75    let mut kind = None;
76    let mut next = Some(error);
77    while let Some(err) = next {
78        next = err.source();
79        if let Some(e) = err.downcast_ref::<io::Error>() {
80            if e.kind() == io::ErrorKind::TimedOut {
81                return "timeout";
82            }
83        } else if let Some(e) = err.downcast_ref::<reqwest::Error>() {
84            if e.is_timeout() {
85                return "timeout";
86            }
87            kind.get_or_insert_with(|| if e.is_decode() { "decode" } else { "transport" });
88        } else if let Some(e) = err.downcast_ref::<K8sError>() {
89            kind.get_or_insert(match e {
90                K8sError::Api(_) => "api",
91                K8sError::SerdeError(_) => "decode",
92                K8sError::HyperError(_) | K8sError::Service(_) => "transport",
93                _ => "other",
94            });
95        }
96    }
97    kind.unwrap_or("other")
98}
99
100#[cfg(test)]
101mod tests;
mz_orchestrator_kubernetes/metrics.rs

mz_orchestrator_kubernetes/
metrics.rs
