|
| 1 | +//! OpenTelemetry OTLP meter-provider construction. |
| 2 | +//! |
| 3 | +//! See [`build_otlp_meter_provider`] for the public entry point used by both |
| 4 | +//! the binary (`main.rs`) and the integration tests. |
| 5 | +
|
| 6 | +use std::time::Duration; |
| 7 | + |
| 8 | +use async_trait::async_trait; |
| 9 | +use opentelemetry::KeyValue; |
| 10 | +use opentelemetry_http::hyper::HyperClient; |
| 11 | +use opentelemetry_http::{Bytes, HttpClient, HttpError, Request, Response}; |
| 12 | +use opentelemetry_otlp::{WithExportConfig, WithHttpConfig}; |
| 13 | +use opentelemetry_sdk::metrics::SdkMeterProvider; |
| 14 | +use opentelemetry_sdk::Resource; |
| 15 | + |
| 16 | +/// Build an OTLP/HTTP `SdkMeterProvider` pinned to the mailroom's `ring` |
| 17 | +/// crypto provider. |
| 18 | +/// |
| 19 | +/// `opentelemetry-otlp`'s `reqwest-rustls` feature pulls in `aws-lc-rs` (plus |
| 20 | +/// its native `cmake` build) via `reqwest` 0.13. We avoid that by enabling only |
| 21 | +/// the `hyper-client` feature and supplying our own `hyper_rustls` |
| 22 | +/// `HttpsConnector`, which reuses the `ring`-backed `rustls` 0.23 already in |
| 23 | +/// the dependency graph. |
| 24 | +/// |
| 25 | +/// The caller is responsible for invoking |
| 26 | +/// `opentelemetry::global::set_meter_provider` and installing any tracing |
| 27 | +/// subscriber; this function deliberately touches no global state so it remains |
| 28 | +/// unit-testable. |
| 29 | +/// |
| 30 | +/// # Panics |
| 31 | +/// Panics if no Tokio runtime is active on the current thread. The OTLP |
| 32 | +/// exporter's `HyperClient` requires a Tokio context, which we capture here |
| 33 | +/// (the SDK's `PeriodicReader` otherwise polls exporters from a bare thread |
| 34 | +/// with no runtime, panicking on the first export). |
| 35 | +pub fn build_otlp_meter_provider( |
| 36 | + endpoint: &str, |
| 37 | + auth_token: &str, |
| 38 | + operator_domain: &str, |
| 39 | +) -> SdkMeterProvider { |
| 40 | + let resource = Resource::builder() |
| 41 | + .with_service_name("payjoin-mailroom") |
| 42 | + .with_attribute(KeyValue::new("operator.domain", operator_domain.to_string())) |
| 43 | + .build(); |
| 44 | + |
| 45 | + let headers: std::collections::HashMap<String, String> = |
| 46 | + [("Authorization".to_string(), format!("Basic {auth_token}"))].into(); |
| 47 | + |
| 48 | + let connector = hyper_rustls::HttpsConnectorBuilder::new() |
| 49 | + .with_webpki_roots() |
| 50 | + .https_or_http() |
| 51 | + .enable_http1() |
| 52 | + .enable_http2() |
| 53 | + .build(); |
| 54 | + |
| 55 | + // The SDK's `PeriodicReader` drives exports on a plain OS thread via |
| 56 | + // `futures_executor::block_on`, with no Tokio context. `HyperClient` calls |
| 57 | + // `tokio::time::timeout`, which panics outside a Tokio runtime. Capture the |
| 58 | + // ambient handle and dispatch each request onto it. |
| 59 | + let handle = tokio::runtime::Handle::try_current() |
| 60 | + .expect("build_otlp_meter_provider must be called from a Tokio runtime"); |
| 61 | + let http_client = TokioDispatchClient { |
| 62 | + inner: HyperClient::new(connector, Duration::from_secs(10), None), |
| 63 | + handle, |
| 64 | + }; |
| 65 | + |
| 66 | + let metric_exporter = opentelemetry_otlp::MetricExporter::builder() |
| 67 | + .with_http() |
| 68 | + .with_endpoint(format!("{endpoint}/v1/metrics")) |
| 69 | + .with_headers(headers) |
| 70 | + .with_http_client(http_client) |
| 71 | + .build() |
| 72 | + .expect("Failed to build OTLP metric exporter"); |
| 73 | + |
| 74 | + SdkMeterProvider::builder() |
| 75 | + .with_periodic_exporter(metric_exporter) |
| 76 | + .with_resource(resource) |
| 77 | + .build() |
| 78 | +} |
| 79 | + |
| 80 | +/// `HttpClient` adapter that runs the inner client on a captured Tokio handle. |
| 81 | +/// |
| 82 | +/// The OpenTelemetry SDK's `PeriodicReader` polls exporters from a standalone |
| 83 | +/// thread (via `futures_executor::block_on`), not from the application's Tokio |
| 84 | +/// runtime. `HyperClient::send_bytes` uses `tokio::time::timeout`, which panics |
| 85 | +/// without a Tokio context. This wrapper spawns each request onto the captured |
| 86 | +/// handle and awaits the `JoinHandle`, which is safe to poll from any executor. |
| 87 | +#[derive(Debug, Clone)] |
| 88 | +struct TokioDispatchClient<C> |
| 89 | +where |
| 90 | + C: hyper_util::client::legacy::connect::Connect |
| 91 | + + Clone |
| 92 | + + Send |
| 93 | + + Sync |
| 94 | + + std::fmt::Debug |
| 95 | + + 'static, |
| 96 | +{ |
| 97 | + inner: HyperClient<C>, |
| 98 | + handle: tokio::runtime::Handle, |
| 99 | +} |
| 100 | + |
| 101 | +#[async_trait] |
| 102 | +impl<C> HttpClient for TokioDispatchClient<C> |
| 103 | +where |
| 104 | + C: hyper_util::client::legacy::connect::Connect |
| 105 | + + Clone |
| 106 | + + Send |
| 107 | + + Sync |
| 108 | + + std::fmt::Debug |
| 109 | + + 'static, |
| 110 | +{ |
| 111 | + async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError> { |
| 112 | + let inner = self.inner.clone(); |
| 113 | + self.handle.spawn(async move { HttpClient::send_bytes(&inner, request).await }).await? |
| 114 | + } |
| 115 | +} |
| 116 | + |
| 117 | +#[cfg(test)] |
| 118 | +mod tests { |
| 119 | + use super::*; |
| 120 | + use crate::metrics::MetricsService; |
| 121 | + |
| 122 | + /// Regression test for the OTLP transport swap (opentelemetry 0.32). |
| 123 | + /// |
| 124 | + /// Replaces the manual `mock_otlp.py` + `curl` + 65s-wait smoke test. |
| 125 | + /// Confirms the telemetry feature functions end-to-end: |
| 126 | + /// |
| 127 | + /// 1. The exporter builds without panicking (`NoHttpClient` is gone, since |
| 128 | + /// we supply a `hyper-client` explicitly instead of the dropped |
| 129 | + /// `reqwest-rustls` feature). |
| 130 | + /// 2. The HTTP transport works under the SDK's `PeriodicReader` -- the |
| 131 | + /// `TokioDispatchClient` bridges the standalone export thread onto the |
| 132 | + /// test's Tokio runtime, so no "no reactor running" panic. |
| 133 | + /// 3. The wire request is well-formed OTLP/protobuf with the configured |
| 134 | + /// Basic-auth header, proving the encoding + auth path survived the |
| 135 | + /// version bump. |
| 136 | + /// |
| 137 | + /// If crypto-provider overlap ever returns (e.g. `aws-lc-rs` sneaks back |
| 138 | + /// in alongside `ring`), the existing compile-time / link-time tests that |
| 139 | + /// guard the rustls provider selection catch that; this test does not. |
| 140 | + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] |
| 141 | + async fn otlp_exporter_posts_metrics_over_http() { |
| 142 | + let mut server = mockito::Server::new_async().await; |
| 143 | + let mock = server |
| 144 | + .mock("POST", "/v1/metrics") |
| 145 | + .match_header("authorization", "Basic dXNlcjpwYXNz") |
| 146 | + .match_header("content-type", "application/x-protobuf") |
| 147 | + .with_status(200) |
| 148 | + .create_async() |
| 149 | + .await; |
| 150 | + |
| 151 | + let provider = build_otlp_meter_provider(&server.url(), "dXNlcjpwYXNz", "test.example.com"); |
| 152 | + |
| 153 | + // Drive a real meter through the same MetricsService the app uses so |
| 154 | + // the export batch is non-empty. |
| 155 | + let metrics = MetricsService::new(Some(provider.clone())); |
| 156 | + metrics.record_http_request("/health", "GET", 200); |
| 157 | + |
| 158 | + provider.force_flush().expect("force_flush"); |
| 159 | + |
| 160 | + mock.assert_async().await; |
| 161 | + |
| 162 | + let _ = provider.shutdown(); |
| 163 | + } |
| 164 | +} |
0 commit comments