featherbit/plugins/native/
opentelemetry.rs

1//! Distributed tracing via OpenTelemetry (`opentelemetry`).
2//!
3//! The featherbit port of Apache APISIX's `opentelemetry` plugin. Unlike the
4//! single-phase APISIX plugin, this is a **start/end node pair** wired around
5//! the `upstream` node, using the shared [`trace`](crate::plugins::util::trace)
6//! helper to carry a [`SpanContext`] through `context.message`:
7//!
8//! - **start node** (`phase: start`, placed right after `listener`, before
9//!   `upstream`): extracts the W3C `traceparent` header. If present, this hop
10//!   *continues* the incoming trace (same `trace_id`, the caller's span becomes
11//!   our parent, and the incoming sampled flag is honored). If absent, a *new*
12//!   trace is started (`new_trace_id()`, no parent, sampled per the `sampler`
13//!   config). Either way this hop gets a fresh `span_id` and start time, the
14//!   span is stored via [`store_span`], and the outgoing `traceparent` header is
15//!   injected so the upstream service continues the trace.
16//! - **end node** (`phase: end`, placed after `upstream`, right before
17//!   `client`): loads the span, computes its duration, and — when sampled —
18//!   builds an OTLP/HTTP JSON payload and **fire-and-forgets** a POST to the
19//!   collector's `/v1/traces` endpoint on a detached `tokio::spawn` task (the
20//!   same best-effort pattern as `proxy-mirror`). The export result is ignored
21//!   and never blocks or fails the request.
22//!
23//! Both nodes always return `Ok`: tracing is observability, so it never
24//! short-circuits the request through an error port. If the end node finds no
25//! stored span (start node absent or misordered) it passes through untouched.
26//!
27//! ## Deviations from APISIX
28//! - Spans are exported one-per-request (fire-and-forget) rather than through a
29//!   batch span processor; `batch_span_processor` config is not supported.
30//! - Sampler strategies are `always_on`, `always_off`, and `trace_id_ratio`
31//!   (`parent_base` is not supported). The new-trace sampling draw is a
32//!   *pseudo-random*, per-trace-consistent hash of the trace id (no `rand`
33//!   crate), so a given trace id always samples the same way in a process.
34//! - OTLP/JSON id fields (`traceId`, `spanId`, `parentSpanId`) are emitted as
35//!   lowercase hex strings, which is the canonical OTLP/JSON encoding accepted
36//!   by modern collectors.
37
38use std::collections::HashMap;
39use std::sync::Arc;
40use std::time::Duration;
41
42use async_trait::async_trait;
43use serde_json::{json, Value};
44
45use crate::context::Context;
46use crate::outbound::{OutboundClient, OutboundRequest};
47use crate::plugins::resources::PluginResources;
48use crate::plugins::util::trace::{
49    build_traceparent, load_span, new_span_id, new_trace_id, now_ms, parse_traceparent, store_span,
50    SpanContext,
51};
52use crate::plugins::{Plugin, PluginOutput, PluginResult};
53
54/// Which node of the tracing pair this instance is.
55#[derive(Debug, Clone, Copy, PartialEq)]
56enum Phase {
57    /// Before `upstream`: extract/create the span and inject `traceparent`.
58    Start,
59    /// After `upstream`: export the finished span (fire-and-forget).
60    End,
61}
62
63/// New-trace sampling strategy (used only when no incoming `traceparent`).
64#[derive(Debug, Clone, Copy)]
65enum Sampler {
66    AlwaysOn,
67    AlwaysOff,
68    /// Sample a fraction of new traces, in `0.0..=1.0`.
69    TraceIdRatio(f64),
70}
71
72impl Sampler {
73    /// Decides whether a brand-new trace should be sampled.
74    ///
75    /// `always_on`/`always_off` are unconditional; `trace_id_ratio` compares a
76    /// pseudo-random, per-trace-consistent draw derived from `trace_id` against
77    /// the fraction (see [`ratio_draw`]).
78    fn sample_new(&self, trace_id: &str) -> bool {
79        match self {
80            Sampler::AlwaysOn => true,
81            Sampler::AlwaysOff => false,
82            Sampler::TraceIdRatio(f) => ratio_draw(trace_id) < *f,
83        }
84    }
85}
86
87/// A pseudo-random draw in `[0.0, 1.0)` derived deterministically from a trace
88/// id, so the same trace id always samples the same way within a process.
89/// Not cryptographic; good enough for ratio sampling without a `rand` crate.
90fn ratio_draw(trace_id: &str) -> f64 {
91    use std::collections::hash_map::DefaultHasher;
92    use std::hash::{Hash, Hasher};
93    let mut h = DefaultHasher::new();
94    trace_id.hash(&mut h);
95    h.finish() as f64 / (u64::MAX as f64 + 1.0)
96}
97
98/// One node of the OpenTelemetry tracing pair.
99pub struct OpenTelemetryPlugin {
100    phase: Phase,
101    /// OTLP/HTTP collector base URL (no trailing slash); traces POST to
102    /// `<collector>/v1/traces`.
103    collector: String,
104    service_name: String,
105    sampler: Sampler,
106    ssl_verify: bool,
107    timeout: Duration,
108    client: Arc<OutboundClient>,
109}
110
111/// Parses the `sampler` config object into a [`Sampler`], failing fast on an
112/// unknown strategy name or a malformed `fraction`.
113fn parse_sampler(config: &HashMap<String, Value>) -> Result<Sampler, String> {
114    let Some(s) = config.get("sampler") else {
115        return Ok(Sampler::AlwaysOn);
116    };
117    let name = s
118        .get("name")
119        .and_then(|v| v.as_str())
120        .unwrap_or("always_on");
121    let fraction = s
122        .get("options")
123        .and_then(|o| o.get("fraction"))
124        .map(|f| {
125            f.as_f64()
126                .filter(|f| (0.0..=1.0).contains(f))
127                .ok_or("opentelemetry `sampler.options.fraction` must be a number in 0.0..=1.0")
128        })
129        .transpose()?
130        .unwrap_or(1.0);
131    match name {
132        "always_on" => Ok(Sampler::AlwaysOn),
133        "always_off" => Ok(Sampler::AlwaysOff),
134        "trace_id_ratio" => Ok(Sampler::TraceIdRatio(fraction)),
135        other => Err(format!(
136            "opentelemetry unknown sampler `{other}` (expected always_on, always_off, trace_id_ratio)"
137        )),
138    }
139}
140
141impl OpenTelemetryPlugin {
142    /// Builds the plugin from node config.
143    ///
144    /// Config keys:
145    ///
146    /// | Key | Type | Default | Description |
147    /// |---|---|---|---|
148    /// | `phase` | string | — (**required**) | `start` (before upstream) or `end` (after upstream). |
149    /// | `endpoint` / `collector` | string | `http://localhost:4318` | OTLP/HTTP collector base URL; traces POST to `<collector>/v1/traces`. |
150    /// | `service_name` | string | `"featherbit"` | `service.name` resource attribute reported on each span. |
151    /// | `sampler.name` | string | `always_on` | `always_on`, `always_off`, or `trace_id_ratio`. |
152    /// | `sampler.options.fraction` | number | `1.0` | Fraction of new traces to sample for `trace_id_ratio`, in `0.0..=1.0`. |
153    /// | `ssl_verify` | bool | `true` | Verify the collector's TLS certificate. |
154    /// | `timeout` | int (seconds) | `3` | Export request timeout (end node). |
155    ///
156    /// Fails fast on an unknown `phase`, an unknown sampler name, or a
157    /// `fraction` outside `0.0..=1.0`.
158    ///
159    /// ```yaml
160    /// # start node — right after listener, before upstream
161    /// - id: otel-start
162    ///   type: opentelemetry
163    ///   config:
164    ///     phase: start
165    ///     service_name: my-gateway
166    ///     sampler: { name: trace_id_ratio, options: { fraction: 0.1 } }
167    /// # end node — after upstream, right before client
168    /// - id: otel-end
169    ///   type: opentelemetry
170    ///   config:
171    ///     phase: end
172    ///     endpoint: http://localhost:4318
173    /// ```
174    pub fn from_config(
175        config: &HashMap<String, Value>,
176        resources: &Arc<PluginResources>,
177    ) -> Result<Self, String> {
178        let phase = match config.get("phase").and_then(|v| v.as_str()) {
179            Some("start") => Phase::Start,
180            Some("end") => Phase::End,
181            Some(other) => {
182                return Err(format!(
183                    "opentelemetry `phase` must be `start` or `end` (got `{other}`)"
184                ))
185            }
186            None => return Err("opentelemetry requires `phase` (`start` or `end`)".to_string()),
187        };
188
189        let collector = config
190            .get("endpoint")
191            .or_else(|| config.get("collector"))
192            .and_then(|v| v.as_str())
193            .filter(|s| !s.is_empty())
194            .unwrap_or("http://localhost:4318")
195            .trim_end_matches('/')
196            .to_string();
197
198        let service_name = config
199            .get("service_name")
200            .and_then(|v| v.as_str())
201            .unwrap_or("featherbit")
202            .to_string();
203
204        let sampler = parse_sampler(config)?;
205
206        let ssl_verify = config
207            .get("ssl_verify")
208            .and_then(|v| v.as_bool())
209            .unwrap_or(true);
210        let timeout =
211            Duration::from_secs(config.get("timeout").and_then(|v| v.as_u64()).unwrap_or(3));
212
213        Ok(Self {
214            phase,
215            collector,
216            service_name,
217            sampler,
218            ssl_verify,
219            timeout,
220            client: resources.outbound.clone(),
221        })
222    }
223
224    /// START: extract or create the span, store it, and inject `traceparent`.
225    fn run_start(&self, ctx: &mut Context) {
226        let incoming = ctx
227            .request
228            .headers
229            .get("traceparent")
230            .and_then(|v| v.first())
231            .and_then(|v| parse_traceparent(v));
232
233        let (trace_id, parent_span_id, sampled) = match incoming {
234            Some((trace_id, parent, sampled)) => (trace_id, Some(parent), sampled),
235            None => {
236                let trace_id = new_trace_id();
237                let sampled = self.sampler.sample_new(&trace_id);
238                (trace_id, None, sampled)
239            }
240        };
241
242        let span = SpanContext {
243            trace_id,
244            span_id: new_span_id(),
245            parent_span_id,
246            sampled,
247            start_ms: now_ms(),
248        };
249
250        // Inject the downstream header so the upstream continues the trace.
251        ctx.request
252            .headers
253            .insert("traceparent".to_string(), vec![build_traceparent(&span)]);
254
255        store_span(ctx, &span);
256    }
257
258    /// END: load the span and fire-and-forget the OTLP export when sampled.
259    fn run_end(&self, ctx: &Context) {
260        let Some(span) = load_span(ctx) else {
261            return;
262        };
263        if !span.sampled {
264            return;
265        }
266
267        let payload = build_otlp(&span, ctx, &self.service_name);
268        let body = match serde_json::to_vec(&payload) {
269            Ok(b) => b,
270            Err(_) => return,
271        };
272
273        let req = OutboundRequest {
274            method: http::Method::POST,
275            url: format!("{}/v1/traces", self.collector),
276            headers: vec![("Content-Type".to_string(), "application/json".to_string())],
277            body: body.into(),
278            timeout: self.timeout,
279            ssl_verify: self.ssl_verify,
280            tls: None,
281        };
282        let client = self.client.clone();
283        // Fire-and-forget: the export never blocks or affects the request path.
284        tokio::spawn(async move {
285            let _ = client.request(req).await;
286        });
287    }
288}
289
290/// Builds the OTLP/HTTP JSON payload for a finished span (pure; no I/O).
291///
292/// Ids are emitted as lowercase hex strings (the canonical OTLP/JSON encoding).
293/// `status.code` is `2` (ERROR) for a 5xx response, else `0` (UNSET).
294fn build_otlp(span: &SpanContext, ctx: &Context, service_name: &str) -> Value {
295    let start_nanos = span.start_ms.saturating_mul(1_000_000);
296    let end_nanos = (span.start_ms + span.duration_ms()).saturating_mul(1_000_000);
297    let status = ctx.response.status_code;
298    let name = format!("{} {}", ctx.request.method, ctx.request.path);
299
300    let mut span_json = json!({
301        "traceId": span.trace_id,
302        "spanId": span.span_id,
303        "name": name,
304        "kind": 2, // SERVER
305        "startTimeUnixNano": start_nanos.to_string(),
306        "endTimeUnixNano": end_nanos.to_string(),
307        "attributes": [
308            str_attr("http.method", &ctx.request.method),
309            int_attr("http.status_code", status as i64),
310            str_attr("http.target", &ctx.request.path),
311            str_attr("http.host", &ctx.request.host),
312        ],
313        "status": { "code": if status >= 500 { 2 } else { 0 } },
314    });
315    if let Some(parent) = &span.parent_span_id {
316        span_json["parentSpanId"] = json!(parent);
317    }
318
319    json!({
320        "resourceSpans": [{
321            "resource": {
322                "attributes": [ str_attr("service.name", service_name) ]
323            },
324            "scopeSpans": [{
325                "spans": [ span_json ]
326            }]
327        }]
328    })
329}
330
331fn str_attr(key: &str, value: &str) -> Value {
332    json!({ "key": key, "value": { "stringValue": value } })
333}
334
335fn int_attr(key: &str, value: i64) -> Value {
336    json!({ "key": key, "value": { "intValue": value.to_string() } })
337}
338
339#[async_trait]
340impl Plugin for OpenTelemetryPlugin {
341    fn plugin_type(&self) -> &str {
342        "opentelemetry"
343    }
344
345    async fn execute(
346        &self,
347        mut ctx: Context,
348        _named_inputs: &HashMap<String, Value>,
349    ) -> PluginResult {
350        match self.phase {
351            Phase::Start => self.run_start(&mut ctx),
352            Phase::End => self.run_end(&ctx),
353        }
354        Ok(PluginOutput {
355            context: ctx,
356            named_outputs: HashMap::new(),
357        })
358    }
359}
360
361#[cfg(test)]
362mod tests {
363    use super::*;
364    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
365    use bytes::Bytes;
366
367    fn plugin(config: Value) -> Result<OpenTelemetryPlugin, String> {
368        let map: HashMap<String, Value> = serde_json::from_value(config).unwrap();
369        OpenTelemetryPlugin::from_config(&map, &PluginResources::empty())
370    }
371
372    fn test_ctx() -> Context {
373        Context {
374            request: GatewayRequest {
375                method: "GET".to_string(),
376                path: "/api/users".to_string(),
377                host: "example.com".to_string(),
378                scheme: "http".to_string(),
379                headers: HashMap::new(),
380                query_params: HashMap::new(),
381                body: Bytes::new(),
382                remote_addr: "10.0.0.1:5555".to_string(),
383                protocol: Protocol::Http1,
384            },
385            response: GatewayResponse {
386                status_code: 200,
387                headers: HashMap::new(),
388                body: Bytes::new(),
389            },
390            message: HashMap::new(),
391            errors: Vec::new(),
392        }
393    }
394
395    fn a_span(sampled: bool, parent: Option<&str>) -> SpanContext {
396        SpanContext {
397            trace_id: "0af7651916cd43dd8448eb211c80319c".to_string(),
398            span_id: "b7ad6b7169203331".to_string(),
399            parent_span_id: parent.map(String::from),
400            sampled,
401            start_ms: 1_700_000_000_000,
402        }
403    }
404
405    #[test]
406    fn config_requires_valid_phase() {
407        assert!(plugin(json!({})).is_err());
408        assert!(plugin(json!({ "phase": "middle" })).is_err());
409        assert!(plugin(json!({ "phase": "start" })).is_ok());
410        assert!(plugin(json!({ "phase": "end" })).is_ok());
411    }
412
413    #[test]
414    fn config_rejects_bad_sampler() {
415        assert!(plugin(json!({ "phase": "start", "sampler": { "name": "bogus" } })).is_err());
416        assert!(plugin(json!({
417            "phase": "start", "sampler": { "name": "trace_id_ratio", "options": { "fraction": 2 } }
418        }))
419        .is_err());
420        assert!(plugin(json!({
421            "phase": "start", "sampler": { "name": "trace_id_ratio", "options": { "fraction": 0.5 } }
422        }))
423        .is_ok());
424    }
425
426    #[test]
427    fn sampler_extremes() {
428        assert!(!Sampler::AlwaysOff.sample_new("abc"));
429        assert!(Sampler::AlwaysOn.sample_new("abc"));
430        assert!(!Sampler::TraceIdRatio(0.0).sample_new("abc"));
431        assert!(Sampler::TraceIdRatio(1.0).sample_new("abc"));
432    }
433
434    #[test]
435    fn otlp_payload_shape() {
436        let mut ctx = test_ctx();
437        ctx.response.status_code = 503;
438        let span = a_span(true, Some("0020000000000001"));
439        let v = build_otlp(&span, &ctx, "svc");
440        let s = &v["resourceSpans"][0]["scopeSpans"][0]["spans"][0];
441        assert_eq!(s["traceId"], "0af7651916cd43dd8448eb211c80319c");
442        assert_eq!(s["spanId"], "b7ad6b7169203331");
443        assert_eq!(s["parentSpanId"], "0020000000000001");
444        assert_eq!(s["kind"], 2);
445        assert_eq!(s["name"], "GET /api/users");
446        // 5xx maps to ERROR (2)
447        assert_eq!(s["status"]["code"], 2);
448        // service.name resource attribute
449        assert_eq!(
450            v["resourceSpans"][0]["resource"]["attributes"][0]["value"]["stringValue"],
451            "svc"
452        );
453    }
454
455    #[test]
456    fn otlp_omits_parent_when_root() {
457        let span = a_span(true, None);
458        let v = build_otlp(&span, &test_ctx(), "svc");
459        let s = &v["resourceSpans"][0]["scopeSpans"][0]["spans"][0];
460        assert!(s.get("parentSpanId").is_none());
461        // 2xx maps to UNSET (0)
462        assert_eq!(s["status"]["code"], 0);
463    }
464
465    #[tokio::test]
466    async fn start_creates_new_trace_and_injects_header() {
467        let p = plugin(json!({ "phase": "start", "sampler": { "name": "always_on" } })).unwrap();
468        let out = p.execute(test_ctx(), &HashMap::new()).await.unwrap();
469        // A span was stored...
470        let span = load_span(&out.context).expect("span stored");
471        assert!(span.parent_span_id.is_none());
472        assert!(span.sampled);
473        assert_eq!(span.trace_id.len(), 32);
474        // ...and the downstream traceparent was injected, matching the span.
475        let hdr = &out.context.request.headers["traceparent"][0];
476        assert_eq!(*hdr, build_traceparent(&span));
477    }
478
479    #[tokio::test]
480    async fn start_continues_incoming_trace() {
481        let p = plugin(json!({ "phase": "start" })).unwrap();
482        let mut ctx = test_ctx();
483        ctx.request.headers.insert(
484            "traceparent".to_string(),
485            vec!["00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01".to_string()],
486        );
487        let out = p.execute(ctx, &HashMap::new()).await.unwrap();
488        let span = load_span(&out.context).unwrap();
489        assert_eq!(span.trace_id, "0af7651916cd43dd8448eb211c80319c");
490        assert_eq!(span.parent_span_id.as_deref(), Some("b7ad6b7169203331"));
491        assert!(span.sampled);
492        // Our own span id replaces the caller's in the injected header.
493        assert_ne!(span.span_id, "b7ad6b7169203331");
494    }
495
496    #[tokio::test]
497    async fn end_passes_through_without_span() {
498        let p = plugin(json!({ "phase": "end" })).unwrap();
499        let out = p.execute(test_ctx(), &HashMap::new()).await.unwrap();
500        assert_eq!(out.context.response.status_code, 200);
501    }
502
503    #[tokio::test]
504    async fn end_returns_ok_with_stored_span() {
505        let p = plugin(json!({ "phase": "end", "endpoint": "http://127.0.0.1:1" })).unwrap();
506        let mut ctx = test_ctx();
507        store_span(&mut ctx, &a_span(true, None));
508        // The spawned export is best-effort; we only assert execute returns Ok.
509        let out = p.execute(ctx, &HashMap::new()).await.unwrap();
510        assert_eq!(out.context.request.path, "/api/users");
511    }
512
513    #[tokio::test]
514    async fn end_skips_export_when_unsampled() {
515        let p = plugin(json!({ "phase": "end" })).unwrap();
516        let mut ctx = test_ctx();
517        store_span(&mut ctx, &a_span(false, None));
518        let out = p.execute(ctx, &HashMap::new()).await.unwrap();
519        assert_eq!(out.context.response.status_code, 200);
520    }
521}
featherbit/plugins/native/opentelemetry.rs

featherbit/plugins/native/
opentelemetry.rs
