arcly_http/observability/

plugin.rs

//! `ArclyObservabilityPlugin` — wires up all four observability backends in one
//! plugin that drops into any arcly-http application.
//!
//! ## What it sets up
//! - **Structured JSON logs** via `tracing-subscriber` (respects `RUST_LOG`).
//! - **Prometheus metrics** at `GET /metrics`.
//! - **OTLP distributed traces** exported to the configured gRPC endpoint.
//! - **Health / readiness probes** at `GET /healthz` and `GET /readyz`.
//!
//! ## Usage
//! ```ignore
//! App::launch_with_plugins::<AppModule>(
//!     "0.0.0.0:3000",
//!     OpenApiInfo { /* ... */ },
//!     vec![
//!         Box::new(ArclyObservabilityPlugin {
//!             service_name:    "my-service",
//!             service_version: env!("CARGO_PKG_VERSION"),
//!             otlp_endpoint:   "http://localhost:4317",
//!         }),
//!     ],
//! ).await

use futures::future::BoxFuture;
use metrics_exporter_prometheus::PrometheusHandle;

use crate::core::engine::{FrozenDiContainer, HttpMethod};
use crate::core::plugins::{ArclyPlugin, ArclyPluginContext, PluginError};

/// Drop-in observability plugin. Add to `App::launch_with_plugins` to get
/// structured logs, Prometheus metrics, OTLP traces, and health endpoints
/// with zero changes to handler code.
pub struct ArclyObservabilityPlugin {
    pub service_name: &'static str,
    pub service_version: &'static str,
    /// gRPC endpoint of the OTLP collector.
    /// Example: `"http://localhost:4317"`
    pub otlp_endpoint: &'static str,
}

impl ArclyPlugin for ArclyObservabilityPlugin {
    fn name(&self) -> &'static str {
        "ArclyObservabilityPlugin"
    }

    fn on_init<'a>(
        &'a mut self,
        ctx: &'a mut ArclyPluginContext,
    ) -> BoxFuture<'a, Result<(), PluginError>> {
        Box::pin(async move {
            // ── 1. Structured JSON logging ─────────────────────────────────
            use tracing_subscriber::{fmt, EnvFilter};
            let _ = fmt()
                .json()
                .with_env_filter(
                    EnvFilter::try_from_default_env()
                        // Suppress opentelemetry_sdk export-failure spam when no
                        // OTLP collector is running (common in dev / CI).
                        // Override: RUST_LOG=info,opentelemetry_sdk=warn
                        .unwrap_or_else(|_| EnvFilter::new("info,opentelemetry_sdk=off")),
                )
                .try_init(); // silently ignore "already set" errors

            // ── 2. Prometheus metrics ──────────────────────────────────────
            let handle: PrometheusHandle = crate::observability::metrics::init_metrics();

            // Provide the handle into DI so the /metrics handler can render it.
            ctx.provide::<PrometheusHandle>(handle.clone());

            // Register GET /metrics route.
            let metrics_handler = crate::observability::metrics::metrics_route_handler(handle);
            ctx.add_route(HttpMethod::GET, "/metrics", metrics_handler);

            // ── 3. OpenTelemetry OTLP traces ───────────────────────────────
            crate::observability::otel::init_tracer(&crate::observability::otel::OtelConfig {
                service_name: self.service_name,
                service_version: self.service_version,
                otlp_endpoint: self.otlp_endpoint,
            });

            // ── 4. Health / readiness endpoints ────────────────────────────
            // Initialise the global registry now so it's ready for other plugins
            // to register checks in their own on_init.
            let _ = crate::observability::health::global();

            ctx.add_route(
                HttpMethod::GET,
                "/healthz",
                crate::observability::health::healthz_handler(),
            );
            ctx.add_route(
                HttpMethod::GET,
                "/readyz",
                crate::observability::health::readyz_handler(),
            );

            // ── 5. Inject plugin routes into the OpenAPI spec ─────────────────
            // Plugin routes are added to the axum router directly (not via the
            // `inventory` system that macro-routes use), so `build_spec` does not
            // see them. The `modify_openapi` mutator runs after `build_spec` and
            // patches the paths object, making all three endpoints visible in the
            // Swagger UI at /docs.
            ctx.modify_openapi(|spec| {
                let Some(paths) = spec
                    .as_object_mut()
                    .and_then(|m| m.get_mut("paths"))
                    .and_then(|v| v.as_object_mut())
                else { return };

                // Shared schema for /healthz and /readyz responses.
                let health_schema = serde_json::json!({
                    "type": "object",
                    "required": ["status", "checks", "uptime_secs"],
                    "properties": {
                        "status": {
                            "type": "string",
                            "enum": ["healthy", "degraded", "unhealthy"],
                            "description": "Overall health of the service."
                        },
                        "checks": {
                            "type": "object",
                            "description": "Per-subsystem health results.",
                            "additionalProperties": {
                                "oneOf": [
                                    { "type": "string", "enum": ["healthy"] },
                                    { "type": "object", "required": ["degraded"],   "properties": { "degraded":   { "type": "string" } } },
                                    { "type": "object", "required": ["unhealthy"], "properties": { "unhealthy": { "type": "string" } } }
                                ]
                            }
                        },
                        "uptime_secs": {
                            "type": "integer",
                            "format": "int64",
                            "description": "Process uptime in seconds."
                        }
                    }
                });

                paths.insert("/healthz".to_string(), serde_json::json!({
                    "get": {
                        "summary": "Liveness probe",
                        "description": "Returns 200 when the service is healthy or degraded, \
                                        503 when any registered health check is Unhealthy.",
                        "operationId": "healthz",
                        "tags": ["observability"],
                        "responses": {
                            "200": {
                                "description": "Healthy or degraded — service is alive.",
                                "content": { "application/json": { "schema": health_schema } }
                            },
                            "503": {
                                "description": "Unhealthy — one or more subsystems have failed."
                            }
                        }
                    }
                }));

                paths.insert("/readyz".to_string(), serde_json::json!({
                    "get": {
                        "summary": "Readiness probe",
                        "description": "Identical to /healthz. Use for Kubernetes `readinessProbe`; \
                                        load-balancers should stop sending traffic when this returns 503.",
                        "operationId": "readyz",
                        "tags": ["observability"],
                        "responses": {
                            "200": { "description": "Ready to serve traffic." },
                            "503": { "description": "Not ready — remove from load-balancer rotation." }
                        }
                    }
                }));

                paths.insert("/metrics".to_string(), serde_json::json!({
                    "get": {
                        "summary": "Prometheus metrics scrape",
                        "description": "Returns Prometheus text-format (version 0.0.4) metrics. \
                                        Counters: `http_requests_total`. \
                                        Histogram: `http_request_duration_seconds`. \
                                        Gauge: `http_requests_in_flight`.",
                        "operationId": "metrics",
                        "tags": ["observability"],
                        "responses": {
                            "200": {
                                "description": "Prometheus text-format metric lines.",
                                "content": {
                                    "text/plain": {
                                        "schema": { "type": "string" }
                                    }
                                }
                            }
                        }
                    }
                }));
            });

            tracing::info!(
                service = self.service_name,
                version = self.service_version,
                otlp = self.otlp_endpoint,
                "ArclyObservabilityPlugin initialised"
            );

            Ok(())
        })
    }

    fn on_shutdown<'a>(
        &'a self,
        _container: &'static FrozenDiContainer,
    ) -> BoxFuture<'a, Result<(), PluginError>> {
        Box::pin(async move {
            opentelemetry::global::shutdown_tracer_provider();
            Ok(())
        })
    }
}