outbound: add stack tests for http (#765)

This branch adds some new stack tests for simple HTTP/1 and HTTP/2
meshed and unmeshed outbound requests. These tests also assert that TCP
load balancers are not constructed when HTTP traffic is detected.

The hope was that the panic on constructing TCP balancers would catch a
bug that exists in #744, but that issue appears to be related to the
cache instead. Still, it's probably worth having these tests.

Signed-off-by: Eliza Weisman <[email protected]>

Author: hawkw

linkerd/app/outbound/src/http/tests.rs

@@ -4,16 +4,31 @@ use crate::test_util::{
     *,
 };
 use crate::Config;
+use hyper::{
+    body::Buf,
+    client::conn::{Builder as ClientBuilder, SendRequest},
+    Body, Request, Response,
+};
 use linkerd2_app_core::{
     drain, metrics,
     proxy::{identity::Name, tap},
     svc::{self, NewService},
-    transport::{io, listen},
+    transport::{
+        io::{self, BoxedIo},
+        listen,
+    },
     Addr, Error,
 };
-use std::{net::SocketAddr, str::FromStr, time::Duration};
+use std::{
+    net::SocketAddr,
+    pin::Pin,
+    str::FromStr,
+    task::{Context, Poll},
+    time::Duration,
+};
 use tokio::time;
 use tower::{Service, ServiceExt};
+use tracing::Instrument;
 
 fn build_server<I>(
     cfg: Config,
@@ -53,11 +68,11 @@ where
         resolver.clone(),
         metrics.outbound.clone(),
     );
-    let svc = crate::server::stack(
+    let svc = crate::server::stack_with_tcp_balancer(
         &cfg,
         profiles,
-        resolver,
         support::connect::NoRawTcp,
+        NoTcpBalancer,
         router,
         metrics.outbound,
         None,
@@ -66,6 +81,32 @@ where
     (svc, drain_tx)
 }
 
+#[derive(Clone, Debug)]
+struct NoTcpBalancer;
+
+impl svc::NewService<crate::tcp::Concrete> for NoTcpBalancer {
+    type Service = Self;
+    fn new_service(&mut self, target: crate::tcp::Concrete) -> Self::Service {
+        panic!(
+            "no TCP load balancer should be created in this test!\n\ttarget = {:?}",
+            target
+        );
+    }
+}
+
+impl<I> svc::Service<I> for NoTcpBalancer {
+    type Response = ();
+    type Error = Error;
+    type Future = Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'static>>;
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        unreachable!("no TCP load balancer should be created in this test!");
+    }
+
+    fn call(&mut self, _: I) -> Self::Future {
+        unreachable!("no TCP load balancer should be created in this test!");
+    }
+}
+
 #[tokio::test(flavor = "current_thread")]
 async fn profile_endpoint_propagates_conn_errors() {
     // This test asserts that when profile resolution returns an endpoint, and
@@ -120,7 +161,7 @@ async fn profile_endpoint_propagates_conn_errors() {
         tracing::info!(?res, "Server complete");
         res
     });
-    let (mut client, conn) = hyper::client::conn::Builder::new()
+    let (mut client, conn) = ClientBuilder::new()
         .handshake(client_io)
         .await
         .expect("Client must connect");
@@ -135,7 +176,7 @@ async fn profile_endpoint_propagates_conn_errors() {
         .await
         .expect("Client must not fail")
         .call(
-            hyper::Request::builder()
+            Request::builder()
                 .header("Host", "foo.ns1.service.cluster.local")
                 .body(hyper::Body::default())
                 .unwrap(),
@@ -159,3 +200,196 @@ async fn profile_endpoint_propagates_conn_errors() {
     drop(client);
     drop(shutdown);
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn unmeshed_http1_hello_world() {
+    let mut server = hyper::server::conn::Http::new();
+    server.http1_only(true);
+    let client = ClientBuilder::new();
+    unmeshed_hello_world(server, client).await;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn unmeshed_http2_hello_world() {
+    let mut server = hyper::server::conn::Http::new();
+    server.http2_only(true);
+    let mut client = ClientBuilder::new();
+    client.http2_only(true);
+    unmeshed_hello_world(server, client).await;
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn meshed_hello_world() {
+    let _trace = support::trace_init();
+
+    let ep1 = SocketAddr::new([10, 0, 0, 41].into(), 5550);
+    let addrs = listen::Addrs::new(
+        ([127, 0, 0, 1], 4140).into(),
+        ([127, 0, 0, 1], 666).into(),
+        Some(ep1),
+    );
+
+    let cfg = default_config(ep1);
+    let id_name = Name::from_str("foo.ns1.serviceaccount.identity.linkerd.cluster.local")
+        .expect("hostname is invalid");
+    let svc_name = profile::Name::from_str("foo.ns1.svc.example.com").unwrap();
+    let meta = support::resolver::Metadata::new(
+        Default::default(),
+        support::resolver::ProtocolHint::Http2,
+        Some(id_name.clone()),
+        10_000,
+        None,
+    );
+
+    // Pretend the upstream is a proxy that supports proto upgrades...
+    let mut server_settings = hyper::server::conn::Http::new();
+    server_settings.http2_only(true);
+    let connect = support::connect().endpoint_fn_boxed(ep1, hello_server(server_settings));
+
+    let profiles = profile::resolver().profile(
+        ep1,
+        profile::Profile {
+            name: Some(svc_name.clone()),
+            ..Default::default()
+        },
+    );
+
+    let resolver = support::resolver::<Addr, support::resolver::Metadata>();
+    let mut dst = resolver.endpoint_tx((svc_name, ep1.port()));
+    dst.add(Some((ep1, meta.clone())))
+        .expect("still listening to resolution");
+
+    // Build the outbound server
+    let (mut s, _shutdown) = build_server(cfg, profiles, resolver, connect);
+    let server = s.new_service(addrs);
+    let (mut client, bg) = connect_client(&mut ClientBuilder::new(), server).await;
+
+    let rsp = http_request(&mut client, Request::default()).await;
+    assert_eq!(rsp.status(), http::StatusCode::OK);
+    let mut body = hyper::body::aggregate(rsp.into_body())
+        .await
+        .expect("body shouldn't error");
+    let mut buf = vec![0u8; body.remaining()];
+    body.copy_to_slice(&mut buf[..]);
+    assert_eq!(std::str::from_utf8(&buf[..]), Ok("Hello world!"));
+
+    drop(client);
+    bg.await.unwrap();
+}
+
+async fn connect_client<S>(
+    client_settings: &mut ClientBuilder,
+    server: S,
+) -> (
+    hyper::client::conn::SendRequest<Body>,
+    tokio::task::JoinHandle<()>,
+)
+where
+    S: svc::Service<support::io::DuplexStream> + Send + Sync + 'static,
+    S::Error: Into<Error>,
+    S::Response: std::fmt::Debug + Send + Sync + 'static,
+    S::Future: Send,
+{
+    tracing::info!(settings = ?client_settings, "connecting client with");
+    let (client_io, server_io) = support::io::duplex(4096);
+    let proxy = server
+        .oneshot(server_io)
+        .map(|res| {
+            let res = res.map_err(Into::into);
+            tracing::info!(?res, "Server complete");
+            res.expect("proxy failed");
+        })
+        .instrument(tracing::info_span!("proxy"));
+    let (client, conn) = client_settings
+        .handshake(client_io)
+        .await
+        .expect("Client must connect");
+    let client_bg = conn
+        .map(|res| {
+            tracing::info!(?res, "Client background complete");
+            res.expect("client bg task failed");
+        })
+        .instrument(tracing::info_span!("client_bg"));
+    let bg = tokio::spawn(async move {
+        tokio::join! {
+            proxy,
+            client_bg,
+        };
+    });
+    (client, bg)
+}
+
+#[tracing::instrument(skip(client))]
+async fn http_request(client: &mut SendRequest<Body>, request: Request<Body>) -> Response<Body> {
+    let rsp = client
+        .ready_and()
+        .await
+        .expect("Client must not fail")
+        .call(request)
+        .await
+        .expect("Request must succeed");
+
+    tracing::info!(?rsp);
+
+    rsp
+}
+
+async fn unmeshed_hello_world(
+    server_settings: hyper::server::conn::Http,
+    mut client_settings: ClientBuilder,
+) {
+    let _trace = support::trace_init();
+
+    let ep1 = SocketAddr::new([10, 0, 0, 41].into(), 5550);
+    let addrs = listen::Addrs::new(
+        ([127, 0, 0, 1], 4140).into(),
+        ([127, 0, 0, 1], 666).into(),
+        Some(ep1),
+    );
+
+    let cfg = default_config(ep1);
+    // Build a mock "connector" that returns the upstream "server" IO.
+    let connect = support::connect().endpoint_fn_boxed(ep1, hello_server(server_settings));
+
+    let profiles = profile::resolver();
+    let profile_tx = profiles.profile_tx(ep1);
+    profile_tx.send(profile::Profile::default()).unwrap();
+
+    let resolver = support::resolver::<Addr, support::resolver::Metadata>();
+
+    // Build the outbound server
+    let (mut s, _shutdown) = build_server(cfg, profiles, resolver, connect);
+    let server = s.new_service(addrs);
+    let (mut client, bg) = connect_client(&mut client_settings, server).await;
+
+    let rsp = http_request(&mut client, Request::default()).await;
+    assert_eq!(rsp.status(), http::StatusCode::OK);
+    let mut body = hyper::body::aggregate(rsp.into_body())
+        .await
+        .expect("body shouldn't error");
+    let mut buf = vec![0u8; body.remaining()];
+    body.copy_to_slice(&mut buf[..]);
+    assert_eq!(std::str::from_utf8(&buf[..]), Ok("Hello world!"));
+
+    drop(client);
+    bg.await.unwrap();
+}
+
+#[tracing::instrument]
+fn hello_server(http: hyper::server::conn::Http) -> impl Fn(Endpoint) -> Result<BoxedIo, Error> {
+    move |endpoint| {
+        let span = tracing::info_span!("hello_server", ?endpoint);
+        let _e = span.enter();
+        tracing::info!("mock connecting");
+        let (client_io, server_io) = support::io::duplex(4096);
+        let hello_svc = hyper::service::service_fn(|request: Request<Body>| async move {
+            tracing::info!(?request);
+            Ok::<_, Error>(Response::new(Body::from("Hello world!")))
+        });
+        tokio::spawn(
+            http.serve_connection(server_io, hello_svc)
+                .in_current_span(),
+        );
+        Ok(BoxedIo::new(client_io))
+    }
+}

linkerd/app/outbound/src/server.rs

@@ -53,6 +53,59 @@ where
     P: profiles::GetProfile<SocketAddr> + Unpin + Clone + Send + Sync + 'static,
     P::Future: Unpin + Send,
     P::Error: Send,
+{
+    let tcp_balance = tcp::balance::stack(&config.proxy, tcp_connect.clone(), resolve);
+    stack_with_tcp_balancer(
+        config,
+        profiles,
+        tcp_connect,
+        tcp_balance,
+        http_router,
+        metrics,
+        span_sink,
+        drain,
+    )
+}
+
+pub fn stack_with_tcp_balancer<P, C, T, H, S, I>(
+    config: &Config,
+    profiles: P,
+    tcp_connect: C,
+    tcp_balance: T,
+    http_router: H,
+    metrics: metrics::Proxy,
+    span_sink: Option<mpsc::Sender<oc::Span>>,
+    drain: drain::Watch,
+) -> impl svc::NewService<
+    listen::Addrs,
+    Service = impl tower::Service<
+        I,
+        Response = (),
+        Error = impl Into<Error>,
+        Future = impl Send + 'static,
+    > + Send
+                  + 'static,
+> + Send
+       + 'static
+where
+    I: io::AsyncRead + io::AsyncWrite + io::PeerAddr + std::fmt::Debug + Unpin + Send + 'static,
+    C: tower::Service<tcp::Endpoint, Error = Error> + Unpin + Clone + Send + Sync + 'static,
+    C::Response: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
+    C::Future: Unpin + Send,
+    T: svc::NewService<tcp::Concrete> + Clone + Unpin + Send + 'static,
+    T::Service: tower::Service<transport::io::PrefixedIo<transport::metrics::SensorIo<I>>, Response = (), Error = Error> + Unpin + Send + 'static,
+    <T::Service as tower::Service<transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>>::Future: Unpin + Send + 'static,
+    H: svc::NewService<http::Logical, Service = S> + Unpin + Clone + Send + Sync + 'static,
+    S: tower::Service<
+            http::Request<http::boxed::BoxBody>,
+            Response = http::Response<http::boxed::BoxBody>,
+            Error = Error,
+        > + Send
+        + 'static,
+    S::Future: Send,
+    P: profiles::GetProfile<SocketAddr> + Unpin + Clone + Send + Sync + 'static,
+    P::Future: Unpin + Send,
+    P::Error: Send,
 {
     let ProxyConfig {
         server: ServerConfig { h2_settings, .. },
@@ -106,11 +159,7 @@ where
         .into_inner();
 
     // Load balances TCP streams that cannot be decoded as HTTP.
-    let tcp_balance = svc::stack(tcp::balance::stack(
-        &config.proxy,
-        tcp_connect.clone(),
-        resolve,
-    ))
+    let tcp_balance = svc::stack(tcp_balance)
     .push_map_target(tcp::Concrete::from)
     .push(profiles::split::layer())
     .check_new_service::<tcp::Logical, transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()

linkerd/app/outbound/src/tcp/tests.rs

@@ -804,7 +804,7 @@ impl Into<Box<dyn FnMut(Endpoint) -> ConnectFuture + Send + 'static>> for Connec
                 .read(b"world")
                 .read_error(std::io::ErrorKind::ConnectionReset.into())
                 .build();
-            Box::pin(async move { Ok(io) })
+            Box::pin(async move { Ok(io::BoxedIo::new(io)) })
         })
     }
 }