featherbit/plugins/native/
skywalking.rs

1//! Distributed tracing with Apache SkyWalking — a **start/end node pair**
2//! wrapped around the `upstream` node.
3//!
4//! Ported from APISIX's `skywalking.lua`. The wire propagation format is the
5//! SkyWalking `sw8` header (see [`crate::plugins::util::trace`]).
6//!
7//! - The **start** node (`phase: start`, placed **before** `upstream`) extracts
8//!   an incoming `sw8` header to continue an existing trace, or begins a new one
9//!   and makes the sampling decision. It creates this hop's
10//!   [`SpanContext`](crate::plugins::util::trace::SpanContext), stores it in
11//!   `context.message`, and injects a fresh downstream `sw8` header so the
12//!   upstream service joins the trace.
13//! - The **end** node (`phase: end`, placed **after** `upstream`) loads the
14//!   span, and — when sampled — fire-and-forget POSTs a SkyWalking trace
15//!   *segment* to `<endpoint_addr>/v3/segments` on a detached task. The request
16//!   path is never blocked by the export.
17//!
18//! Both nodes always continue through the `success` port; neither ever routes
19//! to the error port.
20//!
21//! ## Wiring
22//! ```text
23//! ... -> skywalking(start) -> upstream -> skywalking(end) -> ...
24//! ```
25//!
26//! ## Deviations / simplifications from APISIX
27//! Full SkyWalking correlation (segment references, multi-span segments, the
28//! `ngx.ctx` entry/exit span pair, the background report timer) is
29//! intentionally reduced to a **faithful subset**:
30//! - Each request exports a **single-span** segment (`spanId: 0`,
31//!   `parentSpanId: -1`, `spanType: "Entry"`, `spanLayer: "Http"`). The
32//!   incoming `sw8`'s parent segment/service are honored for propagation
33//!   (trace id + sampling flag are continued) but are **not** emitted as a
34//!   formal `refs` segment-reference on the exported span.
35//! - Export is per-request and immediate (a detached `tokio::spawn`), not
36//!   buffered on APISIX's `report_interval` timer.
37//! - The segment is POSTed as a single JSON object to `/v3/segments`; the real
38//!   OAP endpoint also accepts a batch array — the single-object form is the
39//!   documented subset here.
40//! - `componentId` is reported as `49` (generic HTTP) rather than APISIX's
41//!   `6002`.
42
43use async_trait::async_trait;
44use std::collections::HashMap;
45use std::sync::Arc;
46use std::time::Duration;
47
48use serde_json::{json, Value};
49
50use crate::context::Context;
51use crate::outbound::{OutboundClient, OutboundRequest};
52use crate::plugins::resources::PluginResources;
53use crate::plugins::util::trace::{
54    load_span, new_span_id, new_trace_id, now_ms, parse_sw8, store_span, sw8_encode, SpanContext,
55};
56use crate::plugins::{Plugin, PluginOutput, PluginResult};
57
58/// SkyWalking `componentId` reported for the entry span (49 = generic HTTP).
59const COMPONENT_ID_HTTP: i64 = 49;
60
61/// Which node of the start/end pair this instance is.
62#[derive(Debug, Clone, Copy, PartialEq)]
63enum Phase {
64    /// Before `upstream`: extract/create the span and inject `sw8`.
65    Start,
66    /// After `upstream`: load the span and export the segment.
67    End,
68}
69
70/// SkyWalking tracing node (a start or end half, selected by `phase`).
71pub struct SkywalkingPlugin {
72    phase: Phase,
73    /// OAP HTTP base, e.g. `http://127.0.0.1:12800` (no trailing slash).
74    endpoint_addr: String,
75    service_name: String,
76    service_instance_name: String,
77    /// Fraction of new traces to sample, in `0.0..=1.0`.
78    sample_ratio: f64,
79    ssl_verify: bool,
80    timeout: Duration,
81    /// Shared pooled outbound HTTP client (used by the end node's export).
82    client: Arc<OutboundClient>,
83}
84
85/// Draws a pseudo-random fraction in `[0.0, 1.0)` for sampling — the same
86/// cheap, non-cryptographic source `proxy-mirror`/`fault-injection` use.
87fn roll_fraction() -> f64 {
88    use std::collections::hash_map::RandomState;
89    use std::hash::{BuildHasher, Hasher};
90    let n = RandomState::new().build_hasher().finish();
91    n as f64 / (u64::MAX as f64 + 1.0)
92}
93
94/// Builds a downstream `sw8` header value for `span` (8 hyphen-separated
95/// fields, matching the SkyWalking cross-process propagation header):
96/// `sample-traceId-parentSegmentId-parentSpanId-service-instance-endpoint-peer`.
97/// All fields except the sample flag and the plain-integer parent span id are
98/// base64-encoded.
99fn build_sw8(span: &SpanContext, service: &str, instance: &str, endpoint: &str) -> String {
100    format!(
101        "{}-{}-{}-{}-{}-{}-{}-{}",
102        if span.sampled { "1" } else { "0" },
103        sw8_encode(&span.trace_id),
104        sw8_encode(&span.span_id), // this hop's segment id
105        "0",                       // parent span id within our segment (plain int)
106        sw8_encode(service),
107        sw8_encode(instance),
108        sw8_encode(endpoint),
109        sw8_encode(endpoint), // target address (peer); we reuse the endpoint
110    )
111}
112
113/// Builds the SkyWalking trace *segment* JSON for a finished span (pure; does
114/// no I/O). A single Entry/Http span mirroring the OAP HTTP segment protocol.
115fn build_segment(
116    span: &SpanContext,
117    service: &str,
118    instance: &str,
119    method: &str,
120    path: &str,
121    status: u16,
122    end_ms: u64,
123) -> Value {
124    json!({
125        "traceId": span.trace_id,
126        "traceSegmentId": span.span_id,
127        "service": service,
128        "serviceInstance": instance,
129        "spans": [{
130            "spanId": 0,
131            "parentSpanId": -1,
132            "startTime": span.start_ms,
133            "endTime": end_ms,
134            "operationName": path,
135            "spanType": "Entry",
136            "spanLayer": "Http",
137            "componentId": COMPONENT_ID_HTTP,
138            "isError": status >= 400,
139            "tags": [
140                { "key": "http.method", "value": method },
141                { "key": "http.path", "value": path },
142                { "key": "http.status_code", "value": status.to_string() },
143            ],
144        }],
145    })
146}
147
148impl SkywalkingPlugin {
149    /// Builds the plugin from node config.
150    ///
151    /// Config keys:
152    ///
153    /// | Key | Type | Default | Description |
154    /// |---|---|---|---|
155    /// | `phase` | string | — (**required**) | `start` (before `upstream`) or `end` (after `upstream`). |
156    /// | `endpoint_addr` | string | `http://127.0.0.1:12800` | SkyWalking OAP HTTP base. |
157    /// | `service_name` | string | `"featherbit"` | Service name reported on the segment / `sw8`. |
158    /// | `service_instance_name` | string | `"featherbit Instance Name"` | Service instance name. |
159    /// | `sample_ratio` | number | `1.0` | Fraction of *new* traces to sample, in `0.0..=1.0`. Requests arriving with a sampled `sw8` are always continued. |
160    /// | `ssl_verify` | bool | `true` | Verify the OAP TLS certificate on export. |
161    /// | `timeout` | int (seconds) | `3` | Per-export HTTP timeout (end node). |
162    ///
163    /// ```yaml
164    /// # before upstream
165    /// - id: sw-start
166    ///   type: skywalking
167    ///   config: { phase: start, service_name: my-gateway, sample_ratio: 1.0 }
168    /// # after upstream
169    /// - id: sw-end
170    ///   type: skywalking
171    ///   config: { phase: end, endpoint_addr: http://127.0.0.1:12800, service_name: my-gateway }
172    /// ```
173    pub fn from_config(
174        config: &HashMap<String, Value>,
175        resources: &Arc<PluginResources>,
176    ) -> Result<Self, String> {
177        let phase = match config.get("phase").and_then(|v| v.as_str()) {
178            Some("start") => Phase::Start,
179            Some("end") => Phase::End,
180            _ => {
181                return Err(
182                    "skywalking requires `phase: start` (before upstream) or `phase: end` (after upstream)"
183                        .to_string(),
184                )
185            }
186        };
187
188        let endpoint_addr = config
189            .get("endpoint_addr")
190            .and_then(|v| v.as_str())
191            .filter(|s| !s.is_empty())
192            .unwrap_or("http://127.0.0.1:12800")
193            .trim_end_matches('/')
194            .to_string();
195
196        let service_name = config
197            .get("service_name")
198            .and_then(|v| v.as_str())
199            .unwrap_or("featherbit")
200            .to_string();
201
202        let service_instance_name = config
203            .get("service_instance_name")
204            .and_then(|v| v.as_str())
205            .unwrap_or("featherbit Instance Name")
206            .to_string();
207
208        let sample_ratio = match config.get("sample_ratio") {
209            None => 1.0,
210            Some(v) => v
211                .as_f64()
212                .filter(|r| (0.0..=1.0).contains(r))
213                .ok_or("skywalking `sample_ratio` must be a number in 0.0..=1.0")?,
214        };
215
216        let ssl_verify = config
217            .get("ssl_verify")
218            .and_then(|v| v.as_bool())
219            .unwrap_or(true);
220
221        let timeout =
222            Duration::from_secs(config.get("timeout").and_then(|v| v.as_u64()).unwrap_or(3));
223
224        Ok(Self {
225            phase,
226            endpoint_addr,
227            service_name,
228            service_instance_name,
229            sample_ratio,
230            ssl_verify,
231            timeout,
232            client: resources.outbound.clone(),
233        })
234    }
235
236    /// Sampling decision for a *new* trace (no inbound `sw8`).
237    fn should_sample(&self) -> bool {
238        if self.sample_ratio >= 1.0 {
239            true
240        } else if self.sample_ratio <= 0.0 {
241            false
242        } else {
243            roll_fraction() < self.sample_ratio
244        }
245    }
246
247    /// Start-node logic: continue or begin the trace, store the span, and
248    /// inject the downstream `sw8` header.
249    fn run_start(&self, mut ctx: Context) -> Context {
250        let incoming = ctx
251            .request
252            .headers
253            .get("sw8")
254            .and_then(|v| v.first())
255            .and_then(|s| parse_sw8(s));
256
257        let (trace_id, parent_span_id, sampled) = match incoming {
258            Some((trace_id, parent, sampled)) => (trace_id, Some(parent), sampled),
259            None => (new_trace_id(), None, self.should_sample()),
260        };
261
262        let span = SpanContext {
263            trace_id,
264            span_id: new_span_id(),
265            parent_span_id,
266            sampled,
267            start_ms: now_ms(),
268        };
269        store_span(&mut ctx, &span);
270
271        let sw8 = build_sw8(
272            &span,
273            &self.service_name,
274            &self.service_instance_name,
275            &ctx.request.path,
276        );
277        ctx.request.headers.insert("sw8".to_string(), vec![sw8]);
278
279        ctx
280    }
281
282    /// End-node logic: load the span and, when sampled, fire-and-forget export
283    /// the segment. Returns the context unchanged.
284    fn run_end(&self, ctx: Context) -> Context {
285        if let Some(span) = load_span(&ctx) {
286            if span.sampled {
287                let segment = build_segment(
288                    &span,
289                    &self.service_name,
290                    &self.service_instance_name,
291                    &ctx.request.method,
292                    &ctx.request.path,
293                    ctx.response.status_code,
294                    now_ms(),
295                );
296                let body = serde_json::to_vec(&segment).unwrap_or_default();
297                let req = OutboundRequest {
298                    method: http::Method::POST,
299                    url: format!("{}/v3/segments", self.endpoint_addr),
300                    headers: vec![("Content-Type".to_string(), "application/json".to_string())],
301                    body: body.into(),
302                    timeout: self.timeout,
303                    ssl_verify: self.ssl_verify,
304                    tls: None,
305                };
306                let client = self.client.clone();
307                tokio::spawn(async move {
308                    let _ = client.request(req).await;
309                });
310            }
311        }
312        ctx
313    }
314}
315
316#[async_trait]
317impl Plugin for SkywalkingPlugin {
318    fn plugin_type(&self) -> &str {
319        "skywalking"
320    }
321
322    fn reads_response_body(&self) -> bool {
323        false
324    }
325
326    async fn execute(&self, ctx: Context) -> PluginResult {
327        let ctx = match self.phase {
328            Phase::Start => self.run_start(ctx),
329            Phase::End => self.run_end(ctx),
330        };
331        Ok(PluginOutput::success(ctx))
332    }
333}
334
335#[cfg(test)]
336mod tests {
337    use super::*;
338    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
339    use bytes::Bytes;
340
341    fn test_ctx() -> Context {
342        Context {
343            request: GatewayRequest {
344                method: "GET".to_string(),
345                path: "/api/users".to_string(),
346                host: "api.example.com".to_string(),
347                scheme: "http".to_string(),
348                headers: HashMap::new(),
349                query_params: HashMap::new(),
350                body: Bytes::new(),
351                remote_addr: "10.0.0.1:1234".to_string(),
352                protocol: Protocol::Http1,
353            },
354            response: GatewayResponse {
355                status_code: 200,
356                headers: HashMap::new(),
357                body: Bytes::new(),
358                stream: None,
359            },
360            message: HashMap::new(),
361            errors: Vec::new(),
362        }
363    }
364
365    fn plugin(config: Value) -> Result<SkywalkingPlugin, String> {
366        let map: HashMap<String, Value> = serde_json::from_value(config).unwrap();
367        SkywalkingPlugin::from_config(&map, &PluginResources::empty())
368    }
369
370    #[test]
371    fn from_config_requires_phase() {
372        assert!(plugin(json!({})).is_err());
373        assert!(plugin(json!({ "phase": "middle" })).is_err());
374        assert!(plugin(json!({ "phase": "start" })).is_ok());
375        assert!(plugin(json!({ "phase": "end" })).is_ok());
376        // sample_ratio out of range
377        assert!(plugin(json!({ "phase": "start", "sample_ratio": 2 })).is_err());
378    }
379
380    #[test]
381    fn from_config_defaults() {
382        let p = plugin(json!({ "phase": "end" })).unwrap();
383        assert_eq!(p.endpoint_addr, "http://127.0.0.1:12800");
384        assert_eq!(p.service_name, "featherbit");
385        assert_eq!(p.service_instance_name, "featherbit Instance Name");
386        assert_eq!(p.sample_ratio, 1.0);
387        assert!(p.ssl_verify);
388    }
389
390    #[test]
391    fn sampler_decision() {
392        let never = plugin(json!({ "phase": "start", "sample_ratio": 0 })).unwrap();
393        let always = plugin(json!({ "phase": "start", "sample_ratio": 1 })).unwrap();
394        for _ in 0..100 {
395            assert!(!never.should_sample(), "ratio 0 must never sample");
396            assert!(always.should_sample(), "ratio 1 must always sample");
397        }
398    }
399
400    #[test]
401    fn start_stores_span_and_injects_sw8() {
402        let p = plugin(json!({ "phase": "start", "sample_ratio": 1 })).unwrap();
403        let ctx = p.run_start(test_ctx());
404
405        // span stored for the end node
406        let span = load_span(&ctx).expect("start node must store a span");
407        assert_eq!(span.trace_id.len(), 32);
408        assert_eq!(span.span_id.len(), 16);
409        assert!(span.sampled);
410
411        // downstream sw8 header injected and round-trips back to our span
412        let header = ctx.request.headers.get("sw8").unwrap().first().unwrap();
413        let (trace_id, parent_span, sampled) = parse_sw8(header).expect("injected sw8 must parse");
414        assert_eq!(trace_id, span.trace_id);
415        assert_eq!(parent_span, "0");
416        assert!(sampled);
417    }
418
419    #[test]
420    fn start_continues_incoming_trace() {
421        let p = plugin(json!({ "phase": "start", "sample_ratio": 0 })).unwrap();
422        let incoming_trace = "1f2d4bf47bf711eab794acde48001122";
423        let incoming = format!(
424            "1-{}-{}-7-{}-{}-{}-{}",
425            sw8_encode(incoming_trace),
426            sw8_encode("parent-segment"),
427            sw8_encode("caller-svc"),
428            sw8_encode("caller-inst"),
429            sw8_encode("/caller"),
430            sw8_encode("peer:80"),
431        );
432        let mut ctx = test_ctx();
433        ctx.request
434            .headers
435            .insert("sw8".to_string(), vec![incoming]);
436
437        let ctx = p.run_start(ctx);
438        let span = load_span(&ctx).unwrap();
439        // trace id + sampled flag continued from the incoming header, despite
440        // sample_ratio 0 (which only governs *new* traces)
441        assert_eq!(span.trace_id, incoming_trace);
442        assert!(span.sampled);
443        assert_eq!(span.parent_span_id.as_deref(), Some("7"));
444    }
445
446    #[test]
447    fn segment_payload_shape() {
448        let span = SpanContext {
449            trace_id: "1f2d4bf47bf711eab794acde48001122".to_string(),
450            span_id: "b7ad6b7169203331".to_string(),
451            parent_span_id: None,
452            sampled: true,
453            start_ms: 1_700_000_000_000,
454        };
455        let seg = build_segment(
456            &span,
457            "svc",
458            "inst",
459            "POST",
460            "/api/x",
461            500,
462            1_700_000_000_123,
463        );
464        assert_eq!(seg["traceId"], span.trace_id);
465        assert_eq!(seg["traceSegmentId"], span.span_id);
466        assert_eq!(seg["service"], "svc");
467        assert_eq!(seg["serviceInstance"], "inst");
468        let s = &seg["spans"][0];
469        assert_eq!(s["spanId"], 0);
470        assert_eq!(s["parentSpanId"], -1);
471        assert_eq!(s["spanType"], "Entry");
472        assert_eq!(s["spanLayer"], "Http");
473        assert_eq!(s["componentId"], COMPONENT_ID_HTTP);
474        assert_eq!(s["operationName"], "/api/x");
475        assert_eq!(s["startTime"], 1_700_000_000_000u64);
476        assert_eq!(s["endTime"], 1_700_000_000_123u64);
477        // status 500 -> isError true
478        assert_eq!(s["isError"], true);
479        assert_eq!(s["tags"][0]["key"], "http.method");
480        assert_eq!(s["tags"][0]["value"], "POST");
481        assert_eq!(s["tags"][2]["value"], "500");
482    }
483
484    #[tokio::test]
485    async fn end_returns_ok_without_span() {
486        // No start ran -> no span -> nothing exported, still Ok.
487        let p = plugin(json!({ "phase": "end" })).unwrap();
488        let out = p.execute(test_ctx()).await.unwrap();
489        assert_eq!(out.context.response.status_code, 200);
490    }
491
492    #[tokio::test]
493    async fn end_returns_ok_and_spawns_export_when_sampled() {
494        // Point at a dead port; the spawned export fails and is ignored.
495        let p = plugin(json!({ "phase": "end", "endpoint_addr": "http://127.0.0.1:1" })).unwrap();
496        let start = plugin(json!({ "phase": "start", "sample_ratio": 1 })).unwrap();
497        let ctx = start.run_start(test_ctx());
498        let out = p.execute(ctx).await.unwrap();
499        // context passes through unchanged
500        assert_eq!(out.context.request.path, "/api/users");
501    }
502
503    #[tokio::test]
504    async fn end_skips_export_when_not_sampled() {
505        let p = plugin(json!({ "phase": "end" })).unwrap();
506        let start = plugin(json!({ "phase": "start", "sample_ratio": 0 })).unwrap();
507        let ctx = start.run_start(test_ctx());
508        assert!(!load_span(&ctx).unwrap().sampled);
509        let out = p.execute(ctx).await.unwrap();
510        assert_eq!(out.context.request.path, "/api/users");
511    }
512}
featherbit/plugins/native/skywalking.rs

featherbit/plugins/native/
skywalking.rs
