opendal_layer_chaos/
lib.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18#![doc = include_str!("../README.md")]
19#![cfg_attr(docsrs, feature(doc_cfg))]
20#![cfg_attr(docsrs, doc(auto_cfg))]
21#![deny(missing_docs)]
22use std::sync::Arc;
23use std::sync::Mutex;
24
25use opendal_core::raw::*;
26use opendal_core::*;
27use rand::prelude::*;
28use rand::rngs::StdRng;
29
30/// `ChaosLayer` injects errors into services to test robustness.
31///
32/// # Chaos
33///
34/// Chaos testing complements stress testing. A specified error ratio reproduces
35/// service errors consistently.
36///
37/// Tests that use `ChaosLayer` can expose error-handling weaknesses.
38///
39/// For example: If we specify an error rate of 0.5, there is a 50% chance
40/// of an EOF error for every read operation.
41///
42/// # Note
43///
44/// For now, ChaosLayer only injects read operations. More operations may
45/// be added in the future.
46///
47/// # Examples
48///
49/// ```no_run
50/// # use opendal_core::services;
51/// # use opendal_core::Operator;
52/// # use opendal_core::Result;
53/// # use opendal_layer_chaos::ChaosLayer;
54/// #
55/// # fn main() -> Result<()> {
56/// let _ = Operator::new(services::Memory::default())?
57///     .layer(ChaosLayer::new(0.1));
58/// # Ok(())
59/// # }
60/// ```
61#[derive(Clone, Debug)]
62pub struct ChaosLayer {
63    rng: Arc<Mutex<StdRng>>,
64    error_ratio: f64,
65}
66
67impl ChaosLayer {
68    /// Create a new [`ChaosLayer`] with specified error ratio.
69    ///
70    /// # Panics
71    ///
72    /// Input error_ratio must in [0.0..=1.0]
73    pub fn new(error_ratio: f64) -> Self {
74        assert!(
75            (0.0..=1.0).contains(&error_ratio),
76            "error_ratio must between 0.0 and 1.0"
77        );
78        Self {
79            rng: Arc::new(Mutex::new(StdRng::from_rng(&mut rand::rng()))),
80            error_ratio,
81        }
82    }
83}
84
85impl Layer for ChaosLayer {
86    fn apply_service(&self, inner: Servicer) -> Servicer {
87        Arc::new(self.layer(inner))
88    }
89}
90
91impl ChaosLayer {
92    fn layer(&self, inner: Servicer) -> ChaosService {
93        ChaosService {
94            inner,
95            rng: self.rng.clone(),
96            error_ratio: self.error_ratio,
97        }
98    }
99}
100
101#[doc(hidden)]
102#[derive(Debug)]
103pub struct ChaosService {
104    inner: Servicer,
105    rng: Arc<Mutex<StdRng>>,
106    error_ratio: f64,
107}
108
109impl Service for ChaosService {
110    type Reader = ChaosReader<oio::Reader>;
111    type Writer = oio::Writer;
112    type Lister = oio::Lister;
113    type Deleter = oio::Deleter;
114    type Copier = oio::Copier;
115    type Composer = oio::Composer;
116
117    fn info(&self) -> ServiceInfo {
118        self.inner.info()
119    }
120
121    fn capability(&self) -> Capability {
122        let mut capability = self.inner.capability();
123        capability.write_can_copy_from = false;
124        capability
125    }
126
127    fn compose(&self, ctx: &OperationContext, to: &str, args: OpCompose) -> Result<Self::Composer> {
128        self.inner.compose(ctx, to, args)
129    }
130
131    async fn create_dir(
132        &self,
133        ctx: &OperationContext,
134        path: &str,
135        args: OpCreateDir,
136    ) -> Result<RpCreateDir> {
137        self.inner.create_dir(ctx, path, args).await
138    }
139
140    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
141        self.inner.stat(ctx, path, args).await
142    }
143
144    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
145        self.inner
146            .read(ctx, path, args)
147            .map(|r| ChaosReader::new(r, Arc::clone(&self.rng), self.error_ratio))
148    }
149
150    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
151        self.inner.write(ctx, path, args)
152    }
153
154    fn copy(
155        &self,
156        ctx: &OperationContext,
157        from: &str,
158        to: &str,
159        args: OpCopy,
160    ) -> Result<Self::Copier> {
161        self.inner.copy(ctx, from, to, args)
162    }
163
164    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
165        self.inner.list(ctx, path, args)
166    }
167
168    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
169        self.inner.delete(ctx)
170    }
171
172    async fn rename(
173        &self,
174        ctx: &OperationContext,
175        from: &str,
176        to: &str,
177        args: OpRename,
178    ) -> Result<RpRename> {
179        self.inner.rename(ctx, from, to, args).await
180    }
181
182    async fn restore(
183        &self,
184        ctx: &OperationContext,
185        path: &str,
186        args: OpRestore,
187    ) -> Result<RpRestore> {
188        self.inner.restore(ctx, path, args).await
189    }
190
191    async fn presign(
192        &self,
193        ctx: &OperationContext,
194        path: &str,
195        args: OpPresign,
196    ) -> Result<RpPresign> {
197        self.inner.presign(ctx, path, args).await
198    }
199}
200
201#[doc(hidden)]
202pub struct ChaosReader<R> {
203    inner: R,
204    rng: Arc<Mutex<StdRng>>,
205
206    error_ratio: f64,
207}
208
209impl<R> ChaosReader<R> {
210    fn new(inner: R, rng: Arc<Mutex<StdRng>>, error_ratio: f64) -> Self {
211        Self {
212            inner,
213            rng,
214            error_ratio,
215        }
216    }
217
218    /// If I feel lucky, we can return the correct response. Otherwise,
219    /// we need to generate an error.
220    fn i_feel_lucky(&self) -> bool {
221        let point: u32 = self.rng.lock().unwrap().random_range(0..100);
222        point >= (self.error_ratio * 100.0) as u32
223    }
224
225    fn unexpected_eof() -> Error {
226        Error::new(ErrorKind::Unexpected, "I am your chaos!")
227            .with_operation("chaos")
228            .set_temporary()
229    }
230}
231
232impl<R: oio::ReadStream> oio::ReadStream for ChaosReader<R> {
233    async fn read(&mut self) -> Result<Buffer> {
234        if self.i_feel_lucky() {
235            self.inner.read().await
236        } else {
237            Err(Self::unexpected_eof())
238        }
239    }
240}
241
242impl<R: oio::Read> oio::Read for ChaosReader<R> {
243    async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
244        if self.i_feel_lucky() {
245            let (rp, stream) = self.inner.open(range).await?;
246            Ok((
247                rp,
248                Box::new(ChaosReader::new(stream, self.rng.clone(), self.error_ratio))
249                    as Box<dyn oio::ReadStreamDyn>,
250            ))
251        } else {
252            Err(Self::unexpected_eof())
253        }
254    }
255
256    async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
257        if self.i_feel_lucky() {
258            self.inner.read(range).await
259        } else {
260            Err(Self::unexpected_eof())
261        }
262    }
263}
opendal_layer_chaos/lib.rs

opendal_layer_chaos/
lib.rs
