featherbit/plugins/native/
fault_injection.rs

1//! Fault injection plugin (`fault-injection`) — a faithful subset of Apache
2//! APISIX's `fault-injection` plugin for chaos/resilience testing.
3//!
4//! Injects an artificial `delay` and/or an `abort` response into matching
5//! requests. Both faults are gated independently by a `percentage` sample and
6//! a `vars` condition (APISIX triple-array expressions, see [`crate::vars`]).
7//!
8//! **Early-exit wiring**: an aborted request has the configured response
9//! already written onto `Context.response` and exits through the node's
10//! **error** port with code `FAULT_INJECTED`; non-aborted requests continue
11//! through the **success** port. Wire `error` to a pass-through path (e.g.
12//! straight to `client.in`, or an `error-handler` that preserves the prepared
13//! response) so the injected status/body reach the client, and wire `success`
14//! to the rest of the pipeline. This deviates mechanically from APISIX (where
15//! abort is a direct exit, not an error) because featherbit pipelines need a
16//! distinct port for "stop here" versus "keep going".
17
18use async_trait::async_trait;
19use bytes::Bytes;
20use std::collections::HashMap;
21use std::time::Duration;
22
23use crate::context::{Context, GatewayError};
24use crate::plugins::{Plugin, PluginExecutionError, PluginOutput, PluginResult};
25use crate::vars::{interpolate, Expr};
26
27/// Injects delays and/or abort responses into matching requests.
28///
29/// Order matches APISIX: the delay (if it triggers) is applied first, then
30/// the abort check runs. `body`, and string header values, support `$var`
31/// interpolation against the request.
32pub struct FaultInjectionPlugin {
33    abort: Option<AbortRule>,
34    delay: Option<DelayRule>,
35}
36
37struct AbortRule {
38    http_status: u16,
39    /// Body template (`$var` interpolated); empty body when unset.
40    body: Option<String>,
41    /// Lowercased header name → value template.
42    headers: Vec<(String, String)>,
43    percentage: Option<u8>,
44    /// OR-ed list of expressions (APISIX shape); `None` means always match.
45    vars: Option<Vec<Expr>>,
46}
47
48struct DelayRule {
49    duration: Duration,
50    percentage: Option<u8>,
51    vars: Option<Vec<Expr>>,
52}
53
54/// Draws a pseudo-random number in `0..100`.
55///
56/// Uses a freshly seeded `RandomState` hasher instead of a rand crate: each
57/// `RandomState::new()` carries new per-call keys, so `finish()` yields a
58/// cheap, statistically casual value. Not cryptographic and not a
59/// deterministic sequence — good enough for percentage sampling.
60fn roll_percent() -> u8 {
61    use std::collections::hash_map::RandomState;
62    use std::hash::{BuildHasher, Hasher};
63    (RandomState::new().build_hasher().finish() % 100) as u8
64}
65
66/// APISIX `sample_hit`: no percentage means always hit; `0` never hits and
67/// `100` always hits.
68fn sample_hit(percentage: Option<u8>) -> bool {
69    match percentage {
70        None => true,
71        Some(p) => roll_percent() < p,
72    }
73}
74
75/// True when `vars` is unset, or when any expression in the list matches
76/// (APISIX ORs across the `vars` items and ANDs within each).
77fn vars_match(vars: &Option<Vec<Expr>>, ctx: &Context) -> bool {
78    match vars {
79        None => true,
80        Some(exprs) => exprs.iter().any(|e| e.eval(ctx)),
81    }
82}
83
84/// Parses the APISIX `vars` shape for this plugin: an array whose items are
85/// each a full expression (OR-ed together). Two leniencies over the strict
86/// shape, both resolved at config load:
87/// - an item that is a `["AND"|"OR", rule...]` group is wrapped as a
88///   single-rule expression;
89/// - a "flat" list of bare rules (`[["arg_x", "==", "1"], ...]`) is treated
90///   as one AND-ed expression.
91fn parse_or_vars(v: &serde_json::Value, field: &str) -> Result<Vec<Expr>, String> {
92    let items = v
93        .as_array()
94        .ok_or_else(|| format!("{field}.vars must be an array"))?;
95
96    // Flat shape: first item's first element is a plain string that is not a
97    // logical operator — treat the whole array as a single expression.
98    let first_str = items
99        .first()
100        .and_then(|i| i.as_array())
101        .and_then(|a| a.first())
102        .and_then(|f| f.as_str());
103    if let Some(s) = first_str {
104        if !s.eq_ignore_ascii_case("and") && !s.eq_ignore_ascii_case("or") {
105            let expr = Expr::parse(v).map_err(|e| format!("{field}.vars: {e}"))?;
106            return Ok(vec![expr]);
107        }
108    }
109
110    items
111        .iter()
112        .map(|item| {
113            let starts_with_op = item
114                .as_array()
115                .and_then(|a| a.first())
116                .and_then(|f| f.as_str())
117                .is_some();
118            let parsed = if starts_with_op {
119                // ["AND"/"OR", ...] is one nested rule: wrap it into a rule list.
120                Expr::parse(&serde_json::Value::Array(vec![item.clone()]))
121            } else {
122                Expr::parse(item)
123            };
124            parsed.map_err(|e| format!("{field}.vars: {e}"))
125        })
126        .collect()
127}
128
129/// Parses `percentage` as an integer in `0..=100`.
130fn parse_percentage(
131    obj: &serde_json::Map<String, serde_json::Value>,
132    field: &str,
133) -> Result<Option<u8>, String> {
134    match obj.get("percentage") {
135        None => Ok(None),
136        Some(v) => {
137            let n = v
138                .as_u64()
139                .filter(|n| *n <= 100)
140                .ok_or_else(|| format!("{field}.percentage must be an integer 0-100"))?;
141            Ok(Some(n as u8))
142        }
143    }
144}
145
146/// Accepts `headers` as a map (`{name: value}`) or as the UI editor's array
147/// form (`[{name, value}]`); scalar values are stringified, other shapes are
148/// rejected. Header names are lowercased.
149fn parse_headers(v: &serde_json::Value, field: &str) -> Result<Vec<(String, String)>, String> {
150    let scalar = |v: &serde_json::Value| -> Option<String> {
151        match v {
152            serde_json::Value::String(s) => Some(s.clone()),
153            serde_json::Value::Number(n) => Some(n.to_string()),
154            serde_json::Value::Bool(b) => Some(b.to_string()),
155            _ => None,
156        }
157    };
158    match v {
159        serde_json::Value::Object(m) => m
160            .iter()
161            .map(|(k, v)| {
162                scalar(v)
163                    .map(|s| (k.to_lowercase(), s))
164                    .ok_or_else(|| format!("{field}['{k}'] must be a scalar value"))
165            })
166            .collect(),
167        serde_json::Value::Array(items) => {
168            let mut out = Vec::new();
169            for item in items {
170                let obj = item.as_object().ok_or_else(|| {
171                    format!("{field} entries must be objects with 'name' and 'value'")
172                })?;
173                let name = obj.get("name").and_then(|v| v.as_str()).unwrap_or("");
174                if name.trim().is_empty() {
175                    continue; // blank row left in the UI editor
176                }
177                let value = match obj.get("value") {
178                    None => String::new(),
179                    Some(v) => scalar(v)
180                        .ok_or_else(|| format!("{field}['{name}'] must be a scalar value"))?,
181                };
182                out.push((name.to_lowercase(), value));
183            }
184            Ok(out)
185        }
186        _ => Err(format!(
187            "{field} must be a map of name: value or a list of {{name, value}} objects"
188        )),
189    }
190}
191
192impl FaultInjectionPlugin {
193    /// Builds the plugin from node config. At least one of `abort` / `delay`
194    /// is required; expressions and shapes are validated here at config load.
195    ///
196    /// Accepted keys:
197    /// - `abort` (object): injected response.
198    ///   - `http_status` (integer >= 200, **required**): response status.
199    ///   - `body` (string): response body; supports `$var` interpolation.
200    ///     Empty body when unset.
201    ///   - `headers` (map `{name: value}` or array `[{name, value}]`):
202    ///     response headers; string values support `$var` interpolation.
203    ///   - `percentage` (integer 0-100): chance the abort triggers; unset
204    ///     means always.
205    ///   - `vars` (array): APISIX condition expressions, OR-ed across items.
206    /// - `delay` (object): injected latency.
207    ///   - `duration` (number, seconds, may be fractional, **required**).
208    ///   - `percentage`, `vars`: same gating as for `abort`.
209    ///
210    /// ```yaml
211    /// type: fault-injection
212    /// config:
213    ///   delay:
214    ///     duration: 0.5
215    ///     percentage: 30
216    ///   abort:
217    ///     http_status: 503
218    ///     body: '{"error": "injected for $uri"}'
219    ///     percentage: 10
220    ///     vars:
221    ///       - [["arg_debug", "==", "1"]]
222    /// ```
223    pub fn from_config(config: &HashMap<String, serde_json::Value>) -> Result<Self, String> {
224        let abort = match config.get("abort") {
225            None => None,
226            Some(v) => {
227                let obj = v.as_object().ok_or("abort must be an object")?;
228                let http_status = obj
229                    .get("http_status")
230                    .and_then(|v| v.as_u64())
231                    .filter(|n| (200..=599).contains(n))
232                    .ok_or("abort.http_status is required and must be an integer >= 200")?
233                    as u16;
234                let body = match obj.get("body") {
235                    None => None,
236                    Some(v) => Some(v.as_str().ok_or("abort.body must be a string")?.to_string()),
237                };
238                let headers = match obj.get("headers") {
239                    None => Vec::new(),
240                    Some(v) => parse_headers(v, "abort.headers")?,
241                };
242                let percentage = parse_percentage(obj, "abort")?;
243                let vars = match obj.get("vars") {
244                    None => None,
245                    Some(v) => Some(parse_or_vars(v, "abort")?),
246                };
247                Some(AbortRule {
248                    http_status,
249                    body,
250                    headers,
251                    percentage,
252                    vars,
253                })
254            }
255        };
256
257        let delay = match config.get("delay") {
258            None => None,
259            Some(v) => {
260                let obj = v.as_object().ok_or("delay must be an object")?;
261                let duration = obj
262                    .get("duration")
263                    .and_then(|v| v.as_f64())
264                    .filter(|d| *d >= 0.0 && d.is_finite())
265                    .ok_or(
266                        "delay.duration is required and must be a non-negative number of seconds",
267                    )?;
268                let percentage = parse_percentage(obj, "delay")?;
269                let vars = match obj.get("vars") {
270                    None => None,
271                    Some(v) => Some(parse_or_vars(v, "delay")?),
272                };
273                Some(DelayRule {
274                    duration: Duration::from_secs_f64(duration),
275                    percentage,
276                    vars,
277                })
278            }
279        };
280
281        if abort.is_none() && delay.is_none() {
282            return Err("fault-injection requires at least one of 'abort' or 'delay'".to_string());
283        }
284
285        Ok(Self { abort, delay })
286    }
287}
288
289#[async_trait]
290impl Plugin for FaultInjectionPlugin {
291    fn plugin_type(&self) -> &str {
292        "fault-injection"
293    }
294
295    async fn execute(
296        &self,
297        mut ctx: Context,
298        _named_inputs: &HashMap<String, serde_json::Value>,
299    ) -> PluginResult {
300        if let Some(delay) = &self.delay {
301            if sample_hit(delay.percentage) && vars_match(&delay.vars, &ctx) {
302                tokio::time::sleep(delay.duration).await;
303            }
304        }
305
306        if let Some(abort) = &self.abort {
307            if sample_hit(abort.percentage) && vars_match(&abort.vars, &ctx) {
308                // Interpolate against the request state before mutating the response.
309                let body = abort
310                    .body
311                    .as_ref()
312                    .map(|b| interpolate(&ctx, b))
313                    .unwrap_or_default();
314                let headers: Vec<(String, String)> = abort
315                    .headers
316                    .iter()
317                    .map(|(name, tmpl)| (name.clone(), interpolate(&ctx, tmpl)))
318                    .collect();
319
320                ctx.response.status_code = abort.http_status;
321                ctx.response.body = Bytes::from(body);
322                for (name, value) in headers {
323                    ctx.response.headers.insert(name, vec![value]);
324                }
325
326                return Err(PluginExecutionError {
327                    context: ctx,
328                    error: GatewayError {
329                        node_id: String::new(),
330                        code: "FAULT_INJECTED".to_string(),
331                        message: format!("fault injected: abort with status {}", abort.http_status),
332                        metadata: HashMap::new(),
333                    },
334                });
335            }
336        }
337
338        Ok(PluginOutput {
339            context: ctx,
340            named_outputs: HashMap::new(),
341        })
342    }
343}
344
345#[cfg(test)]
346mod tests {
347    use super::*;
348    use crate::context::{GatewayRequest, GatewayResponse, Protocol};
349    use std::time::Instant;
350
351    fn test_ctx() -> Context {
352        let mut query = HashMap::new();
353        query.insert("name".to_string(), vec!["jack".to_string()]);
354        Context {
355            request: GatewayRequest {
356                method: "GET".to_string(),
357                path: "/api/users".to_string(),
358                host: "example.com".to_string(),
359                scheme: "http".to_string(),
360                headers: HashMap::new(),
361                query_params: query,
362                body: Bytes::new(),
363                remote_addr: "10.1.2.3:44321".to_string(),
364                protocol: Protocol::Http1,
365            },
366            response: GatewayResponse {
367                status_code: 0,
368                headers: HashMap::new(),
369                body: Bytes::new(),
370            },
371            message: HashMap::new(),
372            errors: Vec::new(),
373        }
374    }
375
376    fn plugin(config: serde_json::Value) -> Result<FaultInjectionPlugin, String> {
377        let map: HashMap<String, serde_json::Value> = serde_json::from_value(config).unwrap();
378        FaultInjectionPlugin::from_config(&map)
379    }
380
381    #[tokio::test]
382    async fn test_abort_prepares_response_and_errors() {
383        let p = plugin(serde_json::json!({
384            "abort": {
385                "http_status": 503,
386                "body": "injected for $uri",
387                "headers": { "X-Fault": "yes" }
388            }
389        }))
390        .unwrap();
391
392        let err = p.execute(test_ctx(), &HashMap::new()).await.unwrap_err();
393        assert_eq!(err.error.code, "FAULT_INJECTED");
394        let ctx = err.context;
395        assert_eq!(ctx.response.status_code, 503);
396        assert_eq!(ctx.response.body, Bytes::from("injected for /api/users"));
397        assert_eq!(
398            ctx.response.headers.get("x-fault"),
399            Some(&vec!["yes".to_string()])
400        );
401    }
402
403    #[tokio::test]
404    async fn test_abort_vars_gate() {
405        // OR-of-expressions shape: second expression matches.
406        let p = plugin(serde_json::json!({
407            "abort": {
408                "http_status": 500,
409                "vars": [
410                    [["arg_name", "==", "nope"]],
411                    [["arg_name", "==", "jack"]]
412                ]
413            }
414        }))
415        .unwrap();
416        assert!(p.execute(test_ctx(), &HashMap::new()).await.is_err());
417
418        // No expression matches -> passthrough on success.
419        let p = plugin(serde_json::json!({
420            "abort": {
421                "http_status": 500,
422                "vars": [[["arg_name", "==", "jill"]]]
423            }
424        }))
425        .unwrap();
426        let out = p.execute(test_ctx(), &HashMap::new()).await.unwrap();
427        assert_eq!(out.context.response.status_code, 0);
428    }
429
430    #[tokio::test]
431    async fn test_flat_vars_shape_accepted() {
432        let p = plugin(serde_json::json!({
433            "abort": {
434                "http_status": 500,
435                "vars": [["arg_name", "==", "jack"]]
436            }
437        }))
438        .unwrap();
439        assert!(p.execute(test_ctx(), &HashMap::new()).await.is_err());
440    }
441
442    #[tokio::test]
443    async fn test_percentage_edges() {
444        // 0% never aborts, 100% always aborts.
445        let p0 = plugin(serde_json::json!({
446            "abort": { "http_status": 500, "percentage": 0 }
447        }))
448        .unwrap();
449        let p100 = plugin(serde_json::json!({
450            "abort": { "http_status": 500, "percentage": 100 }
451        }))
452        .unwrap();
453        for _ in 0..20 {
454            assert!(p0.execute(test_ctx(), &HashMap::new()).await.is_ok());
455            assert!(p100.execute(test_ctx(), &HashMap::new()).await.is_err());
456        }
457    }
458
459    #[tokio::test]
460    async fn test_delay_sleeps() {
461        let p = plugin(serde_json::json!({
462            "delay": { "duration": 0.05 }
463        }))
464        .unwrap();
465        let start = Instant::now();
466        let out = p.execute(test_ctx(), &HashMap::new()).await.unwrap();
467        assert!(start.elapsed() >= Duration::from_millis(45));
468        assert_eq!(out.context.response.status_code, 0);
469    }
470
471    #[tokio::test]
472    async fn test_delay_vars_gate_skips_sleep() {
473        let p = plugin(serde_json::json!({
474            "delay": {
475                "duration": 5.0,
476                "vars": [[["arg_name", "==", "nobody"]]]
477            }
478        }))
479        .unwrap();
480        let start = Instant::now();
481        p.execute(test_ctx(), &HashMap::new()).await.unwrap();
482        assert!(start.elapsed() < Duration::from_secs(1));
483    }
484
485    #[test]
486    fn test_config_errors() {
487        // Neither abort nor delay.
488        assert!(plugin(serde_json::json!({})).is_err());
489        // abort without http_status.
490        assert!(plugin(serde_json::json!({ "abort": { "body": "x" } })).is_err());
491        // http_status below 200.
492        assert!(plugin(serde_json::json!({ "abort": { "http_status": 100 } })).is_err());
493        // delay without duration.
494        assert!(plugin(serde_json::json!({ "delay": { "percentage": 50 } })).is_err());
495        // percentage out of range.
496        assert!(plugin(serde_json::json!({
497            "abort": { "http_status": 500, "percentage": 101 }
498        }))
499        .is_err());
500        // invalid vars expression surfaces at load.
501        assert!(plugin(serde_json::json!({
502            "abort": { "http_status": 500, "vars": [[["arg_x", "bogus_op", "1"]]] }
503        }))
504        .is_err());
505    }
506}
featherbit/plugins/native/fault_injection.rs

featherbit/plugins/native/
fault_injection.rs
