mz_orchestratord/
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
10use std::collections::BTreeMap;
11use std::time::Duration;
12
13use axum::{Extension, Router, body::Body, routing::get};
14use http::{HeaderMap, Method, Request, Response, StatusCode};
15use prometheus::{Encoder, TextEncoder};
16use tower_http::{classify::ServerErrorsFailureClass, trace::TraceLayer};
17use tracing::{Level, Span};
18
19use mz_ore::metric;
20use mz_ore::metrics::{MetricsRegistry, UIntGauge};
21
22#[derive(Debug)]
23pub struct Metrics {
24    pub is_leader: UIntGauge,
25    pub environmentd_needs_update: UIntGauge,
26}
27
28impl Metrics {
29    pub fn register_into(registry: &MetricsRegistry) -> Self {
30        Self {
31            is_leader: registry.register(
32                metric! {
33                    name: "orchestratord_is_leader",
34                    help: "Whether this operator replica holds the controller leadership lease, and is therefore the replica reconciling. Summed across the replicas this should be 1. A sustained 0 means no replica can take the lease, for instance because the service account lacks permission on leases, and the operator is reconciling nothing.",
35                }),
36            environmentd_needs_update: registry.register(
37                metric! {
38                    name: "environmentd_needs_update",
39                    help: "Count of organizations in this cluster which are running outdated pod templates. Only the operator replica holding the leadership lease reconciles, so the others report zero.",
40                }),
41        }
42    }
43
44    /// Records that this replica has taken the leadership lease.
45    pub fn leadership_acquired(&self) {
46        self.is_leader.set(1);
47    }
48
49    /// Records that this replica no longer holds the leadership lease, and
50    /// resets the metrics that only mean anything while it is reconciling.
51    ///
52    /// Those are derived from what reconciliation observed, and the process
53    /// outlives its own leadership: it keeps serving the conversion webhook
54    /// after losing the lease. Without this, a former leader would go on
55    /// publishing its last observation forever, so summing across the
56    /// replicas would count the same organizations once per past leader.
57    pub fn leadership_lost(&self) {
58        self.is_leader.set(0);
59        self.environmentd_needs_update.set(0);
60    }
61}
62
63pub fn router(registry: MetricsRegistry) -> Router {
64    add_tracing_layer(
65        Router::new()
66            .route("/metrics", get(metrics))
67            .layer(Extension(registry)),
68    )
69}
70
71#[allow(clippy::unused_async)]
72async fn metrics(Extension(registry): Extension<MetricsRegistry>) -> (StatusCode, Vec<u8>) {
73    let mut buf = vec![];
74    let encoder = TextEncoder::new();
75    let metric_families = registry.gather();
76    encoder.encode(&metric_families, &mut buf).unwrap();
77    (StatusCode::OK, buf)
78}
79
80///   Adds a tracing layer that reports an `INFO` level span per
81///   request and reports a `WARN` event when a handler returns a
82///   server error to the given Axum Router
83///
84///   This accepts a router instead of returning a layer itself
85///   to avoid dealing with defining generics over a bunch of closures
86///   (see <https://users.rust-lang.org/t/how-to-encapsulate-a-builder-that-depends-on-a-closure/71139/6>)
87///
88///   And this also can't be returned as a Router::new()::layer(TraceLayer)...
89///   because the TraceLayer needs to be added to a Router after
90///   all routes are defined, as it won't trace any routes defined
91///   on the router after it's attached.
92fn add_tracing_layer<S>(router: Router<S>) -> Router<S>
93where
94    S: Clone + Send + Sync + 'static,
95{
96    router.layer(
97        TraceLayer::new_for_http()
98            .make_span_with(|request: &Request<Body>| {
99                // This ugly macro is needed, unfortunately (and
100                // copied from tower-http), because
101                // `tracing::span!` required the level argument to
102                // be static. Meaning we can't just pass
103                // `self.level`.
104                macro_rules! make_span {
105                        ($level:expr) => {
106                            tracing::span!(
107                                $level,
108                                "HTTP request",
109                                "request.uri" = %request.uri(),
110                                "request.version" = ?request.version(),
111                                "request.method" = %request.method(),
112                                "request.headers" = tracing::field::Empty,
113                                "response.status" = tracing::field::Empty,
114                                "response.status_code" = tracing::field::Empty,
115                                "response.headers" = tracing::field::Empty,
116                            )
117                        }
118                    }
119                let span = if ["/api/health", "/metrics"].contains(&request.uri().path())
120                    || request.method() == Method::OPTIONS
121                {
122                    make_span!(Level::DEBUG)
123                } else {
124                    make_span!(Level::INFO)
125                };
126
127                if let Ok(s) = serde_json::to_string(&display_headers(request.headers().clone())) {
128                    span.record("request.headers", s);
129                }
130
131                span
132            })
133            .on_response(|response: &Response<Body>, _latency, span: &Span| {
134                span.record(
135                    "response.status",
136                    &tracing::field::display(response.status()),
137                );
138                span.record("response.status_code", response.status().as_u16());
139                if let Ok(s) = serde_json::to_string(&display_headers(response.headers().clone())) {
140                    span.record("response.headers", s);
141                }
142
143                // Emit an event at the same level as the span. For the same reason as noted in the comment
144                // above we can't use `tracing::event!(dynamic_level, ...)` since the level argument
145                // needs to be static
146                if span
147                    .metadata()
148                    .and_then(|m| Some(m.level()))
149                    .unwrap_or(&Level::DEBUG)
150                    == &Level::DEBUG
151                {
152                    tracing::debug!("HTTP response generated");
153                } else {
154                    tracing::info!("HTTP response generated");
155                }
156            })
157            .on_failure(
158                |error: ServerErrorsFailureClass, _latency: Duration, _span: &Span| {
159                    tracing::warn!(error = ?error, "HTTP request handling error");
160                },
161            ),
162    )
163}
164
165fn display_headers(mut headers: HeaderMap) -> BTreeMap<String, String> {
166    // Don't log Authorization headers
167    _ = headers.remove(http::header::AUTHORIZATION);
168
169    headers
170        .into_iter()
171        .filter_map(|(k, v)| {
172            k.map(|k| {
173                (
174                    k.to_string(),
175                    String::from_utf8_lossy(v.as_bytes()).to_string(),
176                )
177            })
178        })
179        .collect()
180}
mz_orchestratord/metrics.rs

mz_orchestratord/
metrics.rs
