Route requests to original dsts when no endpoints exist in the lb

src/app/main.rs

@@ -21,8 +21,9 @@ use never::Never;
 use proxy::{
     self, accept, buffer,
     http::{
-        client, insert_target, metrics as http_metrics, normalize_uri, profiles, router, settings,
-        strip_header,
+        client, insert_target, metrics as http_metrics, normalize_uri, profiles,
+        router::{self, rt::Make},
+        settings, strip_header,
     },
     pending, reconnect,
 };
@@ -35,7 +36,7 @@ use {Addr, Conditional};
 
 use super::admin::{Admin, Readiness};
 use super::config::{Config, H2Settings};
-use super::dst::DstAddr;
+use super::dst::{self, DstAddr};
 use super::identity;
 use super::profiles::Client as ProfilesClient;
 
@@ -404,6 +405,7 @@ where
                 //add_remote_ip_on_rsp, add_server_id_on_rsp,
                 discovery::Resolve,
                 orig_proto_upgrade,
+                Endpoint,
             };
             use proxy::{
                 http::{balance, canonicalize, header_from_target, metrics, retry},
@@ -473,14 +475,32 @@ where
             //    is retryable.
             let dst_route_layer = svc::builder()
                 .layer(buffer::layer(max_in_flight))
-                .layer(pending::layer())
+                .layer(pending::layer::<_, dst::Route, _>())
                 .layer(classify::layer())
                 .layer(metrics::layer::<_, classify::Response>(route_http_metrics))
                 .layer(proxy::http::timeout::layer())
                 .layer(retry::layer(retry_http_metrics.clone()))
                 .layer(metrics::layer::<_, classify::Response>(retry_http_metrics))
                 .layer(insert_target::layer());
 
+            let balancer = svc::builder()
+                .layer(
+                    balance::layer(
+                        EWMA_DEFAULT_RTT,
+                        EWMA_DECAY,
+                        resolve::layer(Resolve::new(resolver)),
+                    )
+                    .with_fallback(max_in_flight, |req: &http::Request<_>| {
+                        let ep = Endpoint::from_orig_dst(req);
+                        debug!("outbound ep={:?}", ep);
+                        ep
+                    }),
+                )
+                .layer(pending::layer())
+                .layer(balance::weight::layer())
+                .service(endpoint_stack)
+                .make(&router::Config::new("out ep", capacity, max_idle_age));
+
             // A per-`DstAddr` stack that does the following:
             //
             // 1. Adds the `CANONICAL_DST_HEADER` from the `DstAddr`.
@@ -497,11 +517,7 @@ where
                 ))
                 .layer(buffer::layer(max_in_flight))
                 .layer(pending::layer())
-                .layer(balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY))
-                .layer(resolve::layer(Resolve::new(resolver)))
-                .layer(pending::layer())
-                .layer(balance::weight::layer())
-                .service(endpoint_stack);
+                .service(balancer);
 
             // Routes request using the `DstAddr` extension.
             //
@@ -653,7 +669,7 @@ where
             // implementations can use the route-specific configuration.
             let dst_route_stack = svc::builder()
                 .layer(buffer::layer(max_in_flight))
-                .layer(pending::layer())
+                .layer(pending::layer::<_, dst::Route, _>())
                 .layer(classify::layer())
                 .layer(http_metrics::layer::<_, classify::Response>(
                     route_http_metrics,
@@ -673,7 +689,7 @@ where
                     dst_route_stack,
                 ))
                 .layer(buffer::layer(max_in_flight))
-                .layer(pending::layer())
+                .layer(pending::layer::<_, DstAddr, _>())
                 .layer(insert_target::layer())
                 .service(svc::shared(endpoint_router));
 

src/app/outbound.rs

@@ -5,8 +5,13 @@ use std::{fmt, hash};
 
 use super::identity;
 use control::destination::{Metadata, ProtocolHint};
-use proxy::http::balance::{HasWeight, Weight};
-use proxy::http::settings;
+use proxy::{
+    self,
+    http::{
+        balance::{HasWeight, Weight},
+        settings,
+    },
+};
 use tap;
 use transport::{connect, tls};
 use {Conditional, NameAddr};
@@ -44,6 +49,23 @@ impl Endpoint {
             }
         }
     }
+
+    pub fn from_orig_dst<B>(req: &http::Request<B>) -> Option<Self> {
+        let addr = req
+            .extensions()
+            .get::<proxy::Source>()?
+            .orig_dst_if_not_local()?;
+        let http_settings = settings::Settings::from_request(req);
+        Some(Self {
+            addr,
+            dst_name: None,
+            identity: Conditional::None(
+                tls::ReasonForNoPeerName::NotProvidedByServiceDiscovery.into(),
+            ),
+            metadata: Metadata::empty(),
+            http_settings,
+        })
+    }
 }
 
 impl From<SocketAddr> for Endpoint {
@@ -207,6 +229,7 @@ pub mod discovery {
         fn poll(&mut self) -> Poll<resolve::Update<Self::Endpoint>, Self::Error> {
             match self.resolving {
                 Resolving::Name(ref name, ref mut res) => match try_ready!(res.poll()) {
+                    resolve::Update::NoEndpoints => Ok(Async::Ready(resolve::Update::NoEndpoints)),
                     resolve::Update::Remove(addr) => {
                         debug!("removing {}", addr);
                         Ok(Async::Ready(resolve::Update::Remove(addr)))

src/control/destination/background/destination_set.rs

@@ -92,13 +92,13 @@ where
                                 .filter_map(|addr| pb_to_sock_addr(addr.clone())),
                         );
                     }
-                    Some(PbUpdate2::NoEndpoints(ref no_endpoints)) if no_endpoints.exists => {
-                        exists = Exists::Yes(());
+                    Some(PbUpdate2::NoEndpoints(ref no_endpoints)) => {
                         self.no_endpoints(auth, no_endpoints.exists);
-                    }
-                    Some(PbUpdate2::NoEndpoints(no_endpoints)) => {
-                        debug_assert!(!no_endpoints.exists);
-                        exists = Exists::No;
+                        if no_endpoints.exists {
+                            exists = Exists::Yes(());
+                        } else {
+                            exists = Exists::No;
+                        }
                     }
                     None => (),
                 },
@@ -224,6 +224,10 @@ where
         );
         match self.addrs.take() {
             Exists::Yes(mut cache) => {
+                self.responders.retain(|r| {
+                    let sent = r.update_tx.unbounded_send(Update::NoEndpoints);
+                    sent.is_ok()
+                });
                 cache.clear(&mut |change| {
                     Self::on_change(&mut self.responders, authority_for_logging, change)
                 });