kmp_observability/
otlp_query_adapter.rs

1use std::future::Future;
2use std::pin::Pin;
3
4use kmp_application::{
5    ObservabilityExemplar, ObservabilityProjection, ObservabilityQuery, ObservabilityQueryPort,
6    ObservabilitySeries,
7};
8use kmp_domain::PortError;
9
10/// Query response produced by a concrete telemetry backend (Prometheus,
11/// Tempo, vendor API, etc.) over data originally exported with OTLP.
12///
13/// OTLP itself is an ingest protocol and deliberately has no query RPC. This
14/// seam keeps that backend-specific API client outside KMP while the adapter
15/// gives every renderer one stable application contract.
16#[derive(Debug, Clone, PartialEq)]
17pub struct OtlpQueryResponse {
18    pub series: Vec<ObservabilitySeries>,
19    pub exemplars: Vec<ObservabilityExemplar>,
20    pub missing: Vec<String>,
21    pub truncated: bool,
22}
23
24pub trait OtlpMetricsQueryClient: Send + Sync {
25    fn query_range<'a>(
26        &'a self,
27        query: &'a ObservabilityQuery,
28    ) -> Pin<Box<dyn Future<Output = Result<OtlpQueryResponse, PortError>> + Send + 'a>>;
29}
30
31/// Maps an OTLP-compatible backend's query API onto the application port.
32/// Values retain backend-provided unit, scope and exemplars; this adapter
33/// derives neither health scores nor causal links.
34#[derive(Debug, Clone)]
35pub struct OtlpObservabilityQueryAdapter<C> {
36    client: C,
37}
38
39impl<C> OtlpObservabilityQueryAdapter<C> {
40    pub fn new(client: C) -> Self {
41        Self { client }
42    }
43}
44
45impl<C> ObservabilityQueryPort for OtlpObservabilityQueryAdapter<C>
46where
47    C: OtlpMetricsQueryClient,
48{
49    fn query<'a>(
50        &'a self,
51        query: ObservabilityQuery,
52    ) -> Pin<Box<dyn Future<Output = Result<ObservabilityProjection, PortError>> + Send + 'a>> {
53        Box::pin(async move {
54            let response = self.client.query_range(&query).await?;
55            Ok(ObservabilityProjection {
56                contract: "kmp.observability.projection.v1".to_string(),
57                from_millis: query.from_millis,
58                to_millis: query.to_millis,
59                series: response.series,
60                exemplars: response.exemplars,
61                missing: response.missing,
62                truncated: response.truncated,
63            })
64        })
65    }
66}
67
68#[cfg(test)]
69mod tests {
70    use super::*;
71
72    struct FakeClient;
73
74    impl OtlpMetricsQueryClient for FakeClient {
75        fn query_range<'a>(
76            &'a self,
77            _query: &'a ObservabilityQuery,
78        ) -> Pin<Box<dyn Future<Output = Result<OtlpQueryResponse, PortError>> + Send + 'a>>
79        {
80            Box::pin(async {
81                Ok(OtlpQueryResponse {
82                    series: vec![ObservabilitySeries {
83                        name: "rpc_duration".to_string(),
84                        unit: "seconds".to_string(),
85                        scope: "rpc".to_string(),
86                        points: Vec::new(),
87                    }],
88                    exemplars: Vec::new(),
89                    missing: Vec::new(),
90                    truncated: false,
91                })
92            })
93        }
94    }
95
96    #[tokio::test]
97    async fn adapter_preserves_exact_metric_semantics() {
98        let adapter = OtlpObservabilityQueryAdapter::new(FakeClient);
99        let result = adapter
100            .query(ObservabilityQuery {
101                about: Some("project:kmp".to_string()),
102                from_millis: 10,
103                to_millis: 20,
104                series: vec!["rpc_duration".to_string()],
105                max_points: 100,
106            })
107            .await
108            .expect("backend query");
109        assert_eq!(result.series[0].unit, "seconds");
110        assert_eq!(result.series[0].scope, "rpc");
111    }
112}
kmp_observability/otlp_query_adapter.rs

kmp_observability/
otlp_query_adapter.rs
