Simplify tap server (#582)

## Description

Remove the Tap daemon that is responsible for handling service registrations and
tap subscriptions.

There is now a registry that contains all active taps.

The tap service has access to this registry and it uses it to retrieve an
updated set of active taps for every request. This set is used to determine
whether the request should be tapped.

New taps are added when the gRPC server receives an observe request. If the
request is valid, the server adds the tap to the registry.

Inactive taps are cleared by a background task. This background task is much
simpler than the one before; it loops on an interval and checks whether each tap
still has an actively used response stream, and if it has not reached its limit.

### Changes

The primary change is the addition of the `Registry` in
`linkerd/proxy/tap/src/registry.rs`.

The tap service holds on to a `Registry` and uses the `taps_recv` field to get
an updated list of taps for each request. The tap service is not responsible for
adding or removing any taps from the registry.

The gRPC service also holds on to a `Registry` and uses the `inner` field to
register new taps with the registry. The field is locked giving access to an
`Inner`. With this, the gRPC server can add new taps to the registry and then
send updates with the `taps_send` sender.

Neither the tap service or the gRPC is responsible for clearing out inactive
taps. To solve this, we spawn a background task with the `async fn clean` method
which loops on an interval and checks that each tap still has an actively used
response stream.


### Traits

The `tap::iface` module now only contains the `Tap`, `TapPayload`, and
`TapResponse` traits. These are helpful for traits that are implemented in
`tap::grpc`.

The `Subscribe` and `Register` traits have been removed. Services no longer
*register* with the tap server. Taps still do subscribe (now called `register`)
with the registry, but it was only implemented by one object and did not
simplify the implementation in terms of understanding or save on duplicated
code.

Signed-off-by: Kevin Leimkuhler <[email protected]>

Author: kleimkuhler

linkerd/app/src/env.rs

@@ -547,7 +547,7 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
     let tap = tap?
         .map(|(addr, ids)| super::tap::Config::Enabled {
             permitted_peer_identities: ids,
-            server: ServerConfig {
+            config: ServerConfig {
                 bind: listen::Bind::new(addr, inbound.proxy.server.bind.keepalive()),
                 h2_settings,
             },

linkerd/app/src/lib.rs

@@ -1,5 +1,4 @@
 //! Configures and executes the proxy
-
 #![recursion_limit = "256"]
 //#![deny(warnings, rust_2018_idioms)]
 
@@ -25,6 +24,7 @@ use linkerd2_app_inbound as inbound;
 use linkerd2_app_outbound as outbound;
 use std::net::SocketAddr;
 use std::pin::Pin;
+use tokio::time::Duration;
 use tracing::{debug, error, info, info_span};
 use tracing_futures::Instrument;
 
@@ -343,8 +343,15 @@ impl App {
                         // Spawn the DNS resolver background task.
                         tokio::spawn(dns.instrument(info_span!("dns")));
 
-                        if let tap::Tap::Enabled { daemon, serve, .. } = tap {
-                            tokio::spawn(daemon.instrument(info_span!("tap")));
+                        if let tap::Tap::Enabled {
+                            registry, serve, ..
+                        } = tap
+                        {
+                            tokio::spawn(
+                                registry
+                                    .clean(tokio::time::interval(Duration::from_secs(60)))
+                                    .instrument(info_span!("tap_clean")),
+                            );
                             tokio::spawn(
                                 serve
                                     .map_err(|error| error!(%error, "server died"))

linkerd/app/src/tap.rs

@@ -14,7 +14,7 @@ use std::pin::Pin;
 pub enum Config {
     Disabled,
     Enabled {
-        server: ServerConfig,
+        config: ServerConfig,
         permitted_peer_identities: IndexSet<identity::Name>,
     },
 }
@@ -26,7 +26,7 @@ pub enum Tap {
     Enabled {
         listen_addr: SocketAddr,
         layer: tap::Layer,
-        daemon: tap::Daemon,
+        registry: tap::Registry,
         serve: Pin<Box<dyn std::future::Future<Output = Result<(), Error>> + Send + 'static>>,
     },
 }
@@ -37,31 +37,30 @@ impl Config {
         identity: tls::Conditional<identity::Local>,
         drain: drain::Watch,
     ) -> Result<Tap, Error> {
-        let (layer, grpc, daemon) = tap::new();
+        let (registry, layer, server) = tap::new();
         match self {
             Config::Disabled => {
-                drop((grpc, daemon));
+                drop((registry, server));
                 Ok(Tap::Disabled { layer })
             }
-
             Config::Enabled {
-                server,
+                config,
                 permitted_peer_identities,
             } => {
-                let (listen_addr, listen) = server.bind.bind()?;
+                let (listen_addr, listen) = config.bind.bind()?;
 
                 let accept = tls::AcceptTls::new(
                     identity,
-                    tap::AcceptPermittedClients::new(permitted_peer_identities.into(), grpc),
+                    tap::AcceptPermittedClients::new(permitted_peer_identities.into(), server),
                 );
 
                 let serve = Box::pin(serve::serve(listen, accept, drain.signal()));
 
                 Ok(Tap::Enabled {
+                    listen_addr,
                     layer,
-                    daemon,
+                    registry,
                     serve,
-                    listen_addr,
                 })
             }
         }

linkerd/proxy/tap/src/accept.rs

@@ -1,3 +1,4 @@
+use crate::grpc::Server;
 use futures::future;
 use indexmap::IndexSet;
 use linkerd2_error::Error;
@@ -17,13 +18,13 @@ use tower::Service;
 #[derive(Clone, Debug)]
 pub struct AcceptPermittedClients {
     permitted_client_ids: Arc<IndexSet<identity::Name>>,
-    server: super::Server,
+    server: Server,
 }
 
 pub type ServeFuture = Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'static>>;
 
 impl AcceptPermittedClients {
-    pub fn new(permitted_client_ids: Arc<IndexSet<identity::Name>>, server: super::Server) -> Self {
+    pub fn new(permitted_client_ids: Arc<IndexSet<identity::Name>>, server: Server) -> Self {
         Self {
             permitted_client_ids,
             server,

linkerd/proxy/tap/src/daemon.rs

@@ -1,5 +1,5 @@
 use super::match_::Match;
-use crate::{iface, Inspect};
+use crate::{iface, Inspect, Registry};
 use bytes::Buf;
 use futures::ready;
 use hyper::body::HttpBody;
@@ -20,9 +20,9 @@ use tonic::{self as grpc, Response};
 use tracing::{debug, trace, warn};
 
 #[derive(Clone, Debug)]
-pub struct Server<T> {
-    subscribe: T,
+pub struct Server {
     base_id: Arc<AtomicUsize>,
+    registry: Registry,
 }
 
 #[pin_project]
@@ -94,10 +94,10 @@ enum ExtractKind {
 
 // === impl Server ===
 
-impl<T: iface::Subscribe<Tap>> Server<T> {
-    pub(in crate) fn new(subscribe: T) -> Self {
+impl Server {
+    pub(in crate) fn new(registry: Registry) -> Self {
         let base_id = Arc::new(0.into());
-        Self { base_id, subscribe }
+        Self { base_id, registry }
     }
 
     fn invalid_arg(message: String) -> grpc::Status {
@@ -106,11 +106,7 @@ impl<T: iface::Subscribe<Tap>> Server<T> {
 }
 
 #[tonic::async_trait]
-impl<T> api::tap_server::Tap for Server<T>
-where
-    T: iface::Subscribe<Tap> + Clone + Send + Sync + 'static,
-    T::Future: Send,
-{
+impl api::tap_server::Tap for Server {
     type ObserveStream = ResponseStream;
 
     #[tracing::instrument(skip(self), level = "debug")]
@@ -178,10 +174,8 @@ where
             shared: Arc::downgrade(&shared),
         };
 
-        // Ensure that tap registers successfully.
-        self.subscribe.subscribe(tap).await.map_err(|_| {
-            grpc::Status::new(grpc::Code::ResourceExhausted, "Too many active taps")
-        })?;
+        // Register the tap with the server's tap registry
+        self.registry.register(tap);
 
         let rsp = ResponseStream {
             shared: Some(shared),