pgroles_operator/
observability.rs

1//! Operator health endpoints and OTLP metrics export.
2
3use std::net::SocketAddr;
4use std::sync::Arc;
5use std::sync::atomic::{AtomicBool, Ordering};
6use std::time::{Duration, Instant};
7
8use axum::extract::State;
9use axum::http::StatusCode;
10use axum::response::IntoResponse;
11use axum::routing::get;
12use axum::{Router, serve};
13use opentelemetry::KeyValue;
14use opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider, UpDownCounter};
15use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig};
16use opentelemetry_sdk::Resource;
17use opentelemetry_sdk::logs::SdkLoggerProvider;
18use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
19use tokio::net::TcpListener;
20
21const SERVICE_NAME: &str = "pgroles-operator";
22
23#[derive(Clone)]
24pub struct OperatorObservability {
25    ready: Arc<AtomicBool>,
26    metrics: Option<Arc<Metrics>>,
27    logger_provider: Option<SdkLoggerProvider>,
28}
29
30struct Metrics {
31    provider: SdkMeterProvider,
32    reconcile_total: Counter<u64>,
33    reconcile_duration_ms: Histogram<u64>,
34    reconcile_inflight: UpDownCounter<i64>,
35    inspect_duration_ms: Histogram<u64>,
36    inspect_items_total: Counter<u64>,
37    wildcard_grantability_queries_total: Counter<u64>,
38    wildcard_unsatisfied_grants_total: Counter<u64>,
39    plan_total: Counter<u64>,
40    plan_changes_total: Counter<u64>,
41    lock_contention_total: Counter<u64>,
42    policy_conflicts_total: Counter<u64>,
43    invalid_spec_total: Counter<u64>,
44    deprecated_approval_unset_total: Counter<u64>,
45    database_connection_failures_total: Counter<u64>,
46    apply_total: Counter<u64>,
47    apply_statements_total: Counter<u64>,
48    ephemeral_transition_total: Counter<u64>,
49    ephemeral_failure_total: Counter<u64>,
50    ephemeral_retained_memberships_total: Counter<u64>,
51    ephemeral_expiry_lag_ms: Histogram<u64>,
52    ephemeral_role_retirement_blocked_total: Counter<u64>,
53    ephemeral_cached_requests: Histogram<u64>,
54    ephemeral_relevant_requests: Histogram<u64>,
55    ephemeral_reconcile_duration_ms: Histogram<u64>,
56    ephemeral_reconcile_inflight: UpDownCounter<i64>,
57}
58
59pub struct ReconcileGuard {
60    metrics: Option<Arc<Metrics>>,
61    started_at: Instant,
62}
63
64pub struct EphemeralReconcileGuard {
65    metrics: Option<Arc<Metrics>>,
66    started_at: Instant,
67    kind: &'static str,
68    request_count_bucket: &'static str,
69}
70
71impl OperatorObservability {
72    pub fn from_env() -> anyhow::Result<Self> {
73        Ok(Self {
74            ready: Arc::new(AtomicBool::new(false)),
75            metrics: init_metrics_from_env()?,
76            logger_provider: None,
77        })
78    }
79
80    pub fn with_logger_provider(mut self, provider: Option<SdkLoggerProvider>) -> Self {
81        self.logger_provider = provider;
82        self
83    }
84
85    pub fn mark_ready(&self) {
86        self.ready.store(true, Ordering::Relaxed);
87    }
88
89    pub fn mark_not_ready(&self) {
90        self.ready.store(false, Ordering::Relaxed);
91    }
92
93    pub fn start_reconcile(&self) -> ReconcileGuard {
94        if let Some(metrics) = &self.metrics {
95            metrics.reconcile_inflight.add(1, &[]);
96            ReconcileGuard {
97                metrics: Some(metrics.clone()),
98                started_at: Instant::now(),
99            }
100        } else {
101            ReconcileGuard {
102                metrics: None,
103                started_at: Instant::now(),
104            }
105        }
106    }
107
108    pub fn record_database_connection_failure(&self) {
109        if let Some(metrics) = &self.metrics {
110            metrics.database_connection_failures_total.add(1, &[]);
111        }
112    }
113
114    pub fn record_policy_conflict(&self) {
115        if let Some(metrics) = &self.metrics {
116            metrics.policy_conflicts_total.add(1, &[]);
117        }
118    }
119
120    pub fn record_lock_contention(&self) {
121        if let Some(metrics) = &self.metrics {
122            metrics.lock_contention_total.add(1, &[]);
123        }
124    }
125
126    pub fn record_plan_result(&self, result: &str) {
127        if let Some(metrics) = &self.metrics {
128            metrics
129                .plan_total
130                .add(1, &[KeyValue::new("result", result.to_string())]);
131        }
132    }
133
134    pub fn record_planned_changes(&self, changes: usize) {
135        if changes == 0 {
136            return;
137        }
138        if let Some(metrics) = &self.metrics {
139            metrics.plan_changes_total.add(changes as u64, &[]);
140        }
141    }
142
143    pub fn record_invalid_spec(&self) {
144        if let Some(metrics) = &self.metrics {
145            metrics.invalid_spec_total.add(1, &[]);
146        }
147    }
148
149    /// Count a reconcile that relied on the deprecated `spec.approval`
150    /// inference, so the remaining exposure is alertable fleet-wide rather than
151    /// only visible per object.
152    pub fn record_deprecated_approval_unset(&self, inferred: &str) {
153        if let Some(metrics) = &self.metrics {
154            metrics
155                .deprecated_approval_unset_total
156                .add(1, &[KeyValue::new("inferred", inferred.to_string())]);
157        }
158    }
159
160    pub fn record_inspection(&self, stats: &pgroles_inspect::InspectionStats) {
161        let Some(metrics) = &self.metrics else {
162            return;
163        };
164
165        for (phase, duration) in &stats.phase_durations {
166            metrics.inspect_duration_ms.record(
167                duration.as_millis() as u64,
168                &[KeyValue::new("phase", *phase)],
169            );
170        }
171
172        for (kind, count) in [
173            ("roles", stats.roles),
174            ("memberships", stats.memberships),
175            ("schemas", stats.schemas),
176            ("grants", stats.grants),
177            ("default_privileges", stats.default_privileges),
178            (
179                "wildcard_configured_grants",
180                stats.wildcard.configured_grants,
181            ),
182            (
183                "wildcard_configured_scopes",
184                stats.wildcard.configured_scopes,
185            ),
186            (
187                "wildcard_inventory_objects",
188                stats.wildcard.inventory_objects,
189            ),
190            (
191                "wildcard_unsatisfied_scopes",
192                stats.wildcard.unsatisfied_scopes,
193            ),
194            (
195                "wildcard_grantability_objects",
196                stats.wildcard.grantability_objects,
197            ),
198        ] {
199            if count > 0 {
200                metrics
201                    .inspect_items_total
202                    .add(count as u64, &[KeyValue::new("kind", kind)]);
203            }
204        }
205
206        if stats.wildcard.grantability_queries > 0 {
207            metrics
208                .wildcard_grantability_queries_total
209                .add(stats.wildcard.grantability_queries as u64, &[]);
210        }
211        if stats.wildcard.unsatisfied_grants > 0 {
212            metrics
213                .wildcard_unsatisfied_grants_total
214                .add(stats.wildcard.unsatisfied_grants as u64, &[]);
215        }
216    }
217
218    pub fn record_apply_result(&self, result: &str) {
219        if let Some(metrics) = &self.metrics {
220            metrics
221                .apply_total
222                .add(1, &[KeyValue::new("result", result.to_string())]);
223        }
224    }
225
226    pub fn record_apply_statements(&self, statements: usize) {
227        if statements == 0 {
228            return;
229        }
230        if let Some(metrics) = &self.metrics {
231            metrics.apply_statements_total.add(statements as u64, &[]);
232        }
233    }
234
235    pub fn record_ephemeral_transition(&self, phase: &str, reason: &str) {
236        if let Some(metrics) = &self.metrics {
237            metrics.ephemeral_transition_total.add(
238                1,
239                &[
240                    KeyValue::new("phase", phase.to_string()),
241                    KeyValue::new("reason", reason.to_string()),
242                ],
243            );
244            if matches!(phase, "Failed" | "Denied" | "ApprovalExpired") {
245                metrics
246                    .ephemeral_failure_total
247                    .add(1, &[KeyValue::new("reason", reason.to_string())]);
248            }
249        }
250    }
251
252    pub fn record_ephemeral_retained_memberships(&self, count: usize) {
253        if count == 0 {
254            return;
255        }
256        if let Some(metrics) = &self.metrics {
257            metrics
258                .ephemeral_retained_memberships_total
259                .add(count as u64, &[]);
260        }
261    }
262
263    pub fn record_ephemeral_expiry_lag(&self, lag: Duration) {
264        if let Some(metrics) = &self.metrics {
265            metrics
266                .ephemeral_expiry_lag_ms
267                .record(lag.as_millis() as u64, &[]);
268        }
269    }
270
271    pub fn record_ephemeral_role_retirement_blocked(&self) {
272        if let Some(metrics) = &self.metrics {
273            metrics.ephemeral_role_retirement_blocked_total.add(1, &[]);
274        }
275    }
276
277    pub fn record_ephemeral_relevant_requests(&self, lookup: &'static str, count: usize) {
278        if let Some(metrics) = &self.metrics {
279            metrics
280                .ephemeral_relevant_requests
281                .record(count as u64, &[KeyValue::new("lookup", lookup)]);
282        }
283    }
284
285    pub fn start_ephemeral_reconcile(
286        &self,
287        kind: &'static str,
288        cached_requests: usize,
289    ) -> EphemeralReconcileGuard {
290        if let Some(metrics) = &self.metrics {
291            metrics
292                .ephemeral_cached_requests
293                .record(cached_requests as u64, &[]);
294            metrics
295                .ephemeral_reconcile_inflight
296                .add(1, &[KeyValue::new("kind", kind)]);
297        }
298        EphemeralReconcileGuard {
299            metrics: self.metrics.clone(),
300            started_at: Instant::now(),
301            kind,
302            request_count_bucket: request_count_bucket(cached_requests),
303        }
304    }
305
306    pub fn shutdown(&self) -> anyhow::Result<()> {
307        if let Some(metrics) = &self.metrics {
308            metrics.provider.shutdown()?;
309        }
310        if let Some(provider) = &self.logger_provider {
311            provider.shutdown()?;
312        }
313        Ok(())
314    }
315}
316
317impl ReconcileGuard {
318    pub fn record_result(self, result: &str, reason: &str) {
319        if let Some(metrics) = &self.metrics {
320            metrics.reconcile_total.add(
321                1,
322                &[
323                    KeyValue::new("result", result.to_string()),
324                    KeyValue::new("reason", reason.to_string()),
325                ],
326            );
327            metrics
328                .reconcile_duration_ms
329                .record(self.started_at.elapsed().as_millis() as u64, &[]);
330        }
331    }
332}
333
334impl Drop for ReconcileGuard {
335    fn drop(&mut self) {
336        if let Some(metrics) = &self.metrics {
337            metrics.reconcile_inflight.add(-1, &[]);
338        }
339    }
340}
341
342impl Drop for EphemeralReconcileGuard {
343    fn drop(&mut self) {
344        if let Some(metrics) = &self.metrics {
345            metrics.ephemeral_reconcile_duration_ms.record(
346                self.started_at.elapsed().as_millis() as u64,
347                &[
348                    KeyValue::new("kind", self.kind),
349                    KeyValue::new("request_count", self.request_count_bucket),
350                ],
351            );
352            metrics
353                .ephemeral_reconcile_inflight
354                .add(-1, &[KeyValue::new("kind", self.kind)]);
355        }
356    }
357}
358
359fn request_count_bucket(count: usize) -> &'static str {
360    match count {
361        0 => "0",
362        1..=10 => "1-10",
363        11..=100 => "11-100",
364        101..=1_000 => "101-1000",
365        _ => "1001+",
366    }
367}
368
369pub async fn serve_health(
370    bind_addr: SocketAddr,
371    observability: OperatorObservability,
372) -> anyhow::Result<()> {
373    let listener = TcpListener::bind(bind_addr).await?;
374    let app = Router::new()
375        .route("/livez", get(livez))
376        .route("/readyz", get(readyz))
377        .with_state(observability);
378
379    serve(listener, app).await?;
380    Ok(())
381}
382
383fn init_metrics_from_env() -> anyhow::Result<Option<Arc<Metrics>>> {
384    if !otel_metrics_enabled() {
385        return Ok(None);
386    }
387
388    let exporter = MetricExporter::builder()
389        .with_tonic()
390        .with_protocol(Protocol::Grpc)
391        .build()?;
392
393    let reader = PeriodicReader::builder(exporter).build();
394    let provider = SdkMeterProvider::builder()
395        .with_reader(reader)
396        .with_resource(
397            Resource::builder_empty()
398                .with_attributes([
399                    KeyValue::new("service.name", SERVICE_NAME),
400                    KeyValue::new("service.version", env!("CARGO_PKG_VERSION")),
401                ])
402                .build(),
403        )
404        .build();
405
406    let meter = provider.meter(SERVICE_NAME);
407    Ok(Some(Arc::new(Metrics::new(provider, meter))))
408}
409
410/// Build the OTLP log provider before the global tracing subscriber is
411/// installed. The caller attaches an `OpenTelemetryTracingBridge` layer and
412/// stores the provider in `OperatorObservability` for graceful shutdown.
413pub fn init_log_provider_from_env() -> anyhow::Result<Option<SdkLoggerProvider>> {
414    let logs_exporter = std::env::var("OTEL_LOGS_EXPORTER").ok();
415    if matches!(logs_exporter.as_deref(), Some("none")) {
416        return Ok(None);
417    }
418    let endpoint_configured = std::env::var_os("OTEL_EXPORTER_OTLP_ENDPOINT").is_some()
419        || std::env::var_os("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT").is_some();
420    if !endpoint_configured && !matches!(logs_exporter.as_deref(), Some("otlp")) {
421        return Ok(None);
422    }
423
424    let exporter = opentelemetry_otlp::LogExporter::builder()
425        .with_tonic()
426        .with_protocol(Protocol::Grpc)
427        .build()?;
428    let provider = SdkLoggerProvider::builder()
429        .with_resource(
430            Resource::builder_empty()
431                .with_attributes([
432                    KeyValue::new("service.name", SERVICE_NAME),
433                    KeyValue::new("service.version", env!("CARGO_PKG_VERSION")),
434                ])
435                .build(),
436        )
437        .with_batch_exporter(exporter)
438        .build();
439    Ok(Some(provider))
440}
441
442fn otel_metrics_enabled() -> bool {
443    let metrics_exporter = std::env::var("OTEL_METRICS_EXPORTER").ok();
444    if matches!(metrics_exporter.as_deref(), Some("none")) {
445        return false;
446    }
447
448    std::env::var_os("OTEL_EXPORTER_OTLP_ENDPOINT").is_some()
449        || std::env::var_os("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT").is_some()
450}
451
452impl Metrics {
453    fn new(provider: SdkMeterProvider, meter: Meter) -> Self {
454        Self {
455            provider,
456            reconcile_total: meter
457                .u64_counter("pgroles.reconcile.total")
458                .with_description("Total reconciliations by result and reason")
459                .build(),
460            reconcile_duration_ms: meter
461                .u64_histogram("pgroles.reconcile.duration")
462                .with_unit("ms")
463                .with_description("Reconciliation duration in milliseconds")
464                .build(),
465            reconcile_inflight: meter
466                .i64_up_down_counter("pgroles.reconcile.inflight")
467                .with_description("In-flight reconciliations")
468                .build(),
469            inspect_duration_ms: meter
470                .u64_histogram("pgroles.inspect.duration")
471                .with_unit("ms")
472                .with_description("Database inspection phase duration in milliseconds")
473                .build(),
474            inspect_items_total: meter
475                .u64_counter("pgroles.inspect.items")
476                .with_description("Database inspection objects observed by kind")
477                .build(),
478            wildcard_grantability_queries_total: meter
479                .u64_counter("pgroles.wildcard.grantability_queries")
480                .with_description("Wildcard grantability catalog queries")
481                .build(),
482            wildcard_unsatisfied_grants_total: meter
483                .u64_counter("pgroles.wildcard.unsatisfied_grants")
484                .with_description("Wildcard grants missing privileges before grantability checks")
485                .build(),
486            plan_total: meter
487                .u64_counter("pgroles.plan.total")
488                .with_description("Successful plan-mode reconciliations by result")
489                .build(),
490            plan_changes_total: meter
491                .u64_counter("pgroles.plan.changes")
492                .with_description("Planned changes discovered during plan-mode reconciliations")
493                .build(),
494            lock_contention_total: meter
495                .u64_counter("pgroles.lock_contention.total")
496                .with_description("Reconciliations delayed by per-database lock contention")
497                .build(),
498            policy_conflicts_total: meter
499                .u64_counter("pgroles.policy.conflicts")
500                .with_description("Conflicting policies targeting the same database")
501                .build(),
502            invalid_spec_total: meter
503                .u64_counter("pgroles.invalid_spec.total")
504                .with_description("Invalid PostgresPolicy specifications")
505                .build(),
506            deprecated_approval_unset_total: meter
507                .u64_counter("pgroles.deprecated.approval_unset")
508                .with_description(
509                    "Reconciles of a PostgresPolicy that omits spec.approval and relies on the \
510                     deprecated inference from spec.mode",
511                )
512                .build(),
513            database_connection_failures_total: meter
514                .u64_counter("pgroles.database.connection_failures")
515                .with_description("Database connection failures during reconciliation")
516                .build(),
517            apply_total: meter
518                .u64_counter("pgroles.apply.total")
519                .with_description("Apply transaction outcomes")
520                .build(),
521            apply_statements_total: meter
522                .u64_counter("pgroles.apply.statements")
523                .with_description("SQL statements executed during successful applies")
524                .build(),
525            ephemeral_transition_total: meter
526                .u64_counter("pgroles.ephemeral_access.transitions")
527                .with_description("Ephemeral access lifecycle transitions by phase and reason")
528                .build(),
529            ephemeral_failure_total: meter
530                .u64_counter("pgroles.ephemeral_access.failures")
531                .with_description("Terminal ephemeral access failures by reason")
532                .build(),
533            ephemeral_retained_memberships_total: meter
534                .u64_counter("pgroles.ephemeral_access.retained_memberships")
535                .with_description("Ephemeral memberships retained because they became durable")
536                .build(),
537            ephemeral_expiry_lag_ms: meter
538                .u64_histogram("pgroles.ephemeral_access.expiry_lag")
539                .with_unit("ms")
540                .with_description("Delay between absolute expiry and revocation processing")
541                .build(),
542            ephemeral_role_retirement_blocked_total: meter
543                .u64_counter("pgroles.ephemeral_access.role_retirement_blocked")
544                .with_description("Role retirements blocked by active access requests")
545                .build(),
546            ephemeral_cached_requests: meter
547                .u64_histogram("pgroles.ephemeral_access.cached_requests")
548                .with_description("Request-cache size sampled at reconcile start")
549                .build(),
550            ephemeral_relevant_requests: meter
551                .u64_histogram("pgroles.ephemeral_access.relevant_requests")
552                .with_description("Requests returned by an indexed lookup")
553                .build(),
554            ephemeral_reconcile_duration_ms: meter
555                .u64_histogram("pgroles.ephemeral_access.reconcile.duration")
556                .with_unit("ms")
557                .with_description("Ephemeral reconcile duration by kind and request-count bucket")
558                .build(),
559            ephemeral_reconcile_inflight: meter
560                .i64_up_down_counter("pgroles.ephemeral_access.reconcile.inflight")
561                .with_description("In-flight ephemeral reconciliations by resource kind")
562                .build(),
563        }
564    }
565}
566
567async fn livez() -> &'static str {
568    "ok"
569}
570
571async fn readyz(State(observability): State<OperatorObservability>) -> impl IntoResponse {
572    if observability.ready.load(Ordering::Relaxed) {
573        (StatusCode::OK, "ready")
574    } else {
575        (StatusCode::SERVICE_UNAVAILABLE, "not ready")
576    }
577}
578
579#[cfg(test)]
580mod tests {
581    use std::sync::{Arc, Mutex};
582    use std::time::Duration;
583
584    use axum::extract::State;
585    use axum::http::StatusCode;
586    use axum::response::IntoResponse;
587    use opentelemetry::metrics::MeterProvider;
588    use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData, ResourceMetrics};
589    use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader, SdkMeterProvider};
590
591    use super::{
592        Metrics, OperatorObservability, ReconcileGuard, SERVICE_NAME, livez, otel_metrics_enabled,
593        readyz,
594    };
595
596    static ENV_LOCK: Mutex<()> = Mutex::new(());
597
598    fn test_observability() -> (
599        OperatorObservability,
600        SdkMeterProvider,
601        InMemoryMetricExporter,
602    ) {
603        let exporter = InMemoryMetricExporter::default();
604        let provider = SdkMeterProvider::builder()
605            .with_reader(PeriodicReader::builder(exporter.clone()).build())
606            .build();
607        let meter = provider.meter(SERVICE_NAME);
608        let observability = OperatorObservability {
609            ready: Arc::new(std::sync::atomic::AtomicBool::new(false)),
610            metrics: Some(Arc::new(Metrics::new(provider.clone(), meter))),
611            logger_provider: None,
612        };
613
614        (observability, provider, exporter)
615    }
616
617    fn metric_exists(metrics: &[ResourceMetrics], name: &str) -> bool {
618        metrics.iter().any(|resource_metrics| {
619            resource_metrics
620                .scope_metrics()
621                .flat_map(|scope_metrics| scope_metrics.metrics())
622                .any(|metric| metric.name() == name)
623        })
624    }
625
626    fn u64_sum_value(metrics: &[ResourceMetrics], name: &str) -> Option<u64> {
627        let mut found = false;
628        let total = metrics
629            .iter()
630            .flat_map(|resource_metrics| resource_metrics.scope_metrics())
631            .flat_map(|scope_metrics| scope_metrics.metrics())
632            .filter(|metric| metric.name() == name)
633            .filter_map(|metric| match metric.data() {
634                AggregatedMetrics::U64(MetricData::Sum(sum)) => {
635                    found = true;
636                    Some(
637                        sum.data_points()
638                            .map(|data_point| data_point.value())
639                            .sum::<u64>(),
640                    )
641                }
642                _ => None,
643            })
644            .sum();
645
646        found.then_some(total)
647    }
648
649    fn i64_sum_value(metrics: &[ResourceMetrics], name: &str) -> Option<i64> {
650        metrics
651            .iter()
652            .flat_map(|resource_metrics| resource_metrics.scope_metrics())
653            .flat_map(|scope_metrics| scope_metrics.metrics())
654            .find(|metric| metric.name() == name)
655            .and_then(|metric| match metric.data() {
656                AggregatedMetrics::I64(MetricData::Sum(sum)) => sum
657                    .data_points()
658                    .next()
659                    .map(|data_point| data_point.value()),
660                _ => None,
661            })
662    }
663
664    #[test]
665    fn otel_metrics_stay_disabled_without_endpoint() {
666        let _guard = ENV_LOCK.lock().expect("env lock should not be poisoned");
667        unsafe {
668            std::env::remove_var("OTEL_EXPORTER_OTLP_ENDPOINT");
669            std::env::remove_var("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT");
670            std::env::remove_var("OTEL_METRICS_EXPORTER");
671        }
672        assert!(!otel_metrics_enabled());
673    }
674
675    #[test]
676    fn otel_metrics_enable_with_explicit_endpoint() {
677        let _guard = ENV_LOCK.lock().expect("env lock should not be poisoned");
678        unsafe {
679            std::env::set_var("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4317");
680            std::env::remove_var("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT");
681            std::env::remove_var("OTEL_METRICS_EXPORTER");
682        }
683        assert!(otel_metrics_enabled());
684        unsafe {
685            std::env::remove_var("OTEL_EXPORTER_OTLP_ENDPOINT");
686        }
687    }
688
689    #[tokio::test]
690    async fn health_endpoints_reflect_readiness() {
691        let (observability, _provider, _exporter) = test_observability();
692
693        assert_eq!(livez().await, "ok");
694
695        let not_ready = readyz(State(observability.clone())).await.into_response();
696        assert_eq!(not_ready.status(), StatusCode::SERVICE_UNAVAILABLE);
697
698        observability.mark_ready();
699        let ready = readyz(State(observability)).await.into_response();
700        assert_eq!(ready.status(), StatusCode::OK);
701    }
702
703    #[test]
704    fn metrics_are_recorded_and_flushed() {
705        let (observability, provider, exporter) = test_observability();
706
707        let guard: ReconcileGuard = observability.start_reconcile();
708        observability.record_lock_contention();
709        observability.record_policy_conflict();
710        observability.record_invalid_spec();
711        observability.record_database_connection_failure();
712        observability.record_inspection(&pgroles_inspect::InspectionStats {
713            roles: 3,
714            memberships: 2,
715            schemas: 1,
716            grants: 5,
717            default_privileges: 1,
718            phase_durations: [
719                ("roles", Duration::from_millis(4)),
720                ("object_privileges", Duration::from_millis(12)),
721            ]
722            .into_iter()
723            .collect(),
724            wildcard: pgroles_inspect::WildcardInspectionStats {
725                configured_grants: 2,
726                configured_scopes: 1,
727                inventory_objects: 100,
728                unsatisfied_grants: 1,
729                unsatisfied_scopes: 1,
730                grantability_queries: 1,
731                grantability_objects: 3,
732            },
733        });
734        observability.record_plan_result("drift");
735        observability.record_planned_changes(2);
736        observability.record_apply_result("success");
737        observability.record_apply_statements(4);
738        observability.record_ephemeral_transition("Active", "MembershipsGranted");
739        observability.record_ephemeral_transition("Failed", "InvalidRequestState");
740        observability.record_ephemeral_retained_memberships(2);
741        observability.record_ephemeral_expiry_lag(Duration::from_millis(250));
742        observability.record_ephemeral_role_retirement_blocked();
743        observability.record_ephemeral_relevant_requests("effective_graph", 3);
744        drop(observability.start_ephemeral_reconcile("access_request", 250));
745        guard.record_result("conflict", "ConflictingPolicy");
746
747        provider.force_flush().expect("flush should succeed");
748
749        let metrics = exporter
750            .get_finished_metrics()
751            .expect("metrics should be exported");
752
753        assert!(metric_exists(&metrics, "pgroles.reconcile.total"));
754        assert!(metric_exists(&metrics, "pgroles.reconcile.duration"));
755        assert!(metric_exists(&metrics, "pgroles.inspect.duration"));
756        assert_eq!(u64_sum_value(&metrics, "pgroles.inspect.items"), Some(119));
757        assert_eq!(
758            u64_sum_value(&metrics, "pgroles.ephemeral_access.transitions"),
759            Some(2)
760        );
761        assert_eq!(
762            u64_sum_value(&metrics, "pgroles.ephemeral_access.failures"),
763            Some(1)
764        );
765        assert_eq!(
766            u64_sum_value(&metrics, "pgroles.ephemeral_access.retained_memberships"),
767            Some(2)
768        );
769        assert!(metric_exists(
770            &metrics,
771            "pgroles.ephemeral_access.expiry_lag"
772        ));
773        assert_eq!(
774            u64_sum_value(&metrics, "pgroles.ephemeral_access.role_retirement_blocked"),
775            Some(1)
776        );
777        assert!(metric_exists(
778            &metrics,
779            "pgroles.ephemeral_access.cached_requests"
780        ));
781        assert!(metric_exists(
782            &metrics,
783            "pgroles.ephemeral_access.relevant_requests"
784        ));
785        assert!(metric_exists(
786            &metrics,
787            "pgroles.ephemeral_access.reconcile.duration"
788        ));
789        assert_eq!(
790            u64_sum_value(&metrics, "pgroles.wildcard.grantability_queries"),
791            Some(1)
792        );
793        assert_eq!(
794            u64_sum_value(&metrics, "pgroles.wildcard.unsatisfied_grants"),
795            Some(1)
796        );
797        assert_eq!(u64_sum_value(&metrics, "pgroles.plan.total"), Some(1));
798        assert_eq!(u64_sum_value(&metrics, "pgroles.plan.changes"), Some(2));
799        assert_eq!(
800            u64_sum_value(&metrics, "pgroles.lock_contention.total"),
801            Some(1)
802        );
803        assert_eq!(u64_sum_value(&metrics, "pgroles.policy.conflicts"), Some(1));
804        assert_eq!(
805            u64_sum_value(&metrics, "pgroles.invalid_spec.total"),
806            Some(1)
807        );
808        assert_eq!(
809            u64_sum_value(&metrics, "pgroles.database.connection_failures"),
810            Some(1)
811        );
812        assert_eq!(u64_sum_value(&metrics, "pgroles.apply.statements"), Some(4));
813        assert_eq!(
814            i64_sum_value(&metrics, "pgroles.reconcile.inflight"),
815            Some(0)
816        );
817    }
818}
pgroles_operator/observability.rs

pgroles_operator/
observability.rs
