policy: refactor opaq stack to use client-policy

Cargo.lock

@@ -1338,6 +1338,7 @@ dependencies = [
  "linkerd-io",
  "linkerd-meshtls",
  "linkerd-meshtls-rustls",
+ "linkerd-opaq-route",
  "linkerd-proxy-client-policy",
  "linkerd-retry",
  "linkerd-stack",
@@ -1738,6 +1739,16 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "linkerd-opaq-route"
+version = "0.1.0"
+dependencies = [
+ "rand",
+ "regex",
+ "thiserror",
+ "tracing",
+]
+
 [[package]]
 name = "linkerd-opencensus"
 version = "0.1.0"
@@ -1902,6 +1913,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

@@ -3,8 +3,15 @@ use inbound::{GatewayAddr, GatewayDomainInvalid};
 use linkerd_app_core::{io, profiles, 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 {
+    orig_dst: OrigDstAddr,
+    // this value is present only if we are using profiles for discovery
+    profiles_logical: Option<profiles::LogicalAddr>,
+    routes: watch::Receiver<outbound::opaq::Routes>,
+}
 
 impl Gateway {
     /// Wrap the provided outbound opaque stack with inbound authorization and
@@ -33,22 +40,62 @@ 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<OrigDstAddr>,
+{
+    type Error = GatewayDomainInvalid;
+
+    fn try_from(opaq: Opaq<T>) -> Result<Self, Self::Error> {
+        let profile = svc::Param::<Option<profiles::Receiver>>::param(&*opaq);
+        let policy = svc::Param::<outbound::policy::Receiver>::param(&*opaq);
+        let orig_dst = svc::Param::<OrigDstAddr>::param(&*opaq);
+        // we error if there is no profiles resolution
+        if profile.is_none() {
+            return Err(GatewayDomainInvalid);
+        }
+
+        let (routes, profiles_logical) =
+            outbound::opaq::routes_from_discovery(orig_dst, profile, policy);
+        Ok(Target {
+            orig_dst,
+            profiles_logical,
+            routes,
+        })
+    }
+}
+
+impl svc::Param<Option<profiles::LogicalAddr>> for Target {
+    fn param(&self) -> Option<profiles::LogicalAddr> {
+        self.profiles_logical.clone()
+    }
+}
+
+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.orig_dst == other.orig_dst
+    }
+}
+
+impl Eq for Target {}
+
+impl std::hash::Hash for Target {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.orig_dst.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,13 @@ 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,
+    // this value is present only if we are using profiles for discovery
+    profiles_logical: Option<profiles::LogicalAddr>,
+    routes: watch::Receiver<opaq::Routes>,
+}
 
 #[derive(Clone, Debug)]
 struct SelectTarget<T>(Http<T>);
@@ -91,7 +96,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 +605,47 @@ impl std::hash::Hash for Logical {
     }
 }
 
-// === impl Opaq ===
-
-impl<T> std::ops::Deref for Opaq<T> {
-    type Target = T;
-
-    fn deref(&self) -> &Self::Target {
-        &self.0
+impl svc::Param<Option<profiles::LogicalAddr>> for Opaq {
+    fn param(&self) -> Option<profiles::LogicalAddr> {
+        self.profiles_logical.clone()
     }
 }
 
-impl<T> svc::Param<Remote<ServerAddr>> for Opaq<T>
+impl<T> From<Discovery<T>> for Opaq
 where
-    T: svc::Param<OrigDstAddr>,
+    T: svc::Param<OrigDstAddr> + Clone,
 {
-    fn param(&self) -> Remote<ServerAddr> {
-        let OrigDstAddr(addr) = (*self.0).param();
-        Remote(ServerAddr(addr))
+    fn from(discovery: Discovery<T>) -> Self {
+        let policy = svc::Param::<policy::Receiver>::param(&discovery);
+        let profile = svc::Param::<Option<watch::Receiver<profiles::Profile>>>::param(&discovery);
+        let orig_dst: OrigDstAddr = discovery.param();
+        let (routes, profiles_logical) =
+            opaq::routes_from_discovery(orig_dst, profile.map(Into::into), policy);
+        Self {
+            routes,
+            profiles_logical,
+            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 PartialEq for Opaq {
+    fn eq(&self, other: &Self) -> bool {
+        self.orig_dst == other.orig_dst
+    }
+}
 
-            if let Some((addr, metadata)) = profile.endpoint() {
-                return opaq::Logical::Forward(Remote(ServerAddr(addr)), metadata);
-            }
-        }
+impl Eq for Opaq {}
+
+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()?),
+    }))
+}