Make the router HTTP-agnostic (#394)

There's nothing HTTP-specific about the router.

This change moves the layer implementations--which unnecessarily enforce
HTTP-specific type requirements--into the generic router crate, relaxing
its type constraints.

Author: olix0r

Cargo.lock

@@ -1035,13 +1035,14 @@ version = "0.1.0"
 dependencies = [
  "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
  "indexmap 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
- "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "linkerd2-error 0.1.0",
  "tokio 0.1.20 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-sync 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-timer 0.2.10 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tower 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "tower-load-shed 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "tower-service 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tracing 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tracing-futures 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]

linkerd/app/core/src/errors.rs

@@ -116,7 +116,7 @@ where
 
 fn map_err_to_5xx(e: Error) -> StatusCode {
     use crate::proxy::buffer;
-    use crate::proxy::http::router::error as router;
+    use linkerd2_router::error as router;
     use tower::load_shed::error as shed;
 
     if let Some(ref c) = e.downcast_ref::<router::NoCapacity>() {

linkerd/app/core/src/lib.rs

@@ -18,6 +18,7 @@ pub use linkerd2_metrics as metrics;
 pub use linkerd2_opencensus as opencensus;
 pub use linkerd2_reconnect as reconnect;
 pub use linkerd2_request_filter as request_filter;
+pub use linkerd2_router as router;
 pub use linkerd2_trace_context as trace_context;
 
 pub mod accept_error;

linkerd/app/inbound/src/endpoint.rs

@@ -4,10 +4,8 @@ use linkerd2_app_core::{
     classify,
     dst::{DstAddr, Route},
     metric_labels::EndpointLabels,
-    proxy::{
-        http::{router, settings},
-        identity, tap,
-    },
+    proxy::{http::settings, identity, tap},
+    router,
     transport::{connect, tls},
     Conditional, NameAddr,
 };
@@ -156,7 +154,8 @@ mod tests {
     use super::{Endpoint, RecognizeEndpoint};
     use http;
     use linkerd2_app_core::{
-        proxy::http::{router::Recognize, Settings},
+        proxy::http::Settings,
+        router::Recognize,
         transport::{listen, tls},
         Conditional,
     };

linkerd/app/inbound/src/lib.rs

@@ -17,14 +17,14 @@ use linkerd2_app_core::{
     proxy::{
         self,
         http::{
-            client, insert, metrics as http_metrics, normalize_uri, profiles, router, settings,
+            client, insert, metrics as http_metrics, normalize_uri, profiles, settings,
             strip_header,
         },
         identity,
         server::{Protocol as ServerProtocol, Server},
         tap, tcp,
     },
-    reconnect, serve,
+    reconnect, router, serve,
     spans::SpanConverter,
     svc, trace, trace_context,
     transport::{self, connect, tls, OrigDstAddr, SysOrigDstAddr},
@@ -138,7 +138,7 @@ impl<A: OrigDstAddr> Config<A> {
                 ))
                 .push_buffer_pending(buffer.max_in_flight, DispatchDeadline::extract)
                 .makes::<Endpoint>()
-                .push(router::layer(
+                .push(router::Layer::new(
                     router::Config::new(router_capacity, router_max_idle_age),
                     RecognizeEndpoint::default(),
                 ))
@@ -194,7 +194,7 @@ impl<A: OrigDstAddr> Config<A> {
             // address is used.
             let dst_router = dst_stack
                 .push_buffer_pending(buffer.max_in_flight, DispatchDeadline::extract)
-                .push(router::layer(
+                .push(router::Layer::new(
                     router::Config::new(router_capacity, router_max_idle_age),
                     |req: &http::Request<_>| {
                         let dst = req

linkerd/app/outbound/src/lib.rs

@@ -20,14 +20,13 @@ use linkerd2_app_core::{
         discover,
         http::{
             balance, canonicalize, client, fallback, header_from_target, insert,
-            metrics as http_metrics, normalize_uri, profiles, retry, router, settings,
-            strip_header,
+            metrics as http_metrics, normalize_uri, profiles, retry, settings, strip_header,
         },
         identity,
         resolve::map_endpoint,
         tap, tcp, Server,
     },
-    reconnect, serve,
+    reconnect, router, serve,
     spans::SpanConverter,
     svc, trace, trace_context,
     transport::{self, connect, tls, OrigDstAddr, SysOrigDstAddr},
@@ -204,7 +203,7 @@ impl<A: OrigDstAddr> Config<A> {
             // used as the server name when connecting to the endpoint.
             let orig_dst_router_layer = svc::layers()
                 .push_buffer_pending(buffer.max_in_flight, DispatchDeadline::extract)
-                .push(router::layer(
+                .push(router::Layer::new(
                     router::Config::new(router_capacity, router_max_idle_age),
                     Endpoint::from_request,
                 ));
@@ -251,7 +250,7 @@ impl<A: OrigDstAddr> Config<A> {
                     |dst: &DstAddr| info_span!("logical", dst.logical = %dst.dst_logical()),
                 ))
                 .push_buffer_pending(buffer.max_in_flight, DispatchDeadline::extract)
-                .push(router::layer(
+                .push(router::Layer::new(
                     router::Config::new(router_capacity, router_max_idle_age),
                     |req: &http::Request<_>| {
                         req.extensions().get::<Addr>().cloned().map(|addr| {
@@ -289,7 +288,7 @@ impl<A: OrigDstAddr> Config<A> {
                 .push(insert::target::layer())
                 .push(trace::layer(|addr: &Addr| info_span!("addr", %addr)))
                 .push_buffer_pending(buffer.max_in_flight, DispatchDeadline::extract)
-                .push(router::layer(
+                .push(router::Layer::new(
                     router::Config::new(router_capacity, router_max_idle_age),
                     |req: &http::Request<_>| {
                         http_request_l5d_override_dst_addr(req)

linkerd/proxy/http/src/lib.rs

@@ -20,7 +20,6 @@ pub mod normalize_uri;
 pub mod orig_proto;
 pub mod profiles;
 pub mod retry;
-pub mod router;
 pub mod settings;
 pub mod strip_header;
 pub mod timeout;

linkerd/router/Cargo.toml

@@ -8,10 +8,11 @@ publish = false
 [dependencies]
 futures = "0.1"
 indexmap = "1.0.0"
-log = "0.4"
+linkerd2-error = { path = "../error" }
 tower-load-shed = "0.1"
 tokio = "0.1.20"
 tokio-sync = "0.1.6"
 tokio-timer = "0.2.4"
-tower-service = "0.2"
+tower = "0.1"
 tracing = "0.1.2"
+tracing-futures = "0.1"

linkerd/proxy/http/src/router.rs renamed to linkerd/router/src/layer.rs

@@ -1,8 +1,6 @@
+use crate::{Recognize, Router};
 use futures::Poll;
-use http;
 use linkerd2_error::{Error, Never};
-use linkerd2_router as rt;
-pub use linkerd2_router::{error, Recognize, Router};
 use std::marker::PhantomData;
 use std::time::Duration;
 use tracing::{info_span, trace};
@@ -27,7 +25,7 @@ pub struct Layer<Req, Rec: Recognize<Req>> {
 }
 
 #[derive(Debug)]
-pub struct Stack<Req, Rec: Recognize<Req>, Mk> {
+pub struct Make<Req, Rec: Recognize<Req>, Mk> {
     config: Config,
     recognize: Rec,
     inner: Mk,
@@ -37,7 +35,7 @@ pub struct Stack<Req, Rec: Recognize<Req>, Mk> {
 pub struct Service<Req, Rec, Mk>
 where
     Rec: Recognize<Req>,
-    Mk: rt::Make<Rec::Target>,
+    Mk: super::Make<Rec::Target>,
     Mk::Value: tower::Service<Req>,
 {
     inner: Router<Req, Rec, Mk>,
@@ -56,29 +54,30 @@ impl Config {
 
 // === impl Layer ===
 
-pub fn layer<Rec, Req>(config: Config, recognize: Rec) -> Layer<Req, Rec>
+impl<Req, Rec> Layer<Req, Rec>
 where
     Rec: Recognize<Req> + Clone + Send + Sync + 'static,
 {
-    Layer {
-        config,
-        recognize,
-        _p: PhantomData,
+    pub fn new(config: Config, recognize: Rec) -> Self {
+        Self {
+            config,
+            recognize,
+            _p: PhantomData,
+        }
     }
 }
 
-impl<Req, Rec, Mk, B> tower::layer::Layer<Mk> for Layer<Req, Rec>
+impl<Req, Rec, Mk> tower::layer::Layer<Mk> for Layer<Req, Rec>
 where
     Rec: Recognize<Req> + Clone + Send + Sync + 'static,
-    Mk: rt::Make<Rec::Target> + Clone + Send + Sync + 'static,
-    Mk::Value: tower::Service<Req, Response = http::Response<B>> + Clone,
+    Mk: super::Make<Rec::Target> + Clone + Send + Sync + 'static,
+    Mk::Value: tower::Service<Req> + Clone,
     <Mk::Value as tower::Service<Req>>::Error: Into<Error>,
-    B: Default + Send + 'static,
 {
-    type Service = Stack<Req, Rec, Mk>;
+    type Service = Make<Req, Rec, Mk>;
 
     fn layer(&self, inner: Mk) -> Self::Service {
-        Stack {
+        Make {
             inner,
             config: self.config.clone(),
             recognize: self.recognize.clone(),
@@ -92,19 +91,18 @@ where
     Rec: Recognize<Req> + Clone + Send + Sync + 'static,
 {
     fn clone(&self) -> Self {
-        layer(self.config.clone(), self.recognize.clone())
+        Self::new(self.config.clone(), self.recognize.clone())
     }
 }
-// === impl Stack ===
+// === impl Make ===
 
-impl<Req, Rec, Mk, B> Stack<Req, Rec, Mk>
+impl<Req, Rec, Mk> Make<Req, Rec, Mk>
 where
     Rec: Recognize<Req> + Clone + Send + Sync + 'static,
     <Rec as Recognize<Req>>::Target: Send + 'static,
-    Mk: rt::Make<Rec::Target> + Clone + Send + Sync + 'static,
-    Mk::Value: tower::Service<Req, Response = http::Response<B>> + Clone + Send + 'static,
+    Mk: super::Make<Rec::Target> + Clone + Send + Sync + 'static,
+    Mk::Value: tower::Service<Req> + Clone + Send + 'static,
     <Mk::Value as tower::Service<Req>>::Error: Into<Error>,
-    B: Default + Send + 'static,
 {
     pub fn make(&self) -> Service<Req, Rec, Mk> {
         let (inner, cache_bg) = Router::new(
@@ -118,14 +116,13 @@ where
     }
 }
 
-impl<Req, Rec, Mk, B, T> tower::Service<T> for Stack<Req, Rec, Mk>
+impl<Req, Rec, Mk, T> tower::Service<T> for Make<Req, Rec, Mk>
 where
     Rec: Recognize<Req> + Clone + Send + Sync + 'static,
     <Rec as Recognize<Req>>::Target: Send + 'static,
-    Mk: rt::Make<Rec::Target> + Clone + Send + Sync + 'static,
-    Mk::Value: tower::Service<Req, Response = http::Response<B>> + Clone + Send + 'static,
+    Mk: super::Make<Rec::Target> + Clone + Send + Sync + 'static,
+    Mk::Value: tower::Service<Req> + Clone + Send + 'static,
     <Mk::Value as tower::Service<Req>>::Error: Into<Error>,
-    B: Default + Send + 'static,
 {
     type Response = Service<Req, Rec, Mk>;
     type Error = Never;
@@ -140,7 +137,7 @@ where
     }
 }
 
-impl<Req, Rec, Mk> Clone for Stack<Req, Rec, Mk>
+impl<Req, Rec, Mk> Clone for Make<Req, Rec, Mk>
 where
     Rec: Recognize<Req> + Clone,
     Mk: Clone,
@@ -156,13 +153,12 @@ where
 }
 // === impl Service ===
 
-impl<Req, Rec, Mk, B> tower::Service<Req> for Service<Req, Rec, Mk>
+impl<Req, Rec, Mk> tower::Service<Req> for Service<Req, Rec, Mk>
 where
     Rec: Recognize<Req> + Send + Sync + 'static,
-    Mk: rt::Make<Rec::Target> + Clone + Send + Sync + 'static,
-    Mk::Value: tower::Service<Req, Response = http::Response<B>> + Clone,
+    Mk: super::Make<Rec::Target> + Clone + Send + Sync + 'static,
+    Mk::Value: tower::Service<Req> + Clone,
     <Mk::Value as tower::Service<Req>>::Error: Into<Error>,
-    B: Default + Send + 'static,
 {
     type Response = <Router<Req, Rec, Mk> as tower::Service<Req>>::Response;
     type Error = <Router<Req, Rec, Mk> as tower::Service<Req>>::Error;
@@ -181,7 +177,7 @@ where
 impl<Req, Rec, Mk> Clone for Service<Req, Rec, Mk>
 where
     Rec: Recognize<Req>,
-    Mk: rt::Make<Rec::Target>,
+    Mk: super::Make<Rec::Target>,
     Mk::Value: tower::Service<Req>,
     Router<Req, Rec, Mk>: Clone,
 {