Add client-policy types (#2236)

This change introduces the client policy types that will be used to
configure outbound proxies. These types are analogous to those used for
server policy in inbound proxies. They support integration with the
Kubernetes Gateway API.

These types are relatively sketchy at the moment. They will change as
they are integrated into the outbound proxy.

Author: olix0r

Cargo.lock

@@ -1349,6 +1349,19 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "linkerd-proxy-client-policy"
+version = "0.1.0"
+dependencies = [
+ "http",
+ "ipnet",
+ "linkerd-http-route",
+ "linkerd-proxy-api-resolve",
+ "linkerd-proxy-core",
+ "maplit",
+ "quickcheck",
+]
+
 [[package]]
 name = "linkerd-proxy-core"
 version = "0.1.0"

Cargo.toml

@@ -40,6 +40,7 @@ members = [
     "linkerd/opencensus",
     "linkerd/proxy/api-resolve",
     "linkerd/proxy/balance",
+    "linkerd/proxy/client-policy",
     "linkerd/proxy/core",
     "linkerd/proxy/dns-resolve",
     "linkerd/proxy/http",

linkerd/proxy/client-policy/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "linkerd-proxy-client-policy"
+version = "0.1.0"
+authors = ["Linkerd Developers <[email protected]>"]
+license = "Apache-2.0"
+edition = "2021"
+publish = false
+
+[features]
+# proto = [
+#     "linkerd-http-route/proto",
+#     "linkerd-proxy-core/proto",
+#     "linkerd2-proxy-api",
+#     "prost-types",
+# ]
+
+[dependencies]
+ipnet = "2"
+http = "0.2"
+linkerd-proxy-api-resolve = { path = "../api-resolve" }
+linkerd-proxy-core = { path = "../core" }
+linkerd-http-route = { path = "../../http-route" }
+# prost-types = { version = "0.11", optional = true }
+# thiserror = "1"
+
+[dev-dependencies]
+maplit = "1"
+quickcheck = { version = "1", default-features = false }

linkerd/proxy/client-policy/src/grpc.rs

@@ -0,0 +1,51 @@
+use linkerd_http_route::{grpc, http};
+use std::sync::Arc;
+
+pub use linkerd_http_route::grpc::{filter, r#match, RouteMatch};
+
+pub type Policy = crate::RoutePolicy<Filter>;
+pub type Route = grpc::Route<Policy>;
+pub type Rule = grpc::Rule<Policy>;
+
+// TODO HTTP2 settings
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct Grpc {
+    pub routes: Arc<[Route]>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum Filter {
+    InjectFailure(filter::InjectFailure),
+    RequestHeaders(http::filter::ModifyHeader),
+    InternalError(&'static str),
+}
+
+#[inline]
+pub fn find<'r, B>(
+    routes: &'r [Route],
+    req: &::http::Request<B>,
+) -> Option<(RouteMatch, &'r Policy)> {
+    grpc::find(routes, req)
+}
+
+pub fn default(distribution: crate::RouteDistribution<Filter>) -> Route {
+    Route {
+        hosts: vec![],
+        rules: vec![Rule {
+            matches: vec![],
+            policy: Policy {
+                meta: crate::Meta::new_default("default"),
+                filters: Arc::new([]),
+                distribution,
+            },
+        }],
+    }
+}
+
+impl Default for Grpc {
+    fn default() -> Self {
+        Self {
+            routes: Arc::new([]),
+        }
+    }
+}

linkerd/proxy/client-policy/src/http.rs

@@ -0,0 +1,66 @@
+use linkerd_http_route::http;
+use std::sync::Arc;
+
+pub use linkerd_http_route::http::{filter, r#match, RouteMatch};
+
+pub type Policy = crate::RoutePolicy<Filter>;
+pub type Route = http::Route<Policy>;
+pub type Rule = http::Rule<Policy>;
+
+// TODO: keepalive settings, etc.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct Http1 {
+    pub routes: Arc<[Route]>,
+}
+
+// TODO: window sizes, etc
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct Http2 {
+    pub routes: Arc<[Route]>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum Filter {
+    InjectFailure(filter::InjectFailure),
+    Redirect(filter::RedirectRequest),
+    RequestHeaders(filter::ModifyHeader),
+    InternalError(&'static str),
+}
+
+#[inline]
+pub fn find<'r, B>(
+    routes: &'r [Route],
+    req: &::http::Request<B>,
+) -> Option<(http::RouteMatch, &'r Policy)> {
+    http::find(routes, req)
+}
+
+pub fn default(distribution: crate::RouteDistribution<Filter>) -> Route {
+    Route {
+        hosts: vec![],
+        rules: vec![Rule {
+            matches: vec![],
+            policy: Policy {
+                meta: crate::Meta::new_default("default"),
+                filters: Arc::new([]),
+                distribution,
+            },
+        }],
+    }
+}
+
+impl Default for Http1 {
+    fn default() -> Self {
+        Self {
+            routes: Arc::new([]),
+        }
+    }
+}
+
+impl Default for Http2 {
+    fn default() -> Self {
+        Self {
+            routes: Arc::new([]),
+        }
+    }
+}

linkerd/proxy/client-policy/src/lib.rs

@@ -0,0 +1,171 @@
+#![deny(rust_2018_idioms, clippy::disallowed_methods, clippy::disallowed_types)]
+#![forbid(unsafe_code)]
+
+use std::{borrow::Cow, hash::Hash, net::SocketAddr, sync::Arc, time};
+
+pub mod grpc;
+pub mod http;
+pub mod opaq;
+
+pub use linkerd_http_route as route;
+pub use linkerd_proxy_api_resolve::Metadata as EndpointMetadata;
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ClientPolicy {
+    pub protocol: Protocol,
+    pub backends: Arc<[Backend]>,
+}
+
+// TODO additional server configs (e.g. concurrency limits, window sizes, etc)
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub enum Protocol {
+    Detect {
+        timeout: time::Duration,
+        http1: http::Http1,
+        http2: http::Http2,
+        opaque: opaq::Opaque,
+    },
+
+    Http1(http::Http1),
+    Http2(http::Http2),
+    Grpc(grpc::Grpc),
+
+    Opaque(opaq::Opaque),
+
+    // TODO(ver) TLS-aware type
+    Tls(opaq::Opaque),
+}
+
+#[derive(Debug, Eq)]
+pub enum Meta {
+    Default {
+        name: Cow<'static, str>,
+    },
+    Resource {
+        group: String,
+        kind: String,
+        name: String,
+        namespace: String,
+        section: Option<String>,
+    },
+}
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub struct RoutePolicy<T> {
+    pub meta: Arc<Meta>,
+    pub filters: Arc<[T]>,
+    pub distribution: RouteDistribution<T>,
+}
+
+// TODO(ver) Weighted random WITHOUT availability awareness, as required by
+// HTTPRoute.
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub enum RouteDistribution<T> {
+    Empty,
+
+    FirstAvailable(Arc<[RouteBackend<T>]>),
+
+    RandomAvailable(Arc<[(RouteBackend<T>, u32)]>),
+}
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub struct RouteBackend<T> {
+    pub filters: Arc<[T]>,
+    pub backend: Backend,
+}
+
+// TODO(ver) how does configuration like failure accrual fit in here? What about
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub struct Backend {
+    pub meta: Arc<Meta>,
+    pub queue: Queue,
+    pub dispatcher: BackendDispatcher,
+}
+
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub struct Queue {
+    pub capacity: usize,
+    pub failfast_timeout: time::Duration,
+}
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub enum BackendDispatcher {
+    Forward(SocketAddr, EndpointMetadata),
+    BalanceP2c(Load, EndpointDiscovery),
+}
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub enum EndpointDiscovery {
+    DestinationGet { path: String },
+}
+
+/// Configures the load balancing strategy for a backend.
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub enum Load {
+    PeakEwma(PeakEwma),
+}
+
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+pub struct PeakEwma {
+    pub decay: time::Duration,
+    pub default_rtt: time::Duration,
+}
+
+// === impl Meta ===
+
+impl Meta {
+    pub fn new_default(name: impl Into<Cow<'static, str>>) -> Arc<Self> {
+        Arc::new(Self::Default { name: name.into() })
+    }
+
+    pub fn group(&self) -> &str {
+        match self {
+            Self::Default { .. } => "",
+            Self::Resource { group, .. } => group,
+        }
+    }
+
+    pub fn kind(&self) -> &str {
+        match self {
+            Self::Default { .. } => "default",
+            Self::Resource { kind, .. } => kind,
+        }
+    }
+
+    pub fn name(&self) -> &str {
+        match self {
+            Self::Default { name } => name,
+            Self::Resource { name, .. } => name,
+        }
+    }
+
+    pub fn namespace(&self) -> &str {
+        match self {
+            Self::Default { .. } => "",
+            Self::Resource { namespace, .. } => namespace,
+        }
+    }
+
+    pub fn section(&self) -> &str {
+        match self {
+            Self::Default { .. } => "",
+            Self::Resource { section, .. } => section.as_deref().unwrap_or(""),
+        }
+    }
+}
+
+impl std::cmp::PartialEq for Meta {
+    fn eq(&self, other: &Self) -> bool {
+        // Resources that look like Defaults are considered equal.
+        self.group() == other.group() && self.kind() == other.kind() && self.name() == other.name()
+    }
+}
+
+impl std::hash::Hash for Meta {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        // Resources that look like Defaults are considered the same.
+        self.group().hash(state);
+        self.kind().hash(state);
+        self.name().hash(state);
+    }
+}

linkerd/proxy/client-policy/src/opaq.rs

@@ -0,0 +1,11 @@
+use crate::RoutePolicy;
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Hash)]
+pub struct Opaque {
+    pub policy: Option<Policy>,
+}
+
+pub type Policy = RoutePolicy<Filter>;
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub enum Filter {}