proxy: detect TLS configuration changes using inotify on Linux (#1077)

This branch adds an inotify-based implementation of filesystem watches
for the TLS config files. On Linux, where inotify is available, this is
used instead of the polling-based code I added in #1056 and #1076.

In order to avoid the issues detecting changes to files in Kubernetes 
ConfigMaps described in #1061, we watch the directory _containing_ the
files we care about rather than the files themselves. I've tested this 
manually in Docker for Mac Kubernetes and can confirm that ConfigMap
changes are detected successfully.

Closes #1061. Closes #369.

Signed-off-by: Eliza Weisman <eliza@buoyant.io>

Cargo.lock

@@ -137,6 +137,7 @@ dependencies = [
  "httparse 1.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "hyper 0.12.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "indexmap 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "inotify 0.5.2-dev (git+https://github.com/inotify-rs/inotify)",
  "ipnet 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "libc 0.2.40 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -456,6 +457,25 @@ name = "indexmap"
 version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
+[[package]]
+name = "inotify"
+version = "0.5.2-dev"
+source = "git+https://github.com/inotify-rs/inotify#901abf4e076e2c96bc4d485d235b7f732bf01b36"
+dependencies = [
+ "bitflags 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
+ "inotify-sys 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.40 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "inotify-sys"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "libc 0.2.40 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "iovec"
 version = "0.1.2"
@@ -1497,6 +1517,8 @@ dependencies = [
 "checksum hyper 0.12.1 (registry+https://github.com/rust-lang/crates.io-index)" = "6416251e6672bff06fe96a3337570772845a44500fba2d178e2e55e0fab58a86"
 "checksum idna 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "014b298351066f1512874135335d62a789ffe78a9974f94b43ed5621951eaf7d"
 "checksum indexmap 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "7b9378f1f3923647a9aea6af4c6b5de68cc8a71415459ad25ef191191c48f5b7"
+"checksum inotify 0.5.2-dev (git+https://github.com/inotify-rs/inotify)" = "<none>"
+"checksum inotify-sys 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7dceb94c43f70baf4c4cd6afbc1e9037d4161dbe68df8a2cd4351a23319ee4fb"
 "checksum iovec 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "dbe6e417e7d0975db6512b90796e8ce223145ac4e33c377e4a42882a0e88bb08"
 "checksum ipconfig 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "9ec4e18c0a0d4340870c14284293632d8421f419008371422dd327892b88877c"
 "checksum ipnet 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "51268c3a27ad46afd1cca0bbf423a5be2e9fd3e6a7534736c195f0f834b763ef"

proxy/Cargo.toml

@@ -59,6 +59,8 @@ untrusted = "0.6.1"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 libc = "0.2"
+# We can use the `crates.io` version of `inotify` once 0.5.2 has been released.
+inotify = { git = "https://github.com/inotify-rs/inotify" }
 
 [dev-dependencies]
 net2 = "0.2"

proxy/src/lib.rs

@@ -15,6 +15,8 @@ extern crate h2;
 extern crate http;
 extern crate httparse;
 extern crate hyper;
+#[cfg(target_os = "linux")]
+extern crate inotify;
 extern crate ipnet;
 #[cfg(target_os = "linux")]
 extern crate libc;

proxy/src/transport/tls/config.rs

@@ -1,7 +1,8 @@
 use std::{
+    collections::HashSet,
     fs::File,
     io::{self, Cursor, Read},
-    path::PathBuf,
+    path::{Path, PathBuf},
     sync::Arc,
     time::{Duration, Instant, SystemTime,},
 };
@@ -63,17 +64,67 @@ pub enum Error {
     EndEntityCertIsNotValid(webpki::Error),
     InvalidPrivateKey,
     TimeConversionFailed,
+    #[cfg(target_os = "linux")]
+    InotifyInit(io::Error),
 }
 
 impl CommonSettings {
+    fn paths(&self) -> [&PathBuf; 3] {
+        [
+            &self.trust_anchors,
+            &self.end_entity_cert,
+            &self.private_key,
+        ]
+    }
 
-    fn change_timestamps(&self, interval: Duration) -> impl Stream<Item = (), Error = ()> {
+    /// Stream changes to the files described by this `CommonSettings`.
-        let paths = [
+    ///
-            self.trust_anchors.clone(),
+    /// The returned stream consists of each subsequent successfully loaded
-            self.end_entity_cert.clone(),
+    /// `CommonSettings` after each change. If the settings could not be
-            self.private_key.clone(),
+    /// reloaded (i.e., they were malformed), nothing is sent.
-        ];
+    pub fn stream_changes(self, interval: Duration)
+        -> impl Stream<Item = CommonConfig, Error = ()>
+    {
+        // If we're on Linux, first atttempt to start an Inotify watch on the
+        // paths. If this fails, fall back to polling the filesystem.
+        #[cfg(target_os = "linux")]
+        let changes: Box<Stream<Item = (), Error = ()> + Send> =
+            match self.stream_changes_inotify() {
+                Ok(s) => Box::new(s),
+                Err(e) => {
+                    warn!(
+                        "inotify init error: {:?}, falling back to polling",
+                        e
+                    );
+                    Box::new(self.stream_changes_polling(interval))
+                },
+            };
 
+        // If we're not on Linux, we can't use inotify, so simply poll the fs.
+        // TODO: Use other FS events APIs (such as `kqueue`) as well, when
+        //       they're available.
+        #[cfg(not(target_os = "linux"))]
+        let changes = self.stream_changes_polling(interval);
+
+        changes.filter_map(move |_|
+            CommonConfig::load_from_disk(&self)
+                .map_err(|e| warn!("error reloading TLS config: {:?}, falling back", e))
+                .ok()
+        )
+
+    }
+
+    /// Stream changes by polling the filesystem.
+    ///
+    /// This will poll the filesystem for changes to the files at the paths
+    /// described by this `CommonSettings` every `interval`, and attempt to
+    /// load a new `CommonConfig` from the files again after each change.
+    ///
+    /// This is used on operating systems other than Linux, or on Linux if
+    /// our attempt to use `inotify` failed.
+    fn stream_changes_polling(&self, interval: Duration)
+        -> impl Stream<Item = (), Error = ()>
+    {
         fn last_modified(path: &PathBuf) -> Option<SystemTime> {
             // We have to canonicalize the path _every_ time we poll the fs,
             // rather than once when we start watching, because if it's a
@@ -94,7 +145,12 @@ impl CommonSettings {
                 .ok()
         }
 
+        let paths = self.paths().iter()
+            .map(|&p| p.clone())
+            .collect::<Vec<PathBuf>>();
+
         let mut max: Option<SystemTime> = None;
+
         Interval::new(Instant::now(), interval)
             .map_err(|e| error!("timer error: {:?}", e))
             .filter_map(move |_| {
@@ -110,27 +166,62 @@ impl CommonSettings {
             })
     }
 
-    /// Stream changes to the files described by this `CommonSettings`.
+    #[cfg(target_os = "linux")]
-    ///
+    fn stream_changes_inotify(&self)
-    /// This will poll the filesystem for changes to the files at the paths
+        -> Result<impl Stream<Item = (), Error = ()>, Error>
-    /// described by this `CommonSettings` every `interval`, and attempt to
-    /// load a new `CommonConfig` from the files again after each change.
-    ///
-    /// The returned stream consists of each subsequent successfully loaded
-    /// `CommonSettings` after each change. If the settings could not be
-    /// reloaded (i.e., they were malformed), nothing is sent.
-    ///
-    /// TODO: On Linux, this should be replaced with an `inotify` watch when
-    ///       available.
-    pub fn stream_changes(self, interval: Duration)
-        -> impl Stream<Item = CommonConfig, Error = ()>
     {
-        self.change_timestamps(interval)
+        use inotify::{Inotify, WatchMask};
-            .filter_map(move |_|
+        // Use a broad watch mask so that we will pick up any events that might
-                CommonConfig::load_from_disk(&self)
+        // indicate a change to the watched files.
-                    .map_err(|e| warn!("error reloading TLS config: {:?}, falling back", e))
+        //
-                    .ok()
+        // Such a broad mask may lead to reloading certs multiple times when k8s
-            )
+        // modifies a ConfigMap or Secret, which is a multi-step process that we
+        // see as a series CREATE, MOVED_TO, MOVED_FROM, and DELETE events.
+        // However, we want to catch single events that might occur when the
+        // files we're watching *don't* live in a k8s ConfigMap/Secret.
+        let mask = WatchMask::CREATE
+                 | WatchMask::MODIFY
+                 | WatchMask::DELETE
+                 | WatchMask::MOVE
+                 ;
+        let mut inotify = Inotify::init().map_err(Error::InotifyInit)?;
+
+        let paths = self.paths();
+        let paths = paths.into_iter()
+            .map(|path| {
+                // If the path to watch has a parent, watch that instead. This
+                // will allow us to pick up events to files in k8s ConfigMaps
+                // or Secrets (which we wouldn't detect if we watch the file
+                // itself, as they are double-symlinked).
+                //
+                // This may also result in some false positives (if a file we
+                // *don't* care about in the same dir changes, we'll still
+                // reload), but that's unlikely to be a problem.
+                let parent = path
+                    .parent()
+                    .map(Path::to_path_buf)
+                    .unwrap_or(path.to_path_buf());
+                trace!("will watch {:?} for {:?}", parent, path);
+                path
+            })
+            // Collect the paths into a `HashSet` eliminates any duplicates, to
+            // conserve the number of inotify watches we create.
+            .collect::<HashSet<_>>();
+
+        for path in paths {
+            inotify.add_watch(path, mask)
+                .map_err(|e| Error::Io(path.to_path_buf(), e))?;
+            trace!("inotify: watch {:?}", path);
+        }
+
+        let events = inotify.into_event_stream()
+            .map(|ev| {
+                trace!("inotify: event={:?}; path={:?};", ev.mask, ev.name);
+            })
+            .map_err(|e| error!("inotify watch error: {}", e));
+        trace!("started inotify watch");
+
+        Ok(events)
     }
 }
 
@@ -202,10 +293,10 @@ impl ServerConfig {
         let f = server_configs.map(Some).forward(store)
             .map(|_| trace!("forwarding to server config watch finished."));
 
-        // NOTE: This function and `no_tls` return `Box<Future<...>>` rather
+        // This function and `no_tls` return `Box<Future<...>>` rather than
-        //       than `impl Future<...>` so that they can have the _same_
+        // `impl Future<...>` so that they can have the _same_ return types
-        //       return types (impl Traits are not the same type unless the
+        // (impl Traits are not the same type unless the original
-        //       original non-anonymized type was the same).
+        // non-anonymized type was the same).
         (watch, Box::new(f))
     }