policy: Serve EgressNetwork responses

Cargo.lock

@@ -1325,8 +1325,7 @@ dependencies = [
 [[package]]
 name = "linkerd2-proxy-api"
 version = "0.14.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "26c72fb98d969e1e94e95d52a6fcdf4693764702c369e577934256e72fb5bc61"
+source = "git+https://github.com/linkerd/linkerd2-proxy-api?rev=c5648ae2a1e405cc6b8aca20522356ebdf20f1ea#c5648ae2a1e405cc6b8aca20522356ebdf20f1ea"
 dependencies = [
  "http",
  "ipnet",

Cargo.toml

@@ -12,3 +12,7 @@ members = [
 
 [profile.release]
 lto = "thin"
+
+[patch.crates-io]
+# TODO(Zahari): switch released version once TLS protocol support is out.
+linkerd2-proxy-api = { git = 'https://github.com/linkerd/linkerd2-proxy-api', rev = 'c5648ae2a1e405cc6b8aca20522356ebdf20f1ea' }

deny.toml

@@ -77,6 +77,7 @@ skip-tree = [
 unknown-registry = "deny"
 unknown-git = "deny"
 allow-registry = ["https://github.com/rust-lang/crates.io-index"]
+allow-git = ["https://github.com/linkerd/linkerd2-proxy-api"]
 
 [sources.allow-org]
 github = []

policy-controller/core/src/outbound.rs

@@ -8,42 +8,44 @@ use chrono::{offset::Utc, DateTime};
 use futures::prelude::*;
 use std::{net::IpAddr, num::NonZeroU16, pin::Pin, time};
 
+mod policy;
+mod target;
+
+type FallbackPolicy = ();
+
+pub use self::{
+    policy::{OutboundPolicy, ParentInfo},
+    target::{Kind, OutboundDiscoverTarget, ResourceTarget},
+};
+
+pub trait Route {
+    fn creation_timestamp(&self) -> Option<DateTime<Utc>>;
+}
+
 /// Models outbound policy discovery.
 #[async_trait::async_trait]
-pub trait DiscoverOutboundPolicy<T> {
-    async fn get_outbound_policy(&self, target: T) -> Result<Option<OutboundPolicy>>;
+pub trait DiscoverOutboundPolicy<R, T> {
+    async fn get_outbound_policy(&self, target: R) -> Result<Option<OutboundPolicy>>;
 
-    async fn watch_outbound_policy(&self, target: T) -> Result<Option<OutboundPolicyStream>>;
+    async fn watch_outbound_policy(&self, target: R) -> Result<Option<OutboundPolicyStream>>;
+
+    async fn watch_external_policy(&self) -> ExternalPolicyStream;
 
     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 type ExternalPolicyStream = Pin<Box<dyn Stream<Item = FallbackPolicy> + Send + Sync + 'static>>;
 
 pub type HttpRoute = OutboundRoute<HttpRouteMatch, HttpRetryCondition>;
 pub type GrpcRoute = OutboundRoute<GrpcRouteMatch, GrpcRetryCondition>;
-pub type RouteSet<T> = HashMap<GroupKindNamespaceName, T>;
 
-pub struct OutboundDiscoverTarget {
-    pub service_name: String,
-    pub service_namespace: String,
-    pub service_port: NonZeroU16,
-    pub source_namespace: String,
-}
+pub type RouteSet<T> = HashMap<GroupKindNamespaceName, T>;
 
-#[derive(Clone, Debug, PartialEq)]
-pub struct OutboundPolicy {
-    pub http_routes: RouteSet<HttpRoute>,
-    pub grpc_routes: RouteSet<GrpcRoute>,
-    pub authority: String,
-    pub name: String,
-    pub namespace: String,
-    pub port: NonZeroU16,
-    pub opaque: bool,
-    pub accrual: Option<FailureAccrual>,
-    pub http_retry: Option<RouteRetry<HttpRetryCondition>>,
-    pub grpc_retry: Option<RouteRetry<GrpcRetryCondition>>,
-    pub timeouts: RouteTimeouts,
+#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq)]
+pub enum TrafficPolicy {
+    Allow,
+    Deny,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -56,6 +58,24 @@ pub struct OutboundRoute<M, R> {
     pub creation_timestamp: Option<DateTime<Utc>>,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TlsRoute {
+    pub hostnames: Vec<HostMatch>,
+    pub rule: TcpRouteRule,
+    /// This is required for ordering returned routes
+    /// by their creation timestamp.
+    pub creation_timestamp: Option<DateTime<Utc>>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TcpRoute {
+    pub rule: TcpRouteRule,
+
+    /// This is required for ordering returned routes
+    /// by their creation timestamp.
+    pub creation_timestamp: Option<DateTime<Utc>>,
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct OutboundRouteRule<M, R> {
     pub matches: Vec<M>,
@@ -65,10 +85,16 @@ pub struct OutboundRouteRule<M, R> {
     pub filters: Vec<Filter>,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TcpRouteRule {
+    pub backends: Vec<Backend>,
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum Backend {
     Addr(WeightedAddr),
     Service(WeightedService),
+    EgressNetwork(WeightedEgressNetwork),
     Invalid { weight: u32, message: String },
 }
 
@@ -90,6 +116,16 @@ pub struct WeightedService {
     pub exists: bool,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct WeightedEgressNetwork {
+    pub weight: u32,
+    pub name: String,
+    pub namespace: String,
+    pub port: Option<NonZeroU16>,
+    pub filters: Vec<Filter>,
+    pub exists: bool,
+}
+
 #[derive(Copy, Clone, Debug, PartialEq)]
 pub enum FailureAccrual {
     Consecutive { max_failures: u32, backoff: Backoff },
@@ -138,3 +174,21 @@ pub enum GrpcRetryCondition {
     Internal,
     Unavailable,
 }
+
+impl<M, R> Route for OutboundRoute<M, R> {
+    fn creation_timestamp(&self) -> Option<DateTime<Utc>> {
+        self.creation_timestamp
+    }
+}
+
+impl Route for TcpRoute {
+    fn creation_timestamp(&self) -> Option<DateTime<Utc>> {
+        self.creation_timestamp
+    }
+}
+
+impl Route for TlsRoute {
+    fn creation_timestamp(&self) -> Option<DateTime<Utc>> {
+        self.creation_timestamp
+    }
+}

policy-controller/core/src/outbound/policy.rs

@@ -0,0 +1,63 @@
+use super::{
+    FailureAccrual, GrpcRetryCondition, GrpcRoute, HttpRetryCondition, HttpRoute, RouteRetry,
+    RouteSet, RouteTimeouts, TcpRoute, TlsRoute, TrafficPolicy,
+};
+
+use std::num::NonZeroU16;
+
+// ParentInfo carries resource-specific information about
+// the parent to which outbound policy is associated.
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+pub enum ParentInfo {
+    Service {
+        name: String,
+        namespace: String,
+        authority: String,
+    },
+    EgressNetwork {
+        name: String,
+        namespace: String,
+        traffic_policy: TrafficPolicy,
+    },
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct OutboundPolicy {
+    pub parent_info: ParentInfo,
+    pub http_routes: RouteSet<HttpRoute>,
+    pub grpc_routes: RouteSet<GrpcRoute>,
+    pub tls_routes: RouteSet<TlsRoute>,
+    pub tcp_routes: RouteSet<TcpRoute>,
+    pub port: NonZeroU16,
+    pub opaque: bool,
+    pub accrual: Option<FailureAccrual>,
+    pub http_retry: Option<RouteRetry<HttpRetryCondition>>,
+    pub grpc_retry: Option<RouteRetry<GrpcRetryCondition>>,
+    pub timeouts: RouteTimeouts,
+}
+
+impl ParentInfo {
+    pub fn name(&self) -> &str {
+        match self {
+            Self::EgressNetwork { name, .. } => name,
+            Self::Service { name, .. } => name,
+        }
+    }
+
+    pub fn namespace(&self) -> &str {
+        match self {
+            Self::EgressNetwork { namespace, .. } => namespace,
+            Self::Service { namespace, .. } => namespace,
+        }
+    }
+}
+
+impl OutboundPolicy {
+    pub fn parent_name(&self) -> &str {
+        self.parent_info.name()
+    }
+
+    pub fn parent_namespace(&self) -> &str {
+        self.parent_info.namespace()
+    }
+}

policy-controller/core/src/outbound/target.rs

@@ -0,0 +1,35 @@
+use std::{net::SocketAddr, num::NonZeroU16};
+
+/// OutboundDiscoverTarget allows us to express the fact that
+/// a policy resolution can be fulfilled by either a resource
+/// we know about (a specific EgressNetwork or a Service) or
+/// by our fallback mechanism.
+#[derive(Clone, Debug)]
+pub enum OutboundDiscoverTarget {
+    Resource(ResourceTarget),
+    External(SocketAddr),
+}
+
+#[derive(Clone, Debug)]
+pub struct ResourceTarget {
+    pub name: String,
+    pub namespace: String,
+    pub port: NonZeroU16,
+    pub source_namespace: String,
+    pub kind: Kind,
+}
+
+#[derive(Debug, Copy, Clone, Hash, PartialEq, Eq)]
+pub enum Kind {
+    EgressNetwork(SocketAddr),
+    Service,
+}
+
+impl ResourceTarget {
+    pub fn original_dst(&self) -> Option<SocketAddr> {
+        match self.kind {
+            Kind::EgressNetwork(original_dst) => Some(original_dst),
+            Kind::Service => None,
+        }
+    }
+}