GITGUD

neiam neiam /mutant-operator

An Experimental Operator

public · Issues · Pulls · Labels · Forks · Compare · Actions success · Badges · Packages

⭐

Log in to mark this repository.

Prometheus metrics on :9568

ec509f5 · gmorell · 2026-11-09 20:01

✓ Build and Push Docker Image / build-and-push (push) ✓ clippy / clippy (push) ✓ image / image (push)
10 files +523 -16
Message 
{commit_body(@commit)}

Files changed

modified Cargo.lock
+60 −2
@@ -1798,6 +1798,12 @@ dependencies = [
1798 1798 "vcpkg",
1799 1799 ]
1800 1800
1801 +[[package]]
1802 +name = "linux-raw-sys"
1803 +version = "0.4.15"
1804 +source = "registry+https://github.com/rust-lang/crates.io-index"
1805 +checksum = "d26c52dbd32dccf2d10cac7725f8eae5296885fb5703b261f7d0a0739ec807ab"
1806 +
1801 1807 [[package]]
1802 1808 name = "linux-raw-sys"
1803 1809 version = "0.9.4"
@@ -1897,6 +1903,7 @@ dependencies = [
1897 1903 "getrandom 0.3.2",
1898 1904 "k8s-openapi",
1899 1905 "kube",
1906 + "prometheus",
1900 1907 "schemars",
1901 1908 "sea-orm",
1902 1909 "serde",
@@ -2299,6 +2306,44 @@ dependencies = [
2299 2306 "yansi",
2300 2307 ]
2301 2308
2309 +[[package]]
2310 +name = "procfs"
2311 +version = "0.17.0"
2312 +source = "registry+https://github.com/rust-lang/crates.io-index"
2313 +checksum = "cc5b72d8145275d844d4b5f6d4e1eef00c8cd889edb6035c21675d1bb1f45c9f"
2314 +dependencies = [
2315 + "bitflags",
2316 + "hex",
2317 + "procfs-core",
2318 + "rustix 0.38.44",
2319 +]
2320 +
2321 +[[package]]
2322 +name = "procfs-core"
2323 +version = "0.17.0"
2324 +source = "registry+https://github.com/rust-lang/crates.io-index"
2325 +checksum = "239df02d8349b06fc07398a3a1697b06418223b1c7725085e801e7c0fc6a12ec"
2326 +dependencies = [
2327 + "bitflags",
2328 + "hex",
2329 +]
2330 +
2331 +[[package]]
2332 +name = "prometheus"
2333 +version = "0.14.0"
2334 +source = "registry+https://github.com/rust-lang/crates.io-index"
2335 +checksum = "3ca5326d8d0b950a9acd87e6a3f94745394f62e4dae1b1ee22b2bc0c394af43a"
2336 +dependencies = [
2337 + "cfg-if",
2338 + "fnv",
2339 + "lazy_static",
2340 + "libc",
2341 + "memchr",
2342 + "parking_lot",
2343 + "procfs",
2344 + "thiserror 2.0.21",
2345 +]
2346 +
2302 2347 [[package]]
2303 2348 name = "quote"
2304 2349 version = "1.0.40"
@@ -2474,6 +2519,19 @@ dependencies = [
2474 2519 "semver",
2475 2520 ]
2476 2521
2522 +[[package]]
2523 +name = "rustix"
2524 +version = "0.38.44"
2525 +source = "registry+https://github.com/rust-lang/crates.io-index"
2526 +checksum = "fdb5bc1ae2baa591800df16c9ca78619bf65c0488b41b96ccec5d11220d8c154"
2527 +dependencies = [
2528 + "bitflags",
2529 + "errno",
2530 + "libc",
2531 + "linux-raw-sys 0.4.15",
2532 + "windows-sys 0.59.0",
2533 +]
2534 +
2477 2535 [[package]]
2478 2536 name = "rustix"
2479 2537 version = "1.0.5"
@@ -2483,7 +2541,7 @@ dependencies = [
2483 2541 "bitflags",
2484 2542 "errno",
2485 2543 "libc",
2486 − "linux-raw-sys",
2544 + "linux-raw-sys 0.9.4",
2487 2545 "windows-sys 0.59.0",
2488 2546 ]
2489 2547
@@ -3289,7 +3347,7 @@ dependencies = [
3289 3347 "fastrand",
3290 3348 "getrandom 0.3.2",
3291 3349 "once_cell",
3292 − "rustix",
3350 + "rustix 1.0.5",
3293 3351 "windows-sys 0.59.0",
3294 3352 ]
3295 3353
modified Cargo.toml
+3 −0
@@ -20,6 +20,9 @@ sea-orm = { version = "2.0.4", features = ["sqlx-postgres", "runtime-tokio-nativ
20 20 futures = "0.3.31"
21 21 axum = { version = "0.8", default-features = false, features = ["http1", "json", "tokio", "tower-log", "query"] }
22 22 getrandom = "0.3"
23 +# /metrics for the cluster's Prometheus (see src/metrics.rs); "process" adds
24 +# the standard process_* series on Linux.
25 +prometheus = { version = "0.14", default-features = false, features = ["process"] }
23 26
24 27
25 28 [package.metadata.sqlx]
modified README.md
+22 −0
@@ -235,3 +235,25 @@ A listed variant looks like:
235 235
236 236 A Mutation adds its `name`, `namespace`, base `deployment`, `variants`,
237 237 `secret`, `summary`, `message` and `readOnly`.
238 +
239 +## Metrics
240 +
241 +Prometheus metrics are served at `/metrics` on a listener of their own,
242 +`MUTANT_METRICS_ADDR` (default `0.0.0.0:9568`), which the API's Service
243 +does not name. `operator.yml` annotates the pod for scraping. `kind` is
244 +`MutantDeployment` or `Mutation`.
245 +
246 +| Metric | Type | Labels | What it says |
247 +| --- | --- | --- | --- |
248 +| `mutant_operator_reconciliations_total` | counter | `kind`, `outcome` (`ok`/`error`) | Reconciles finished |
249 +| `mutant_operator_reconcile_duration_seconds` | histogram | `kind`, `outcome` | Time one reconcile took |
250 +| `mutant_operator_reconcile_errors_total` | counter | `kind`, `error` (`invalid`/`api`/`client`) | Failed reconciles, retried after 10s |
251 +| `mutant_operator_controller_errors_total` | counter | `kind`, `error` (`queue`/`object_not_found`/`runner`) | Other controller errors; `queue` means a watch failed |
252 +| `mutant_operator_managed_objects` | gauge | `kind` | Resources in the controller's cache |
253 +| `mutant_operator_variants` | gauge | `kind` | Variants declared in specs, summed |
254 +| `mutant_operator_mutation_variants` | gauge | `state` | Mutation variants by reported state |
255 +| `mutant_operator_build_info` | gauge | `version` | Always 1 |
256 +
257 +Plus the standard `process_*` series (CPU, memory, file descriptors). The
258 +gauges are read from the controllers' caches at scrape time; the counters
259 +reset when the pod restarts.
modified operator.yml
+7 −0
@@ -12,6 +12,10 @@ spec:
12 12 metadata:
13 13 labels:
14 14 app: mutantops-operator
15 + annotations:
16 + prometheus.io/scrape: "true"
17 + prometheus.io/path: /metrics
18 + prometheus.io/port: "9568"
15 19 spec:
16 20 serviceAccountName: mutantops-operator
17 21 containers:
@@ -43,6 +47,9 @@ spec:
43 47 ports:
44 48 - name: http
45 49 containerPort: 8080
50 + # /metrics, on its own port; the Service does not name it.
51 + - name: metrics
52 + containerPort: 9568
46 53 ---
47 54 apiVersion: v1
48 55 kind: Service
modified src/bin/main-deployment.rs
+19 −2
@@ -1,7 +1,7 @@
1 1 use kube::Client;
2 −use mutantops::{api, controller};
2 +use mutantops::{api, controller, metrics};
3 3 use std::net::SocketAddr;
4 −use tracing::{info, warn};
4 +use tracing::{error, info, warn};
5 5
6 6 /// Address the HTTP API listens on.
7 7 const DEFAULT_API_ADDR: &str = "0.0.0.0:8080";
@@ -27,6 +27,23 @@ async fn main() -> anyhow::Result<()> {
27 27 .unwrap_or_else(|_| DEFAULT_API_ADDR.into())
28 28 .parse()?;
29 29
30 + // Metrics get their own listener, which the API's Service does not name.
31 + // A failure here is logged, not fatal: the operator matters more than
32 + // its metrics.
33 + let metrics_addr: SocketAddr = std::env::var("MUTANT_METRICS_ADDR")
34 + .unwrap_or_else(|_| metrics::DEFAULT_METRICS_ADDR.into())
35 + .parse()?;
36 + tokio::spawn(async move {
37 + let res = async {
38 + let listener = tokio::net::TcpListener::bind(metrics_addr).await?;
39 + info!(addr = %metrics_addr, "metrics listening");
40 + metrics::serve(listener).await
41 + };
42 + if let Err(e) = res.await {
43 + error!(addr = %metrics_addr, "metrics listener failed: {e}");
44 + }
45 + });
46 +
30 47 let server = async {
31 48 let listener = tokio::net::TcpListener::bind(addr).await?;
32 49 info!(%addr, "HTTP API listening");
modified src/controller.rs
+47 −12
@@ -8,6 +8,7 @@ use crate::crd::{
8 8 validate_base, MutantDeployment, Mutation, MutationStatus, Variant, VariantState, VariantStatus,
9 9 MUTATION_LABEL, OWNER_LABEL,
10 10 };
11 +use crate::metrics::{self, KIND_MUTANT, KIND_MUTATION};
11 12 use futures::StreamExt;
12 13 use k8s_openapi::api::apps::v1::Deployment;
13 14 use k8s_openapi::api::core::v1::Secret;
@@ -15,7 +16,7 @@ use k8s_openapi::apimachinery::pkg::apis::meta::v1::OwnerReference;
15 16 use kube::{
16 17 api::{Api, DeleteParams, ListParams, Patch, PatchParams, PostParams},
17 18 runtime::{
18 − controller::{Action, Controller},
19 + controller::{self, Action, Controller},
19 20 watcher,
20 21 },
21 22 Client, Resource, ResourceExt,
@@ -45,6 +46,17 @@ impl fmt::Display for MutantopsError {
45 46 }
46 47 }
47 48
49 +impl MutantopsError {
50 + /// A short, bounded name for the error, for metrics.
51 + pub fn label(&self) -> &'static str {
52 + match self {
53 + MutantopsError::Invalid(_) => "invalid",
54 + MutantopsError::Kube(kube::Error::Api(_)) => "api",
55 + MutantopsError::Kube(_) => "client",
56 + }
57 + }
58 +}
59 +
48 60 impl From<kube::Error> for MutantopsError {
49 61 fn from(err: kube::Error) -> Self {
50 62 MutantopsError::Kube(err)
@@ -216,11 +228,24 @@ async fn ensure_secret(api: &Api<Secret>, m: &Mutation) -> Result<Result<String,
216 228 }
217 229 }
218 230
219 −fn error_policy<K: ResourceExt>(obj: Arc<K>, error: &MutantopsError, _ctx: Arc<Ctx>) -> Action {
231 +fn error_policy<K: Resource<DynamicType = ()>>(obj: Arc<K>, error: &MutantopsError, _ctx: Arc<Ctx>) -> Action {
232 + metrics::reconcile_error(&K::kind(&()), error.label());
220 233 warn!(name = %obj.name_any(), ns = ?obj.namespace(), "reconcile failed: {error}");
221 234 Action::requeue(Duration::from_secs(10))
222 235 }
223 236
237 +/// Counts what the controller's stream reports, beyond the reconcile
238 +/// errors `error_policy` has already counted.
239 +fn stream_error<E, Q>(kind: &'static str, err: &controller::Error<E, Q>) {
240 + let label = match err {
241 + controller::Error::ReconcilerFailed(..) => return,
242 + controller::Error::ObjectNotFound(_) => "object_not_found",
243 + controller::Error::QueueError(_) => "queue",
244 + controller::Error::RunnerError(_) => "runner",
245 + };
246 + metrics::controller_error(kind, label);
247 +}
248 +
224 249 /// Runs both controllers until a shutdown signal.
225 250 pub async fn run(client: Client) {
226 251 let ctx = Arc::new(Ctx { client: client.clone() });
@@ -230,23 +255,33 @@ pub async fn run(client: Client) {
230 255
231 256 let mutants = Controller::new(Api::<MutantDeployment>::all(client.clone()), watcher::Config::default())
232 257 .owns(Api::<Deployment>::all(client.clone()), owned())
233 − .shutdown_on_signal()
234 − .run(reconcile_mutant, error_policy, ctx.clone())
258 + .shutdown_on_signal();
259 + let mutations = Controller::new(Api::<Mutation>::all(client.clone()), watcher::Config::default())
260 + .owns(Api::<Deployment>::all(client.clone()), owned())
261 + .shutdown_on_signal();
262 + metrics::track(mutants.store(), mutations.store());
263 +
264 + let mutants = mutants
265 + .run(|o, c| metrics::instrument(KIND_MUTANT, reconcile_mutant(o, c)), error_policy, ctx.clone())
235 266 .for_each(|res| async move {
236 267 match res {
237 − Ok((obj, _)) => info!(kind = "MutantDeployment", name = %obj.name, ns = ?obj.namespace, "reconciled"),
238 − Err(err) => warn!(kind = "MutantDeployment", "reconcile failed: {err}"),
268 + Ok((obj, _)) => info!(kind = KIND_MUTANT, name = %obj.name, ns = ?obj.namespace, "reconciled"),
269 + Err(err) => {
270 + stream_error(KIND_MUTANT, &err);
271 + warn!(kind = KIND_MUTANT, "reconcile failed: {err}");
272 + }
239 273 }
240 274 });
241 275
242 − let mutations = Controller::new(Api::<Mutation>::all(client.clone()), watcher::Config::default())
243 − .owns(Api::<Deployment>::all(client.clone()), owned())
244 − .shutdown_on_signal()
245 − .run(reconcile_mutation, error_policy, ctx)
276 + let mutations = mutations
277 + .run(|o, c| metrics::instrument(KIND_MUTATION, reconcile_mutation(o, c)), error_policy, ctx)
246 278 .for_each(|res| async move {
247 279 match res {
248 − Ok((obj, _)) => info!(kind = "Mutation", name = %obj.name, ns = ?obj.namespace, "reconciled"),
249 − Err(err) => warn!(kind = "Mutation", "reconcile failed: {err}"),
280 + Ok((obj, _)) => info!(kind = KIND_MUTATION, name = %obj.name, ns = ?obj.namespace, "reconciled"),
281 + Err(err) => {
282 + stream_error(KIND_MUTATION, &err);
283 + warn!(kind = KIND_MUTATION, "reconcile failed: {err}");
284 + }
250 285 }
251 286 });
252 287
modified src/lib.rs
+1 −0
@@ -1,3 +1,4 @@
1 1 pub mod api;
2 2 pub mod controller;
3 3 pub mod crd;
4 +pub mod metrics;
added src/metrics.rs
+364 −0
@@ -0,0 +1,364 @@
1 +//! Prometheus metrics, on their own listener (`MUTANT_METRICS_ADDR`, default
2 +//! `0.0.0.0:9568`).
3 +//!
4 +//! Not on the HTTP API's port: that one is behind a Service other namespaces
5 +//! reach, and `/metrics` there would be one more thing to authorize. The
6 +//! cluster's Prometheus finds this listener from the pod's `prometheus.io/*`
7 +//! annotations.
8 +//!
9 +//! Every name is `mutant_operator_*`, except the process collector's standard
10 +//! `process_*` series. `kind` is the custom resource's kind, `MutantDeployment`
11 +//! or `Mutation`.
12 +//!
13 +//! - **Reconciles** (`mutant_operator_reconciliations_total`,
14 +//! `mutant_operator_reconcile_duration_seconds`,
15 +//! `mutant_operator_reconcile_errors_total`) are counted as they happen and
16 +//! reset when the pod restarts: use `rate`/`increase`.
17 +//! - **What is managed** (`mutant_operator_managed_objects`,
18 +//! `mutant_operator_variants`, `mutant_operator_mutation_variants`) is read
19 +//! from the controllers' caches at scrape time, so a deleted object drops
20 +//! out at once rather than waiting for a reconcile.
21 +
22 +use std::future::Future;
23 +use std::sync::{LazyLock, Mutex};
24 +use std::time::Instant;
25 +
26 +use axum::http::{header, StatusCode};
27 +use axum::response::IntoResponse;
28 +use axum::routing::get;
29 +use axum::Router;
30 +use kube::runtime::reflector::Store;
31 +use prometheus::{
32 + Encoder, HistogramOpts, HistogramVec, IntCounterVec, IntGauge, IntGaugeVec, Opts, Registry, TextEncoder,
33 +};
34 +use tracing::error;
35 +
36 +use crate::crd::{MutantDeployment, Mutation, VariantState};
37 +
38 +/// Default listen address for `/metrics`: the fleet's port for a separate
39 +/// metrics listener.
40 +pub const DEFAULT_METRICS_ADDR: &str = "0.0.0.0:9568";
41 +
42 +pub const KIND_MUTANT: &str = "MutantDeployment";
43 +pub const KIND_MUTATION: &str = "Mutation";
44 +
45 +/// A reconcile runs a handful of API calls: tens of milliseconds normally,
46 +/// seconds when the API server is slow.
47 +const DURATION_BUCKETS: &[f64] = &[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0];
48 +
49 +const VARIANT_STATES: [VariantState; 4] =
50 + [VariantState::Progressing, VariantState::Ready, VariantState::Invalid, VariantState::Conflict];
51 +
52 +struct Metrics {
53 + registry: Registry,
54 + reconciliations: IntCounterVec,
55 + reconcile_duration: HistogramVec,
56 + reconcile_errors: IntCounterVec,
57 + controller_errors: IntCounterVec,
58 + managed_objects: IntGaugeVec,
59 + variants: IntGaugeVec,
60 + mutation_variants: IntGaugeVec,
61 +}
62 +
63 +fn register<C: prometheus::core::Collector + Clone + 'static>(registry: &Registry, c: C) -> C {
64 + registry.register(Box::new(c.clone())).expect("unique metric");
65 + c
66 +}
67 +
68 +static METRICS: LazyLock<Metrics> = LazyLock::new(|| {
69 + let r = Registry::new();
70 +
71 + #[cfg(target_os = "linux")]
72 + r.register(Box::new(prometheus::process_collector::ProcessCollector::for_self()))
73 + .expect("process collector");
74 +
75 + let build_info = register(
76 + &r,
77 + IntGauge::with_opts(
78 + Opts::new("mutant_operator_build_info", "Always 1; version is the crate version running.")
79 + .const_label("version", env!("CARGO_PKG_VERSION")),
80 + )
81 + .expect("valid gauge"),
82 + );
83 + build_info.set(1);
84 +
85 + let m = Metrics {
86 + reconciliations: register(
87 + &r,
88 + IntCounterVec::new(
89 + Opts::new(
90 + "mutant_operator_reconciliations_total",
91 + "Reconciles finished. outcome: ok or error.",
92 + ),
93 + &["kind", "outcome"],
94 + )
95 + .expect("valid counter"),
96 + ),
97 + reconcile_duration: register(
98 + &r,
99 + HistogramVec::new(
100 + HistogramOpts::new(
101 + "mutant_operator_reconcile_duration_seconds",
102 + "Time one reconcile took. outcome: ok or error.",
103 + )
104 + .buckets(DURATION_BUCKETS.to_vec()),
105 + &["kind", "outcome"],
106 + )
107 + .expect("valid histogram"),
108 + ),
109 + reconcile_errors: register(
110 + &r,
111 + IntCounterVec::new(
112 + Opts::new(
113 + "mutant_operator_reconcile_errors_total",
114 + "Failed reconciles, retried after 10s. error: invalid (the resource itself), \
115 + api (the API server refused a request) or client (it was never answered).",
116 + ),
117 + &["kind", "error"],
118 + )
119 + .expect("valid counter"),
120 + ),
121 + controller_errors: register(
122 + &r,
123 + IntCounterVec::new(
124 + Opts::new(
125 + "mutant_operator_controller_errors_total",
126 + "Controller errors other than a failed reconcile. error: queue (a watch failed, \
127 + so changes are not being seen), object_not_found or runner.",
128 + ),
129 + &["kind", "error"],
130 + )
131 + .expect("valid counter"),
132 + ),
133 + managed_objects: register(
134 + &r,
135 + IntGaugeVec::new(
136 + Opts::new(
137 + "mutant_operator_managed_objects",
138 + "Custom resources in the controller's cache, cluster-wide.",
139 + ),
140 + &["kind"],
141 + )
142 + .expect("valid gauge"),
143 + ),
144 + variants: register(
145 + &r,
146 + IntGaugeVec::new(
147 + Opts::new(
148 + "mutant_operator_variants",
149 + "Variants declared in specs, summed over every resource of the kind: \
150 + the Deployments the operator is asked to run.",
151 + ),
152 + &["kind"],
153 + )
154 + .expect("valid gauge"),
155 + ),
156 + mutation_variants: register(
157 + &r,
158 + IntGaugeVec::new(
159 + Opts::new(
160 + "mutant_operator_mutation_variants",
161 + "Mutation variants by the state their Mutation's status reports: \
162 + Progressing, Ready, Invalid or Conflict.",
163 + ),
164 + &["state"],
165 + )
166 + .expect("valid gauge"),
167 + ),
168 + registry: r,
169 + };
170 +
171 + // Present from the first scrape, so a panel shows 0 rather than nothing.
172 + for kind in [KIND_MUTANT, KIND_MUTATION] {
173 + m.managed_objects.with_label_values(&[kind]).set(0);
174 + m.variants.with_label_values(&[kind]).set(0);
175 + for outcome in ["ok", "error"] {
176 + m.reconciliations.with_label_values(&[kind, outcome]);
177 + }
178 + }
179 + for state in VARIANT_STATES {
180 + m.mutation_variants.with_label_values(&[state_label(state)]).set(0);
181 + }
182 + m
183 +});
184 +
185 +fn state_label(state: VariantState) -> &'static str {
186 + match state {
187 + VariantState::Progressing => "Progressing",
188 + VariantState::Ready => "Ready",
189 + VariantState::Invalid => "Invalid",
190 + VariantState::Conflict => "Conflict",
191 + }
192 +}
193 +
194 +/// The controllers' caches, read at scrape time.
195 +static STORES: Mutex<Option<Stores>> = Mutex::new(None);
196 +
197 +struct Stores {
198 + mutants: Store<MutantDeployment>,
199 + mutations: Store<Mutation>,
200 +}
201 +
202 +/// Hands the controllers' caches to the scrape handler.
203 +pub fn track(mutants: Store<MutantDeployment>, mutations: Store<Mutation>) {
204 + *STORES.lock().unwrap_or_else(|e| e.into_inner()) = Some(Stores { mutants, mutations });
205 +}
206 +
207 +/// Sets the gauges from the resources as the controllers last saw them.
208 +fn record_managed(mutants: &[impl AsRef<MutantDeployment>], mutations: &[impl AsRef<Mutation>]) {
209 + let m = &*METRICS;
210 + m.managed_objects.with_label_values(&[KIND_MUTANT]).set(mutants.len() as i64);
211 + m.managed_objects.with_label_values(&[KIND_MUTATION]).set(mutations.len() as i64);
212 +
213 + let mutant_variants: usize = mutants.iter().map(|md| md.as_ref().spec.mutations.len()).sum();
214 + let mutation_variants: usize = mutations.iter().map(|mu| mu.as_ref().spec.variants.len()).sum();
215 + m.variants.with_label_values(&[KIND_MUTANT]).set(mutant_variants as i64);
216 + m.variants.with_label_values(&[KIND_MUTATION]).set(mutation_variants as i64);
217 +
218 + let mut by_state = [0i64; VARIANT_STATES.len()];
219 + for status in mutations.iter().filter_map(|mu| mu.as_ref().status.as_ref()) {
220 + for v in &status.variants {
221 + if let Some(i) = VARIANT_STATES.iter().position(|s| *s == v.state) {
222 + by_state[i] += 1;
223 + }
224 + }
225 + }
226 + for (state, n) in VARIANT_STATES.iter().zip(by_state) {
227 + m.mutation_variants.with_label_values(&[state_label(*state)]).set(n);
228 + }
229 +}
230 +
231 +fn refresh() {
232 + let guard = STORES.lock().unwrap_or_else(|e| e.into_inner());
233 + if let Some(stores) = guard.as_ref() {
234 + record_managed(&stores.mutants.state(), &stores.mutations.state());
235 + }
236 +}
237 +
238 +/// Times a reconcile and counts it by outcome.
239 +pub async fn instrument<T, E>(kind: &'static str, reconcile: impl Future<Output = Result<T, E>>) -> Result<T, E> {
240 + let start = Instant::now();
241 + let res = reconcile.await;
242 + let outcome = if res.is_ok() { "ok" } else { "error" };
243 + let m = &*METRICS;
244 + m.reconcile_duration.with_label_values(&[kind, outcome]).observe(start.elapsed().as_secs_f64());
245 + m.reconciliations.with_label_values(&[kind, outcome]).inc();
246 + res
247 +}
248 +
249 +/// Counts a failed reconcile; `error` is a short, bounded label.
250 +pub fn reconcile_error(kind: &str, error: &str) {
251 + METRICS.reconcile_errors.with_label_values(&[kind, error]).inc();
252 +}
253 +
254 +/// Counts an error the controller's stream reported other than a failed
255 +/// reconcile, which [`reconcile_error`] has already counted.
256 +pub fn controller_error(kind: &str, error: &str) {
257 + METRICS.controller_errors.with_label_values(&[kind, error]).inc();
258 +}
259 +
260 +/// The text exposition of every metric, refreshed from the caches first.
261 +pub fn render() -> Result<Vec<u8>, prometheus::Error> {
262 + refresh();
263 + let mut body = Vec::new();
264 + TextEncoder::new().encode(&METRICS.registry.gather(), &mut body)?;
265 + Ok(body)
266 +}
267 +
268 +async fn metrics() -> impl IntoResponse {
269 + match render() {
270 + Ok(body) => ([(header::CONTENT_TYPE, TextEncoder::new().format_type().to_string())], body).into_response(),
271 + Err(e) => {
272 + error!("metrics: encoding: {e}");
273 + StatusCode::INTERNAL_SERVER_ERROR.into_response()
274 + }
275 + }
276 +}
277 +
278 +/// The metrics listener's router.
279 +pub fn router() -> Router {
280 + Router::new().route("/metrics", get(metrics))
281 +}
282 +
283 +/// Serves [`router`] on `listener` until the process exits.
284 +pub async fn serve(listener: tokio::net::TcpListener) -> std::io::Result<()> {
285 + axum::serve(listener, router()).await
286 +}
287 +
288 +#[cfg(test)]
289 +mod tests {
290 + use super::*;
291 + use crate::crd::{MutationSpec, MutationStatus, VariantStatus};
292 + use tokio::io::{AsyncReadExt, AsyncWriteExt};
293 +
294 + fn text() -> String {
295 + String::from_utf8(render().unwrap()).unwrap()
296 + }
297 +
298 + #[tokio::test]
299 + async fn serves_metrics_over_http() {
300 + instrument(KIND_MUTATION, async { Ok::<_, ()>(()) }).await.unwrap();
301 + // The real listener, over a real socket.
302 + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
303 + let addr = listener.local_addr().unwrap();
304 + tokio::spawn(serve(listener));
305 + let mut conn = tokio::net::TcpStream::connect(addr).await.unwrap();
306 + conn.write_all(b"GET /metrics HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").await.unwrap();
307 + let mut body = String::new();
308 + conn.read_to_string(&mut body).await.unwrap();
309 + assert!(body.starts_with("HTTP/1.1 200"), "{body}");
310 + assert!(body.to_ascii_lowercase().contains("content-type: text/plain; version=0.0.4"), "{body}");
311 + for name in [
312 + "mutant_operator_build_info",
313 + "mutant_operator_reconciliations_total",
314 + "mutant_operator_reconcile_duration_seconds_bucket",
315 + "mutant_operator_managed_objects",
316 + "mutant_operator_mutation_variants",
317 + ] {
318 + assert!(body.contains(name), "{name} missing from:\n{body}");
319 + }
320 + #[cfg(target_os = "linux")]
321 + assert!(body.contains("process_resident_memory_bytes"), "{body}");
322 + }
323 +
324 + #[tokio::test]
325 + async fn counts_reconciles_and_errors() {
326 + // Its own kind label, so the parallel test cannot disturb the count.
327 + const KIND: &str = "TestKind";
328 + instrument(KIND, async { Ok::<_, ()>(()) }).await.unwrap();
329 + instrument(KIND, async { Err::<(), _>("boom") }).await.unwrap_err();
330 + reconcile_error(KIND, "api");
331 + let t = text();
332 + assert!(t.contains(r#"mutant_operator_reconciliations_total{kind="TestKind",outcome="ok"} 1"#), "{t}");
333 + assert!(t.contains(r#"mutant_operator_reconciliations_total{kind="TestKind",outcome="error"} 1"#), "{t}");
334 + assert!(t.contains(r#"mutant_operator_reconcile_duration_seconds_count{kind="TestKind",outcome="ok"} 1"#));
335 + assert!(t.contains(r#"mutant_operator_reconcile_errors_total{error="api",kind="TestKind"} 1"#), "{t}");
336 + }
337 +
338 + #[test]
339 + fn managed_gauges_follow_the_resources() {
340 + let variant = |state| VariantStatus::rejected("x", state, String::new());
341 + let spec: MutationSpec = serde_json::from_value(serde_json::json!({
342 + "deployment": {},
343 + "variants": [
344 + {"label": "a", "instances": 1},
345 + {"label": "b", "instances": 1},
346 + {"label": "c", "instances": 1},
347 + ],
348 + }))
349 + .unwrap();
350 + let mut m = Mutation::new("web", spec);
351 + m.status = Some(MutationStatus {
352 + variants: vec![variant(VariantState::Ready), variant(VariantState::Ready), variant(VariantState::Conflict)],
353 + ..MutationStatus::default()
354 + });
355 + record_managed(&[] as &[std::sync::Arc<MutantDeployment>], &[std::sync::Arc::new(m)]);
356 + let t = text();
357 + assert!(t.contains(r#"mutant_operator_managed_objects{kind="Mutation"} 1"#), "{t}");
358 + assert!(t.contains(r#"mutant_operator_managed_objects{kind="MutantDeployment"} 0"#), "{t}");
359 + assert!(t.contains(r#"mutant_operator_variants{kind="Mutation"} 3"#), "{t}");
360 + assert!(t.contains(r#"mutant_operator_mutation_variants{state="Ready"} 2"#), "{t}");
361 + assert!(t.contains(r#"mutant_operator_mutation_variants{state="Conflict"} 1"#), "{t}");
362 + assert!(t.contains(r#"mutant_operator_mutation_variants{state="Invalid"} 0"#), "{t}");
363 + }
364 +}

Parents: f91d5bc