featherbit/plugins/native/
proxy_mirror.rs

1//! Request-mirroring plugin (`proxy-mirror`).
2//!
3//! The featherbit port of Apache APISIX's `proxy-mirror`: fires a fire-and-
4//! forget clone of the incoming request at a shadow upstream for traffic
5//! shadowing (comparing a new backend, capturing traffic, ...). The mirror is
6//! sent on a detached background task via the shared outbound client; its
7//! response and any error are ignored, and the request path is **never**
8//! blocked or affected by it.
9//!
10//! **Placement**: put this node before `upstream` so it observes the request
11//! as it will be proxied. It always continues through the `success` port and
12//! never errors.
13
14use async_trait::async_trait;
15use std::collections::HashMap;
16use std::sync::Arc;
17use std::time::Duration;
18
19use crate::context::Context;
20use crate::outbound::{OutboundClient, OutboundRequest};
21use crate::plugins::resources::PluginResources;
22use crate::plugins::{Plugin, PluginOutput, PluginResult};
23use crate::vars::template::Template;
24
25/// Whole-call deadline for a mirrored (best-effort) request.
26const MIRROR_TIMEOUT: Duration = Duration::from_secs(60);
27
28/// Mirrors matching requests to a shadow host, best-effort.
29pub struct ProxyMirrorPlugin {
30    /// Shadow base URL, e.g. `http://shadow:8080` (scheme + host + optional
31    /// port, no trailing path). The `http://`/`https://` prefix is validated
32    /// against the raw config string at load time; the value itself supports
33    /// `{{namespace.path}}` template references, rendered per request.
34    host: Template,
35    /// Optional path override; when unset the original request path is used.
36    /// Supports `{{namespace.path}}` references.
37    path: Option<Template>,
38    /// Fraction of requests to mirror, in `0.0..=1.0`.
39    sample_ratio: f64,
40    /// Shared pooled outbound HTTP client (from `PluginResources`).
41    client: Arc<OutboundClient>,
42}
43
44/// Draws a pseudo-random fraction in `[0.0, 1.0)`.
45///
46/// Uses a freshly seeded `RandomState` hasher (new per-call keys) rather than
47/// a rand crate — the same cheap, statistically casual source the
48/// `fault-injection` plugin uses for sampling. Not cryptographic and not a
49/// deterministic sequence; good enough for `sample_ratio`.
50fn roll_fraction() -> f64 {
51    use std::collections::hash_map::RandomState;
52    use std::hash::{BuildHasher, Hasher};
53    let n = RandomState::new().build_hasher().finish();
54    // Map the u64 into [0, 1); dividing by 2^64 keeps it strictly below 1.
55    n as f64 / (u64::MAX as f64 + 1.0)
56}
57
58impl ProxyMirrorPlugin {
59    /// Builds the plugin from node config.
60    ///
61    /// Accepted keys:
62    /// - `host` (string, **required**): shadow base URL — scheme + host +
63    ///   optional port, e.g. `http://shadow:8080`. Must start with `http://`
64    ///   or `https://`; no trailing path.
65    /// - `path` (string, optional): overrides the mirrored request path; when
66    ///   unset the original request path is mirrored. The original query
67    ///   string is always appended.
68    /// - `sample_ratio` (number, default `1.0`): fraction of requests to
69    ///   mirror, in `0.0..=1.0`. `1.0` mirrors every request; `0.0` mirrors
70    ///   none. Sampling uses a pseudo-random draw (see [`roll_fraction`]).
71    ///
72    /// ```yaml
73    /// type: proxy-mirror
74    /// config:
75    ///   host: "http://shadow:8080"
76    ///   path: "/mirror"
77    ///   sample_ratio: 0.5
78    /// ```
79    pub fn from_config(
80        config: &HashMap<String, serde_json::Value>,
81        resources: &Arc<PluginResources>,
82    ) -> Result<Self, String> {
83        let host = config
84            .get("host")
85            .and_then(|v| v.as_str())
86            .filter(|s| !s.is_empty())
87            .ok_or("proxy-mirror requires 'host' (e.g. \"http://shadow:8080\")")?
88            .trim_end_matches('/')
89            .to_string();
90
91        if !host.starts_with("http://") && !host.starts_with("https://") {
92            return Err(format!(
93                "proxy-mirror 'host' must start with http:// or https:// (got '{host}')"
94            ));
95        }
96        // Discard warnings here — the compile-time walk (a later task)
97        // reports well-formed-but-unknown references; execution must not.
98        let host = Template::parse(&host).0;
99
100        let path = config
101            .get("path")
102            .and_then(|v| v.as_str())
103            .filter(|s| !s.is_empty())
104            .map(|s| Template::parse(s).0);
105
106        let sample_ratio = match config.get("sample_ratio") {
107            None => 1.0,
108            Some(v) => v
109                .as_f64()
110                .filter(|r| (0.0..=1.0).contains(r))
111                .ok_or("proxy-mirror 'sample_ratio' must be a number in 0.0..=1.0")?,
112        };
113
114        Ok(Self {
115            host,
116            path,
117            sample_ratio,
118            client: resources.outbound.clone(),
119        })
120    }
121
122    /// Decides whether this request should be mirrored, per `sample_ratio`.
123    ///
124    /// `>= 1.0` always mirrors, `0.0` never mirrors; otherwise a pseudo-random
125    /// draw is compared against the ratio.
126    fn should_mirror(&self) -> bool {
127        if self.sample_ratio >= 1.0 {
128            true
129        } else {
130            roll_fraction() < self.sample_ratio
131        }
132    }
133
134    /// Builds the outbound mirror request from the current context (pure; does
135    /// no I/O). The mirrored URL is `host` + (`path` override or the original
136    /// path) + the original query string. All request headers and the body are
137    /// copied.
138    fn build_request(&self, ctx: &Context) -> OutboundRequest {
139        let host = self.host.render(ctx);
140        let path = match &self.path {
141            Some(tpl) => tpl.render(ctx),
142            None => std::borrow::Cow::Borrowed(ctx.request.path.as_str()),
143        };
144        let url = match crate::vars::resolve(ctx, "query_string") {
145            Some(qs) => format!("{}{}?{}", host, path, qs),
146            None => format!("{}{}", host, path),
147        };
148
149        let method: http::Method = ctx.request.method.parse().unwrap_or(http::Method::GET);
150
151        let mut headers: Vec<(String, String)> = Vec::new();
152        for (name, values) in &ctx.request.headers {
153            for value in values {
154                headers.push((name.clone(), value.clone()));
155            }
156        }
157
158        OutboundRequest {
159            method,
160            url,
161            headers,
162            body: ctx.request.body.clone(),
163            timeout: MIRROR_TIMEOUT,
164            ssl_verify: true,
165            tls: None,
166        }
167    }
168}
169
170#[async_trait]
171impl Plugin for ProxyMirrorPlugin {
172    fn plugin_type(&self) -> &str {
173        "proxy-mirror"
174    }
175
176    async fn execute(&self, ctx: Context) -> PluginResult {
177        if self.should_mirror() {
178            // Build everything the detached task needs, then spawn it. The
179            // mirror never blocks or affects the request path: its response
180            // and any error are dropped.
181            let request = self.build_request(&ctx);
182            let client = self.client.clone();
183            tokio::spawn(async move {
184                let _ = client.request(request).await;
185            });
186        }
187
188        Ok(PluginOutput::success(ctx))
189    }
190}
191
192#[cfg(test)]
193mod tests {
194    use super::*;
195    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
196    use bytes::Bytes;
197
198    fn test_ctx() -> Context {
199        let mut headers = HashMap::new();
200        headers.insert("x-trace".to_string(), vec!["t1".to_string()]);
201        let mut query = HashMap::new();
202        query.insert("q".to_string(), vec!["1".to_string()]);
203        Context {
204            request: GatewayRequest {
205                method: "POST".to_string(),
206                path: "/api/users".to_string(),
207                host: "example.com".to_string(),
208                scheme: "http".to_string(),
209                headers,
210                query_params: query,
211                body: Bytes::from_static(b"payload"),
212                remote_addr: "10.1.2.3:44321".to_string(),
213                protocol: Protocol::Http1,
214            },
215            response: GatewayResponse {
216                status_code: 0,
217                headers: HashMap::new(),
218                body: Bytes::new(),
219                stream: None,
220            },
221            message: HashMap::new(),
222            errors: Vec::new(),
223        }
224    }
225
226    fn plugin(config: serde_json::Value) -> Result<ProxyMirrorPlugin, String> {
227        let map: HashMap<String, serde_json::Value> = serde_json::from_value(config).unwrap();
228        ProxyMirrorPlugin::from_config(&map, &PluginResources::empty())
229    }
230
231    #[test]
232    fn test_config_requires_valid_host() {
233        assert!(plugin(serde_json::json!({})).is_err());
234        assert!(plugin(serde_json::json!({ "host": "" })).is_err());
235        assert!(plugin(serde_json::json!({ "host": "shadow:8080" })).is_err());
236        assert!(plugin(serde_json::json!({ "host": "http://shadow:8080" })).is_ok());
237        // sample_ratio out of range
238        assert!(plugin(serde_json::json!({ "host": "http://s", "sample_ratio": 2 })).is_err());
239    }
240
241    #[test]
242    fn test_sampling_decision() {
243        let never = plugin(serde_json::json!({ "host": "http://s", "sample_ratio": 0 })).unwrap();
244        let always = plugin(serde_json::json!({ "host": "http://s", "sample_ratio": 1 })).unwrap();
245        for _ in 0..100 {
246            assert!(!never.should_mirror(), "ratio 0 must never mirror");
247            assert!(always.should_mirror(), "ratio 1 must always mirror");
248        }
249    }
250
251    #[test]
252    fn test_build_request_default_path() {
253        let p = plugin(serde_json::json!({ "host": "http://shadow:8080" })).unwrap();
254        let req = p.build_request(&test_ctx());
255        assert_eq!(req.url, "http://shadow:8080/api/users?q=1");
256        assert_eq!(req.method, http::Method::POST);
257        assert_eq!(req.body, Bytes::from_static(b"payload"));
258        assert!(req.headers.iter().any(|(k, v)| k == "x-trace" && v == "t1"));
259    }
260
261    #[test]
262    fn test_build_request_host_and_path_render_template() {
263        let p = plugin(serde_json::json!({
264            "host": "http://{{request.headers.x-shadow-host}}",
265            "path": "/mirror/{{request.headers.x-tenant}}"
266        }))
267        .unwrap();
268        let mut ctx = test_ctx();
269        ctx.request
270            .headers
271            .insert("x-shadow-host".to_string(), vec!["shadow:9090".to_string()]);
272        ctx.request
273            .headers
274            .insert("x-tenant".to_string(), vec!["acme".to_string()]);
275        let req = p.build_request(&ctx);
276        assert_eq!(req.url, "http://shadow:9090/mirror/acme?q=1");
277    }
278
279    #[test]
280    fn test_build_request_path_override() {
281        let p = plugin(serde_json::json!({
282            "host": "http://shadow:8080/", "path": "/mirror"
283        }))
284        .unwrap();
285        let req = p.build_request(&test_ctx());
286        // trailing slash on host is trimmed; query string is preserved
287        assert_eq!(req.url, "http://shadow:8080/mirror?q=1");
288    }
289
290    #[tokio::test]
291    async fn test_execute_returns_ok_and_leaves_context() {
292        // ratio 1 spawns a task that fails to connect; that's ignored and
293        // execute still returns Ok with the context unchanged.
294        let p = plugin(serde_json::json!({
295            "host": "http://127.0.0.1:1", "sample_ratio": 1
296        }))
297        .unwrap();
298        let out = p.execute(test_ctx()).await.unwrap();
299        assert_eq!(out.context.request.path, "/api/users");
300        assert_eq!(out.context.response.status_code, 0);
301    }
302}
featherbit/plugins/native/proxy_mirror.rs

featherbit/plugins/native/
proxy_mirror.rs
