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    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
323        let ctx = match self.phase {
324            Phase::Start => self.run_start(ctx),
325            Phase::End => self.run_end(ctx),
326        };
327        Ok(PluginOutput {
328            context: ctx,
329            named_outputs: HashMap::new(),
330        })
331    }
332}
333
334#[cfg(test)]
335mod tests {
336    use super::*;
337    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
338    use bytes::Bytes;
339
340    fn test_ctx() -> Context {
341        Context {
342            request: GatewayRequest {
343                method: "GET".to_string(),
344                path: "/api/users".to_string(),
345                host: "api.example.com".to_string(),
346                scheme: "http".to_string(),
347                headers: HashMap::new(),
348                query_params: HashMap::new(),
349                body: Bytes::new(),
350                remote_addr: "10.0.0.1:1234".to_string(),
351                protocol: Protocol::Http1,
352            },
353            response: GatewayResponse {
354                status_code: 200,
355                headers: HashMap::new(),
356                body: Bytes::new(),
357            },
358            message: HashMap::new(),
359            errors: Vec::new(),
360        }
361    }
362
363    fn plugin(config: Value) -> Result<SkywalkingPlugin, String> {
364        let map: HashMap<String, Value> = serde_json::from_value(config).unwrap();
365        SkywalkingPlugin::from_config(&map, &PluginResources::empty())
366    }
367
368    #[test]
369    fn from_config_requires_phase() {
370        assert!(plugin(json!({})).is_err());
371        assert!(plugin(json!({ "phase": "middle" })).is_err());
372        assert!(plugin(json!({ "phase": "start" })).is_ok());
373        assert!(plugin(json!({ "phase": "end" })).is_ok());
374        // sample_ratio out of range
375        assert!(plugin(json!({ "phase": "start", "sample_ratio": 2 })).is_err());
376    }
377
378    #[test]
379    fn from_config_defaults() {
380        let p = plugin(json!({ "phase": "end" })).unwrap();
381        assert_eq!(p.endpoint_addr, "http://127.0.0.1:12800");
382        assert_eq!(p.service_name, "featherbit");
383        assert_eq!(p.service_instance_name, "featherbit Instance Name");
384        assert_eq!(p.sample_ratio, 1.0);
385        assert!(p.ssl_verify);
386    }
387
388    #[test]
389    fn sampler_decision() {
390        let never = plugin(json!({ "phase": "start", "sample_ratio": 0 })).unwrap();
391        let always = plugin(json!({ "phase": "start", "sample_ratio": 1 })).unwrap();
392        for _ in 0..100 {
393            assert!(!never.should_sample(), "ratio 0 must never sample");
394            assert!(always.should_sample(), "ratio 1 must always sample");
395        }
396    }
397
398    #[test]
399    fn start_stores_span_and_injects_sw8() {
400        let p = plugin(json!({ "phase": "start", "sample_ratio": 1 })).unwrap();
401        let ctx = p.run_start(test_ctx());
402
403        // span stored for the end node
404        let span = load_span(&ctx).expect("start node must store a span");
405        assert_eq!(span.trace_id.len(), 32);
406        assert_eq!(span.span_id.len(), 16);
407        assert!(span.sampled);
408
409        // downstream sw8 header injected and round-trips back to our span
410        let header = ctx.request.headers.get("sw8").unwrap().first().unwrap();
411        let (trace_id, parent_span, sampled) = parse_sw8(header).expect("injected sw8 must parse");
412        assert_eq!(trace_id, span.trace_id);
413        assert_eq!(parent_span, "0");
414        assert!(sampled);
415    }
416
417    #[test]
418    fn start_continues_incoming_trace() {
419        let p = plugin(json!({ "phase": "start", "sample_ratio": 0 })).unwrap();
420        let incoming_trace = "1f2d4bf47bf711eab794acde48001122";
421        let incoming = format!(
422            "1-{}-{}-7-{}-{}-{}-{}",
423            sw8_encode(incoming_trace),
424            sw8_encode("parent-segment"),
425            sw8_encode("caller-svc"),
426            sw8_encode("caller-inst"),
427            sw8_encode("/caller"),
428            sw8_encode("peer:80"),
429        );
430        let mut ctx = test_ctx();
431        ctx.request
432            .headers
433            .insert("sw8".to_string(), vec![incoming]);
434
435        let ctx = p.run_start(ctx);
436        let span = load_span(&ctx).unwrap();
437        // trace id + sampled flag continued from the incoming header, despite
438        // sample_ratio 0 (which only governs *new* traces)
439        assert_eq!(span.trace_id, incoming_trace);
440        assert!(span.sampled);
441        assert_eq!(span.parent_span_id.as_deref(), Some("7"));
442    }
443
444    #[test]
445    fn segment_payload_shape() {
446        let span = SpanContext {
447            trace_id: "1f2d4bf47bf711eab794acde48001122".to_string(),
448            span_id: "b7ad6b7169203331".to_string(),
449            parent_span_id: None,
450            sampled: true,
451            start_ms: 1_700_000_000_000,
452        };
453        let seg = build_segment(
454            &span,
455            "svc",
456            "inst",
457            "POST",
458            "/api/x",
459            500,
460            1_700_000_000_123,
461        );
462        assert_eq!(seg["traceId"], span.trace_id);
463        assert_eq!(seg["traceSegmentId"], span.span_id);
464        assert_eq!(seg["service"], "svc");
465        assert_eq!(seg["serviceInstance"], "inst");
466        let s = &seg["spans"][0];
467        assert_eq!(s["spanId"], 0);
468        assert_eq!(s["parentSpanId"], -1);
469        assert_eq!(s["spanType"], "Entry");
470        assert_eq!(s["spanLayer"], "Http");
471        assert_eq!(s["componentId"], COMPONENT_ID_HTTP);
472        assert_eq!(s["operationName"], "/api/x");
473        assert_eq!(s["startTime"], 1_700_000_000_000u64);
474        assert_eq!(s["endTime"], 1_700_000_000_123u64);
475        // status 500 -> isError true
476        assert_eq!(s["isError"], true);
477        assert_eq!(s["tags"][0]["key"], "http.method");
478        assert_eq!(s["tags"][0]["value"], "POST");
479        assert_eq!(s["tags"][2]["value"], "500");
480    }
481
482    #[tokio::test]
483    async fn end_returns_ok_without_span() {
484        // No start ran -> no span -> nothing exported, still Ok.
485        let p = plugin(json!({ "phase": "end" })).unwrap();
486        let out = p.execute(test_ctx(), &HashMap::new()).await.unwrap();
487        assert_eq!(out.context.response.status_code, 200);
488    }
489
490    #[tokio::test]
491    async fn end_returns_ok_and_spawns_export_when_sampled() {
492        // Point at a dead port; the spawned export fails and is ignored.
493        let p = plugin(json!({ "phase": "end", "endpoint_addr": "http://127.0.0.1:1" })).unwrap();
494        let start = plugin(json!({ "phase": "start", "sample_ratio": 1 })).unwrap();
495        let ctx = start.run_start(test_ctx());
496        let out = p.execute(ctx, &HashMap::new()).await.unwrap();
497        // context passes through unchanged
498        assert_eq!(out.context.request.path, "/api/users");
499    }
500
501    #[tokio::test]
502    async fn end_skips_export_when_not_sampled() {
503        let p = plugin(json!({ "phase": "end" })).unwrap();
504        let start = plugin(json!({ "phase": "start", "sample_ratio": 0 })).unwrap();
505        let ctx = start.run_start(test_ctx());
506        assert!(!load_span(&ctx).unwrap().sampled);
507        let out = p.execute(ctx, &HashMap::new()).await.unwrap();
508        assert_eq!(out.context.request.path, "/api/users");
509    }
510}
featherbit/plugins/native/skywalking.rs

featherbit/plugins/native/
skywalking.rs
