introduce tests for isolated services (#655)

This branch introduces a new style of tests to the proxy codebase, and
an initial implementation of new test support code for writing tests in
this style. Rather than running an entire proxy alongside a simulated
control plane, and binding actual network socket, as the present
integration tests do, the new tests run individual proxy components in
isolation, using simpler mock implementations of components like name
resolution and IO. 

This approach has a few advantages. It should reduce flakiness
significantly, since we don't perform any IO and don't need to
synchronize events between multiple threads running test support
servers/controllers (the new mocks all just synchronously provide values
immediately). We can run everything in a single thread using Tokio's
basic scheduler. Additionally, since we are testing individual
components in isolation, these tests can live within the crate for the
part of the proxy being tested. This means we have access to more
internal state to make assertions on, rather than having to make
assertions on side effects like metrics.

I've added a new `linkerd2-app-test` crate that adds some initial mocks
for this kind of testing, including a simple mock resolver and a mock
connector using `tokio-test`'s mock IO. I've also added a simple "hello
world" test for the outbound TCP stack written in this style. The
intention is to use the new test support code for testing the recent
changes adding TCP mTLS and load-balancing.

In order to make this code more testable, it was necessary to move a few
things around. In particular, the outbound TCP balancer stack is now
constructed in a separate method, rather than in `Config::build_server`,
so that it can be tested without requiring all the dependencies of the
full HTTP server stack. A few other similar changes were also necessary.

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

Author: hawkw

Cargo.lock

@@ -906,6 +906,7 @@ dependencies = [
  "hyper",
  "linkerd2-app",
  "linkerd2-app-core",
+ "linkerd2-app-test",
  "linkerd2-metrics",
  "linkerd2-proxy-api",
  "quickcheck",
@@ -932,6 +933,7 @@ dependencies = [
  "http 0.2.1",
  "indexmap",
  "linkerd2-app-core",
+ "linkerd2-app-test",
  "linkerd2-identity",
  "linkerd2-retry",
  "pin-project",
@@ -949,6 +951,28 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "linkerd2-app-test"
+version = "0.1.0"
+dependencies = [
+ "futures 0.3.5",
+ "h2 0.2.6",
+ "http 0.2.1",
+ "http-body",
+ "hyper",
+ "linkerd2-app-core",
+ "linkerd2-proxy-api-resolve",
+ "linkerd2-proxy-core",
+ "linkerd2-proxy-transport",
+ "regex 0.1.80",
+ "tokio",
+ "tokio-test",
+ "tower",
+ "tracing",
+ "tracing-futures",
+ "tracing-subscriber",
+]
+
 [[package]]
 name = "linkerd2-buffer"
 version = "0.1.0"
@@ -1663,12 +1687,24 @@ dependencies = [
  "kernel32-sys",
  "libc",
  "log",
- "miow",
+ "miow 0.2.1",
  "net2",
  "slab",
  "winapi 0.2.8",
 ]
 
+[[package]]
+name = "mio-named-pipes"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0840c1c50fd55e521b247f949c241c9997709f23bd7f023b9762cd561e935656"
+dependencies = [
+ "log",
+ "mio",
+ "miow 0.3.5",
+ "winapi 0.3.8",
+]
+
 [[package]]
 name = "mio-uds"
 version = "0.6.7"
@@ -1692,6 +1728,16 @@ dependencies = [
  "ws2_32-sys",
 ]
 
+[[package]]
+name = "miow"
+version = "0.3.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "07b88fb9795d4d36d62a012dfbf49a8f5cf12751f36d31a9dbe66d528e58979e"
+dependencies = [
+ "socket2",
+ "winapi 0.3.8",
+]
+
 [[package]]
 name = "multimap"
 version = "0.8.1"
@@ -2508,6 +2554,7 @@ dependencies = [
  "libc",
  "memchr 2.3.3",
  "mio",
+ "mio-named-pipes",
  "mio-uds",
  "num_cpus",
  "parking_lot",

Cargo.toml

@@ -9,6 +9,7 @@ members = [
     "linkerd/app/integration",
     "linkerd/app/outbound",
     "linkerd/app/profiling",
+    "linkerd/app/test",
     "linkerd/app",
     "linkerd/cache",
     "linkerd/buffer",

linkerd/app/core/src/lib.rs

@@ -18,7 +18,6 @@ pub use linkerd2_drain as drain;
 pub use linkerd2_error::{Error, Never, Recover};
 pub use linkerd2_exp_backoff as exp_backoff;
 pub use linkerd2_http_metrics as http_metrics;
-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;
@@ -37,6 +36,7 @@ pub mod dst;
 pub mod errors;
 pub mod handle_time;
 pub mod metric_labels;
+pub mod metrics;
 pub mod proxy;
 pub mod retry;
 pub mod serve;

linkerd/app/src/metrics.rs renamed to linkerd/app/core/src/metrics.rs

@@ -1,10 +1,10 @@
-pub use linkerd2_app_core::{
+pub use crate::{
     classify::Class,
     errors, handle_time, http_metrics as metrics,
     metric_labels::{ControlLabels, EndpointLabels, RouteLabels},
-    metrics::FmtMetrics,
     opencensus, proxy, stack_metrics, telemetry, transport, ControlHttpMetrics, ProxyMetrics,
 };
+pub use linkerd2_metrics::*;
 use std::time::{Duration, SystemTime};
 
 pub struct Metrics {

linkerd/app/integration/Cargo.toml

@@ -28,6 +28,7 @@ linkerd2-app = { path = "..", features = ["mock-orig-dst"] }
 linkerd2-app-core = { path = "../core", features = ["mock-orig-dst"] }
 linkerd2-metrics = { path = "../../metrics", features = ["test_util"] }
 linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", tag = "v0.1.13", features = ["arbitrary"] }
+linkerd2-app-test = { path = "../test" }
 regex = "0.1"
 socket2 = "0.3.12"
 quickcheck = { version = "0.9", default-features = false }

linkerd/app/integration/src/lib.rs

@@ -12,8 +12,11 @@ pub use futures::{future, FutureExt, TryFuture, TryFutureExt};
 
 pub use http::{HeaderMap, Request, Response, StatusCode};
 pub use http_body::Body as HttpBody;
-pub use linkerd2_app as app;
-pub use linkerd2_app_core::drain;
+pub use linkerd2_app::{
+    self as app,
+    core::{drain, Addr},
+};
+pub use linkerd2_app_test::*;
 use socket2::Socket;
 pub use std::collections::HashMap;
 use std::fmt;
@@ -39,31 +42,6 @@ pub const DEFAULT_TEST_PATIENCE: Duration = Duration::from_millis(15);
 
 pub type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
 
-/// By default, disable logging in modules that are expected to error in tests.
-const DEFAULT_LOG: &'static str = "error,\
-                                   linkerd2_proxy_http=off,\
-                                   linkerd2_proxy_transport=off";
-
-pub fn trace_subscriber() -> (Dispatch, app::core::trace::Handle) {
-    use std::env;
-    let log_level = env::var("LINKERD2_PROXY_LOG")
-        .or_else(|_| env::var("RUST_LOG"))
-        .unwrap_or_else(|_| DEFAULT_LOG.to_owned());
-    env::set_var("RUST_LOG", &log_level);
-    env::set_var("LINKERD2_PROXY_LOG", &log_level);
-    let log_format = env::var("LINKERD2_PROXY_LOG_FORMAT").unwrap_or_else(|_| "PLAIN".to_string());
-    env::set_var("LINKERD2_PROXY_LOG_FORMAT", &log_format);
-    // This may fail, since the global log compat layer may have been
-    // initialized by another test.
-    let _ = app::core::trace::init_log_compat();
-    app::core::trace::with_filter_and_format(&log_level, &log_format)
-}
-
-pub fn trace_init() -> tracing::dispatcher::DefaultGuard {
-    let (d, _) = trace_subscriber();
-    tracing::dispatcher::set_default(&d)
-}
-
 /// Retry an assertion up to a specified number of times, waiting
 /// `RUST_TEST_PATIENCE_MS` between retries.
 ///

linkerd/app/outbound/Cargo.toml

@@ -31,3 +31,5 @@ features = [
 
 [dev-dependencies]
 quickcheck = { version = "0.9", default-features = false }
+linkerd2-app-test = { path = "../test" }
+tokio = { version = "0.2", features = ["full", "macros"]}

linkerd/app/outbound/src/lib.rs

@@ -35,6 +35,8 @@ use tracing::{info, info_span};
 pub mod endpoint;
 mod prevent_loop;
 mod require_identity_on_endpoint;
+#[cfg(test)]
+mod tests;
 
 use self::prevent_loop::PreventLoop;
 use self::require_identity_on_endpoint::MakeRequireIdentityLayer;
@@ -357,6 +359,82 @@ impl Config {
             .into_inner()
     }
 
+    /// Constructs a TCP load balancer.
+    pub fn build_tcp_balance<C, E, I>(
+        &self,
+        tcp_connect: &C,
+        resolve: E,
+        prevent_loop: PreventLoop,
+        metrics: &ProxyMetrics,
+    ) -> impl tower::Service<
+        SocketAddr,
+        Error = Error,
+        Future = impl Unpin + Send + 'static,
+        Response = impl tower::Service<
+            I,
+            Response = (),
+            Future = impl Unpin + Send + 'static,
+            Error = Error,
+        > + Unpin
+                       + Clone
+                       + Send
+                       + 'static,
+    > + Unpin
+           + Clone
+           + Send
+           + Sync
+           + 'static
+    where
+        C: tower::Service<TcpEndpoint, Error = Error> + Unpin + Clone + Send + Sync + 'static,
+        C::Response: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
+        C::Future: Unpin + Send,
+        E: Resolve<Addr, Endpoint = proxy::api_resolve::Metadata> + Unpin + Clone + Send + 'static,
+        E::Future: Unpin + Send,
+        E::Resolution: Unpin + Send,
+        I: tokio::io::AsyncRead + tokio::io::AsyncWrite + std::fmt::Debug + Unpin + Send + 'static,
+    {
+        let ProxyConfig {
+            dispatch_timeout,
+            cache_max_idle_age,
+            buffer_capacity,
+            ..
+        } = self.proxy;
+        svc::stack(tcp_connect.clone())
+            .push_make_thunk()
+            .instrument(|t: &TcpEndpoint| info_span!("endpoint", peer.addr = %t.addr, peer.id = ?t.identity))
+            .push(admit::AdmitLayer::new(prevent_loop))
+            .check_make_service::<TcpEndpoint, ()>()
+            .push(discover::resolve(map_endpoint::Resolve::new(
+                endpoint::FromMetadata,
+                resolve,
+            )))
+            .push(discover::buffer(1_000, cache_max_idle_age))
+            .push_map_target(Addr::from)
+            .push_on_response(tcp::balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY))
+            .push_fallback_with_predicate(
+                svc::stack(tcp_connect.clone())
+                    .push_make_thunk()
+                    .push(admit::AdmitLayer::new(prevent_loop))
+                    .push_map_target(TcpEndpoint::from)
+                    .instrument(|_: &SocketAddr| info_span!("forward")),
+                is_discovery_rejected,
+            )
+            .into_new_service()
+            .check_new_service::<SocketAddr, ()>()
+            .cache(
+                svc::layers().push_on_response(
+                    svc::layers()
+                        .push_failfast(dispatch_timeout)
+                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age)
+                        .push(metrics.stack.layer(stack_labels("tcp"))),
+                ),
+            )
+            .spawn_buffer(buffer_capacity)
+            .check_make_service::<SocketAddr, ()>()
+            .push(svc::layer::mk(tcp::Forward::new))
+            .instrument(|a: &SocketAddr| info_span!("tcp", dst = %a))
+    }
+
     pub async fn build_server<E, R, C, H, S>(
         self,
         listen_addr: std::net::SocketAddr,
@@ -398,17 +476,21 @@ impl Config {
     {
         let ProxyConfig {
             server: ServerConfig { h2_settings, .. },
-            disable_protocol_detection_for_ports: skip_detect,
+            disable_protocol_detection_for_ports: ref skip_detect,
             dispatch_timeout,
             max_in_flight_requests,
             detect_protocol_timeout,
-            cache_max_idle_age,
-            buffer_capacity,
             ..
         } = self.proxy;
         let canonicalize_timeout = self.canonicalize_timeout;
         let prevent_loop = PreventLoop::from(listen_addr.port());
 
+        // Load balances TCP streams that cannot be decoded as HTTP.
+        let tcp_balance =
+            svc::stack(self.build_tcp_balance(&tcp_connect, resolve, prevent_loop, &metrics))
+                .push_map_target(|a: listen::Addrs| a.target_addr())
+                .into_inner();
+
         let http_admit_request = svc::layers()
             // Limits the number of in-flight requests.
             .push_concurrency_limit(max_in_flight_requests)
@@ -448,43 +530,6 @@ impl Config {
             .into_inner()
             .into_make_service();
 
-        // Load balances TCP streams that cannot be decoded as HTTP.
-        let tcp_balance = svc::stack(tcp_connect.clone())
-            .push_make_thunk()
-            .instrument(|t: &TcpEndpoint| info_span!("endpoint", peer.addr = %t.addr, peer.id = ?t.identity))
-            .push(admit::AdmitLayer::new(prevent_loop))
-            .check_make_service::<TcpEndpoint, ()>()
-            .push(discover::resolve(map_endpoint::Resolve::new(
-                endpoint::FromMetadata,
-                resolve,
-            )))
-            .push(discover::buffer(1_000, cache_max_idle_age))
-            .push_map_target(Addr::from)
-            .push_on_response(tcp::balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY))
-            .push_fallback_with_predicate(
-                svc::stack(tcp_connect.clone())
-                    .push_make_thunk()
-                    .push(admit::AdmitLayer::new(prevent_loop))
-                    .push_map_target(TcpEndpoint::from)
-                    .instrument(|_: &SocketAddr| info_span!("forward")),
-                is_discovery_rejected,
-            )
-            .into_new_service()
-            .check_new_service::<SocketAddr, ()>()
-            .cache(
-                svc::layers().push_on_response(
-                    svc::layers()
-                        .push_failfast(dispatch_timeout)
-                        .push_spawn_buffer_with_idle_timeout(buffer_capacity, cache_max_idle_age)
-                        .push(metrics.stack.layer(stack_labels("tcp"))),
-                ),
-            )
-            .spawn_buffer(buffer_capacity)
-            .check_make_service::<SocketAddr, ()>()
-            .push(svc::layer::mk(tcp::Forward::new))
-            .instrument(|a: &SocketAddr| info_span!("tcp", dst = %a))
-            .push_map_target(|a: listen::Addrs| a.target_addr());
-
         let http = http::DetectHttp::new(
             h2_settings,
             detect_protocol_timeout,
@@ -499,8 +544,12 @@ impl Config {
             .push(admit::AdmitLayer::new(prevent_loop))
             .push_map_target(TcpEndpoint::from);
 
-        let accept = svc::stack(svc::stack::MakeSwitch::new(skip_detect, http, tcp_forward))
-            .push(metrics.transport.layer_accept(TransportLabels));
+        let accept = svc::stack(svc::stack::MakeSwitch::new(
+            skip_detect.clone(),
+            http,
+            tcp_forward,
+        ))
+        .push(metrics.transport.layer_accept(TransportLabels));
 
         info!(addr = %listen_addr, "Serving");
         serve::serve(listen, accept, drain.signal()).await