Upstream/v2.210.4

Cargo.lock

@@ -16,14 +16,15 @@ checksum = "aae1277d39aeec15cb388266ecc24b11c80469deae6067e17a1a7aa9e5c1f234"
 
 [[package]]
 name = "ahash"
-version = "0.8.3"
+version = "0.8.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2c99f64d1e06488f620f932677e24bc6e2897582980441ae90a671415bd7ec2f"
+checksum = "cd7d5a2cecb58716e47d67d5703a249964b14c7be1ec3cad3affc295b2d1c35d"
 dependencies = [
  "cfg-if",
  "getrandom",
  "once_cell",
  "version_check",
+ "zerocopy",
 ]
 
 [[package]]
@@ -1010,6 +1011,7 @@ dependencies = [
  "linkerd-app-core",
  "linkerd-app-test",
  "linkerd-http-access-log",
+ "linkerd-http-metrics",
  "linkerd-idle-cache",
  "linkerd-io",
  "linkerd-meshtls",
@@ -3341,3 +3343,23 @@ checksum = "0120db82e8a1e0b9fb3345a539c478767c0048d842860994d96113d5b667bd69"
 dependencies = [
  "winapi",
 ]
+
+[[package]]
+name = "zerocopy"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a7af71d8643341260a65f89fa60c0eeaa907f34544d8f6d9b0df72f069b5e74"
+dependencies = [
+ "zerocopy-derive",
+]
+
+[[package]]
+name = "zerocopy-derive"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9731702e2f0617ad526794ae28fbc6f6ca8849b5ba729666c2a5bc4b6ddee2cd"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.12",
+]

deny.toml

