http: Simplify stacks and target types (#656)

Currently, various properties of HTTP requests are exposed into the
stack target type, meaning that stacks (including their resolutions)
cannot be shared across requests that vary by any of these settings and,
worse, that per-request determinations must be made to determine which
stack to use.

In order to bind stacks according to connection-level metadata, where
these request-level parameters cannot be known, we need to remove
this metadata from the stack target.

The HTTP server now applies URI normalization as an internal
implementation detail (because it's a detail of how hyper clients and
servers differ). This eliminates the `ShouldNormalizeUri` trait, as that
determination can be made entirely based on the properties of the
request without any additional stack-specific metadata.

The HTTP client now takes an updated `Settings` type that explicitly
handles the orig-proto upgrade logic (removing this from the client
stack).

Stack targets now hold an `http::Version` on the accept-side, and map
that to a per-endpoint `http::client::Settings` for use by the http
client.

Author: olix0r

linkerd/app/core/src/control.rs

@@ -34,7 +34,7 @@ impl fmt::Display for ControlAddr {
 }
 
 type BalanceBody =
-    http::balance::PendingUntilFirstDataBody<tower::load::peak_ewma::Handle, hyper::Body>;
+    http::balance::PendingUntilFirstDataBody<tower::load::peak_ewma::Handle, http::Payload>;
 
 type RspBody = linkerd2_http_metrics::requests::ResponseBody<BalanceBody, classify::Eos>;
 

linkerd/app/gateway/src/lib.rs

@@ -153,7 +153,7 @@ mod test {
                 dst: dst_name
                     .map(|n| NameAddr::from_str(n).unwrap().into())
                     .unwrap_or_else(|| socket_addr.into()),
-                http_settings: http::Settings::Http2,
+                http_version: http::Version::Http1,
                 tls_client_id: peer_id,
             };
             assert_ready_ok!(make_gateway.poll_ready());

linkerd/app/gateway/src/make.rs

@@ -71,7 +71,7 @@ where
     fn call(&mut self, target: inbound::Target) -> Self::Future {
         let inbound::Target {
             dst,
-            http_settings,
+            http_version,
             tls_client_id,
             ..
         } = target;
@@ -120,7 +120,7 @@ where
                     let endpoint = outbound::HttpLogical {
                         dst: dst_name.clone().into(),
                         orig_dst: dst_addr,
-                        settings: http_settings,
+                        version: http_version,
                         require_identity: None,
                     };
 

linkerd/app/inbound/src/endpoint.rs

@@ -7,16 +7,14 @@ use linkerd2_app_core::{
     transport::{listen, tls},
     Addr, Conditional, CANONICAL_DST_HEADER, DST_OVERRIDE_HEADER,
 };
-use std::fmt;
-use std::net::SocketAddr;
-use std::sync::Arc;
+use std::{convert::TryInto, fmt, net::SocketAddr, sync::Arc};
 use tracing::debug;
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct Target {
     pub dst: Addr,
     pub socket_addr: SocketAddr,
-    pub http_settings: http::Settings,
+    pub http_version: http::Version,
     pub tls_client_id: tls::PeerIdentity,
 }
 
@@ -29,7 +27,7 @@ pub struct Logical {
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct HttpEndpoint {
     pub port: u16,
-    pub settings: http::Settings,
+    pub settings: http::client::Settings,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
@@ -53,17 +51,17 @@ impl Into<SocketAddr> for HttpEndpoint {
     }
 }
 
-impl AsRef<http::Settings> for HttpEndpoint {
-    fn as_ref(&self) -> &http::Settings {
-        &self.settings
+impl Into<http::client::Settings> for &'_ HttpEndpoint {
+    fn into(self) -> http::client::Settings {
+        self.settings
     }
 }
 
 impl From<Target> for HttpEndpoint {
     fn from(target: Target) -> Self {
         Self {
             port: target.socket_addr.port(),
-            settings: target.http_settings,
+            settings: target.http_version.into(),
         }
     }
 }
@@ -122,25 +120,6 @@ impl AsRef<Addr> for Target {
     }
 }
 
-impl http::normalize_uri::ShouldNormalizeUri for Target {
-    fn should_normalize_uri(&self) -> Option<http::uri::Authority> {
-        if let http::Settings::Http1 {
-            was_absolute_form: false,
-            ..
-        } = self.http_settings
-        {
-            return Some(self.dst.to_http_authority());
-        }
-        None
-    }
-}
-
-impl AsRef<http::Settings> for Target {
-    fn as_ref(&self) -> &http::Settings {
-        &self.http_settings
-    }
-}
-
 impl tls::HasPeerIdentity for Target {
     fn peer_identity(&self) -> tls::PeerIdentity {
         Conditional::None(tls::ReasonForNoPeerName::Loopback.into())
@@ -219,8 +198,8 @@ impl stack_tracing::GetSpan<()> for Target {
     fn get_span(&self, _: &()) -> tracing::Span {
         use tracing::info_span;
 
-        match self.http_settings {
-            http::Settings::Http2 => match self.dst.name_addr() {
+        match self.http_version {
+            http::Version::H2 => match self.dst.name_addr() {
                 None => info_span!(
                     "http2",
                     port = %self.socket_addr.port(),
@@ -231,25 +210,15 @@ impl stack_tracing::GetSpan<()> for Target {
                     port = %self.socket_addr.port(),
                 ),
             },
-            http::Settings::Http1 {
-                keep_alive,
-                wants_h1_upgrade,
-                was_absolute_form,
-            } => match self.dst.name_addr() {
+            http::Version::Http1 => match self.dst.name_addr() {
                 None => info_span!(
                     "http1",
                     port = %self.socket_addr.port(),
-                    keep_alive,
-                    wants_h1_upgrade,
-                    was_absolute_form,
                 ),
                 Some(name) => info_span!(
                     "http1",
                     %name,
                     port = %self.socket_addr.port(),
-                    keep_alive,
-                    wants_h1_upgrade,
-                    was_absolute_form,
                 ),
             },
         }
@@ -295,7 +264,10 @@ impl<A> router::Recognize<http::Request<A>> for RequestTarget {
             dst,
             socket_addr: self.accept.addrs.target_addr(),
             tls_client_id: self.accept.peer_identity.clone(),
-            http_settings: http::Settings::from_request(req),
+            http_version: req
+                .version()
+                .try_into()
+                .expect("HTTP version must be valid"),
         }
     }
 }

linkerd/app/inbound/src/lib.rs

@@ -16,7 +16,7 @@ use linkerd2_app_core::{
     opencensus::proto::trace::v1 as oc,
     profiles,
     proxy::{
-        http::{self, normalize_uri, orig_proto, strip_header, DetectHttp},
+        http::{self, orig_proto, strip_header, DetectHttp},
         identity, tap, tcp, SkipDetect,
     },
     reconnect, router, serve,
@@ -179,7 +179,7 @@ impl Config {
 
         // Creates HTTP clients for each inbound port & HTTP settings.
         let endpoint = svc::stack(tcp_connect)
-            .push(http::MakeClientLayer::new(connect.h2_settings))
+            .push(http::client::layer(connect.h2_settings))
             .push(reconnect::layer({
                 let backoff = connect.backoff.clone();
                 move |_| Ok(backoff.stream())
@@ -199,9 +199,6 @@ impl Config {
 
         let target = endpoint
             .push_map_target(HttpEndpoint::from)
-            // Normalizes the URI, i.e. if it was originally in
-            // absolute-form on the outbound side.
-            .push(normalize_uri::layer())
             .push(observe)
             .into_new_service()
             .check_new_service::<Target, http::Request<_>>();

linkerd/app/outbound/src/endpoint.rs

@@ -7,7 +7,7 @@ use linkerd2_app_core::{
     proxy::{
         api_resolve::{Metadata, ProtocolHint},
         http::override_authority::CanOverrideAuthority,
-        http::{self, identity_from_header, Settings},
+        http::{self, identity_from_header},
         identity,
         resolve::map_endpoint::MapEndpoint,
         tap,
@@ -16,8 +16,7 @@ use linkerd2_app_core::{
     transport::{listen, tls},
     Addr, Conditional, L5D_REQUIRE_ID,
 };
-use std::net::SocketAddr;
-use std::sync::Arc;
+use std::{convert::TryInto, net::SocketAddr, sync::Arc};
 
 #[derive(Copy, Clone, Debug)]
 pub struct FromMetadata;
@@ -26,7 +25,7 @@ pub struct FromMetadata;
 pub struct HttpLogical {
     pub dst: Addr,
     pub orig_dst: SocketAddr,
-    pub settings: http::Settings,
+    pub version: http::Version,
     pub require_identity: Option<identity::Name>,
 }
 
@@ -48,7 +47,7 @@ pub struct Profile {
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct HttpEndpoint {
     pub addr: SocketAddr,
-    pub settings: http::Settings,
+    pub settings: http::client::Settings,
     pub identity: tls::PeerIdentity,
     pub metadata: Metadata,
     pub concrete: HttpConcrete,
@@ -130,7 +129,7 @@ impl From<HttpLogical> for HttpEndpoint {
     fn from(logical: HttpLogical) -> Self {
         Self {
             addr: logical.orig_dst,
-            settings: logical.settings,
+            settings: logical.version.into(),
             identity: logical
                 .require_identity
                 .clone()
@@ -146,22 +145,6 @@ impl From<HttpLogical> for HttpEndpoint {
     }
 }
 
-impl HttpEndpoint {
-    pub fn can_use_orig_proto(&self) -> bool {
-        if let ProtocolHint::Unknown = self.metadata.protocol_hint() {
-            return false;
-        }
-
-        // Look at the original settings, ignoring any authority overrides.
-        match self.settings {
-            http::Settings::Http2 => false,
-            http::Settings::Http1 {
-                wants_h1_upgrade, ..
-            } => !wants_h1_upgrade,
-        }
-    }
-}
-
 impl std::hash::Hash for HttpEndpoint {
     fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
         self.addr.hash(state);
@@ -182,22 +165,9 @@ impl Into<SocketAddr> for HttpEndpoint {
     }
 }
 
-impl AsRef<http::Settings> for HttpEndpoint {
-    fn as_ref(&self) -> &http::Settings {
-        &self.settings
-    }
-}
-
-impl http::normalize_uri::ShouldNormalizeUri for HttpEndpoint {
-    fn should_normalize_uri(&self) -> Option<http::uri::Authority> {
-        if let http::Settings::Http1 {
-            was_absolute_form: false,
-            ..
-        } = self.settings
-        {
-            return Some(self.concrete.logical.dst.to_http_authority());
-        }
-        None
+impl Into<http::client::Settings> for &'_ HttpEndpoint {
+    fn into(self) -> http::client::Settings {
+        self.settings
     }
 }
 
@@ -260,18 +230,12 @@ impl MapEndpoint<HttpConcrete, Metadata> for FromMetadata {
                 Conditional::None(tls::ReasonForNoPeerName::NotProvidedByServiceDiscovery.into())
             });
 
-        let settings = match concrete.logical.settings {
-            Settings::Http1 {
-                keep_alive,
-                wants_h1_upgrade,
-                was_absolute_form,
-            } => Settings::Http1 {
-                keep_alive,
-                wants_h1_upgrade,
-                // Always use absolute form when an onverride is present.
-                was_absolute_form: metadata.authority_override().is_some() || was_absolute_form,
+        let settings = match concrete.logical.version {
+            http::Version::H2 => http::client::Settings::H2,
+            http::Version::Http1 => match metadata.protocol_hint() {
+                ProtocolHint::Unknown => http::client::Settings::Http1,
+                ProtocolHint::Http2 => http::client::Settings::OrigProtoUpgrade,
             },
-            settings => settings,
         };
 
         HttpEndpoint {
@@ -406,17 +370,18 @@ impl<B> router::Recognize<http::Request<B>> for LogicalPerRequest {
                 addr.into()
             });
 
-        let settings = http::Settings::from_request(req);
-
-        tracing::debug!(headers = ?req.headers(), uri = %req.uri(), dst = %dst, http.settings = ?settings, "Setting target for request");
+        tracing::debug!(headers = ?req.headers(), uri = %req.uri(), dst = %dst, version = ?req.version(), "Setting target for request");
 
         let require_identity = identity_from_header(req, L5D_REQUIRE_ID);
 
         HttpLogical {
             dst,
             orig_dst: self.0.target_addr(),
-            settings,
             require_identity,
+            version: req
+                .version()
+                .try_into()
+                .expect("HTTP version must be valid"),
         }
     }
 }

linkerd/app/outbound/src/lib.rs

@@ -34,11 +34,9 @@ use tokio::sync::mpsc;
 use tracing::{info, info_span};
 
 pub mod endpoint;
-mod orig_proto_upgrade;
 mod prevent_loop;
 mod require_identity_on_endpoint;
 
-use self::orig_proto_upgrade::OrigProtoUpgradeLayer;
 use self::prevent_loop::PreventLoop;
 use self::require_identity_on_endpoint::MakeRequireIdentityLayer;
 
@@ -166,7 +164,7 @@ impl Config {
             // Initiates an HTTP client on the underlying transport. Prior-knowledge HTTP/2
             // is typically used (i.e. when communicating with other proxies); though
             // HTTP/1.x fallback is supported as needed.
-            .push(http::MakeClientLayer::new(self.proxy.connect.h2_settings))
+            .push(http::client::layer(self.proxy.connect.h2_settings))
             // Re-establishes a connection when the client fails.
             .push(reconnect::layer({
                 let backoff = self.proxy.connect.backoff.clone();
@@ -179,13 +177,6 @@ impl Config {
                 HOST.as_str(),
                 CANONICAL_DST_HEADER,
             ]))
-            // Ensures that the request's URI is in the proper form.
-            .push(http::normalize_uri::layer())
-            // Upgrades HTTP/1 requests to be transported over HTTP/2 connections.
-            //
-            // This sets headers so that the inbound proxy can downgrade the request
-            // properly.
-            .push(OrigProtoUpgradeLayer::new())
             .push_on_response(svc::layers().box_http_response())
             .check_service::<HttpEndpoint>()
             .instrument(|e: &HttpEndpoint| info_span!("endpoint", peer.addr = %e.addr))