Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

v2.210.3 #2522

Merged
merged 3 commits into from
Nov 16, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 24 additions & 2 deletions Cargo.lock
Original file line number Diff line number Diff line change
Expand Up @@ -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]]
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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",
]
15 changes: 12 additions & 3 deletions deny.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ ignore = []
unlicensed = "deny"
allow = [
"Apache-2.0",
"BSD-2-Clause",
"BSD-3-Clause",
"ISC",
"MIT",
Expand All @@ -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]]
Expand Down
17 changes: 15 additions & 2 deletions linkerd/app/admin/src/stack.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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)?;

Expand All @@ -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())
Expand Down Expand Up @@ -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()
Expand Down
1 change: 1 addition & 0 deletions linkerd/app/core/src/control.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
22 changes: 16 additions & 6 deletions linkerd/app/core/src/serve.rs
Original file line number Diff line number Diff line change
@@ -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};

Expand All @@ -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>,
Expand All @@ -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
Expand All @@ -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
Expand Down
1 change: 1 addition & 0 deletions linkerd/app/inbound/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"] }
Expand Down
21 changes: 8 additions & 13 deletions linkerd/app/inbound/src/http/router.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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 })
Expand All @@ -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<_>>();

Expand Down Expand Up @@ -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)
Expand All @@ -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);
Expand Down Expand Up @@ -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)
Expand Down
Loading