rskit-observability 0.2.0-alpha.1

use std::time::Duration;

use opentelemetry::metrics::{Counter, Gauge, Histogram, UpDownCounter};
use serde::Deserialize;

use rskit_errors::{AppError, AppResult, ErrorCode};

use crate::tracer::{OtlpProtocol, SERVICE_NAME};

/// Configuration for the OpenTelemetry metrics pipeline.
#[derive(Debug, Clone, Deserialize)]
pub struct MetricsConfig {
    /// Logical service name attached to every metric.
    pub service_name: String,
    /// How often metrics are exported to the collector.
    pub export_interval: Duration,
    /// OTLP endpoint. When `None`, metrics are collected but not exported.
    pub otlp_endpoint: Option<String>,
}

/// Handle to the metrics pipeline — use it to create instruments.
pub struct MetricsHandle {
    meter: opentelemetry::metrics::Meter,
    _provider: opentelemetry_sdk::metrics::SdkMeterProvider,
}

impl MetricsHandle {
    /// Create a monotonic `Counter<u64>`.
    pub fn counter(&self, name: impl Into<String>, description: impl Into<String>) -> Counter<u64> {
        self.meter
            .u64_counter(name.into())
            .with_description(description.into())
            .build()
    }

    /// Create a `Histogram<f64>` for latency distributions etc.
    pub fn histogram(
        &self,
        name: impl Into<String>,
        description: impl Into<String>,
    ) -> Histogram<f64> {
        self.meter
            .f64_histogram(name.into())
            .with_description(description.into())
            .build()
    }

    /// Create a `Gauge<f64>` for point-in-time values.
    pub fn gauge(&self, name: impl Into<String>, description: impl Into<String>) -> Gauge<f64> {
        self.meter
            .f64_gauge(name.into())
            .with_description(description.into())
            .build()
    }

    /// Create an `UpDownCounter<i64>` for values that go up and down.
    pub fn up_down_counter(
        &self,
        name: impl Into<String>,
        description: impl Into<String>,
    ) -> UpDownCounter<i64> {
        self.meter
            .i64_up_down_counter(name.into())
            .with_description(description.into())
            .build()
    }
}

/// Initialise an OpenTelemetry metrics pipeline with optional OTLP export.
pub fn init_metrics(cfg: &MetricsConfig) -> AppResult<MetricsHandle> {
    init_metrics_with_protocol(cfg, OtlpProtocol::Grpc)
}

/// Initialise an OpenTelemetry metrics pipeline with an explicit OTLP protocol.
pub fn init_metrics_with_protocol(
    cfg: &MetricsConfig,
    protocol: OtlpProtocol,
) -> AppResult<MetricsHandle> {
    use opentelemetry::metrics::MeterProvider as _;
    use opentelemetry::{InstrumentationScope, KeyValue};
    use opentelemetry_sdk::Resource;
    use opentelemetry_sdk::metrics::SdkMeterProvider;

    let resource = Resource::builder_empty()
        .with_attributes([KeyValue::new(SERVICE_NAME, cfg.service_name.clone())])
        .build();

    let builder = SdkMeterProvider::builder().with_resource(resource);

    if let Some(endpoint) = &cfg.otlp_endpoint {
        #[cfg(not(feature = "otlp"))]
        {
            let _ = (endpoint, protocol);
            Err(AppError::new(
                ErrorCode::InvalidInput,
                "OTLP metric exporter requires the `otlp` feature",
            ))?;
        }

        #[cfg(feature = "otlp")]
        {
            use opentelemetry_otlp::{MetricExporter, WithExportConfig};
            use opentelemetry_sdk::metrics::PeriodicReader;

            let exporter = match protocol {
                OtlpProtocol::Grpc => MetricExporter::builder()
                    .with_tonic()
                    .with_endpoint(endpoint)
                    .build(),
                OtlpProtocol::HttpBinary => MetricExporter::builder()
                    .with_http()
                    .with_protocol(opentelemetry_otlp::Protocol::HttpBinary)
                    .with_endpoint(endpoint)
                    .build(),
            }
            .map_err(|e| {
                AppError::new(ErrorCode::Internal, format!("OTLP metric exporter: {e}"))
            })?;

            let reader = PeriodicReader::builder(exporter)
                .with_interval(cfg.export_interval)
                .build();

            let builder = builder.with_reader(reader);
            let provider = builder.build();
            let scope = InstrumentationScope::builder(cfg.service_name.clone()).build();
            let meter = provider.meter_with_scope(scope);

            return Ok(MetricsHandle {
                meter,
                _provider: provider,
            });
        }
    }

    let provider = builder.build();
    let scope = InstrumentationScope::builder(cfg.service_name.clone()).build();
    let meter = provider.meter_with_scope(scope);

    Ok(MetricsHandle {
        meter,
        _provider: provider,
    })
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::*;

    fn config(endpoint: Option<&str>) -> MetricsConfig {
        MetricsConfig {
            service_name: "metrics-test".to_string(),
            export_interval: Duration::from_millis(50),
            otlp_endpoint: endpoint.map(str::to_string),
        }
    }

    #[test]
    fn no_export_metrics_pipeline_creates_all_instrument_kinds() {
        let metrics = init_metrics(&config(None)).expect("build metrics provider");

        metrics.counter("requests", "request count").add(1, &[]);
        metrics
            .histogram("latency", "request latency")
            .record(12.5, &[]);
        metrics.gauge("load", "current load").record(0.75, &[]);
        metrics
            .up_down_counter("inflight", "inflight requests")
            .add(1, &[]);
    }

    #[cfg(not(feature = "otlp"))]
    #[test]
    fn otlp_endpoint_requires_feature() {
        let error = match init_metrics_with_protocol(
            &config(Some("http://127.0.0.1:4317")),
            OtlpProtocol::Grpc,
        ) {
            Ok(_) => panic!("otlp disabled should reject metric exporter configuration"),
            Err(error) => error,
        };

        assert_eq!(error.code(), ErrorCode::InvalidInput);
        assert!(error.message().contains("otlp"));
    }

    #[cfg(feature = "otlp")]
    #[tokio::test]
    async fn otlp_metrics_pipeline_builds_for_supported_protocols() {
        for protocol in [OtlpProtocol::Grpc, OtlpProtocol::HttpBinary] {
            let metrics =
                init_metrics_with_protocol(&config(Some("http://127.0.0.1:4317")), protocol)
                    .expect("build otlp metrics provider");
            metrics.counter("requests", "request count").add(1, &[]);
        }
    }
}