fmt: Add support for runtime filter reloading (#45)

This branch adds support for runtime filter reloading to
`tokio-trace-fmt`. Filters may be reloaded through a `Handle`
type that can be constructed when the subscriber or filter are
constructed.

In the future, when PR #44 merges, it will also be possible to add a
free function `record_current` that downcasts and reloads the current
subscriber.

Currently, when using a reloadable filter, `Interest` is never cached.
When tokio-rs/tokio#1039 is merged, we can invalidate the callsite cache
when reloading, allowing us to cache interests.

Closes #42

Signed-off-by: Eliza Weisman <[email protected]>

Author: hawkw

tokio-trace-fmt/src/filter.rs renamed to tokio-trace-fmt/src/filter/env.rs

@@ -1,24 +1,11 @@
-use span;
-
 use regex::Regex;
 use tokio_trace_core::{subscriber::Interest, Level, Metadata};
+use {filter::Filter, span::Context};
 
 use std::env;
 
 pub const DEFAULT_FILTER_ENV: &'static str = "RUST_LOG";
 
-pub trait Filter<N> {
-    fn callsite_enabled(&self, metadata: &Metadata, ctx: &span::Context<N>) -> Interest {
-        if self.enabled(metadata, ctx) {
-            Interest::always()
-        } else {
-            Interest::never()
-        }
-    }
-
-    fn enabled(&self, metadata: &Metadata, ctx: &span::Context<N>) -> bool;
-}
-
 #[derive(Debug)]
 pub struct EnvFilter {
     directives: Vec<Directive>,
@@ -105,7 +92,7 @@ where
 }
 
 impl<N> Filter<N> for EnvFilter {
-    fn callsite_enabled(&self, metadata: &Metadata, _: &span::Context<N>) -> Interest {
+    fn callsite_enabled(&self, metadata: &Metadata, _: &Context<N>) -> Interest {
         if !self.includes_span_directive && metadata.level() > &self.max_level {
             return Interest::never();
         }
@@ -134,7 +121,7 @@ impl<N> Filter<N> for EnvFilter {
         interest
     }
 
-    fn enabled<'a>(&self, metadata: &Metadata, ctx: &span::Context<'a, N>) -> bool {
+    fn enabled<'a>(&self, metadata: &Metadata, ctx: &Context<'a, N>) -> bool {
         for directive in self.directives_for(metadata) {
             let accepts_level = metadata.level() <= &directive.level;
             match directive.in_span.as_ref() {
@@ -319,16 +306,6 @@ impl Default for Directive {
     }
 }
 
-impl<'a, F, N> Filter<N> for F
-where
-    F: Fn(&Metadata, &span::Context<N>) -> bool,
-    N: ::NewVisitor<'a>,
-{
-    fn enabled(&self, metadata: &Metadata, ctx: &span::Context<N>) -> bool {
-        (self)(metadata, ctx)
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

tokio-trace-fmt/src/filter/mod.rs

@@ -0,0 +1,30 @@
+use span;
+
+use tokio_trace_core::{subscriber::Interest, Metadata};
+
+pub trait Filter<N> {
+    fn callsite_enabled(&self, metadata: &Metadata, ctx: &span::Context<N>) -> Interest {
+        if self.enabled(metadata, ctx) {
+            Interest::always()
+        } else {
+            Interest::never()
+        }
+    }
+
+    fn enabled(&self, metadata: &Metadata, ctx: &span::Context<N>) -> bool;
+}
+
+pub mod env;
+pub mod reload;
+
+pub use self::{env::EnvFilter, reload::ReloadFilter};
+
+impl<'a, F, N> Filter<N> for F
+where
+    F: Fn(&Metadata, &span::Context<N>) -> bool,
+    N: ::NewVisitor<'a>,
+{
+    fn enabled(&self, metadata: &Metadata, ctx: &span::Context<N>) -> bool {
+        (self)(metadata, ctx)
+    }
+}

tokio-trace-fmt/src/filter/reload.rs

@@ -0,0 +1,170 @@
+use parking_lot::RwLock;
+use std::{
+    error, fmt,
+    sync::{Arc, Weak},
+};
+use tokio_trace_core::{subscriber::Interest, Metadata};
+use {filter::Filter, span::Context};
+
+#[derive(Debug)]
+pub struct ReloadFilter<F> {
+    inner: Arc<RwLock<F>>,
+}
+
+#[derive(Debug, Clone)]
+pub struct Handle<F> {
+    inner: Weak<RwLock<F>>,
+}
+
+#[derive(Debug)]
+pub struct Error {
+    kind: ErrorKind,
+}
+
+#[derive(Debug)]
+enum ErrorKind {
+    SubscriberGone,
+}
+
+// ===== impl ReloadFilter =====
+
+impl<F, N> Filter<N> for ReloadFilter<F>
+where
+    F: Filter<N>,
+{
+    fn callsite_enabled(&self, _: &Metadata, _: &Context<N>) -> Interest {
+        // TODO(eliza): When tokio-rs/tokio#1039 lands, we can allow our
+        // interest to be cached. For now, we must always return `sometimes`.
+        Interest::sometimes()
+    }
+
+    fn enabled(&self, metadata: &Metadata, ctx: &Context<N>) -> bool {
+        self.inner.read().enabled(metadata, ctx)
+    }
+}
+
+impl<F: 'static> ReloadFilter<F> {
+    pub fn new<N>(f: F) -> Self
+    where
+        F: Filter<N>,
+    {
+        Self {
+            inner: Arc::new(RwLock::new(f)),
+        }
+    }
+
+    pub fn handle(&self) -> Handle<F> {
+        Handle {
+            inner: Arc::downgrade(&self.inner),
+        }
+    }
+}
+
+// ===== impl Handle =====
+
+impl<F: 'static> Handle<F> {
+    pub fn reload<N>(&self, new_filter: impl Into<F>) -> Result<(), Error>
+    where
+        F: Filter<N>,
+    {
+        self.modify(|filter| {
+            *filter = new_filter.into();
+        })
+    }
+
+    /// Invokes a closure with a mutable reference to the current filter,
+    /// allowing it to be modified in place.
+    pub fn modify<N>(&self, f: impl FnOnce(&mut F)) -> Result<(), Error>
+    where
+        F: Filter<N>,
+    {
+        let inner = self.inner.upgrade().ok_or(Error {
+            kind: ErrorKind::SubscriberGone,
+        })?;
+        let mut inner = inner.write();
+        f(&mut *inner);
+        // TODO(eliza): When tokio-rs/tokio#1039 lands, this is where we would
+        // invalidate the callsite cache.
+        Ok(())
+    }
+
+    /// Returns a clone of the filter's current value if it still exists.
+    /// Otherwise, if the filter has been dropped, returns `None`.
+    pub fn clone_current(&self) -> Option<F>
+    where
+        F: Clone,
+    {
+        self.inner.upgrade().map(|inner| inner.read().clone())
+    }
+}
+
+// ===== impl Error =====
+
+impl fmt::Display for Error {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        error::Error::description(self).fmt(f)
+    }
+}
+
+impl error::Error for Error {
+    fn description(&self) -> &str {
+        match self.kind {
+            ErrorKind::SubscriberGone => "subscriber no longer exists",
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use tokio_trace_core::{
+        dispatcher::{self, Dispatch},
+        Metadata,
+    };
+    use *;
+
+    #[test]
+    fn reload_handle() {
+        static FILTER1_CALLS: AtomicUsize = AtomicUsize::new(0);
+        static FILTER2_CALLS: AtomicUsize = AtomicUsize::new(0);
+
+        fn filter1<N>(_: &Metadata, _: &span::Context<N>) -> bool {
+            FILTER1_CALLS.fetch_add(1, Ordering::Relaxed);
+            true
+        }
+
+        fn filter2<N>(_: &Metadata, _: &span::Context<N>) -> bool {
+            FILTER2_CALLS.fetch_add(1, Ordering::Relaxed);
+            true
+        }
+
+        fn event() {
+            trace!("my event");
+        }
+
+        let subscriber = FmtSubscriber::builder()
+            .with_filter(filter1 as fn(&Metadata, &span::Context<_>) -> bool)
+            .with_filter_reloading();
+        let handle = subscriber.reload_handle();
+        let subscriber = Dispatch::new(subscriber.finish());
+
+        dispatcher::with_default(&subscriber, || {
+            assert_eq!(FILTER1_CALLS.load(Ordering::Relaxed), 0);
+            assert_eq!(FILTER2_CALLS.load(Ordering::Relaxed), 0);
+
+            event();
+
+            assert_eq!(FILTER1_CALLS.load(Ordering::Relaxed), 1);
+            assert_eq!(FILTER2_CALLS.load(Ordering::Relaxed), 0);
+
+            handle
+                .reload(filter2 as fn(&Metadata, &span::Context<_>) -> bool)
+                .expect("should reload");
+
+            event();
+
+            assert_eq!(FILTER1_CALLS.load(Ordering::Relaxed), 1);
+            assert_eq!(FILTER2_CALLS.load(Ordering::Relaxed), 1);
+        })
+    }
+}

tokio-trace-fmt/src/lib.rs

@@ -1,4 +1,7 @@
 extern crate tokio_trace_core;
+#[cfg(test)]
+#[macro_use]
+extern crate tokio_trace;
 
 #[cfg(feature = "ansi")]
 extern crate ansi_term;
@@ -76,6 +79,17 @@ where
     }
 }
 
+impl<N, E, F> FmtSubscriber<N, E, filter::ReloadFilter<F>>
+where
+    F: Filter<N> + 'static,
+{
+    /// Returns a `Handle` that may be used to reload this subscriber's
+    /// filter.
+    pub fn reload_handle(&self) -> filter::reload::Handle<F> {
+        self.filter.handle()
+    }
+}
+
 impl<N, E, F> tokio_trace_core::Subscriber for FmtSubscriber<N, E, F>
 where
     N: for<'a> NewVisitor<'a> + 'static,
@@ -205,6 +219,33 @@ where
     }
 }
 
+impl<N, E, F> Builder<N, E, F>
+where
+    F: Filter<N> + 'static,
+{
+    /// Configures the subscriber being built to allow filter reloading at
+    /// runtime.
+    pub fn with_filter_reloading(self) -> Builder<N, E, filter::ReloadFilter<F>> {
+        Builder {
+            new_visitor: self.new_visitor,
+            fmt_event: self.fmt_event,
+            filter: filter::ReloadFilter::new(self.filter),
+            settings: self.settings,
+        }
+    }
+}
+
+impl<N, E, F> Builder<N, E, filter::ReloadFilter<F>>
+where
+    F: Filter<N> + 'static,
+{
+    /// Returns a `Handle` that may be used to reload the constructed subscriber's
+    /// filter.
+    pub fn reload_handle(&self) -> filter::reload::Handle<F> {
+        self.filter.handle()
+    }
+}
+
 impl<N, E, F> Builder<N, E, F> {
     /// Sets the Visitor that the subscriber being built will use to record
     /// fields.