@@ -17,6 +17,7 @@ ignore = []
 unlicensed = "deny"
 allow = [
     "Apache-2.0",
+    "BSD-2-Clause",
     "BSD-3-Clause",
     "ISC",
     "MIT",
@@ -27,13 +28,21 @@ allow-osi-fsf-free = "neither"
 default = "deny"
 confidence-threshold = 0.8
 exceptions = [
-    { allow = ["Zlib"], name = "adler32", version = "*" },
-    { allow = ["ISC", "MIT", "OpenSSL"], name = "ring", version = "*" },
+    { allow = [
+        "Zlib",
+    ], name = "adler32", version = "*" },
+    { allow = [
+        "ISC",
+        "MIT",
+        "OpenSSL",
+    ], name = "ring", version = "*" },
     # The Unicode-DFS-2016 license is necessary for unicode-ident because they
     # use data from the unicode tables to generate the tables which are
     # included in the application. We do not distribute those data files so
     # this is not a problem for us. See https://github.com/dtolnay/unicode-ident/pull/9/files
-    { allow = ["Unicode-DFS-2016"], name = "unicode-ident", version = "*"},
+    { allow = [
+        "Unicode-DFS-2016",
+    ], name = "unicode-ident", version = "*" },
 ]
 
 [[licenses.clarify]]

linkerd/app/admin/src/stack.rs

@@ -7,7 +7,9 @@ use linkerd_app_core::{
     serve,
     svc::{self, ExtractParam, InsertParam, Param},
     tls, trace,
-    transport::{self, listen::Bind, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr},
+    transport::{
+        self, addrs::AddrPair, listen::Bind, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr,
+    },
     Error, Result,
 };
 use linkerd_app_inbound as inbound;
@@ -84,7 +86,9 @@ impl Config {
     where
         R: FmtMetrics + Clone + Send + Sync + Unpin + 'static,
         B: Bind<ServerConfig>,
-        B::Addrs: svc::Param<Remote<ClientAddr>> + svc::Param<Local<ServerAddr>>,
+        B::Addrs: svc::Param<Remote<ClientAddr>>,
+        B::Addrs: svc::Param<Local<ServerAddr>>,
+        B::Addrs: svc::Param<AddrPair>,
     {
         let (listen_addr, listen) = bind.bind(&self.server)?;
 
@@ -95,6 +99,7 @@ impl Config {
         let admin = crate::server::Admin::new(report, ready, shutdown, trace);
         let admin = svc::stack(move |_| admin.clone())
             .push(metrics.proxy.http_endpoint.to_layer::<classify::Response, _, Permitted>())
+            .push(classify::NewClassify::layer_default())
             .push_map_target(|(permit, http)| Permitted { permit, http })
             .push(inbound::policy::NewHttpPolicy::layer(metrics.http_authz.clone()))
             .push(Rescue::layer())
@@ -201,6 +206,14 @@ impl Param<Remote<ClientAddr>> for Http {
     }
 }
 
+impl Param<AddrPair> for Http {
+    fn param(&self) -> AddrPair {
+        let Remote(client) = self.tcp.client;
+        let Local(server) = self.tcp.addr;
+        AddrPair(client, server)
+    }
+}
+
 impl Param<tls::ConditionalServerTls> for Http {
     fn param(&self) -> tls::ConditionalServerTls {
         self.tcp.tls.clone()

linkerd/app/core/src/control.rs

@@ -124,6 +124,7 @@ impl Config {
             .lift_new()
             .push(self::balance::layer(dns, resolve_backoff))
             .push(metrics.to_layer::<classify::Response, _, _>())
+            .push(classify::NewClassify::layer_default())
             // This buffer allows a resolver client to be shared across stacks.
             // No load shed is applied here, however, so backpressure may leak
             // into the caller task.

linkerd/app/core/src/serve.rs

@@ -1,11 +1,11 @@
 use crate::{
     io, is_caused_by,
     svc::{self, Param},
-    transport::{ClientAddr, Remote},
     Result,
 };
 use futures::prelude::*;
 use linkerd_error::Error;
+use linkerd_proxy_transport::AddrPair;
 use tower::util::ServiceExt;
 use tracing::{debug, debug_span, info, instrument::Instrument, warn};
 
@@ -18,7 +18,7 @@ pub async fn serve<M, S, I, A>(
     shutdown: impl Future,
 ) where
     I: Send + 'static,
-    A: Param<Remote<ClientAddr>>,
+    A: Param<AddrPair>,
     M: svc::NewService<A, Service = S>,
     S: tower::Service<io::ScopedIo<I>, Response = ()> + Send + 'static,
     S::Error: Into<Error>,
@@ -40,8 +40,8 @@ pub async fn serve<M, S, I, A>(
                     };
 
                     // The local addr should be instrumented from the listener's context.
-                    let Remote(ClientAddr(client_addr)) = addrs.param();
-                    let span = debug_span!("accept", client.addr = %client_addr).entered();
+                    let AddrPair(client_addr, server_addr) = addrs.param();
+                    let span = debug_span!("accept", client.addr = %client_addr, server.addr = %server_addr).entered();
                     let accept = new_accept.new_service(addrs);
 
                     // Dispatch all of the work for a given connection onto a
@@ -57,10 +57,20 @@ pub async fn serve<M, S, I, A>(
                                     {
                                         Ok(()) => debug!("Connection closed"),
                                         Err(reason) if is_caused_by::<std::io::Error>(&*reason) => {
-                                            debug!(%reason, "Connection closed")
+                                            debug!(
+                                                reason,
+                                                client.addr = %client_addr,
+                                                server.addr = %server_addr,
+                                                "Connection closed"
+                                            );
                                         }
                                         Err(error) => {
-                                            info!(error, client.addr = %client_addr, "Connection closed")
+                                            info!(
+                                                error,
+                                                client.addr = %client_addr,
+                                                server.addr = %server_addr,
+                                                "Connection closed"
+                                            );
                                         }
                                     }
                                     // Hold the service until the connection is complete. This

linkerd/app/inbound/Cargo.toml

@@ -52,6 +52,7 @@ libfuzzer-sys = { version = "0.4", features = ["arbitrary-derive"] }
 [dev-dependencies]
 hyper = { version = "0.14", features = ["http1", "http2"] }
 linkerd-app-test = { path = "../test" }
+linkerd-http-metrics = { path = "../../http-metrics", features = ["test-util"] }
 linkerd-idle-cache = { path = "../../idle-cache", features = ["test-util"] }
 linkerd-io = { path = "../../io", features = ["tokio-test"] }
 linkerd-meshtls = { path = "../../meshtls", features = ["rustls"] }

linkerd/app/inbound/src/http/router.rs

@@ -148,7 +148,8 @@ impl<C> Inbound<C> {
             // Attempts to discover a service profile for each logical target (as
             // informed by the request's headers). The stack is cached until a
             // request has not been received for `cache_max_idle_age`.
-            let router = http.clone()
+            let router = http
+                .clone()
                 .check_new_service::<Logical, http::Request<http::BoxBody>>()
                 .push_map_target(|p: Profile| p.logical)
                 .push(profiles::http::NewProxyRouter::layer(
@@ -164,6 +165,7 @@ impl<C> Inbound<C> {
                                 .to_layer::<classify::Response, _, _>(),
                         )
                         .push_on_service(http::BoxResponse::layer())
+                        // Configure a per-route response classifier based on the profile.
                         .push(classify::NewClassify::layer())
                         .push_http_insert_target::<profiles::http::Route>()
                         .push_map_target(|(route, profile)| ProfileRoute { route, profile })
@@ -186,10 +188,7 @@ impl<C> Inbound<C> {
                         }
                         Ok(svc::Either::B(logical))
                     },
-                    http.clone()
-                    .push_on_service(svc::MapErr::layer(Error::from))
-                        .check_new_service::<Logical, http::Request<_>>()
-                        .into_inner(),
+                    http.clone().into_inner(),
                 )
                 .check_new_service::<(Option<profiles::Receiver>, Logical), http::Request<_>>();
 
@@ -229,8 +228,7 @@ impl<C> Inbound<C> {
                 // Skip the profile stack if it takes too long to become ready.
                 .push_when_unready(config.profile_skip_timeout, http.into_inner())
                 .push_on_service(
-                    svc::layers()
-                        .push(rt.metrics.proxy.stack.layer(stack_labels("http", "logical")))
+                    rt.metrics.proxy.stack.layer(stack_labels("http", "logical")),
                 )
                 .push(svc::NewQueue::layer_via(config.http_request_queue))
                 .push_new_idle_cached(config.discovery_idle_timeout)
@@ -239,6 +237,9 @@ impl<C> Inbound<C> {
                         .push(http::Retain::layer())
                         .push(http::BoxResponse::layer()),
                 )
+                // Configure default response classification early. It may be
+                // overridden by profile routes above.
+                .push(classify::NewClassify::layer_default())
                 .check_new_service::<Logical, http::Request<http::BoxBody>>()
                 .instrument(|t: &Logical| {
                     let name = t.logical.as_ref().map(tracing::field::display);
@@ -414,12 +415,6 @@ impl Param<metrics::EndpointLabels> for Logical {
     }
 }
 
-impl Param<classify::Request> for Logical {
-    fn param(&self) -> classify::Request {
-        classify::Request::default()
-    }
-}
-
 impl tap::Inspect for Logical {
     fn src_addr<B>(&self, req: &http::Request<B>) -> Option<SocketAddr> {
         req.extensions().get::<Remote<ClientAddr>>().map(|a| **a)