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

featherbit/plugins/native/
proxy_mirror.rs
