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.

Sign up for GitHub

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

outbound: Avoid redundant TCP endpoint resolution #742

Merged
merged 7 commits into from
Nov 17, 2020
Merged

outbound: Avoid redundant TCP endpoint resolution #742

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
46d150c
outbound: Avoid redundant TCP endpoint resolution
olix0r Nov 14, 2020
7482487
fixup tcp balancer resolution in tests
olix0r Nov 15, 2020
8c143bf
Add a FromStr for dns::Name
olix0r Nov 17, 2020
a3a30cd
Use Name::from_str in TCP tests
olix0r Nov 17, 2020
f372851
use Name::from_str where possible
olix0r Nov 17, 2020
c5a4d52
Merge branch 'main' into ver/tcp-skip-resolution
olix0r Nov 17, 2020
4976e95
Remove Addr boilerplate
olix0r Nov 17, 2020
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
15 changes: 15 additions & 0 deletions linkerd/app/outbound/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,21 @@ where
))
.push_map_target(tcp::Concrete::from)
.push(profiles::split::layer())
.check_new_service::<tcp::Logical, transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()
.push_switch(
tcp::Logical::should_resolve,
svc::stack(tcp_connect.clone())
.push_make_thunk()
.check_make_service::<tcp::Endpoint, ()>()
.push_on_response(svc::layer::mk(tcp::Forward::new))
.into_new_service()
.check_new_service::<tcp::Endpoint, transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()
.push_map_target(tcp::Endpoint::from_logical(
tls::ReasonForNoPeerName::NotProvidedByServiceDiscovery,
))
.check_new_service::<tcp::Logical, transport::io::PrefixedIo<transport::metrics::SensorIo<I>>>()
.into_inner(),
)
.push_on_response(
svc::layers()
.push_failfast(dispatch_timeout)
Expand Down
65 changes: 54 additions & 11 deletions linkerd/app/outbound/src/tcp/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ use linkerd2_app_core::{
Addr, Error, IpMatch,
};
use std::{
convert::TryFrom,
future::Future,
net::SocketAddr,
str::FromStr,
Expand Down Expand Up @@ -173,6 +174,7 @@ async fn resolutions_are_reused() {

let addr = SocketAddr::new([0, 0, 0, 0].into(), 5550);
let cfg = default_config(addr);
let svc_name = profile::Name::try_from("foo.ns1.svc.example.com".as_bytes()).unwrap();
let id_name = linkerd2_identity::Name::from_hostname(
b"foo.ns1.serviceaccount.identity.linkerd.cluster.local",
)
Expand All @@ -197,10 +199,20 @@ async fn resolutions_are_reused() {

// Configure the mock destination resolver to just give us a single endpoint
// for the target, which always exists and has no metadata.
let resolver = support::resolver().endpoint_exists(Addr::from(addr), addr, meta);
let resolver = support::resolver().endpoint_exists(
Addr::Name((svc_name.clone(), addr.port()).into()),
addr,
meta,
);
let resolve_state = resolver.handle();

let profiles = support::profiles().profile(addr, Default::default());
let profiles = support::profiles().profile(
addr,
profile::Profile {
name: Some(svc_name),
..profile::Profile::default()
},
);
let profile_state = profiles.handle();

// Build the outbound server
Expand Down Expand Up @@ -254,6 +266,7 @@ async fn load_balances() {
];

let cfg = default_config(svc_addr);
let svc_name = profile::Name::try_from("foo.ns1.svc.example.com".as_bytes()).unwrap();
let id_name = linkerd2_identity::Name::from_hostname(
b"foo.ns1.serviceaccount.identity.linkerd.cluster.local",
)
Expand All @@ -272,7 +285,13 @@ async fn load_balances() {
);
}

let profiles = support::profile::resolver().profile(svc_addr, Default::default());
let profiles = support::profile::resolver().profile(
svc_addr,
profile::Profile {
name: Some(svc_name.clone()),
..Default::default()
},
);
let profile_state = profiles.handle();

let meta = support::resolver::Metadata::new(
Expand All @@ -284,7 +303,7 @@ async fn load_balances() {
);

let resolver = support::resolver();
let mut dst = resolver.endpoint_tx(Addr::Socket(svc_addr));
let mut dst = resolver.endpoint_tx(Addr::Name((svc_name, svc_addr.port()).into()));
dst.add(endpoints.iter().map(|&(addr, _)| (addr, meta.clone())))
.expect("still listening");
let resolve_state = resolver.handle();
Expand Down Expand Up @@ -345,6 +364,7 @@ async fn load_balancer_add_endpoints() {
];

let cfg = default_config(svc_addr);
let svc_name = profile::Name::try_from("foo.ns1.svc.example.com".as_bytes()).unwrap();
let id_name = linkerd2_identity::Name::from_hostname(
b"foo.ns1.serviceaccount.identity.linkerd.cluster.local",
)
Expand All @@ -362,7 +382,13 @@ async fn load_balancer_add_endpoints() {
);
}

let profiles = support::profile::resolver().profile(svc_addr, Default::default());
let profiles = support::profile::resolver().profile(
svc_addr,
profile::Profile {
name: Some(svc_name.clone()),
..Default::default()
},
);

let meta = support::resolver::Metadata::new(
Default::default(),
Expand All @@ -373,7 +399,7 @@ async fn load_balancer_add_endpoints() {
);

let resolver = support::resolver();
let mut dst = resolver.endpoint_tx(Addr::Socket(svc_addr));
let mut dst = resolver.endpoint_tx(Addr::Name((svc_name, svc_addr.port()).into()));
dst.add(Some((endpoints[0].0, meta.clone())))
.expect("still listening");

Expand Down Expand Up @@ -454,6 +480,7 @@ async fn load_balancer_remove_endpoints() {
];

let cfg = default_config(svc_addr);
let svc_name = profile::Name::try_from("foo.ns1.svc.example.com".as_bytes()).unwrap();
let id_name = linkerd2_identity::Name::from_hostname(
b"foo.ns1.serviceaccount.identity.linkerd.cluster.local",
)
Expand All @@ -471,7 +498,13 @@ async fn load_balancer_remove_endpoints() {
);
}

let profiles = support::profile::resolver().profile(svc_addr, Default::default());
let profiles = support::profile::resolver().profile(
svc_addr,
profile::Profile {
name: Some(svc_name.clone()),
..Default::default()
},
);

let meta = support::resolver::Metadata::new(
Default::default(),
Expand All @@ -482,7 +515,7 @@ async fn load_balancer_remove_endpoints() {
);

let resolver = support::resolver();
let mut dst = resolver.endpoint_tx(Addr::Socket(svc_addr));
let mut dst = resolver.endpoint_tx(Addr::Name((svc_name, svc_addr.port()).into()));
dst.add(Some((endpoints[0].0, meta.clone())))
.expect("still listening");

Expand Down Expand Up @@ -544,6 +577,7 @@ async fn no_profiles_when_outside_search_nets() {
allow_discovery: IpMatch::new(Some(IpNet::from_str("10.0.0.0/8").unwrap())).into(),
..default_config(profile_addr)
};
let svc_name = profile::Name::try_from("foo.ns1.svc.example.com".as_bytes()).unwrap();
let id_name = linkerd2_identity::Name::from_hostname(
b"foo.ns1.serviceaccount.identity.linkerd.cluster.local",
)
Expand Down Expand Up @@ -584,11 +618,20 @@ async fn no_profiles_when_outside_search_nets() {

// Configure the mock destination resolver to just give us a single endpoint
// for the target, which always exists and has no metadata.
let resolver =
support::resolver().endpoint_exists(Addr::from(profile_addr), profile_addr, meta);
let resolver = support::resolver().endpoint_exists(
Addr::Name((svc_name.clone(), profile_addr.port()).into()),
profile_addr,
meta,
);
let resolve_state = resolver.handle();

let profiles = support::profiles().profile(profile_addr, Default::default());
let profiles = support::profiles().profile(
profile_addr,
profile::Profile {
name: Some(svc_name),
..Default::default()
},
);
let profile_state = profiles.handle();

// Build the outbound server
Expand Down
9 changes: 7 additions & 2 deletions linkerd/app/test/src/connect.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,9 +35,14 @@ where
let span = tracing::info_span!("connect", %addr);
let f = span.in_scope(|| {
tracing::trace!("connecting...");
match self.endpoints.lock().unwrap().get_mut(&addr) {
let mut endpoints = self.endpoints.lock().unwrap();
match endpoints.get_mut(&addr) {
Some(f) => (f)(endpoint),
None => panic!("did not expect to connect to the endpoint {:?}", endpoint),
None => panic!(
"did not expect to connect to the endpoint {} not in {:?}",
addr,
endpoints.keys().collect::<Vec<_>>()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

<3

),
}
});
f.instrument(span)
Expand Down
2 changes: 1 addition & 1 deletion linkerd/service-profiles/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
#![deny(warnings, rust_2018_idioms)]

use linkerd2_addr::Addr;
use linkerd2_dns_name::Name;
pub use linkerd2_dns_name::Name;
use linkerd2_error::Error;
use linkerd2_proxy_api_resolve::Metadata;
use std::{
Expand Down