outbound: Require ClientPolicy discovery (#2265)

To support Gateway API-style routes in the outbound proxy, we need to begin
discovering this route configuration from the control plane (via the new
`OutboundPolicies` API).

This change updates the proxy as follows:

1. Policy controller configuration is now required for the proxy.
   Previously, the policy API was optionally configured for the inbound
   proxy.
2. The sidecar and ingress proxies are updated to use client policies.
   Service profile configurations continue to be used when they include
   HTTP routes and/or traffic split. Otherwise, a client policy is used
   to route traffic.

Outbound policies are currently discovered for *all* outbound IP addresses. Over
time, the policy controller will assume responsibility to make *all* routing
decisions.  It does not yet serve responses for all cases, however, so some
fallback behavior exists to use endpoint metadata from profile discovery,
if it exists.

The multi-cluster gateway configuration does not yet use policies for
outbound routing. Furthermore, the proxy reports an IP logical address for
policy routes (instead of a named address, as is done with profiles). There
are no new metrics or labels introduced in this PR. Metrics changes will be made
in follow-up changes.

Author: hawkw

Cargo.lock

@@ -982,6 +982,7 @@ dependencies = [
  "linkerd-io",
  "linkerd-meshtls",
  "linkerd-meshtls-rustls",
+ "linkerd-proxy-client-policy",
  "linkerd-proxy-server-policy",
  "linkerd-tonic-watch",
  "linkerd-tracing",
@@ -1051,12 +1052,16 @@ dependencies = [
  "linkerd-meshtls-rustls",
  "linkerd-proxy-client-policy",
  "linkerd-retry",
+ "linkerd-tonic-watch",
  "linkerd-tracing",
+ "linkerd2-proxy-api",
+ "once_cell",
  "parking_lot",
  "pin-project",
  "thiserror",
  "tokio",
  "tokio-test",
+ "tonic",
  "tower",
  "tower-test",
  "tracing",

linkerd/app/gateway/src/lib.rs

@@ -54,7 +54,6 @@ impl Gateway {
         T: svc::Param<tls::ClientId>,
         T: svc::Param<inbound::policy::AllowPolicy>,
         T: svc::Param<Option<SessionProtocol>>,
-        T: svc::Param<profiles::LookupAddr>,
         T: Clone + Send + Sync + Unpin + 'static,
         // Server-side socket
         I: io::AsyncRead + io::AsyncWrite + io::PeerAddr,

linkerd/app/gateway/src/server.rs

@@ -38,7 +38,6 @@ impl Gateway {
         T: svc::Param<tls::ClientId>,
         T: svc::Param<inbound::policy::AllowPolicy>,
         T: svc::Param<Option<SessionProtocol>>,
-        T: svc::Param<profiles::LookupAddr>,
         T: Clone + Send + Sync + Unpin + 'static,
         // Server-side socket
         I: io::AsyncRead + io::AsyncWrite + io::PeerAddr,
@@ -80,10 +79,12 @@ impl Gateway {
 
             // Apply the gateway's allowlist to the profile discovery service.
             let allowlist = self.config.allow_discovery.clone().into();
-            let profiles = profiles::WithAllowlist::new(profiles, allowlist);
+            let mut profiles = profiles::WithAllowlist::new(profiles, allowlist);
             self.outbound
                 .with_stack(protocol)
-                .push_discover(profiles.into_service())
+                .push_discover(svc::mk(move |GatewayAddr(addr)| {
+                    profiles.get_profile(profiles::LookupAddr(addr.into()))
+                }))
                 .into_stack()
         };
 

linkerd/app/inbound/Cargo.toml

@@ -27,6 +27,7 @@ linkerd-http-access-log = { path = "../../http-access-log" }
 linkerd-idle-cache = { path = "../../idle-cache" }
 linkerd-meshtls = { path = "../../meshtls", optional = true }
 linkerd-meshtls-rustls = { path = "../../meshtls/rustls", optional = true }
+linkerd-proxy-client-policy = { path = "../../proxy/client-policy" }
 linkerd-tonic-watch = { path = "../../tonic-watch" }
 linkerd2-proxy-api = { version = "0.8", features = ["inbound"] }
 once_cell = "1"

linkerd/app/inbound/src/lib.rs

@@ -70,6 +70,7 @@ struct Runtime {
     drain: drain::Watch,
 }
 
+/// Indicates the name to be used to route gateway connections.
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 pub struct GatewayAddr(pub NameAddr);
 

linkerd/app/integration/src/client.rs

@@ -119,6 +119,7 @@ impl Client {
         }
     }
 
+    #[track_caller]
     pub async fn get(&self, path: &str) -> String {
         let req = self.request_builder(path);
         let res = self.request(req.method("GET")).await.expect("response");

linkerd/app/integration/src/controller.rs

@@ -181,32 +181,39 @@ fn grpc_unexpected_request() -> grpc::Status {
 }
 
 impl DstSender {
+    #[track_caller]
     pub fn send(&self, up: impl Into<pb::Update>) {
         self.0.send(Ok(up.into())).expect("send dst update")
     }
 
+    #[track_caller]
     pub fn send_err(&self, e: grpc::Status) {
         self.0.send(Err(e)).expect("send dst err")
     }
 
+    #[track_caller]
     pub fn send_addr(&self, addr: SocketAddr) {
         self.send(destination_add(addr))
     }
 
+    #[track_caller]
     pub fn send_h2_hinted(&self, addr: SocketAddr) {
         self.send(destination_add(addr).hint(Hint::H2));
     }
 
+    #[track_caller]
     pub fn send_no_endpoints(&self) {
         self.send(destination_exists_with_no_endpoints())
     }
 }
 
 impl ProfileSender {
+    #[track_caller]
     pub fn send(&self, up: pb::DestinationProfile) {
         self.0.send(Ok(up)).expect("send profile update")
     }
 
+    #[track_caller]
     pub fn send_err(&self, err: grpc::Status) {
         self.0.send(Err(err)).expect("send profile update")
     }