Add support for consumer routes

cli/cmd/policy.go

@@ -132,7 +132,7 @@ displayed.`,
 				client := outbound.NewOutboundPoliciesClient(conn)
 
 				result, err = client.Get(cmd.Context(), &outbound.TrafficSpec{
-					SourceWorkload: "diagnostics",
+					SourceWorkload: "default:diagnostics",
 					Target:         &outbound.TrafficSpec_Authority{Authority: fmt.Sprintf("%s.%s.svc:%d", name, namespace, port)},
 				})
 				if err != nil {

policy-controller/core/src/http_route.rs

@@ -15,6 +15,14 @@ pub struct GroupKindName {
     pub name: Cow<'static, str>,
 }
 
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub struct GroupKindNamespaceName {
+    pub group: Cow<'static, str>,
+    pub kind: Cow<'static, str>,
+    pub namespace: Cow<'static, str>,
+    pub name: Cow<'static, str>,
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum HostMatch {
     Exact(String),
@@ -107,6 +115,15 @@ impl GroupKindName {
             && self.kind.eq_ignore_ascii_case(&other.kind)
             && self.name.eq_ignore_ascii_case(&other.name)
     }
+
+    pub fn namespaced(self, namespace: String) -> GroupKindNamespaceName {
+        GroupKindNamespaceName {
+            group: self.group,
+            kind: self.kind,
+            namespace: namespace.into(),
+            name: self.name,
+        }
+    }
 }
 
 // === impl PathMatch ===

policy-controller/core/src/outbound.rs

@@ -1,5 +1,6 @@
 use crate::http_route::{
-    GroupKindName, HostMatch, HttpRouteMatch, RequestHeaderModifierFilter, RequestRedirectFilter,
+    GroupKindNamespaceName, HostMatch, HttpRouteMatch, RequestHeaderModifierFilter,
+    RequestRedirectFilter,
 };
 use ahash::AHashMap as HashMap;
 use anyhow::Result;
@@ -14,14 +15,21 @@ pub trait DiscoverOutboundPolicy<T> {
 
     async fn watch_outbound_policy(&self, target: T) -> Result<Option<OutboundPolicyStream>>;
 
-    fn lookup_ip(&self, addr: IpAddr, port: NonZeroU16) -> Option<T>;
+    fn lookup_ip(&self, addr: IpAddr, port: NonZeroU16, source_namespace: String) -> Option<T>;
 }
 
 pub type OutboundPolicyStream = Pin<Box<dyn Stream<Item = OutboundPolicy> + Send + Sync + 'static>>;
 
+pub struct OutboundDiscoverTarget {
+    pub service_name: String,
+    pub service_namespace: String,
+    pub service_port: NonZeroU16,
+    pub source_namespace: String,
+}
+
 #[derive(Clone, Debug, PartialEq)]
 pub struct OutboundPolicy {
-    pub http_routes: HashMap<GroupKindName, HttpRoute>,
+    pub http_routes: HashMap<GroupKindNamespaceName, HttpRoute>,
     pub authority: String,
     pub name: String,
     pub namespace: String,

policy-controller/grpc/src/outbound.rs

@@ -9,10 +9,10 @@ use linkerd2_proxy_api::{
     },
 };
 use linkerd_policy_controller_core::{
-    http_route::GroupKindName,
+    http_route::GroupKindNamespaceName,
     outbound::{
-        Backend, DiscoverOutboundPolicy, Filter, HttpRoute, HttpRouteRule, OutboundPolicy,
-        OutboundPolicyStream,
+        Backend, DiscoverOutboundPolicy, Filter, HttpRoute, HttpRouteRule, OutboundDiscoverTarget,
+        OutboundPolicy, OutboundPolicyStream,
     },
 };
 use std::{net::SocketAddr, num::NonZeroU16, sync::Arc, time};
@@ -27,7 +27,7 @@ pub struct OutboundPolicyServer<T> {
 
 impl<T> OutboundPolicyServer<T>
 where
-    T: DiscoverOutboundPolicy<(String, String, NonZeroU16)> + Send + Sync + 'static,
+    T: DiscoverOutboundPolicy<OutboundDiscoverTarget> + Send + Sync + 'static,
 {
     pub fn new(discover: T, cluster_domain: impl Into<Arc<str>>, drain: drain::Watch) -> Self {
         Self {
@@ -41,16 +41,33 @@ where
         OutboundPoliciesServer::new(self)
     }
 
-    fn lookup(
-        &self,
-        spec: outbound::TrafficSpec,
-    ) -> Result<(String, String, NonZeroU16), tonic::Status> {
+    fn lookup(&self, spec: outbound::TrafficSpec) -> Result<OutboundDiscoverTarget, tonic::Status> {
         let target = spec
             .target
             .ok_or_else(|| tonic::Status::invalid_argument("target is required"))?;
+        let source_namespace = spec
+            .source_workload
+            .split_once(':')
+            .ok_or_else(|| {
+                tonic::Status::invalid_argument(format!(
+                    "failed to parse source workload: {}",
+                    spec.source_workload
+                ))
+            })?
+            .0
+            .to_string();
         let target = match target {
             outbound::traffic_spec::Target::Addr(target) => target,
-            outbound::traffic_spec::Target::Authority(auth) => return self.lookup_authority(&auth),
+            outbound::traffic_spec::Target::Authority(auth) => {
+                return self.lookup_authority(&auth).map(
+                    |(service_namespace, service_name, service_port)| OutboundDiscoverTarget {
+                        service_name,
+                        service_namespace,
+                        service_port,
+                        source_namespace,
+                    },
+                )
+            }
         };
 
         let port = target
@@ -69,7 +86,7 @@ where
             })?;
 
         self.index
-            .lookup_ip(addr, port)
+            .lookup_ip(addr, port, source_namespace)
             .ok_or_else(|| tonic::Status::not_found("No such service"))
     }
 
@@ -119,7 +136,7 @@ where
 #[async_trait::async_trait]
 impl<T> OutboundPolicies for OutboundPolicyServer<T>
 where
-    T: DiscoverOutboundPolicy<(String, String, NonZeroU16)> + Send + Sync + 'static,
+    T: DiscoverOutboundPolicy<OutboundDiscoverTarget> + Send + Sync + 'static,
 {
     async fn get(
         &self,
@@ -215,9 +232,7 @@ fn to_service(outbound: OutboundPolicy) -> outbound::OutboundPolicy {
 
         let mut http_routes: Vec<_> = http_routes
             .into_iter()
-            .map(|(gkn, route)| {
-                convert_outbound_http_route(outbound.namespace.clone(), gkn, route, backend.clone())
-            })
+            .map(|(gknn, route)| convert_outbound_http_route(gknn, route, backend.clone()))
             .collect();
 
         if http_routes.is_empty() {
@@ -291,8 +306,7 @@ fn to_service(outbound: OutboundPolicy) -> outbound::OutboundPolicy {
 }
 
 fn convert_outbound_http_route(
-    namespace: String,
-    gkn: GroupKindName,
+    gknn: GroupKindNamespaceName,
     HttpRoute {
         hostnames,
         rules,
@@ -302,10 +316,10 @@ fn convert_outbound_http_route(
 ) -> outbound::HttpRoute {
     let metadata = Some(Metadata {
         kind: Some(metadata::Kind::Resource(api::meta::Resource {
-            group: gkn.group.to_string(),
-            kind: gkn.kind.to_string(),
-            namespace,
-            name: gkn.name.to_string(),
+            group: gknn.group.to_string(),
+            kind: gknn.kind.to_string(),
+            namespace: gknn.namespace.to_string(),
+            name: gknn.name.to_string(),
             ..Default::default()
         })),
     });

policy-controller/k8s/index/src/http_route.rs

@@ -1,7 +1,7 @@
 use anyhow::{anyhow, bail, Result};
 use k8s_gateway_api as api;
 use kube::{Resource, ResourceExt};
-use linkerd_policy_controller_core::http_route::{self, GroupKindName};
+use linkerd_policy_controller_core::http_route::{self, GroupKindName, GroupKindNamespaceName};
 use linkerd_policy_controller_k8s_api::policy;
 use std::num::NonZeroU16;
 
@@ -44,10 +44,12 @@ impl HttpRouteResource {
         }
     }
 
-    pub(crate) fn gkn(&self) -> GroupKindName {
+    pub(crate) fn gknn(&self) -> GroupKindNamespaceName {
         match self {
-            HttpRouteResource::Linkerd(route) => gkn_for_resource(route),
-            HttpRouteResource::Gateway(route) => gkn_for_resource(route),
+            HttpRouteResource::Linkerd(route) => gkn_for_resource(route)
+                .namespaced(route.namespace().expect("Route must have namespace")),
+            HttpRouteResource::Gateway(route) => gkn_for_resource(route)
+                .namespaced(route.namespace().expect("Route must have namespace")),
         }
     }
 }