slog_retry/lib.rs
1#![doc(html_root_url = "https://docs.rs/slog-retry/0.1.1/slog-retry/")]
2#![warn(missing_docs)]
3
4//! A slog adapter for retrying on errors.
5//!
6//! The [slog](https://crates.io/crates/slog) infrastructure is very powerful and can be bent to
7//! many scenarios. Many of the loggers there may fail, for example because they log over the
8//! network ‒ like [slog-json](https://crates.io/crates/slog-json) logging into a TCP stream.
9//!
10//! The basic framework allows for three options:
11//!
12//! * Handle the error manually, which is uncomfortable.
13//! * Ignore the error, but then all the future records are lost.
14//! * Fuse the drain, making it explode on the first error and killing the whole application.
15//!
16//! This crate brings an adapter that initializes the drain anew each time an error happens, adding
17//! the ability to recover from errors.
18//!
19//! # Warning
20//!
21//! The adapter blocks the current thread on reconnects. Therefore, you want to wrap it inside
22//! [slog-async](https://crates.io/crates/slog-async) and not it directly as the root drain.
23//!
24//! # Examples
25//!
26//! ```rust,no_run
27//! #[macro_use]
28//! extern crate slog;
29//! extern crate slog_async;
30//! extern crate slog_json;
31//! extern crate slog_retry;
32//!
33//! use std::net::TcpStream;
34//!
35//! use slog::Drain;
36//!
37//! fn main() {
38//! let retry = slog_retry::Retry::new(|| -> Result<_, std::io::Error> {
39//! let connection = TcpStream::connect("127.0.0.1:1234")?;
40//! Ok(slog_json::Json::default(connection))
41//! }, None, true)
42//! // Kill the application if the initial connection fails
43//! .unwrap()
44//! // Ignore if it isn't possible to log some of the messages, we'll try again
45//! .ignore_res();
46//! let async = slog_async::Async::default(retry)
47//! .fuse();
48//! let root = slog::Logger::root(async, o!());
49//! info!(root, "Everything is set up");
50//! }
51//! ```
52
53extern crate failure;
54extern crate slog;
55
56use std::cell::{Cell, RefCell, RefMut};
57use std::fmt::{Debug, Display, Formatter, Result as FmtResult};
58use std::iter;
59use std::thread;
60use std::time::Duration;
61
62use failure::Fail;
63use slog::{Drain, OwnedKVList, Record};
64
65/// An error when the retry adaptor fails.
66///
67/// It wasn't possible to log the record (or initialize it when starting). Usually that means it
68/// wasn't possible to create the drain at all or that each newly created drain failed.
69#[derive(Debug)]
70pub struct Error<FactoryError: Fail + Debug, SlaveError: Fail + Debug> {
71 /// The last error during creation of a new drain, if any.
72 pub factory: Option<FactoryError>,
73 /// The last error during logging attempt, if any.
74 pub slave: Option<SlaveError>,
75}
76
77impl<FactoryError, SlaveError> Fail for Error<FactoryError, SlaveError>
78where
79 FactoryError: Fail + Debug,
80 SlaveError: Fail + Debug,
81{
82 fn cause(&self) -> Option<&Fail> {
83 if let Some(ref slave) = self.slave {
84 return Some(slave);
85 }
86 if let Some(ref fact) = self.factory {
87 return Some(fact);
88 }
89 None
90 }
91}
92
93impl<FactoryError, SlaveError> Display for Error<FactoryError, SlaveError>
94where
95 FactoryError: Fail + Debug,
96 SlaveError: Fail + Debug,
97{
98 fn fmt(&self, fmt: &mut Formatter) -> FmtResult {
99 let factory = self.factory
100 .as_ref()
101 .map(|f| format!("{}", f))
102 .unwrap_or_else(|| "()".to_owned());
103 let slave = self.slave
104 .as_ref()
105 .map(|s| format!("{}", s))
106 .unwrap_or_else(|| "()".to_owned());
107 write!(
108 fmt,
109 "Failed to reconnect the logging drain: {}/{}",
110 factory, slave
111 )
112 }
113}
114
115/// A retry strategy.
116///
117/// The iterator describes how long to wait before reconnection attempts. Once the iterator runs
118/// out of items, the adapter gives up trying to reconnect. Therefore, it specifies both the
119/// waiting intervals and number of retries.
120pub type Strategy = Box<Iterator<Item = Duration>>;
121
122/// A constructor of a new instance of a retry strategy.
123///
124/// Every time the adapter needs to retry connection, it calls the constructor and gets a fresh
125/// retry strategy.
126pub type NewStrategy = Box<Fn() -> Strategy + Send>;
127
128/// The retry adapter.
129///
130/// This wraps another drain and forwards log records into that. However, if the drain returns an
131/// error, it discards it and tries to create a new one and log the message into it.
132///
133/// It uses the [retry strategy](type.Strategy.html) to decide how long to wait before retrying and
134/// how many times. If the retry strategy runs out of items, it gives up, returns an error and the
135/// log record is lost.
136///
137/// However, it is not destroyed by the error and if it is called to log another record, it tries
138/// to reconnect again (using a fresh instance of the strategy).
139///
140/// # Warning
141///
142/// This adapter is *synchronous* and *blocks* during the retry attempts. Unless you provide a
143/// retry strategy with a single zero item, you don't want to use it directly. Wrap it inside
144/// [slog-async](https://crates.io/crates/slog-async), where it'll only slow down the logging
145/// thread and the channel into that thread will be used as a buffer for messages waiting to be
146/// written after the reconnect.
147pub struct Retry<Slave, Factory> {
148 slave: RefCell<Option<Slave>>,
149 factory: Factory,
150 strategy: NewStrategy,
151 initialized: Cell<bool>,
152}
153
154impl<Slave, FactoryError, Factory> Retry<Slave, Factory>
155where
156 Slave: Drain,
157 FactoryError: Fail + Debug,
158 Slave::Err: Fail + Debug,
159 Factory: Fn() -> Result<Slave, FactoryError>,
160{
161 /// Creates a new retry adapter.
162 ///
163 /// # Parameters
164 ///
165 /// * `factory`: A factory function that is used to produce new instance of the slave drain on
166 /// every (re)connection attempt.
167 /// * `strategy`: A reconnect strategy, describing how long to wait between attempts and how
168 /// many attempts to make. If set to `None` a default strategy with 4 increasingly delayed
169 /// attemps is used.
170 /// * `connect_now`: Should a connection be made right away. If it is set to `true`, it may
171 /// block (it uses the reconnect strategy provided) and it may return an error. If set to
172 /// `false`, the connection is made on the first logged message. No matter if connecting now
173 /// or later, the first connection attempt is without waiting.
174 pub fn new(
175 factory: Factory,
176 strategy: Option<NewStrategy>,
177 connect_now: bool,
178 ) -> Result<Self, Error<FactoryError, Slave::Err>> {
179 let result = Self {
180 slave: RefCell::new(None),
181 factory,
182 strategy: strategy.unwrap_or_else(|| Box::new(|| default_new_strategy())),
183 initialized: Cell::new(false),
184 };
185 if connect_now {
186 result
187 .init(&mut result.slave.borrow_mut(), &mut (result.strategy)())
188 .map_err(|factory| {
189 Error {
190 factory,
191 slave: None,
192 }
193 })?;
194 }
195 Ok(result)
196 }
197 fn init(
198 &self,
199 slave: &mut RefMut<Option<Slave>>,
200 strategy: &mut Strategy,
201 ) -> Result<(), Option<FactoryError>> {
202 let prefix: Strategy = if self.initialized.get() {
203 Box::new(iter::empty())
204 } else {
205 self.initialized.set(true);
206 Box::new(iter::once(Duration::from_secs(0)))
207 };
208 let mut last_err = None;
209 for sleep in prefix.chain(strategy) {
210 thread::sleep(sleep);
211 match (self.factory)() {
212 Ok(ok) => {
213 **slave = Some(ok);
214 return Ok(());
215 },
216 Err(err) => last_err = Some(err),
217 }
218 }
219 Err(last_err)
220 }
221}
222
223impl<Slave, FactoryError, Factory> Drain for Retry<Slave, Factory>
224where
225 Slave: Drain,
226 FactoryError: Fail + Debug,
227 Slave::Err: Fail + Debug,
228 Factory: Fn() -> Result<Slave, FactoryError>,
229{
230 type Ok = Slave::Ok;
231 type Err = Error<FactoryError, Slave::Err>;
232 fn log(&self, record: &Record, values: &OwnedKVList) -> Result<Self::Ok, Self::Err> {
233 let mut borrowed = self.slave.borrow_mut();
234 let mut slave_err = None;
235
236 if let Some(ref slave) = *borrowed {
237 match slave.log(record, values) {
238 Ok(ok) => return Ok(ok),
239 Err(err) => slave_err = Some(err),
240 }
241 }
242 // By now there was no slave to start with or it failed, so we recreate it.
243 borrowed.take();
244
245 // Try creating a new one and retry with that.
246 let mut strategy = (self.strategy)();
247 loop {
248 match self.init(&mut borrowed, &mut strategy) {
249 Err(factory) =>
250 return Err(Error {
251 factory,
252 slave: slave_err,
253 }),
254 Ok(()) => match borrowed.as_ref().unwrap().log(record, values) {
255 Ok(ok) => return Ok(ok),
256 Err(err) => {
257 slave_err = Some(err);
258 borrowed.take();
259 },
260 },
261 }
262 }
263 }
264}
265
266fn default_new_strategy() -> Strategy {
267 let iterator = (1..5).map(Duration::from_secs);
268 Box::new(iterator)
269}