dns: Run DNS resolutions on the main runtime (#637)

DNS resolutions are run on the admin runtime. This requires an
unnecessary layer of indirection around the resolver, including an MPSC.

Now that we allow the main runtime to use more than one thread, it's
preferable to do this discovery on the main runtime and we can simplify
the implementation.

Author: olix0r

linkerd/app/core/src/dns.rs

@@ -11,16 +11,15 @@ pub struct Config {
 
 pub struct Dns {
     pub resolver: Resolver,
-    pub task: Task,
 }
 
 // === impl Config ===
 
 impl Config {
     pub fn build(self) -> Dns {
-        let (resolver, task) =
+        let resolver =
             Resolver::from_system_config_with(&self).expect("system DNS config must be valid");
-        Dns { resolver, task }
+        Dns { resolver }
     }
 }
 

linkerd/app/src/lib.rs

@@ -57,7 +57,6 @@ pub struct Config {
 pub struct App {
     admin: admin::Admin,
     drain: drain::Signal,
-    dns: dns::Task,
     dst: ControlAddr,
     identity: identity::Identity,
     inbound_addr: SocketAddr,
@@ -223,7 +222,6 @@ impl Config {
             admin,
             dst: dst_addr,
             drain: drain_tx,
-            dns: dns.task,
             identity,
             inbound_addr,
             oc_collector,
@@ -283,7 +281,6 @@ impl App {
         let App {
             admin,
             drain,
-            dns,
             identity,
             oc_collector,
             start_proxy,
@@ -340,9 +337,6 @@ impl App {
                             admin.latch.release()
                         }
 
-                        // Spawn the DNS resolver background task.
-                        tokio::spawn(dns.instrument(info_span!("dns")));
-
                         if let tap::Tap::Enabled {
                             registry, serve, ..
                         } = tap

linkerd/dns/src/lib.rs

@@ -7,17 +7,19 @@ pub use linkerd2_dns_name::{InvalidName, Name, Suffix};
 use std::future::Future;
 use std::pin::Pin;
 use std::{fmt, net};
-use tokio::sync::{mpsc, oneshot};
-use tracing::{info_span, trace, Span};
+use tracing::{info_span, trace};
 use tracing_futures::Instrument;
-pub use trust_dns_resolver::config::ResolverOpts;
-pub use trust_dns_resolver::error::{ResolveError, ResolveErrorKind};
-use trust_dns_resolver::lookup_ip::LookupIp;
-use trust_dns_resolver::{config::ResolverConfig, system_conf, AsyncResolver};
+use trust_dns_resolver::{
+    config::ResolverConfig, lookup_ip::LookupIp, system_conf, AsyncResolver, TokioAsyncResolver,
+};
+pub use trust_dns_resolver::{
+    config::ResolverOpts,
+    error::{ResolveError, ResolveErrorKind},
+};
 
 #[derive(Clone)]
 pub struct Resolver {
-    tx: mpsc::UnboundedSender<ResolveRequest>,
+    dns: TokioAsyncResolver,
 }
 
 pub trait ConfigureResolver {
@@ -28,19 +30,10 @@ pub trait ConfigureResolver {
 pub enum Error {
     NoAddressesFound,
     ResolutionFailed(ResolveError),
-    TaskLost,
 }
 
-pub type Task = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
-
 pub type IpAddrFuture = Pin<Box<dyn Future<Output = Result<net::IpAddr, Error>> + Send + 'static>>;
 
-struct ResolveRequest {
-    name: Name,
-    result_tx: oneshot::Sender<Result<LookupIp, ResolveError>>,
-    span: tracing::Span,
-}
-
 impl Resolver {
     /// Construct a new `Resolver` from environment variables and system
     /// configuration.
@@ -51,61 +44,24 @@ impl Resolver {
     /// could not be parsed.
     ///
     /// TODO: This should be infallible like it is in the `domain` crate.
-    pub fn from_system_config_with<C: ConfigureResolver>(
-        c: &C,
-    ) -> Result<(Self, Task), ResolveError> {
+    pub fn from_system_config_with<C: ConfigureResolver>(c: &C) -> Result<Self, ResolveError> {
         let (config, mut opts) = system_conf::read_system_conf()?;
         c.configure_resolver(&mut opts);
         trace!("DNS config: {:?}", &config);
         trace!("DNS opts: {:?}", &opts);
-        Self::new(config, opts)
+        Ok(Self::new(config, opts))
     }
 
-    pub fn new(
-        config: ResolverConfig,
-        mut opts: ResolverOpts,
-    ) -> Result<(Self, Task), ResolveError> {
+    pub fn new(config: ResolverConfig, mut opts: ResolverOpts) -> Self {
         // Disable Trust-DNS's caching.
         opts.cache_size = 0;
-
-        // XXX(eliza): figure out an appropriate bound for the channel...
-        let (tx, mut rx) = mpsc::unbounded_channel();
-        let task = Box::pin(async move {
-            let resolver = match AsyncResolver::tokio(config, opts) {
-                Ok(resolver) => resolver,
-                Err(e) => unreachable!("constructing resolver should not fail: {}", e),
-            };
-            while let Some(ResolveRequest {
-                name,
-                result_tx,
-                span,
-            }) = rx.recv().await
-            {
-                let resolver = resolver.clone();
-                tokio::spawn(
-                    async move {
-                        let res = resolver.lookup_ip(name.as_ref()).await;
-                        if result_tx.send(res).is_err() {
-                            tracing::debug!("resolution canceled");
-                        }
-                    }
-                    .instrument(span),
-                );
-            }
-            tracing::debug!("all resolver handles dropped; terminating.");
-        });
-        Ok((Resolver { tx }, task))
+        let dns = AsyncResolver::tokio(config, opts).expect("Resolver must be valid");
+        Resolver { dns }
     }
 
-    async fn lookup_ip(&self, name: Name, span: Span) -> Result<LookupIp, Error> {
-        let (result_tx, rx) = oneshot::channel();
-        self.tx.send(ResolveRequest {
-            name,
-            result_tx,
-            span,
-        })?;
-        let ips = rx.await??;
-        Ok(ips)
+    async fn lookup_ip(&self, name: Name) -> Result<LookupIp, Error> {
+        let lookup = self.dns.lookup_ip(name.as_ref()).await?;
+        Ok(lookup)
     }
 
     pub fn resolve_one_ip(
@@ -116,7 +72,7 @@ impl Resolver {
         let resolver = self.clone();
         Box::pin(async move {
             let span = info_span!("resolve_one_ip", %name);
-            let ips = resolver.lookup_ip(name, span).await?;
+            let ips = resolver.lookup_ip(name).instrument(span).await?;
             ips.iter().next().ok_or_else(|| Error::NoAddressesFound)
         })
     }
@@ -137,18 +93,6 @@ impl fmt::Debug for Resolver {
     }
 }
 
-impl<T> From<mpsc::error::SendError<T>> for Error {
-    fn from(_: mpsc::error::SendError<T>) -> Self {
-        Self::TaskLost
-    }
-}
-
-impl From<oneshot::error::RecvError> for Error {
-    fn from(_: oneshot::error::RecvError) -> Self {
-        Self::TaskLost
-    }
-}
-
 impl From<ResolveError> for Error {
     fn from(e: ResolveError) -> Self {
         Self::ResolutionFailed(e)
@@ -160,7 +104,6 @@ impl fmt::Display for Error {
         match self {
             Self::NoAddressesFound => f.pad("no addresses found"),
             Self::ResolutionFailed(e) => fmt::Display::fmt(e, f),
-            Self::TaskLost => f.pad("background task terminated unexpectedly"),
         }
     }
 }

linkerd/dns/src/refine.rs

@@ -9,6 +9,7 @@ use std::net::IpAddr;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::time::Instant;
+use tracing_futures::Instrument;
 use trust_dns_resolver::lookup_ip::LookupIp;
 
 /// A `MakeService` that produces a `Refine` for a given name.
@@ -54,12 +55,11 @@ impl tower::Service<()> for Refine {
         loop {
             self.state = match self.state {
                 State::Init => {
-                    let resolver = self.resolver.clone();
                     let name = self.name.clone();
-                    let span = tracing::Span::current();
-                    State::Pending(Box::pin(
-                        async move { resolver.lookup_ip(name, span).await },
-                    ))
+                    let dns = self.resolver.clone();
+                    State::Pending(Box::pin(async move {
+                        dns.lookup_ip(name).in_current_span().await
+                    }))
                 }
                 State::Pending(ref mut fut) => {
                     let lookup = ready!(fut.as_mut().poll(cx))?;