Introduce the tonic-watch crate (#1034)

The service-profiles client includes some complicated manual-polling
machinery for publish a `tokio::sync::watch::Receiver` from a streaming
gRPC endpoint.

This change moves this logic into a `linkerd-tonic-watch` crate that
is generic over the watch type and the underlying gRPC lookup
functionality. Furthermore, this is code is ported to use async/await
syntax to dramatically simplify state management.

This is being done in anticipation of upcoming inbound-discovery
infrastructure, where a similar watcher pattern will be used.

Author: olix0r

Cargo.lock

@@ -1314,6 +1314,7 @@ dependencies = [
  "linkerd-error",
  "linkerd-proxy-api-resolve",
  "linkerd-stack",
+ "linkerd-tonic-watch",
  "linkerd2-proxy-api",
  "pin-project",
  "prost-types",
@@ -1423,6 +1424,22 @@ dependencies = [
  "webpki",
 ]
 
+[[package]]
+name = "linkerd-tonic-watch"
+version = "0.1.0"
+dependencies = [
+ "futures",
+ "linkerd-error",
+ "linkerd-stack",
+ "linkerd-tracing",
+ "tokio",
+ "tokio-stream",
+ "tokio-test",
+ "tonic",
+ "tower-test",
+ "tracing",
+]
+
 [[package]]
 name = "linkerd-trace-context"
 version = "0.1.0"

Cargo.toml

@@ -53,6 +53,7 @@ members = [
     "linkerd/stack/tracing",
     "linkerd/system",
     "linkerd/timeout",
+    "linkerd/tonic-watch",
     "linkerd/tls",
     "linkerd/tracing",
     "linkerd2-proxy",

linkerd/app/Cargo.toml

@@ -23,8 +23,8 @@ linkerd-app-core = { path = "./core" }
 linkerd-app-gateway = { path = "./gateway" }
 linkerd-app-inbound = { path = "./inbound" }
 linkerd-app-outbound = { path = "./outbound" }
-linkerd-opencensus = { path = "../opencensus" }
 linkerd-error = { path = "../error" }
+linkerd-opencensus = { path = "../opencensus" }
 regex = "1.0.0"
 thiserror = "1.0"
 tokio = { version = "1", features = ["rt"] }

linkerd/app/core/src/svc.rs

@@ -7,8 +7,8 @@ use linkerd_error::Recover;
 use linkerd_exp_backoff::{ExponentialBackoff, ExponentialBackoffStream};
 pub use linkerd_reconnect::NewReconnect;
 pub use linkerd_stack::{
-    self as stack, layer, BoxNewService, BoxService, BoxServiceLayer, Fail, Filter, MapTargetLayer,
-    NewRouter, NewService, Param, Predicate, UnwrapOr,
+    self as stack, layer, BoxNewService, BoxService, BoxServiceLayer, Either, Fail, Filter,
+    MapErrLayer, MapTargetLayer, NewRouter, NewService, Param, Predicate, UnwrapOr,
 };
 pub use linkerd_stack_tracing::{NewInstrument, NewInstrumentLayer};
 pub use linkerd_timeout::{self as timeout, FailFast};
@@ -21,13 +21,7 @@ use tower::{
     layer::util::{Identity, Stack as Pair},
     make::MakeService,
 };
-pub use tower::{
-    layer::Layer,
-    service_fn as mk,
-    spawn_ready::SpawnReady,
-    util::{Either, MapErrLayer},
-    Service, ServiceExt,
-};
+pub use tower::{layer::Layer, service_fn as mk, spawn_ready::SpawnReady, Service, ServiceExt};
 
 #[derive(Copy, Clone, Debug)]
 pub struct AlwaysReconnect(ExponentialBackoff);

linkerd/app/inbound/fuzz/Cargo.lock

@@ -1143,6 +1143,7 @@ dependencies = [
  "linkerd-error",
  "linkerd-proxy-api-resolve",
  "linkerd-stack",
+ "linkerd-tonic-watch",
  "linkerd2-proxy-api",
  "pin-project",
  "rand",
@@ -1235,6 +1236,18 @@ dependencies = [
  "webpki",
 ]
 
+[[package]]
+name = "linkerd-tonic-watch"
+version = "0.1.0"
+dependencies = [
+ "futures",
+ "linkerd-error",
+ "linkerd-stack",
+ "tokio",
+ "tonic",
+ "tracing",
+]
+
 [[package]]
 name = "linkerd-trace-context"
 version = "0.1.0"

linkerd/app/src/dst.rs

@@ -21,7 +21,7 @@ pub struct Dst {
     pub addr: control::ControlAddr,
 
     /// Resolves profiles.
-    pub profiles: profiles::Client<control::Client<BoxBody>, BackoffUnlessInvalidArgument>,
+    pub profiles: profiles::Client<BackoffUnlessInvalidArgument, control::Client<BoxBody>>,
 
     /// Resolves endpoints.
     pub resolve:
@@ -46,7 +46,7 @@ impl Config {
 
         Ok(Dst {
             addr,
-            profiles: profiles::Client::new(svc.clone(), backoff, self.context.clone()),
+            profiles: profiles::Client::new(backoff, svc.clone(), self.context.clone()),
             resolve: recover::Resolve::new(backoff, api::Resolve::new(svc, self.context)),
         })
     }
@@ -66,3 +66,20 @@ impl Recover<Error> for BackoffUnlessInvalidArgument {
         Ok(self.0.stream())
     }
 }
+
+impl Recover<tonic::Status> for BackoffUnlessInvalidArgument {
+    type Backoff = ExponentialBackoffStream;
+
+    fn recover(&self, status: tonic::Status) -> Result<Self::Backoff, tonic::Status> {
+        // Address is not resolvable
+        if status.code() == tonic::Code::InvalidArgument
+                    // Unexpected cluster state
+                    || status.code() == tonic::Code::FailedPrecondition
+        {
+            return Err(status);
+        }
+
+        tracing::trace!(%status, "Recovering");
+        Ok(self.0.stream())
+    }
+}

linkerd/error/src/recover.rs

@@ -9,8 +9,8 @@ pub trait Recover<E: Into<Error> = Error> {
     ///
     /// If it is, a backoff stream is returned. When the backoff becomes ready,
     /// it signals that the caller should retry its operation. If the backoff is
-    /// polled agian, it is assumed that the operation failed and a new (possibly
-    /// longer) backoff is initated.
+    /// polled again, it is assumed that the operation failed and a new (possibly
+    /// longer) backoff is initiated.
     ///
     /// If the error is not recoverable, it is returned immediately.
     fn recover(&self, err: E) -> Result<Self::Backoff, E>;

linkerd/service-profiles/Cargo.toml

@@ -21,6 +21,7 @@ linkerd-error = { path = "../error" }
 linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", branch = "main", features = ["destination", "client"] }
 linkerd-proxy-api-resolve = { path = "../proxy/api-resolve" }
 linkerd-stack = { path = "../stack" }
+linkerd-tonic-watch = { path = "../tonic-watch" }
 rand = { version = "0.8", features = ["small_rng"] }
 regex = "1.0.0"
 tokio = { version = "1", features = ["macros", "rt", "sync", "time"] }