Adds failfast mode for unroutable requests.

* Fixes Issue #26
* Adds a failfast mode for currently unroutable requests.
* Adds an integration test failfast.
* Adds a mock namerd.
* Adds a mock static webserver.
* Improves some comments.

Cargo.toml

@@ -34,3 +34,10 @@ tokio-io = "0.1"
 tokio-service = "0.1"
 tokio-timer = "0.1"
 url = "1.4"
+
+[dev-dependencies]
+env_logger = { version = "0.3", default-features = false }
+futures = "0.1"
+# We use not-yet-released tokio integration on master:
+hyper = { git = "https://github.com/hyperium/hyper", rev = "5a3743c1" }
+tokio-core = "0.1"

src/app/mod.rs

@@ -9,7 +9,7 @@ use std::cell::RefCell;
 use std::collections::{VecDeque, HashMap};
 use std::fs::File;
 use std::io::{self, BufReader};
-use std::net;
+use std::net::{self, SocketAddr};
 use std::rc::Rc;
 use std::time::Duration;
 use tacho::{self, Tacho};
@@ -21,12 +21,12 @@ mod admin_http;
 mod sni;
 pub mod config;
 
-use self::config::*;
-use self::sni::Sni;
 use WeightedAddr;
 use lb::{Balancer, Acceptor, Connector, PlainAcceptor, PlainConnector, SecureAcceptor,
          SecureConnector};
 use namerd;
+use self::config::*;
+use self::sni::Sni;
 
 const DEFAULT_BUFFER_SIZE: usize = 8 * 1024;
 const DEFAULT_MAX_WAITERS: usize = 8;
