outbound: Add a policy router

Cargo.lock

@@ -945,10 +945,12 @@ dependencies = [
  "linkerd-distribute",
  "linkerd-http-classify",
  "linkerd-http-retry",
+ "linkerd-http-route",
  "linkerd-identity",
  "linkerd-io",
  "linkerd-meshtls",
  "linkerd-meshtls-rustls",
+ "linkerd-proxy-client-policy",
  "linkerd-retry",
  "linkerd-tracing",
  "parking_lot",

linkerd/app/outbound/Cargo.toml

@@ -10,11 +10,11 @@ Configures and runs the outbound proxy
 """
 
 [features]
-test-util = ["linkerd-app-test", "linkerd-meshtls-rustls/test-util"]
-
 default = []
 allow-loopback = []
 test-subscriber = []
+test-util = ["linkerd-app-test", "linkerd-meshtls-rustls/test-util"]
+
 [dependencies]
 bytes = "1"
 http = "0.2"
@@ -24,8 +24,10 @@ linkerd-app-test = { path = "../test", optional = true }
 linkerd-distribute = { path = "../../distribute" }
 linkerd-http-classify = { path = "../../http-classify" }
 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-proxy-client-policy = { path = "../../proxy/client-policy" }
 linkerd-retry = { path = "../../retry" }
 parking_lot = "0.12"
 thiserror = "1"

linkerd/app/outbound/src/http.rs

@@ -25,7 +25,7 @@ mod retry;
 mod server;
 mod strip_proxy_error;
 
-pub use self::logical::{profile, LogicalAddr, Routes};
+pub use self::logical::{policy, profile, LogicalAddr, Routes};
 pub(crate) use self::require_id_header::IdentityRequired;
 pub use linkerd_app_core::proxy::http::{self as http, *};
 

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

@@ -13,6 +13,7 @@ use linkerd_distribute as distribute;
 use std::{fmt::Debug, hash::Hash};
 use tokio::sync::watch;
 
+pub mod policy;
 pub mod profile;
 
 /// Indicates the address used for logical routing.
@@ -22,6 +23,9 @@ pub struct LogicalAddr(pub Addr);
 /// Configures the flavor of HTTP routing.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum Routes {
+    /// Policy routes.
+    Policy(policy::Routes),
+
     /// Service profile routes.
     Profile(profile::Routes),
 
@@ -51,6 +55,8 @@ pub struct LogicalError {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 enum RouterParams<T: Clone + Debug + Eq + Hash> {
+    Policy(policy::Params<T>),
+
     Profile(profile::Params<T>),
 
     // TODO(ver) Remove this variant when policy routes are fully wired up.
@@ -155,6 +161,7 @@ where
         S::Future: Send,
     {
         svc::layer::mk(move |concrete: N| {
+            let policy = svc::stack(concrete.clone()).push(policy::Params::layer());
             let profile =
                 svc::stack(concrete.clone()).push(profile::Params::layer(metrics.clone()));
             svc::stack(concrete)
@@ -168,10 +175,16 @@ where
                                     parent,
                                 })
                             }
-                            RouterParams::Profile(profile) => svc::Either::B(profile),
+                            RouterParams::Profile(profile) => {
+                                svc::Either::B(svc::Either::A(profile))
+                            }
+                            RouterParams::Policy(policy) => svc::Either::B(svc::Either::B(policy)),
                         })
                     },
-                    profile.into_inner(),
+                    // Switch profile and policy routing.
+                    profile
+                        .push_switch(Ok::<_, Infallible>, policy.into_inner())
+                        .into_inner(),
                 )
                 .push(svc::NewMapErr::layer_from_target::<LogicalError, _>())
                 .push_on_service(svc::MapErr::layer_boxed())
@@ -187,6 +200,7 @@ where
 {
     fn from((routes, parent): (Routes, T)) -> Self {
         match routes {
+            Routes::Policy(routes) => Self::Policy((routes, parent).into()),
             Routes::Profile(routes) => Self::Profile((routes, parent).into()),
             Routes::Endpoint(addr, metadata) => Self::Endpoint(addr, metadata, parent),
         }
@@ -199,6 +213,7 @@ where
 {
     fn param(&self) -> LogicalAddr {
         match self {
+            Self::Policy(ref p) => p.param(),
             Self::Profile(ref p) => {
                 let profile::LogicalAddr(addr) = p.param();
                 LogicalAddr(addr.into())