refactor(proxy/tap): remove unused trait system (#4477)

`linkerd-proxy-tap` contains data plane facilities that undergird "tap", a
system for inspecting traffic inside of a service mesh.

this crate contains an `iface` submodule that defines a number of
traits intended to provide a set of generic interfaces for uses like
testing, or other transport protocols. however, this suite has not been
used for these purposes. these traits are only used internally. some types
implement these traits, only to provide empty implementations that do nothing.

this means that we pay for the complexity associated with this abstraction,
for no current benefit. this branch removes these traits, to help simplify
the `linkerd-proxy-tap` crate.

* refactor(tap): remove unused `Tap` trait

this commit concerns the `linkerd-proxy-tap::iface::Tap` trait.

this trait was, per the documentation of the `iface` submodule:

> These interfaces are provided to decouple the service implementation from any
> Protobuf or gRPC concerns, hopefully to make this module more testable and
> easier to change.
>
> This module is necessary to seal the traits, which must be public
> for Registry/Layer/grpc, but need not be implemented outside of the `tap`
> module.

traits like this *are* a useful way to permit e.g. tests to mock i/o,
decouple implementations from the particulars of certain protocols, and
so forth.

however, this `Tap` trait is only used within the context of the
`Registry` structure, and is not used for tests.

traits like this have a cost, in the form of complexity in the
implementation. a registry that is generic across a "tap" then leaks
complexity into our middleware, which must include e.g. bounds for the
tap, its request payload, and its response payload.

this commit removes this generic flexibility, and phrases calling code
in terms of this concrete structure. the `can_tap_more()` and `tap()`
methods are now inherent methods, rather than trait methods
implementations. see `linkerd/proxy/tap/src/grpc/server.rs`.

doing this simplifies bounds in `linkerd/proxy/tap/src/service.rs`.

Signed-off-by: katelyn martin <kate@buoyant.io>

* refactor(tap): remove noöp `TapRequestPayload` type

this commit removes `TapRequestPayload`.

we define a `TapPayload` trait, ostensibly to abstract over the task of
"tapping" streams going in and out of workloads. this trait is
implemented by two types, `TapRequestPayload` and `TapResponsePayload`.

note this trait implementation, however:

```rust
 impl iface::TapPayload for TapRequestPayload {
     fn data<B: Buf>(&mut self, _: &B) {}

     fn eos(self, _: Option<&http::HeaderMap>) {}

     fn fail<E: HasH2Reason>(self, _: &E) {}
 }
```

..and this struct definiction:

```rust
 #[derive(Debug)]
 pub struct TapRequestPayload {}
```

we do not tap the request payload. we don't store any state in this
type, not is it affiliated with a meaningful trait implementation.

all of `Body<B, T>`'s relevant logic is related to managing its
`TapPayload`s, which as we can see above, are inert. we do not need to
wrap the body in a `Body<B, T>` wrapper if `T` does not apply any
behavior. as in the removal of the `Tap` trait, this is introducing
complexity without any concrete motivating need.

we can remove this trait implementation, and the TapRequestPayload type.

Signed-off-by: katelyn martin <kate@buoyant.io>

* refactor(tap): remove unused `TapResponse` trait

again, we have a trait that is only implemented once, by a single type.
we define an interface here, without any consumers using it. this commit
removes the `TapResponse` trait and replaces calling code with
invocations of inherent methods.

Signed-off-by: katelyn martin <kate@buoyant.io>

* refactor(tap): remove `TapPayload` trait

in a previous commit, we removed the `TapRequestPayload` trait. this
leaves us with a single implementer of this trait. as before, this is
defining an interface with no other consumers.

not only can we remove this trait, but in doing so, we'll discover that
one of these trait methods was never being used!

by removing this unused abstraction, we not only simplify the code
relying on that interface, but the compiler can note that certain parts
of that interface aren't serving us any purpose.

Signed-off-by: katelyn martin <kate@buoyant.io>

* nit(tap): remove inert `Registry` type alias

now that we aren't defining `Registry` as a generic structure, this type
alias is inert. we can instead reëxport it.

Signed-off-by: katelyn martin <kate@buoyant.io>

* refactor(tap): `Inner` is not generic

this type does not need to be defined as a generic.

see <#4477 (comment)>.

Co-Authored-By: Co-authored-by: Alejandro Martinez Ruiz <amr@buoyant.io>
Signed-off-by: katelyn martin <kate@buoyant.io>

* refactor(tap): remove `TapRequestPayload`

see:

- #4477 (comment)
- #4477 (comment)

we removed trait implementations for this type in 337ceda. we can go
further than this, and remove it altogether.

Co-Authored-By: Co-authored-by: Alejandro Martinez Ruiz <amr@buoyant.io>
Signed-off-by: katelyn martin <kate@buoyant.io>

---------

Signed-off-by: katelyn martin <kate@buoyant.io>
Co-authored-by: Co-authored-by: Alejandro Martinez Ruiz <amr@buoyant.io>

Author: cratelyn

linkerd/proxy/tap/src/grpc.rs

@@ -1,4 +1,5 @@
 mod match_;
 mod server;
 
+pub(crate) use self::server::TapResponsePayload;
 pub use self::server::{Server, Tap};

linkerd/proxy/tap/src/grpc/server.rs

@@ -1,6 +1,5 @@
 use super::match_::Match;
-use crate::{iface, Inspect, Registry};
-use bytes::Buf;
+use crate::{Inspect, Registry};
 use futures::ready;
 use futures::stream::Stream;
 use http_body::Body;
@@ -63,9 +62,6 @@ pub struct TapResponse {
     tap: TapTx,
 }
 
-#[derive(Debug)]
-pub struct TapRequestPayload {}
-
 #[derive(Debug)]
 pub struct TapResponsePayload {
     base_event: api::TapEvent,
@@ -219,23 +215,15 @@ impl Shared {
 
 // === impl Tap ===
 
-impl iface::Tap for Tap {
-    type TapRequestPayload = TapRequestPayload;
-    type TapResponse = TapResponse;
-    type TapResponsePayload = TapResponsePayload;
-
-    fn can_tap_more(&self) -> bool {
+impl Tap {
+    pub(crate) fn can_tap_more(&self) -> bool {
         self.shared
             .upgrade()
             .map(|shared| shared.is_under_limit())
             .unwrap_or(false)
     }
 
-    fn tap<B, I>(
-        &mut self,
-        req: &http::Request<B>,
-        inspect: &I,
-    ) -> Option<(TapRequestPayload, TapResponse)>
+    pub(crate) fn tap<B, I>(&mut self, req: &http::Request<B>, inspect: &I) -> Option<TapResponse>
     where
         B: Body,
         I: Inspect,
@@ -331,23 +319,19 @@ impl iface::Tap for Tap {
 
         let tap = TapTx { id, tx: events_tx };
 
-        let req = TapRequestPayload {};
-        let rsp = TapResponse {
+        Some(TapResponse {
             tap,
             base_event,
             request_init_at,
             extract_headers,
-        };
-        Some((req, rsp))
+        })
     }
 }
 
 // === impl TapResponse ===
 
-impl iface::TapResponse for TapResponse {
-    type TapPayload = TapResponsePayload;
-
-    fn tap<B: Body>(self, rsp: &http::Response<B>) -> TapResponsePayload {
+impl TapResponse {
+    pub(crate) fn tap<B: Body>(self, rsp: &http::Response<B>) -> TapResponsePayload {
         let response_init_at = Instant::now();
 
         let headers = if self.extract_headers {
@@ -395,7 +379,7 @@ impl iface::TapResponse for TapResponse {
         }
     }
 
-    fn fail<E: HasH2Reason>(self, err: &E) {
+    pub(crate) fn fail<E: HasH2Reason>(self, err: &E) {
         let response_end_at = Instant::now();
         let reason = err.h2_reason();
         let since_request_init = response_end_at.saturating_duration_since(self.request_init_at);
@@ -420,24 +404,10 @@ impl iface::TapResponse for TapResponse {
     }
 }
 
-// === impl TapRequestPayload ===
-
-impl iface::TapPayload for TapRequestPayload {
-    fn data<B: Buf>(&mut self, _: &B) {}
-
-    fn eos(self, _: Option<&http::HeaderMap>) {}
-
-    fn fail<E: HasH2Reason>(self, _: &E) {}
-}
-
 // === impl TapResponsePayload ===
 
-impl iface::TapPayload for TapResponsePayload {
-    fn data<B: Buf>(&mut self, data: &B) {
-        self.response_bytes += data.remaining();
-    }
-
-    fn eos(self, trls: Option<&http::HeaderMap>) {
+impl TapResponsePayload {
+    pub(crate) fn eos(self, trls: Option<&http::HeaderMap>) {
         let status = match trls {
             None => self.grpc_status,
             Some(t) => t
@@ -449,15 +419,13 @@ impl iface::TapPayload for TapResponsePayload {
         self.send(status.map(api::eos::End::GrpcStatusCode), trls);
     }
 
-    fn fail<E: HasH2Reason>(self, e: &E) {
+    pub(crate) fn fail<E: HasH2Reason>(self, e: &E) {
         let end = e
             .h2_reason()
             .map(|r| api::eos::End::ResetErrorCode(r.into()));
         self.send(end, None);
     }
-}
 
-impl TapResponsePayload {
     fn send(self, end: Option<api::eos::End>, trls: Option<&http::HeaderMap>) {
         let response_end_at = Instant::now();
         let trailers = if self.extract_headers {

linkerd/proxy/tap/src/lib.rs

@@ -9,11 +9,7 @@ mod grpc;
 mod registry;
 mod service;
 
-pub use self::{accept::AcceptPermittedClients, service::NewTapHttp};
-
-/// A registry containing all the active taps that have registered with the
-/// gRPC server.
-pub type Registry = registry::Registry<grpc::Tap>;
+pub use self::{accept::AcceptPermittedClients, registry::Registry, service::NewTapHttp};
 
 // The number of events that may be buffered for a given response.
 const PER_RESPONSE_EVENT_BUFFER_CAPACITY: usize = 400;
@@ -61,54 +57,3 @@ pub trait Inspect {
             })
     }
 }
-
-/// The internal interface used between Registry, Layer, and grpc.
-///
-/// These interfaces are provided to decouple the service implementation from any
-/// Protobuf or gRPC concerns, hopefully to make this module more testable and
-/// easier to change.
-///
-/// This module is necessary to seal the traits, which must be public
-/// for Registry/Layer/grpc, but need not be implemented outside of the `tap`
-/// module.
-mod iface {
-    use bytes::Buf;
-    use linkerd_proxy_http::HasH2Reason;
-
-    pub trait Tap: Clone {
-        type TapRequestPayload: TapPayload;
-        type TapResponse: TapResponse<TapPayload = Self::TapResponsePayload>;
-        type TapResponsePayload: TapPayload;
-
-        /// Returns `true` as l
-        fn can_tap_more(&self) -> bool;
-
-        /// Initiate a tap, if it matches.
-        ///
-        /// If the tap cannot be initialized, for instance because the tap has
-        /// completed or been canceled, then `None` is returned.
-        fn tap<B: http_body::Body, I: super::Inspect>(
-            &mut self,
-            req: &http::Request<B>,
-            inspect: &I,
-        ) -> Option<(Self::TapRequestPayload, Self::TapResponse)>;
-    }
-
-    pub trait TapPayload {
-        fn data<B: Buf>(&mut self, data: &B);
-
-        fn eos(self, headers: Option<&http::HeaderMap>);
-
-        fn fail<E: HasH2Reason>(self, error: &E);
-    }
-
-    pub trait TapResponse {
-        type TapPayload: TapPayload;
-
-        /// Record a response and obtain a handle to tap its body.
-        fn tap<B: http_body::Body>(self, rsp: &http::Response<B>) -> Self::TapPayload;
-
-        /// Record a service failure.
-        fn fail<E: HasH2Reason>(self, error: &E);
-    }
-}

linkerd/proxy/tap/src/registry.rs

@@ -1,23 +1,25 @@
-use crate::iface;
+use crate::grpc::Tap;
 use futures::{Stream, StreamExt};
 use parking_lot::Mutex;
 use std::sync::Arc;
 use tokio::sync::watch;
 use tracing::trace;
 
+/// A registry containing all the active taps that have registered with the
+/// gRPC server.
 #[derive(Debug)]
-pub struct Registry<T> {
-    inner: Arc<Mutex<Inner<T>>>,
-    taps_recv: watch::Receiver<Vec<T>>,
+pub struct Registry {
+    inner: Arc<Mutex<Inner>>,
+    taps_recv: watch::Receiver<Vec<Tap>>,
 }
 
 #[derive(Debug)]
-struct Inner<T> {
-    taps: Vec<T>,
-    taps_send: watch::Sender<Vec<T>>,
+struct Inner {
+    taps: Vec<Tap>,
+    taps_send: watch::Sender<Vec<Tap>>,
 }
 
-impl<T> Default for Registry<T> {
+impl Default for Registry {
     fn default() -> Self {
         let (taps_send, taps_recv) = watch::channel(vec![]);
         let inner = Inner {
@@ -31,19 +33,16 @@ impl<T> Default for Registry<T> {
     }
 }
 
-impl<T> Registry<T>
-where
-    T: iface::Tap + Clone,
-{
+impl Registry {
     pub fn new() -> Self {
         Self::default()
     }
 
-    pub fn get_taps(&self) -> Vec<T> {
+    pub fn get_taps(&self) -> Vec<Tap> {
         self.taps_recv.borrow().clone()
     }
 
-    pub fn register(&self, tap: T) {
+    pub fn register(&self, tap: Tap) {
         let mut inner = self.inner.lock();
         inner.taps.push(tap);
         let _ = inner.taps_send.send(inner.taps.clone());
@@ -60,7 +59,7 @@ where
     }
 }
 
-impl<T> Clone for Registry<T> {
+impl Clone for Registry {
     fn clone(&self) -> Self {
         Self {
             inner: self.inner.clone(),