@@ -95,7 +95,7 @@ pub fn configure(app: AppConfig) -> (Admin, Proxies) {
 
 pub trait Loader: Sized {
     type Run: Future<Item = (), Error = io::Error>;
-    fn load(self, handle: Handle) -> io::Result<Self::Run>;
+    fn load(self, handle: Handle) -> io::Result<(SocketAddr, Self::Run)>;
 }
 pub trait Runner: Sized {
     fn run(self) -> io::Result<()>;
@@ -104,7 +104,7 @@ pub trait Runner: Sized {
 impl<L: Loader> Runner for L {
     fn run(self) -> io::Result<()> {
         let mut core = Core::new()?;
-        let fut = self.load(core.handle())?;
+        let (_, fut) = self.load(core.handle())?;
         core.run(fut)
     }
 }
@@ -118,12 +118,12 @@ pub struct Admin {
 }
 impl Loader for Admin {
     type Run = Running;
-    fn load(self, handle: Handle) -> io::Result<Running> {
+    fn load(self, handle: Handle) -> io::Result<(SocketAddr, Running)> {
         let mut running = Running::new();
         {
             let mut namerds = self.namerds;
             for _ in 0..namerds.len() {
-                let f = namerds.pop_front().unwrap().load(handle.clone())?;
+                let (_, f) = namerds.pop_front().unwrap().load(handle.clone())?;
                 running.register(f.map_err(|_| io::ErrorKind::Other.into()));
             }
         }
@@ -163,19 +163,19 @@ impl Loader for Admin {
             });
             running.register(srv);
         }
-        Ok(running)
+        Ok((self.addr, running))
     }
 }
 
 
-struct Namerd {
-    config: NamerdConfig,
-    sender: mpsc::Sender<Vec<WeightedAddr>>,
-    metrics: tacho::Metrics,
+pub struct Namerd {
+    pub config: NamerdConfig,
+    pub sender: mpsc::Sender<Vec<WeightedAddr>>,
+    pub metrics: tacho::Metrics,
 }
 impl Loader for Namerd {
     type Run = Box<Future<Item = (), Error = io::Error>>;
-    fn load(self, handle: Handle) -> io::Result<Self::Run> {
+    fn load(self, handle: Handle) -> io::Result<(SocketAddr, Self::Run)> {
         let path = self.config.path;
         let addr = self.config.addr;
         let interval_secs = self.config.interval_secs.unwrap_or(DEFAULT_NAMERD_SECONDS);
@@ -194,7 +194,7 @@ impl Loader for Namerd {
             let sink = self.sender.sink_map_err(|_| error!("sink error"));
             addrs.forward(sink).map_err(|_| io::ErrorKind::Other.into()).map(|_| {})
         };
-        Ok(Box::new(driver))
+        Ok((addr, Box::new(driver)))
     }
 }
 
@@ -203,29 +203,32 @@ pub struct Proxies {
 }
 impl Loader for Proxies {
     type Run = Running;
-    fn load(self, handle: Handle) -> io::Result<Running> {
+    fn load(self, handle: Handle) -> io::Result<(SocketAddr, Running)> {
         let mut running = Running::new();
         let mut proxies = self.proxies;
+        let mut addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
         for _ in 0..proxies.len() {
             let p = proxies.pop_front().unwrap();
-            let f = p.load(handle.clone())?;
+            let (_addr, f) = p.load(handle.clone())?;
+            addr = _addr;
             running.register(f);
         }
-        Ok(running)
+        Ok((addr, running))
     }
 }
 
-struct Proxy {
-    client: Option<ClientConfig>,
-    server: ProxyServer,
+pub struct Proxy {
+    pub client: Option<ClientConfig>,
+    pub server: ProxyServer,
 }
 impl Loader for Proxy {
     type Run = Running;
-    fn load(self, handle: Handle) -> io::Result<Running> {
+    fn load(self, handle: Handle) -> io::Result<(SocketAddr, Running)> {
         match self.client.and_then(|c| c.tls) {
             None => {
                 let conn = PlainConnector::new(handle.clone());
-                self.server.load(&handle, conn)
+                let f = self.server.load(&handle, conn).expect("b");
+                Ok(f)
             }
             Some(ref c) => {
                 let mut tls = rustls::ClientConfig::new();
@@ -238,29 +241,33 @@ impl Loader for Proxy {
                     }
                 };
                 let conn = SecureConnector::new(c.dns_name.clone(), tls, handle.clone());
-                self.server.load(&handle, conn)
+                let f = self.server.load(&handle, conn).expect("a");
+                Ok(f)
             }
         }
     }
 }
 
-struct ProxyServer {
-    label: String,
-    servers: Vec<ServerConfig>,
-    addrs: Box<Stream<Item = Vec<WeightedAddr>, Error = ()>>,
-    buf: Rc<RefCell<Vec<u8>>>,
-    max_waiters: usize,
-    metrics: tacho::Metrics,
+pub struct ProxyServer {
+    pub label: String,
+    pub servers: Vec<ServerConfig>,
+    pub addrs: Box<Stream<Item = Vec<WeightedAddr>, Error = ()>>,
+    pub buf: Rc<RefCell<Vec<u8>>>,
+    pub max_waiters: usize,
+    pub metrics: tacho::Metrics,
 }
 impl ProxyServer {
-    fn load<C>(self, handle: &Handle, conn: C) -> io::Result<Running>
+    fn load<C>(self, handle: &Handle, conn: C) -> io::Result<(SocketAddr, Running)>
         where C: Connector + 'static
     {
         let addrs = self.addrs.map_err(|_| io::ErrorKind::Other.into());
         let metrics = self.metrics.clone().labeled("proxy".into(), self.label.into());
         let bal = Balancer::new(addrs, conn, self.buf.clone(), metrics.clone())
             .into_shared(self.max_waiters, handle.clone());
 
+        // Placeholder for our local listening SocketAddr.
+        let mut local_addr: SocketAddr = "127.0.0.1:0".parse().expect("unable to parse addr");
+
         // TODO scope/tag stats for servers.
 
         let mut running = Running::new();
@@ -271,7 +278,9 @@ impl ProxyServer {
                 ServerConfig::Tcp { ref addr } => {
                     let metrics = metrics.clone().labeled("srv".into(), format!("{}", addr));
                     let acceptor = PlainAcceptor::new(handle, metrics);
-                    let f = acceptor.accept(addr).forward(bal).map(|_| {});
+                    let (bound_addr, forwarder) = acceptor.accept(addr);
+                    local_addr = bound_addr;
+                    let f = forwarder.forward(bal).map(|_| {});
                     running.register(f);
                 }
                 ServerConfig::Tls { ref addr,
@@ -287,12 +296,14 @@ impl ProxyServer {
 
                     let metrics = metrics.clone().labeled("srv".into(), format!("{}", addr));
                     let acceptor = SecureAcceptor::new(handle, tls, metrics);
-                    let f = acceptor.accept(addr).forward(bal).map(|_| {});
+                    let (bound_addr, forwarder) = acceptor.accept(addr);
+                    local_addr = bound_addr;
+                    let f = forwarder.forward(bal).map(|_| {});
                     running.register(f);
                 }
             }
         }
-        Ok(running)
+        Ok((local_addr, running))
     }
 }
 

src/lb/balancer.rs

@@ -44,6 +44,8 @@ pub struct Balancer<A, C> {
     retired: VecDeque<Endpoint>,
 
     stats: Stats,
+
+    fail_fast_mode: bool,
 }
 
 impl<A, C> Balancer<A, C>
@@ -64,6 +66,7 @@ impl<A, C> Balancer<A, C>
             ready: VecDeque::new(),
             retired: VecDeque::new(),
             stats: Stats::new(metrics),
+            fail_fast_mode: false,
         }
     }
 
@@ -140,6 +143,9 @@ impl<A, C> Balancer<A, C>
         if let Async::Ready(addrs) = self.addrs.poll()? {
             trace!("addr update");
             let addrs = addrs.expect("addr stream must be infinite");
+            // If there are no addrs to route to, drop requests quickly.
+            // TODO: validate that fail_fast_mode is being disabled once addrs exist.
+            self.fail_fast_mode = addrs.is_empty();
             let new = addr_weight_map(&addrs);
             self.update_endpoints(&new);
         }
@@ -357,12 +363,18 @@ impl<A, C> Sink for Balancer<A, C>
                 self.evict_retirees(&mut rec)?;
                 self.promote_unready(&mut rec)?;
                 self.discover_and_retire()?;
-                trace!("retrying {} unready={} ready={} retired={}",
+                trace!("retrying {} unready={} ready={} retired={} failfast={}",
                        src_addr,
                        self.unready.len(),
                        self.ready.len(),
-                       self.retired.len());
-                self.dispatch(src, &mut rec)
+                       self.retired.len(),
+                       self.fail_fast_mode);
+                if self.fail_fast_mode {
+                    trace!("in fail fast mode, dropping traffic");
+                    Err(io::ErrorKind::Other.into())
+                } else {
+                    self.dispatch(src, &mut rec)
+                }
             }
         };
 

src/lb/mod.rs

@@ -45,9 +45,11 @@ impl WithAddr for Src {
     }
 }
 
-/// Binds on `addr` and produces `U`-typed src connections.
+/// Binds on `addr` and produces the bound `SocketAddr` and a` Stream` of `Src` connections.
 pub trait Acceptor {
-    fn accept(&self, addr: &SocketAddr) -> Box<Stream<Item = Src, Error = io::Error>>;
+    fn accept(&self,
+              addr: &SocketAddr)
+              -> (SocketAddr, Box<Stream<Item = Src, Error = io::Error>>);
 }
 
 /// Establishes a `D`-typed connection to `addr`.
@@ -71,17 +73,20 @@ impl PlainAcceptor {
     }
 }
 impl Acceptor for PlainAcceptor {
-    fn accept(&self, addr: &SocketAddr) -> Box<Stream<Item = Src, Error = io::Error>> {
+    fn accept(&self,
+              addr: &SocketAddr)
+              -> (SocketAddr, Box<Stream<Item = Src, Error = io::Error>>) {
         let metrics = self.metrics.clone();
         let connects_key = self.connects_key.clone();
-        TcpListener::bind(addr, &self.handle)
-            .unwrap()
-            .incoming()
+        let listener = TcpListener::bind(addr, &self.handle).expect("could not bind to address");
+        let local_addr = listener.local_addr().expect("could not get local_addr from listener");
+        let worker = listener.incoming()
             .map(move |(s, a)| {
                 metrics.recorder().incr(&connects_key, 1);
                 Src(Socket::plain(a, s))
             })
-            .boxed()
+            .boxed();
+        (local_addr, worker)
     }
 }
 
@@ -119,9 +124,13 @@ impl SecureAcceptor {
     }
 }
 impl Acceptor for SecureAcceptor {
-    fn accept(&self, addr: &SocketAddr) -> Box<Stream<Item = Src, Error = io::Error>> {
+    fn accept(&self,
+              addr: &SocketAddr)
+              -> (SocketAddr, Box<Stream<Item = Src, Error = io::Error>>) {
         let tls = self.config.clone();
-        let l = TcpListener::bind(addr, &self.handle).unwrap();
+        let l = TcpListener::bind(addr, &self.handle)
+            .expect("could not bind listener for SecureAcceptor");
+        let local_addr = l.local_addr().expect("could not get local_addr from listener");
 
         let metrics = self.metrics.clone();
         let connects_key = self.connects_key.clone();
@@ -143,7 +152,7 @@ impl Acceptor for SecureAcceptor {
                 }
             }
         });
-        Box::new(srcs)
+        (local_addr, Box::new(srcs))
     }
 }
 

