featherbit/plugins/native/
skywalking_logger.rs

1//! The `skywalking-logger` node — ships access-log entries to a SkyWalking OAP
2//! (Observability Analysis Platform) HTTP endpoint in batches.
3//!
4//! Ported from APISIX's `skywalking-logger.lua`. On the request path the node
5//! builds a log entry (via the shared [`build_entry`](crate::plugins::util::log_entry::build_entry)),
6//! wraps it in a SkyWalking `LogData` item, and hands it to a [`BatchSink`].
7//! A background task POSTs the buffered items as a JSON array to
8//! `<endpoint_addr>/v3/logs`; the node itself never blocks and passes the
9//! context through unchanged.
10//!
11//! ## Deviations from APISIX
12//! - The APISIX plugin parses the inbound `sw8` trace header to attach a
13//!   `traceContext` to each item. featherbit does not thread trace propagation
14//!   through the log entry, so `traceContext` is omitted.
15//! - APISIX wraps the entry as `body.json.json`; we mirror that shape. A
16//!   millisecond `timestamp` is added (SkyWalking's `LogData.timestamp`), which
17//!   the APISIX plugin leaves to the OAP server to stamp.
18
19use std::collections::HashMap;
20use std::sync::Arc;
21use std::time::Duration;
22
23use async_trait::async_trait;
24use serde_json::{json, Value};
25
26use crate::batch::{BatchConfig, BatchFlusher, BatchSink, FlushError};
27use crate::context::Context;
28use crate::outbound::{OutboundClient, OutboundRequest};
29use crate::plugins::resources::PluginResources;
30use crate::plugins::util::log_entry::{build_entry, parse_log_format};
31use crate::plugins::{Plugin, PluginOutput, PluginResult};
32
33/// Ships log entries to a SkyWalking OAP endpoint in batches.
34pub struct SkywalkingLoggerPlugin {
35    sink: BatchSink,
36    log_format: Option<HashMap<String, Value>>,
37    include_req_body: bool,
38    include_resp_body: bool,
39    service_name: String,
40    service_instance_name: String,
41}
42
43/// Delivers batched SkyWalking `LogData` items to `<endpoint>/v3/logs`.
44struct SkywalkingFlusher {
45    client: Arc<OutboundClient>,
46    url: String,
47    timeout: Duration,
48    ssl_verify: bool,
49}
50
51impl SkywalkingLoggerPlugin {
52    /// Builds the plugin from node config.
53    ///
54    /// Config keys:
55    ///
56    /// | Key | Type | Default | Description |
57    /// |---|---|---|---|
58    /// | `endpoint_addr` | string | — (required) | SkyWalking OAP HTTP base, e.g. `http://127.0.0.1:12800`. |
59    /// | `service_name` | string | `"featherbit"` | SkyWalking service name reported on each item. |
60    /// | `service_instance_name` | string | `"featherbit Instance Name"` | SkyWalking service instance name. |
61    /// | `ssl_verify` | bool | `true` | Verify the OAP TLS certificate. |
62    /// | `timeout` | int (seconds) | `3` | Per-flush HTTP timeout. |
63    /// | `log_format` | object | — | Custom `name -> "$var template"` entry; when set the default entry is replaced. |
64    /// | `include_req_body` | bool | `false` | Include the (lossy UTF-8) request body in the default entry. |
65    /// | `include_resp_body` | bool | `false` | Include the (lossy UTF-8) response body in the default entry. |
66    ///
67    /// Batch-processor keys (`batch_max_size`, `inactive_timeout`,
68    /// `buffer_duration`, `max_retry_count`, `retry_delay`,
69    /// `max_pending_entries`) are parsed by [`BatchConfig::from_config`].
70    ///
71    /// ```yaml
72    /// type: skywalking-logger
73    /// config:
74    ///   endpoint_addr: http://127.0.0.1:12800
75    ///   service_name: my-gateway
76    ///   service_instance_name: gw-1
77    /// ```
78    pub fn from_config(
79        config: &HashMap<String, Value>,
80        resources: &Arc<PluginResources>,
81    ) -> Result<Self, String> {
82        let endpoint_addr = config
83            .get("endpoint_addr")
84            .and_then(|v| v.as_str())
85            .filter(|s| !s.is_empty())
86            .ok_or("skywalking-logger: `endpoint_addr` is required")?
87            .trim_end_matches('/')
88            .to_string();
89
90        let service_name = config
91            .get("service_name")
92            .and_then(|v| v.as_str())
93            .unwrap_or("featherbit")
94            .to_string();
95        let service_instance_name = config
96            .get("service_instance_name")
97            .and_then(|v| v.as_str())
98            .unwrap_or("featherbit Instance Name")
99            .to_string();
100        let ssl_verify = config
101            .get("ssl_verify")
102            .and_then(|v| v.as_bool())
103            .unwrap_or(true);
104        let timeout =
105            Duration::from_secs(config.get("timeout").and_then(|v| v.as_u64()).unwrap_or(3));
106
107        let log_format = parse_log_format(config)?;
108        let include_req_body = bool_key(config, "include_req_body");
109        let include_resp_body = bool_key(config, "include_resp_body");
110
111        let batch_cfg =
112            BatchConfig::from_config(config).map_err(|e| format!("skywalking-logger: {e}"))?;
113
114        let flusher = Arc::new(SkywalkingFlusher {
115            client: resources.outbound.clone(),
116            url: format!("{endpoint_addr}/v3/logs"),
117            timeout,
118            ssl_verify,
119        });
120        let sink = BatchSink::spawn("skywalking-logger", batch_cfg, flusher);
121
122        Ok(Self {
123            sink,
124            log_format,
125            include_req_body,
126            include_resp_body,
127            service_name,
128            service_instance_name,
129        })
130    }
131}
132
133/// Builds one SkyWalking `LogData` item wrapping `entry` (JSON-encoded into
134/// `body.json.json`, mirroring APISIX).
135fn build_log_item(
136    entry: &Value,
137    service: &str,
138    service_instance: &str,
139    endpoint: &str,
140    timestamp_ms: u64,
141) -> Value {
142    let entry_str = serde_json::to_string(entry).unwrap_or_else(|_| "{}".to_string());
143    json!({
144        "service": service,
145        "serviceInstance": service_instance,
146        "endpoint": endpoint,
147        "timestamp": timestamp_ms,
148        "body": { "json": { "json": entry_str } },
149    })
150}
151
152fn bool_key(config: &HashMap<String, Value>, key: &str) -> bool {
153    config.get(key).and_then(|v| v.as_bool()).unwrap_or(false)
154}
155
156fn now_ms() -> u64 {
157    std::time::SystemTime::now()
158        .duration_since(std::time::UNIX_EPOCH)
159        .map(|d| d.as_millis() as u64)
160        .unwrap_or(0)
161}
162
163#[async_trait]
164impl BatchFlusher for SkywalkingFlusher {
165    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
166        let body = serde_json::to_vec(&Value::Array(entries.to_vec())).map_err(|e| FlushError {
167            message: format!("failed to encode log batch: {e}"),
168            first_fail: None,
169        })?;
170
171        let req = OutboundRequest {
172            method: http::Method::POST,
173            url: self.url.clone(),
174            headers: vec![("Content-Type".to_string(), "application/json".to_string())],
175            body: body.into(),
176            timeout: self.timeout,
177            ssl_verify: self.ssl_verify,
178            tls: None,
179        };
180
181        match self.client.request(req).await {
182            Ok(resp) if resp.status < 400 => Ok(()),
183            Ok(resp) => Err(FlushError {
184                message: format!(
185                    "skywalking OAP returned status {}: {}",
186                    resp.status,
187                    String::from_utf8_lossy(&resp.body)
188                ),
189                first_fail: None,
190            }),
191            Err(e) => Err(FlushError {
192                message: e.to_string(),
193                first_fail: None,
194            }),
195        }
196    }
197}
198
199#[async_trait]
200impl Plugin for SkywalkingLoggerPlugin {
201    fn plugin_type(&self) -> &str {
202        "skywalking-logger"
203    }
204
205    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
206        let entry = build_entry(
207            &ctx,
208            self.log_format.as_ref(),
209            self.include_req_body,
210            self.include_resp_body,
211        );
212        let item = build_log_item(
213            &entry,
214            &self.service_name,
215            &self.service_instance_name,
216            &ctx.request.path,
217            now_ms(),
218        );
219        self.sink.push(item);
220
221        Ok(PluginOutput {
222            context: ctx,
223            named_outputs: HashMap::new(),
224        })
225    }
226}
227
228#[cfg(test)]
229mod tests {
230    use super::*;
231
232    fn cfg(pairs: &[(&str, Value)]) -> HashMap<String, Value> {
233        pairs
234            .iter()
235            .map(|(k, v)| (k.to_string(), v.clone()))
236            .collect()
237    }
238
239    #[test]
240    fn from_config_requires_endpoint() {
241        let res = SkywalkingLoggerPlugin::from_config(&HashMap::new(), &PluginResources::empty());
242        let Err(e) = res else {
243            panic!("expected error")
244        };
245        assert!(e.contains("endpoint_addr"));
246    }
247
248    #[tokio::test]
249    async fn from_config_defaults() {
250        let c = cfg(&[("endpoint_addr", json!("http://oap:12800/"))]);
251        let p = SkywalkingLoggerPlugin::from_config(&c, &PluginResources::empty()).unwrap();
252        assert_eq!(p.service_name, "featherbit");
253        assert_eq!(p.service_instance_name, "featherbit Instance Name");
254        assert!(!p.include_req_body);
255    }
256
257    #[test]
258    fn log_item_shape() {
259        let entry = json!({ "request": { "method": "GET" }, "response": { "status": 200 } });
260        let item = build_log_item(&entry, "svc", "inst", "/api/x", 1_700_000_000_000);
261        assert_eq!(item["service"], "svc");
262        assert_eq!(item["serviceInstance"], "inst");
263        assert_eq!(item["endpoint"], "/api/x");
264        assert_eq!(item["timestamp"], 1_700_000_000_000u64);
265        // The entry is embedded as a JSON *string* under body.json.json.
266        let embedded = item["body"]["json"]["json"].as_str().unwrap();
267        let reparsed: Value = serde_json::from_str(embedded).unwrap();
268        assert_eq!(reparsed["request"]["method"], "GET");
269        assert_eq!(reparsed["response"]["status"], 200);
270    }
271}
featherbit/plugins/native/skywalking_logger.rs

featherbit/plugins/native/
skywalking_logger.rs
