Add basic tap integration tests (#154)

Signed-off-by: Sean McArthur <sean@buoyant.io>

tests/support/client.rs

@@ -11,12 +11,15 @@ use self::tokio::{
     net::TcpStream,
     io::{AsyncRead, AsyncWrite},
 };
+use support::hyper::body::Payload;
 
 type Request = http::Request<Bytes>;
-type Response = http::Response<BodyStream>;
-type BodyStream = Box<Stream<Item=Bytes, Error=String> + Send>;
+type Response = http::Response<BytesBody>;
 type Sender = mpsc::UnboundedSender<(Request, oneshot::Sender<Result<Response, String>>)>;
 
+#[derive(Debug)]
+pub struct BytesBody(hyper::Body);
+
 pub fn new<T: Into<String>>(addr: SocketAddr, auth: T) -> Client {
     http2(addr, auth.into())
 }
@@ -84,9 +87,7 @@ impl Client {
     }
 
     pub fn request_async(&self, builder: &mut http::request::Builder) -> Box<Future<Item=Response, Error=String> + Send> {
-        let (tx, rx) = oneshot::channel();
-        let _ = self.tx.unbounded_send((builder.body(Bytes::new()).unwrap(), tx));
-        Box::new(rx.then(|oneshot_result| oneshot_result.expect("request canceled")))
+        self.send_req(builder.body(Bytes::new()).unwrap())
     }
 
     pub fn request(&self, builder: &mut http::request::Builder) -> Response {
@@ -96,14 +97,15 @@ impl Client {
     }
 
     pub fn request_body(&self, req: Request) -> Response {
-        let (tx, rx) = oneshot::channel();
-        let _ = self.tx.unbounded_send((req, tx));
-        rx
-            .then(|oneshot_result| oneshot_result.expect("request canceled"))
+        self.send_req(req)
             .wait()
             .expect("response")
     }
 
+    pub fn request_body_async(&self, req: Request) -> Box<Future<Item=Response, Error=String> + Send> {
+        self.send_req(req)
+    }
+
     pub fn request_builder(&self, path: &str) -> http::request::Builder {
         let mut b = ::http::Request::builder();
         b.uri(format!("http://{}{}", self.authority, path).as_str())
@@ -111,6 +113,16 @@ impl Client {
         b
     }
 
+    fn send_req(&self, mut req: Request) -> Box<Future<Item=Response, Error=String> + Send> {
+        if req.uri().scheme_part().is_none() {
+            let absolute = format!("http://{}{}", self.authority, req.uri().path()).parse().unwrap();
+            *req.uri_mut() = absolute;
+        }
+        let (tx, rx) = oneshot::channel();
+        let _ = self.tx.unbounded_send((req, tx));
+        Box::new(rx.then(|oneshot_result| oneshot_result.expect("request canceled")))
+    }
+
     pub fn wait_for_closed(self) {
         self.running
             .wait()
@@ -159,13 +171,7 @@ fn run(addr: SocketAddr, version: Run) -> (Sender, Running) {
             let req = req.map(hyper::Body::from);
             let fut = client.request(req).then(move |result| {
                 let result = result
-                    .map(|res| {
-                        let res = http::Response::from(res);
-                        res.map(|body| -> BodyStream {
-                            Box::new(body.map(|chunk| chunk.into())
-                                .map_err(|e| e.to_string()))
-                        })
-                    })
+                    .map(|resp| resp.map(BytesBody))
                     .map_err(|e| e.to_string());
                 let _ = cb.send(result);
                 Ok(())
@@ -263,3 +269,25 @@ impl AsyncWrite for RunningIo {
     }
 }
 
+impl BytesBody {
+    pub fn poll_data(&mut self) -> Poll<Option<Bytes>, hyper::Error> {
+        match try_ready!(self.0.poll_data()) {
+            Some(chunk) => Ok(Async::Ready(Some(chunk.into()))),
+            None => Ok(Async::Ready(None)),
+        }
+    }
+
+    pub fn poll_trailers(&mut self) -> Poll<Option<http::HeaderMap>, hyper::Error> {
+        self.0.poll_trailers()
+    }
+}
+
+impl Stream for BytesBody {
+    type Item = Bytes;
+    type Error = hyper::Error;
+
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        self.poll_data()
+    }
+}
+

tests/support/mod.rs

@@ -148,6 +148,7 @@ pub mod client;
 pub mod controller;
 pub mod proxy;
 pub mod server;
+pub mod tap;
 pub mod tcp;
 
 pub fn shutdown_signal() -> (Shutdown, ShutdownRx) {

tests/support/tap.rs

@@ -0,0 +1,217 @@
+use support::*;
+
+use bytes::BytesMut;
+use linkerd2_proxy_api::tap as pb;
+
+pub fn client(addr: SocketAddr) -> Client {
+    let api = pb::client::Tap::new(SyncSvc(client::http2(addr, "localhost")));
+    Client {
+        api,
+    }
+}
+
+pub struct Client {
+    api: pb::client::Tap<SyncSvc>,
+}
+
+impl Client {
+    pub fn observe(&mut self, req: ObserveBuilder) -> impl Stream<Item = pb::TapEvent, Error = tower_grpc::Error> {
+        let req = tower_grpc::Request::new(req.0);
+        self.api.observe(req)
+            .wait()
+            .expect("tap observe wait")
+            .into_inner()
+    }
+}
+
+pub fn observe_request() -> ObserveBuilder {
+    ObserveBuilder(pb::ObserveRequest {
+        limit: 100,
+        match_: Some(pb::observe_request::Match {
+            match_: Some(pb::observe_request::match_::Match::Http(
+                pb::observe_request::match_::Http {
+                    match_: Some(pb::observe_request::match_::http::Match::Path(
+                        pb::observe_request::match_::http::StringMatch {
+                            match_: Some(pb::observe_request::match_::http::string_match::Match::Prefix(
+                                "/".to_string()
+                            )),
+                        },
+                    )),
+                },
+            )),
+        }),
+    })
+}
+
+#[derive(Debug)]
+pub struct ObserveBuilder(pb::ObserveRequest);
+
+impl ObserveBuilder {
+    pub fn limit(mut self, limit: u32) -> Self {
+        self.0.limit = limit;
+        self
+    }
+
+    pub fn ports(mut self, min: u16, max: u16) -> Self {
+        self.0.match_ = Some(pb::observe_request::Match {
+            match_: Some(
+                pb::observe_request::match_::Match::Destination(
+                    pb::observe_request::match_::Tcp {
+                        match_: Some(pb::observe_request::match_::tcp::Match::Ports(
+                            pb::observe_request::match_::tcp::PortRange {
+                                min: min.into(),
+                                max: max.into(),
+                            },
+                        )),
+                    },
+                ),
+            ),
+        });
+        self
+    }
+}
+
+pub trait TapEventExt {
+    fn is_inbound(&self) -> bool;
+    fn is_outbound(&self) -> bool;
+    //fn id(&self) -> (u32, u64);
+    fn event(&self) -> &pb::tap_event::http::Event;
+
+    fn request_init_method(&self) -> String;
+    fn request_init_authority(&self) -> &str;
+    fn request_init_path(&self) -> &str;
+
+    fn response_init_status(&self) -> u16;
+
+    fn response_end_bytes(&self) -> u64;
+    fn response_end_eos_grpc(&self) -> u32;
+}
+
+impl TapEventExt for pb::TapEvent {
+    fn is_inbound(&self) -> bool {
+        self.proxy_direction == pb::tap_event::ProxyDirection::Inbound as i32
+    }
+
+    fn is_outbound(&self) -> bool {
+        self.proxy_direction == pb::tap_event::ProxyDirection::Outbound as i32
+    }
+
+    fn event(&self) -> &pb::tap_event::http::Event {
+        match self.event {
+            Some(
+                pb::tap_event::Event::Http(
+                    pb::tap_event::Http {
+                        event: Some(ref ev),
+                    }
+                )
+            ) => ev,
+            _ => panic!("unknown event: {:?}", self.event),
+        }
+    }
+
+    fn request_init_method(&self) -> String {
+        match self.event() {
+            pb::tap_event::http::Event::RequestInit(_ev) => {
+                //TODO: ugh
+                unimplemented!("method");
+            },
+            _ => panic!("not RequestInit event"),
+        }
+    }
+
+    fn request_init_authority(&self) -> &str {
+        match self.event() {
+            pb::tap_event::http::Event::RequestInit(ev) => {
+                &ev.authority
+            },
+            _ => panic!("not RequestInit event"),
+        }
+    }
+
+    fn request_init_path(&self) -> &str {
+        match self.event() {
+            pb::tap_event::http::Event::RequestInit(ev) => {
+                &ev.path
+            },
+            _ => panic!("not RequestInit event"),
+        }
+    }
+
+    fn response_init_status(&self) -> u16 {
+        match self.event() {
+            pb::tap_event::http::Event::ResponseInit(ev) => {
+                ev.http_status as u16
+            },
+            _ => panic!("not ResponseInit event"),
+        }
+    }
+
+    fn response_end_bytes(&self) -> u64 {
+        match self.event() {
+            pb::tap_event::http::Event::ResponseEnd(ev) => {
+                ev.response_bytes
+            },
+            _ => panic!("not ResponseEnd event"),
+        }
+    }
+
+    fn response_end_eos_grpc(&self) -> u32 {
+        match self.event() {
+            pb::tap_event::http::Event::ResponseEnd(ev) => {
+                match ev.eos {
+                    Some(pb::Eos {
+                        end: Some(pb::eos::End::GrpcStatusCode(code)),
+                    }) => code,
+                    _ => panic!("not Eos GrpcStatusCode: {:?}", ev.eos),
+                }
+            },
+            _ => panic!("not ResponseEnd event"),
+        }
+    }
+}
+
+struct SyncSvc(client::Client);
+
+impl<B> tower_service::Service<http::Request<B>> for SyncSvc
+where
+    B: tower_grpc::Body<Data = Bytes>,
+{
+    type Response = http::Response<GrpcBody>;
+    type Error = String;
+    type Future = Box<Future<Item = Self::Response, Error = Self::Error> + Send>;
+
+    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
+        unreachable!("tap SyncSvc poll_ready");
+    }
+
+    fn call(&mut self, req: http::Request<B>) -> Self::Future {
+        let req = req.map(|mut body| {
+            let mut buf = BytesMut::new();
+            while let Some(bytes) = future::poll_fn(|| body.poll_data()).wait().expect("req body") {
+                buf.extend_from_slice(&bytes);
+            }
+
+            buf.freeze()
+        });
+        Box::new(self.0.request_body_async(req)
+            .map(|res| res.map(GrpcBody)))
+    }
+}
+
+struct GrpcBody(client::BytesBody);
+
+impl tower_grpc::Body for GrpcBody {
+    type Data = Bytes;
+
+    fn poll_data(&mut self) -> Poll<Option<Self::Data>, tower_grpc::Error> {
+        self.0.poll_data().map_err(|err| {
+            unimplemented!("grpc poll_data error: {}", err)
+        })
+    }
+
+    fn poll_metadata(&mut self) -> Poll<Option<http::HeaderMap>, tower_grpc::Error> {
+        self.0.poll_trailers().map_err(|err| {
+            unimplemented!("grpc poll_trailers error: {}", err)
+        })
+    }
+}

tests/tap.rs

@@ -0,0 +1,83 @@
+#![deny(warnings)]
+#[macro_use]
+mod support;
+use self::support::*;
+use support::tap::TapEventExt;
+
+// Flaky: sometimes the admin thread hasn't had a chance to register
+// the Taps before the `client.get` is called.
+#[test]
+#[cfg_attr(not(feature = "flaky_tests"), ignore)]
+fn inbound_http1() {
+    let _ = env_logger_init();
+    let srv = server::http1()
+        .route("/", "hello")
+        .run();
+
+    let proxy = proxy::new()
+        .inbound(srv)
+        .run();
+
+    let mut tap = tap::client(proxy.control);
+    let events = tap.observe(
+        tap::observe_request()
+    );
+
+    let authority = "tap.test.svc.cluster.local";
+    let client = client::http1(proxy.inbound, authority);
+
+    assert_eq!(client.get("/"), "hello");
+
+    let mut events = events.wait().take(3);
+
+    let ev1 = events.next().expect("next1").expect("stream1");
+    assert!(ev1.is_inbound());
+    assert_eq!(ev1.request_init_authority(), authority);
+    assert_eq!(ev1.request_init_path(), "/");
+
+    let ev2 = events.next().expect("next2").expect("stream2");
+    assert!(ev2.is_inbound());
+    assert_eq!(ev2.response_init_status(), 200);
+
+    let ev3 = events.next().expect("next3").expect("stream3");
+    assert!(ev3.is_inbound());
+    assert_eq!(ev3.response_end_bytes(), 5);
+}
+
+#[test]
+#[cfg_attr(not(feature = "flaky_tests"), ignore)]
+fn grpc_headers_end() {
+    let _ = env_logger_init();
+    let srv = server::http2()
+        .route_fn("/", |_req| {
+            Response::builder()
+                .header("grpc-status", "1")
+                .body(Default::default())
+                .unwrap()
+        })
+        .run();
+
+    let proxy = proxy::new()
+        .inbound(srv)
+        .run();
+
+    let mut tap = tap::client(proxy.control);
+    let events = tap.observe(
+        tap::observe_request()
+    );
+
+    let authority = "tap.test.svc.cluster.local";
+    let client = client::http2(proxy.inbound, authority);
+
+    let res = client.request(client
+        .request_builder("/")
+        .header("content-type", "application/grpc+nope")
+    );
+    assert_eq!(res.status(), 200);
+    assert_eq!(res.headers()["grpc-status"], "1");
+    assert_eq!(res.into_body().concat2().wait().unwrap().len(), 0);
+
+    let ev = events.wait().nth(2).expect("nth").expect("stream");
+
+    assert_eq!(ev.response_end_eos_grpc(), 1);
+}