src/lb/proxy_stream.rs

@@ -12,8 +12,8 @@ use tokio_io::AsyncWrite;
 /// A future representing reading all data from one side of a proxy connection and writing
 /// it to another.
 ///
-/// In the typical case, nothing allocations are required.  If the write side exhibits
-/// backpressure, however, a buffer is allocated to
+/// In the typical case, no allocations are required.  If the write side exhibits
+/// backpressure, however, a buffer is allocated.
 pub struct ProxyStream {
     reader: Rc<RefCell<Socket>>,
     writer: Rc<RefCell<Socket>>,

src/namerd.rs

@@ -83,7 +83,7 @@ fn request<C: Connect>(client: Rc<Client<C>>, url: Url, stats: Stats) -> AddrsFu
                     }
                 }
                 Err(e) => {
-                    error!("failed to read response: {}", e);
+                    error!("failed to read response from remote namerd: {}", e);
                     future::ok(None).boxed()
                 }
             })
@@ -136,9 +136,10 @@ fn parse_chunks(chunks: &[Chunk]) -> Option<Vec<::WeightedAddr>> {
     let result: json::Result<NamerdResponse> = json::from_reader(r);
     match result {
         Ok(ref nrsp) if nrsp.kind == "bound" => Some(to_weighted_addrs(&nrsp.addrs)),
+        Ok(ref nrsp) if nrsp.kind == "neg" => Some(vec![]),
         Ok(_) => Some(vec![]),
         Err(e) => {
-            info!("error parsing response: {}", e);
+            error!("error parsing response: {}", e);
             None
         }
     }
@@ -159,7 +160,9 @@ fn to_weighted_addrs(namerd_addrs: &[NamerdAddr]) -> Vec<::WeightedAddr> {
 struct NamerdResponse {
     #[serde(rename = "type")]
     kind: String,
+    #[serde(default)]
     addrs: Vec<NamerdAddr>,
+    #[serde(default)]
     meta: HashMap<String, String>,
 }
 

tests/lib.rs

@@ -0,0 +1,13 @@
+#[cfg(tests)]
+mod tests;
+extern crate log;
+
+extern crate env_logger;
+extern crate futures;
+extern crate hyper;
+extern crate tokio_core;
+extern crate tokio_io;
+extern crate linkerd_tcp;
+
+mod mocks;
+pub use mocks::MockNamerd;