feat(outbound): Use the policy API for opaq routing (#3306)

The OutboundPolicy API is able to express configuration for opaq routing, but
the proxy only currently uses the ServiceProfile to configure routing. To
support configuring routing via the Gateway API's TCPRoute, we need to begin
using the OutboundPolicy API to configure opaq routing without breaking
existing configurations that use profiles.

With this change, the outbound proxy is updated so that route configurations are
required for the proxy to handle outbound opaq traffic. Route configurations may
be provided by:

* The Profile API, when it returns an Endpoint, so that the proxy synthesizes a
  route configuration that forwards traffic to the endpoint.
* The Profile API, when it returns a traffic split configuration, so that the
  proxy synthesizes a route configuration that forwards traffic over backend
  services.
* The OutboundPolicy API when it returns opaq routes.

When no routes are returned by the API, the proxy fails to route the
connections.

Follow up changes will enhance this functionality with metrics and improved
metadata.

Author: zaharidichev

Cargo.lock

@@ -1491,6 +1491,7 @@ dependencies = [
  "linkerd-io",
  "linkerd-meshtls",
  "linkerd-meshtls-rustls",
+ "linkerd-opaq-route",
  "linkerd-proxy-client-policy",
  "linkerd-retry",
  "linkerd-stack",
@@ -1890,6 +1891,10 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "linkerd-opaq-route"
+version = "0.1.0"
+
 [[package]]
 name = "linkerd-opencensus"
 version = "0.1.0"
@@ -2054,6 +2059,7 @@ dependencies = [
  "linkerd-error",
  "linkerd-exp-backoff",
  "linkerd-http-route",
+ "linkerd-opaq-route",
  "linkerd-proxy-api-resolve",
  "linkerd-proxy-core",
  "linkerd-tls-route",

Cargo.toml

@@ -40,6 +40,7 @@ members = [
     "linkerd/meshtls/rustls",
     "linkerd/meshtls/verifier",
     "linkerd/metrics",
+    "linkerd/opaq-route",
     "linkerd/opencensus",
     "linkerd/opentelemetry",
     "linkerd/pool",

linkerd/app/gateway/src/opaq.rs

@@ -1,10 +1,15 @@
 use super::{server::Opaq, Gateway};
 use inbound::{GatewayAddr, GatewayDomainInvalid};
-use linkerd_app_core::{io, profiles, svc, tls, transport::addrs::*, Error};
+use linkerd_app_core::{io, svc, tls, transport::addrs::*, Error};
 use linkerd_app_inbound as inbound;
 use linkerd_app_outbound as outbound;
+use tokio::sync::watch;
 
-pub type Target = outbound::opaq::Logical;
+#[derive(Clone, Debug)]
+pub struct Target {
+    addr: GatewayAddr,
+    routes: watch::Receiver<outbound::opaq::Routes>,
+}
 
 impl Gateway {
     /// Wrap the provided outbound opaque stack with inbound authorization and
@@ -33,22 +38,55 @@ impl Gateway {
             .push_filter(
                 |(_, opaq): (_, Opaq<T>)| -> Result<_, GatewayDomainInvalid> {
                     // Fail connections were not resolved.
-                    let profile = svc::Param::<Option<profiles::Receiver>>::param(&*opaq)
-                        .ok_or(GatewayDomainInvalid)?;
-                    if let Some(profiles::LogicalAddr(addr)) = profile.logical_addr() {
-                        Ok(outbound::opaq::Logical::Route(addr, profile))
-                    } else if let Some((addr, metadata)) = profile.endpoint() {
-                        Ok(outbound::opaq::Logical::Forward(
-                            Remote(ServerAddr(addr)),
-                            metadata,
-                        ))
-                    } else {
-                        Err(GatewayDomainInvalid)
-                    }
+                    Target::try_from(opaq)
                 },
             )
             // Authorize connections to the gateway.
             .push(self.inbound.authorize_tcp())
             .arc_new_tcp()
     }
 }
+
+impl<T> TryFrom<Opaq<T>> for Target
+where
+    T: svc::Param<GatewayAddr>,
+{
+    type Error = GatewayDomainInvalid;
+
+    fn try_from(opaq: Opaq<T>) -> Result<Self, Self::Error> {
+        use svc::Param;
+
+        let addr: GatewayAddr = (**opaq).param();
+        let Some(profile) = (*opaq).param() else {
+            // The gateway address must be resolvable via the profile API.
+            return Err(GatewayDomainInvalid);
+        };
+        let routes = outbound::opaq::routes_from_discovery(
+            addr.0.clone().into(),
+            Some(profile),
+            (*opaq).param(),
+        );
+
+        Ok(Target { addr, routes })
+    }
+}
+
+impl svc::Param<watch::Receiver<outbound::opaq::Routes>> for Target {
+    fn param(&self) -> watch::Receiver<outbound::opaq::Routes> {
+        self.routes.clone()
+    }
+}
+
+impl PartialEq for Target {
+    fn eq(&self, other: &Self) -> bool {
+        self.addr == other.addr
+    }
+}
+
+impl Eq for Target {}
+
+impl std::hash::Hash for Target {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.addr.hash(state);
+    }
+}

linkerd/app/outbound/Cargo.toml

@@ -40,6 +40,7 @@ linkerd-http-retry = { path = "../../http/retry" }
 linkerd-http-route = { path = "../../http/route" }
 linkerd-identity = { path = "../../identity" }
 linkerd-meshtls-rustls = { path = "../../meshtls/rustls", optional = true }
+linkerd-opaq-route = { path = "../../opaq-route" }
 linkerd-proxy-client-policy = { path = "../../proxy/client-policy", features = [
     "proto",
 ] }

linkerd/app/outbound/src/discover.rs

@@ -234,16 +234,18 @@ fn policy_for_backend(
     static NO_OPAQ_FILTERS: Lazy<Arc<[policy::opaq::Filter]>> = Lazy::new(|| Arc::new([]));
 
     let opaque = policy::opaq::Opaque {
-        policy: Some(policy::opaq::Policy {
-            meta: meta.clone(),
-            filters: NO_OPAQ_FILTERS.clone(),
-            params: Default::default(),
-            distribution: policy::RouteDistribution::FirstAvailable(Arc::new([
-                policy::RouteBackend {
-                    filters: NO_OPAQ_FILTERS.clone(),
-                    backend: backend.clone(),
-                },
-            ])),
+        routes: Some(policy::opaq::Route {
+            policy: policy::opaq::Policy {
+                meta: meta.clone(),
+                filters: NO_OPAQ_FILTERS.clone(),
+                params: Default::default(),
+                distribution: policy::RouteDistribution::FirstAvailable(Arc::new([
+                    policy::RouteBackend {
+                        filters: NO_OPAQ_FILTERS.clone(),
+                        backend: backend.clone(),
+                    },
+                ])),
+            },
         }),
     };
 
@@ -327,6 +329,15 @@ impl<T> svc::Param<Option<profiles::Receiver>> for Discovery<T> {
     }
 }
 
+impl<T> svc::Param<OrigDstAddr> for Discovery<T>
+where
+    T: svc::Param<OrigDstAddr>,
+{
+    fn param(&self) -> OrigDstAddr {
+        self.parent.param()
+    }
+}
+
 impl<T> svc::Param<Option<watch::Receiver<profiles::Profile>>> for Discovery<T> {
     fn param(&self) -> Option<watch::Receiver<profiles::Profile>> {
         self.profile.clone().map(Into::into)

linkerd/app/outbound/src/http/logical/profile.rs

@@ -2,11 +2,11 @@ use super::{
     super::{concrete, retry},
     CanonicalDstHeader, Concrete, NoRoute,
 };
-use crate::{policy, BackendRef, ParentRef, UNKNOWN_META};
+use crate::{service_meta, BackendRef, ParentRef, UNKNOWN_META};
 use linkerd_app_core::{
     classify, metrics,
     proxy::http::{self, balance},
-    svc, Error, NameAddr,
+    svc, Error,
 };
 use linkerd_distribute as distribute;
 use std::{fmt::Debug, hash::Hash, sync::Arc, time};
@@ -107,26 +107,6 @@ where
             targets,
         } = routes;
 
-        fn service_meta(addr: &NameAddr) -> Option<Arc<policy::Meta>> {
-            let mut parts = addr.name().split('.');
-
-            let name = parts.next()?;
-            let namespace = parts.next()?;
-
-            if !parts.next()?.eq_ignore_ascii_case("svc") {
-                return None;
-            }
-
-            Some(Arc::new(policy::Meta::Resource {
-                group: "core".to_string(),
-                kind: "Service".to_string(),
-                namespace: namespace.to_string(),
-                name: name.to_string(),
-                section: None,
-                port: Some(addr.port().try_into().ok()?),
-            }))
-        }
-
         let parent_meta = service_meta(&addr).unwrap_or_else(|| UNKNOWN_META.clone());
 
         // Create concrete targets for all of the profile's routes.

linkerd/app/outbound/src/ingress.rs

@@ -22,8 +22,11 @@ struct Http<T> {
     version: http::Version,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq, Hash)]
-struct Opaq<T>(Discovery<T>);
+#[derive(Clone, Debug)]
+struct Opaq {
+    orig_dst: OrigDstAddr,
+    routes: watch::Receiver<opaq::Routes>,
+}
 
 #[derive(Clone, Debug)]
 struct SelectTarget<T>(Http<T>);
@@ -91,7 +94,7 @@ impl Outbound<()> {
             let discover = discover.clone();
             self.to_tcp_connect()
                 .push_opaq_cached(resolve.clone())
-                .map_stack(|_, _, stk| stk.push_map_target(Opaq))
+                .map_stack(|_, _, stk| stk.push_map_target(Opaq::from))
                 .push_discover(svc::mk(move |OrigDstAddr(addr)| {
                     discover.clone().oneshot(DiscoverAddr(addr.into()))
                 }))
@@ -600,42 +603,41 @@ impl std::hash::Hash for Logical {
     }
 }
 
-// === impl Opaq ===
+impl<T> From<Discovery<T>> for Opaq
+where
+    T: svc::Param<OrigDstAddr>,
+{
+    fn from(discovery: Discovery<T>) -> Self {
+        use svc::Param;
 
-impl<T> std::ops::Deref for Opaq<T> {
-    type Target = T;
+        let orig_dst: OrigDstAddr = discovery.param();
+        let routes = opaq::routes_from_discovery(
+            Addr::Socket(orig_dst.into()),
+            discovery.param(),
+            discovery.param(),
+        );
 
-    fn deref(&self) -> &Self::Target {
-        &self.0
+        Self { routes, orig_dst }
     }
 }
 
-impl<T> svc::Param<Remote<ServerAddr>> for Opaq<T>
-where
-    T: svc::Param<OrigDstAddr>,
-{
-    fn param(&self) -> Remote<ServerAddr> {
-        let OrigDstAddr(addr) = (*self.0).param();
-        Remote(ServerAddr(addr))
+impl PartialEq for Opaq {
+    fn eq(&self, other: &Self) -> bool {
+        self.orig_dst == other.orig_dst
     }
 }
 
-impl<T> svc::Param<opaq::Logical> for Opaq<T>
-where
-    T: svc::Param<OrigDstAddr>,
-{
-    fn param(&self) -> opaq::Logical {
-        if let Some(profile) = svc::Param::<Option<profiles::Receiver>>::param(&self.0) {
-            if let Some(profiles::LogicalAddr(addr)) = profile.logical_addr() {
-                return opaq::Logical::Route(addr, profile);
-            }
+impl Eq for Opaq {}
 
-            if let Some((addr, metadata)) = profile.endpoint() {
-                return opaq::Logical::Forward(Remote(ServerAddr(addr)), metadata);
-            }
-        }
+impl std::hash::Hash for Opaq {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.orig_dst.hash(state);
+    }
+}
 
-        opaq::Logical::Forward(self.param(), Default::default())
+impl svc::Param<watch::Receiver<opaq::Routes>> for Opaq {
+    fn param(&self) -> watch::Receiver<opaq::Routes> {
+        self.routes.clone()
     }
 }
 

linkerd/app/outbound/src/lib.rs

@@ -23,7 +23,7 @@ use linkerd_app_core::{
     svc::{self, ServiceExt},
     tls::ConnectMeta as TlsConnectMeta,
     transport::addrs::*,
-    AddrMatch, Error, ProxyRuntime,
+    AddrMatch, Error, NameAddr, ProxyRuntime,
 };
 use linkerd_tonic_stream::ReceiveLimits;
 use std::{
@@ -342,3 +342,23 @@ impl EndpointRef {
 
 static UNKNOWN_META: once_cell::sync::Lazy<Arc<policy::Meta>> =
     once_cell::sync::Lazy::new(|| policy::Meta::new_default("unknown"));
+
+pub(crate) fn service_meta(addr: &NameAddr) -> Option<Arc<policy::Meta>> {
+    let mut parts = addr.name().split('.');
+
+    let name = parts.next()?;
+    let namespace = parts.next()?;
+
+    if !parts.next()?.eq_ignore_ascii_case("svc") {
+        return None;
+    }
+
+    Some(Arc::new(policy::Meta::Resource {
+        group: "core".to_string(),
+        kind: "Service".to_string(),
+        namespace: namespace.to_string(),
+        name: name.to_string(),
+        section: None,
+        port: Some(addr.port().try_into().ok()?),
+    }))
+}