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
116    fn info(&self) -> ServiceInfo {
117        self.inner.info()
118    }
119
120    fn capability(&self) -> Capability {
121        self.inner.capability()
122    }
123
124    async fn create_dir(
125        &self,
126        ctx: &OperationContext,
127        path: &str,
128        args: OpCreateDir,
129    ) -> Result<RpCreateDir> {
130        self.inner.create_dir(ctx, path, args).await
131    }
132
133    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
134        self.inner.stat(ctx, path, args).await
135    }
136
137    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
138        self.inner
139            .read(ctx, path, args)
140            .map(|r| ChaosReader::new(r, Arc::clone(&self.rng), self.error_ratio))
141    }
142
143    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
144        self.inner.write(ctx, path, args)
145    }
146
147    fn copy(
148        &self,
149        ctx: &OperationContext,
150        from: &str,
151        to: &str,
152        args: OpCopy,
153        opts: OpCopier,
154    ) -> Result<Self::Copier> {
155        self.inner.copy(ctx, from, to, args, opts)
156    }
157
158    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
159        self.inner.list(ctx, path, args)
160    }
161
162    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
163        self.inner.delete(ctx)
164    }
165
166    async fn rename(
167        &self,
168        ctx: &OperationContext,
169        from: &str,
170        to: &str,
171        args: OpRename,
172    ) -> Result<RpRename> {
173        self.inner.rename(ctx, from, to, args).await
174    }
175
176    async fn presign(
177        &self,
178        ctx: &OperationContext,
179        path: &str,
180        args: OpPresign,
181    ) -> Result<RpPresign> {
182        self.inner.presign(ctx, path, args).await
183    }
184}
185
186#[doc(hidden)]
187pub struct ChaosReader<R> {
188    inner: R,
189    rng: Arc<Mutex<StdRng>>,
190
191    error_ratio: f64,
192}
193
194impl<R> ChaosReader<R> {
195    fn new(inner: R, rng: Arc<Mutex<StdRng>>, error_ratio: f64) -> Self {
196        Self {
197            inner,
198            rng,
199            error_ratio,
200        }
201    }
202
203    /// If I feel lucky, we can return the correct response. Otherwise,
204    /// we need to generate an error.
205    fn i_feel_lucky(&self) -> bool {
206        let point: u32 = self.rng.lock().unwrap().random_range(0..100);
207        point >= (self.error_ratio * 100.0) as u32
208    }
209
210    fn unexpected_eof() -> Error {
211        Error::new(ErrorKind::Unexpected, "I am your chaos!")
212            .with_operation("chaos")
213            .set_temporary()
214    }
215}
216
217impl<R: oio::ReadStream> oio::ReadStream for ChaosReader<R> {
218    async fn read(&mut self) -> Result<Buffer> {
219        if self.i_feel_lucky() {
220            self.inner.read().await
221        } else {
222            Err(Self::unexpected_eof())
223        }
224    }
225}
226
227impl<R: oio::Read> oio::Read for ChaosReader<R> {
228    async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
229        if self.i_feel_lucky() {
230            let (rp, stream) = self.inner.open(range).await?;
231            Ok((
232                rp,
233                Box::new(ChaosReader::new(stream, self.rng.clone(), self.error_ratio))
234                    as Box<dyn oio::ReadStreamDyn>,
235            ))
236        } else {
237            Err(Self::unexpected_eof())
238        }
239    }
240
241    async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
242        if self.i_feel_lucky() {
243            self.inner.read(range).await
244        } else {
245            Err(Self::unexpected_eof())
246        }
247    }
248}
opendal_layer_chaos/lib.rs

opendal_layer_chaos/
lib.rs
