inbound: determine default policies using the opaque ports env var (#2395)

The proxy injector populates an environment variable,
`LINKERD2_PROXY_INBOUND_PORTS_DISABLE_PROTOCOL_DETECTION`, with a list
of all ports marked as opaque. Currently, however, _the proxy _does not
actually use this environment variable_. Instead, opaque ports are
discovered from the policy controller. The opaque ports environment
variable was used only when running in the "fixed" inbound policy mode,
where all inbound policies are determined from environment variables,
and no policy controller address is provided. This mode is no longer
supported, and the policy controller address is now required, so the
`LINKERD2_PROXY_INBOUND_PORTS_DISABLE_PROTOCOL_DETECTION` environment
variable is not currently used to discover inbound opaque ports.

There are two issues with the current state of things. One is that
inbound policy discovery is _non-blocking_: when an inbound proxy
receives a connection on a port that it has not previously discovered a
policy for, it uses the default policy until it has successfully
discovered a policy for that port from the policy controller. This means
that the proxy may perform protocol detection on the first connection to
an opaque port. This isn't great, as it may result in a protocol
detection timeout error on a port that the user had previously marked as
opaque. It would be preferable for the proxy to read the environment
variable, and use it to determine whether the default policy for a port
is opaque, so that ports marked as opaque disable protocol detection
even before the "actual" policy is discovered.

The other issue with the
`LINKERD2_PROXY_INBOUND_PORTS_DISABLE_PROTOCOL_DETECTION` environment
variable is that it is currently a list of _individual port numbers_,
while the proxy injector can accept annotations that specify _ranges_ of
opaque ports. This means that when a very large number of ports are
marked as opaque, the proxy manifest must contain a list of each
individual port number in those ranges, making it potentially quite
large. See linkerd/linkerd2#9803 for details on this issue.

This branch addresses both of these problems. The proxy is changed so
that it will once again read the
`LINKERD2_PROXY_INBOUND_PORTS_DISABLE_PROTOCOL_DETECTION` environment
variable, and use it to determine which ports should have opaque
policies by default. The parsing of the environment variable is changed
to support specifying ports as a list of ranges, rather than a list of
individual port numbers. Along with a proxy-injector change, this would
resolve the manifest size issue described in linkerd/linkerd2#9803.

This is implemented by changing the `inbound::policy::Store` type to
also include a set of port ranges that are marked as opaque. When the
`Store` handles a `get_policy` call for a port that is not already in
the cache, it starts a control plane watch for that port just as it did
previously. However, when determining the initial _default_ value for
the policy, before the control plane discovery provides one, it checks
whether the port is in a range that is marked as opaque, and, if it is,
uses an opaque default policy instead.

This approach was chosen rather than pre-populating the `Store` with
policies for all opaque ports to better handle the case where very large
ranges are marked as opaque and are used infrequently. If the `Store`
was pre-populated with default policies for all such ports, it would
essentially behave as though all ports in
`LINKERD2_PROXY_INBOUND_PORTS_DISABLE_PROTOCOL_DETECTION` were also in
`LINKERD2_PROXY_INBOUND_PORTS`, and the proxy would immediately start a
policy controller discovery watch for all opaque ports, which would be
kept open for the proxy's entire lifetime. In cases where the opaque
ports ranges include ~10,000s of ports, this causes significant
unnecessary load on the policy controller. Storing opaque port ranges
separately and using them to determine the default policy as needed
allows opaque port policies to be treated the same as non-default ports,
which are discovered as needed and can be evicted from the cache if they
are unused. If a port is in both
`LINKERD2_PROXY_INBOUND_PORTS_DISABLE_PROTOCOL_DETECTION` *and*
`LINKERD2_PROXY_INBOUND_PORTS`, the proxy will start discovery eagerly
and retain the port in the cache forever, but the default policy will be
opaque.

I've also added a test for the behavior of opaque ports where the port's
policy has not been discovered from the policy controller. That test
fails on `main`, as the proxy attempts protocol detection, but passes on
this branch.

In addition, I changed the parsing of the `LINKERD2_PROXY_INBOUND_PORTS`
environment variable to also accept ranges, because it seemed like a
nice thing to do while I was here. :)

Author: hawkw

Cargo.lock

@@ -849,6 +849,7 @@ dependencies = [
  "linkerd-app-outbound",
  "linkerd-error",
  "linkerd-opencensus",
+ "rangemap",
  "regex",
  "thiserror",
  "tokio",
@@ -985,6 +986,7 @@ dependencies = [
  "linkerd2-proxy-api",
  "once_cell",
  "parking_lot",
+ "rangemap",
  "thiserror",
  "tokio",
  "tokio-test",
@@ -2300,6 +2302,12 @@ dependencies = [
  "getrandom",
 ]
 
+[[package]]
+name = "rangemap"
+version = "1.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b9283c6b06096b47afc7109834fdedab891175bb5241ee5d4f7d2546549f263"
+
 [[package]]
 name = "redox_syscall"
 version = "0.2.15"

linkerd/app/Cargo.toml

@@ -24,6 +24,7 @@ linkerd-app-inbound = { path = "./inbound" }
 linkerd-app-outbound = { path = "./outbound" }
 linkerd-error = { path = "../error" }
 linkerd-opencensus = { path = "../opencensus" }
+rangemap = "1"
 regex = "1"
 thiserror = "1"
 tokio = { version = "1", features = ["rt"] }

linkerd/app/inbound/Cargo.toml

@@ -32,6 +32,7 @@ linkerd-tonic-watch = { path = "../../tonic-watch" }
 linkerd2-proxy-api = { version = "0.9", features = ["inbound"] }
 once_cell = "1"
 parking_lot = "0.12"
+rangemap = "1"
 thiserror = "1"
 tokio = { version = "1", features = ["sync"] }
 tonic = { version = "0.8", default-features = false }

linkerd/app/inbound/src/policy/config.rs

@@ -1,5 +1,6 @@
 use super::{api::Api, DefaultPolicy, GetPolicy, Protocol, ServerPolicy, Store};
 use linkerd_app_core::{exp_backoff::ExponentialBackoff, proxy::http, Error};
+use rangemap::RangeInclusiveSet;
 use std::{
     collections::{HashMap, HashSet},
     sync::Arc,
@@ -17,11 +18,13 @@ pub enum Config {
         default: DefaultPolicy,
         cache_max_idle_age: Duration,
         ports: HashSet<u16>,
+        opaque_ports: RangeInclusiveSet<u16>,
     },
     Fixed {
         default: DefaultPolicy,
         cache_max_idle_age: Duration,
         ports: HashMap<u16, ServerPolicy>,
+        opaque_ports: RangeInclusiveSet<u16>,
     },
 }
 
@@ -46,12 +49,14 @@ impl Config {
                 default,
                 ports,
                 cache_max_idle_age,
-            } => Store::spawn_fixed(default, cache_max_idle_age, ports),
+                opaque_ports,
+            } => Store::spawn_fixed(default, cache_max_idle_age, ports, opaque_ports),
 
             Self::Discover {
                 default,
                 ports,
                 cache_max_idle_age,
+                opaque_ports,
             } => {
                 let watch = {
                     let detect_timeout = match default {
@@ -63,7 +68,7 @@ impl Config {
                     };
                     Api::new(workload, detect_timeout, client).into_watch(backoff)
                 };
-                Store::spawn_discover(default, cache_max_idle_age, watch, ports)
+                Store::spawn_discover(default, cache_max_idle_age, watch, ports, opaque_ports)
             }
         }
     }

linkerd/app/inbound/src/policy/store.rs

@@ -4,9 +4,11 @@ use linkerd_idle_cache::IdleCache;
 pub use linkerd_proxy_server_policy::{
     authz::Suffix, Authentication, Authorization, Protocol, ServerPolicy,
 };
+use rangemap::RangeInclusiveSet;
 use std::{
     collections::HashSet,
     hash::{BuildHasherDefault, Hasher},
+    sync::Arc,
 };
 use tokio::{sync::watch, time::Duration};
 use tracing::info_span;
@@ -15,6 +17,8 @@ use tracing::info_span;
 pub struct Store<S> {
     cache: IdleCache<u16, Rx, BuildHasherDefault<PortHasher>>,
     default_rx: Rx,
+    opaque_ports: Arc<RangeInclusiveSet<u16>>,
+    opaque_default_rx: Rx,
     discover: Option<api::Watch<S>>,
 }
 
@@ -34,22 +38,32 @@ impl<S> Store<S> {
         default: DefaultPolicy,
         idle_timeout: Duration,
         ports: impl IntoIterator<Item = (u16, ServerPolicy)>,
+        opaque_ports: RangeInclusiveSet<u16>,
     ) -> Self {
+        let opaque_default_rx = Self::spawn_default(Self::make_opaque(default.clone()));
         let cache = {
+            let opaque_rxs = opaque_ports.iter().flat_map(|range| {
+                range
+                    .clone()
+                    .into_iter()
+                    .map(|port| (port, opaque_default_rx.clone()))
+            });
             let rxs = ports.into_iter().map(|(p, s)| {
                 // When using a fixed policy, we don't need to watch for changes. It's
                 // safe to discard the sender, as the receiver will continue to let us
                 // borrow/clone each fixed policy.
                 let (_, rx) = watch::channel(s);
                 (p, rx)
             });
-            IdleCache::with_permanent_from_iter(idle_timeout, rxs)
+            IdleCache::with_permanent_from_iter(idle_timeout, opaque_rxs.chain(rxs))
         };
 
         Self {
             cache,
             discover: None,
             default_rx: Self::spawn_default(default),
+            opaque_ports: Arc::new(opaque_ports),
+            opaque_default_rx,
         }
     }
 
@@ -62,6 +76,7 @@ impl<S> Store<S> {
         idle_timeout: Duration,
         discover: api::Watch<S>,
         ports: HashSet<u16>,
+        opaque_ports: RangeInclusiveSet<u16>,
     ) -> Self
     where
         S: tonic::client::GrpcService<tonic::body::BoxBody, Error = Error>,
@@ -70,6 +85,7 @@ impl<S> Store<S> {
         S::ResponseBody:
             http::HttpBody<Data = tonic::codegen::Bytes, Error = Error> + Default + Send + 'static,
     {
+        let opaque_default = Self::make_opaque(default.clone());
         // The initial set of policies never expire from the cache.
         //
         // Policies that are dynamically discovered at runtime will expire after
@@ -78,7 +94,11 @@ impl<S> Store<S> {
         let cache = {
             let rxs = ports.into_iter().map(|port| {
                 let discover = discover.clone();
-                let default = default.clone();
+                let default = if opaque_ports.contains(&port) {
+                    opaque_default.clone()
+                } else {
+                    default.clone()
+                };
                 let rx = info_span!("watch", port)
                     .in_scope(|| discover.spawn_with_init(port, default.into()));
                 (port, rx)
@@ -90,6 +110,24 @@ impl<S> Store<S> {
             cache,
             discover: Some(discover),
             default_rx: Self::spawn_default(default),
+            opaque_ports: Arc::new(opaque_ports),
+            opaque_default_rx: Self::spawn_default(opaque_default),
+        }
+    }
+
+    fn make_opaque(default: DefaultPolicy) -> DefaultPolicy {
+        match default {
+            DefaultPolicy::Allow(mut policy) => {
+                policy.protocol = match policy.protocol {
+                    Protocol::Detect { tcp_authorizations, .. } => {
+                        Protocol::Opaque(tcp_authorizations)
+                    }
+                    opaq @ Protocol::Opaque(_) => opaq,
+                    _ => unreachable!("default policy must have been configured to detect prior to marking it opaque"),
+                };
+                DefaultPolicy::Allow(policy)
+            }
+            DefaultPolicy::Deny => DefaultPolicy::Deny,
         }
     }
 
@@ -113,15 +151,23 @@ where
         http::HttpBody<Data = tonic::codegen::Bytes, Error = Error> + Default + Send + 'static,
 {
     fn get_policy(&self, dst: OrigDstAddr) -> AllowPolicy {
-        // Lookup the polcify for the target port in the cache. If it doesn't
+        // Lookup the policy for the target port in the cache. If it doesn't
         // already exist, we spawn a watch on the API (if it is configured). If
         // no discovery API is configured we use the default policy.
         let server =
             self.cache
                 .get_or_insert_with(dst.port(), |port| match self.discover.clone() {
                     Some(disco) => info_span!("watch", port).in_scope(|| {
-                        tracing::trace!(%port, "spawning policy discovery");
-                        disco.spawn_with_init(*port, self.default_rx.borrow().clone())
+                        let is_default_opaque = self.opaque_ports.contains(port);
+                        tracing::trace!(%port, is_default_opaque, "spawning policy discovery");
+                        // If the port is in the range of ports marked as
+                        // opaque, use the opaque default policy instead.
+                        let init = if is_default_opaque {
+                            self.opaque_default_rx.borrow().clone()
+                        } else {
+                            self.default_rx.borrow().clone()
+                        };
+                        disco.spawn_with_init(*port, init)
                     }),
 
                     // If no discovery API is configured, then we use the

linkerd/app/inbound/src/policy/tcp/tests.rs

@@ -272,6 +272,11 @@ impl Store<MockSvc> {
         default: impl Into<DefaultPolicy>,
         ports: impl IntoIterator<Item = (u16, ServerPolicy)>,
     ) -> Self {
-        Self::spawn_fixed(default.into(), std::time::Duration::MAX, ports)
+        Self::spawn_fixed(
+            default.into(),
+            std::time::Duration::MAX,
+            ports,
+            Default::default(),
+        )
     }
 }

linkerd/app/inbound/src/test_util.rs

@@ -49,6 +49,7 @@ pub fn default_config() -> Config {
         }
         .into(),
         ports: Default::default(),
+        opaque_ports: Default::default(),
     };
 
     Config {

linkerd/app/integration/src/tests/transparency.rs

@@ -225,6 +225,80 @@ async fn inbound_tcp_server_first() {
     test_inbound_server_speaks_first(TestEnv::default()).await;
 }
 
+/// Like `tcp_server_first`, but the opaque port configuration is not discovered
+/// from the policy controller before accepting the connection (i.e. the port is
+/// *not* in `LINKERD2_PROXY_INBOUND_PORTS`).
+#[tokio::test]
+async fn inbound_tcp_server_first_no_discovery() {
+    const TIMEOUT: Duration = Duration::from_secs(5);
+
+    let _trace = trace_init();
+
+    let (tx, rx) = mpsc::channel(1);
+    let srv = server::tcp()
+        .accept_fut(move |sock| serve_server_first(sock, tx))
+        .run()
+        .await;
+
+    let mut env = TestEnv::default();
+    env.put(
+        app::env::ENV_INBOUND_PORTS_DISABLE_PROTOCOL_DETECTION,
+        srv.addr.port().to_string(),
+    );
+
+    let proxy = proxy::new().inbound(srv).run_with_test_env(env).await;
+
+    server_first_client(proxy.inbound, rx).await;
+
+    // ensure panics from the server are propagated
+    proxy.join_servers().await;
+}
+
+// FIXME(ver) this test doesn't actually test TLS functionality.
+#[ignore]
+#[tokio::test]
+async fn inbound_tcp_server_first_tls() {
+    use std::path::PathBuf;
+
+    let (_cert, _key, _trust_anchors) = {
+        let path_to_string = |path: &PathBuf| {
+            path.as_path()
+                .to_owned()
+                .into_os_string()
+                .into_string()
+                .unwrap()
+        };
+        let mut tls = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+        tls.push("src");
+        tls.push("transport");
+        tls.push("tls");
+        tls.push("testdata");
+
+        tls.push("foo-ns1-ca1.crt");
+        let cert = path_to_string(&tls);
+
+        tls.set_file_name("foo-ns1-ca1.p8");
+        let key = path_to_string(&tls);
+
+        tls.set_file_name("ca1.pem");
+        let trust_anchors = path_to_string(&tls);
+        (cert, key, trust_anchors)
+    };
+
+    let env = TestEnv::default();
+
+    // FIXME
+    //env.put(app::env::ENV_TLS_CERT, cert);
+    //env.put(app::env::ENV_TLS_PRIVATE_KEY, key);
+    //env.put(app::env::ENV_TLS_TRUST_ANCHORS, trust_anchors);
+    //env.put(
+    //    app::env::ENV_TLS_LOCAL_IDENTITY,
+    //    "foo.deployment.ns1.linkerd-managed.linkerd.svc.cluster.local".to_string(),
+    //);
+
+    test_inbound_server_speaks_first(env).await
+}
+
 /// Tests that the outbound proxy does not attempt to perform protocol detection
 /// when the policy controller returns an `OutboundPolicy` indicating that the
 /// destination is opaque.
@@ -304,52 +378,6 @@ async fn server_first_client(addr: SocketAddr, mut rx: mpsc::Receiver<()>) {
     // TCP client must close first
     tcp_client.shutdown().await;
 }
-
-// FIXME(ver) this test doesn't actually test TLS functionality.
-#[ignore]
-#[tokio::test]
-async fn inbound_tcp_server_first_tls() {
-    use std::path::PathBuf;
-
-    let (_cert, _key, _trust_anchors) = {
-        let path_to_string = |path: &PathBuf| {
-            path.as_path()
-                .to_owned()
-                .into_os_string()
-                .into_string()
-                .unwrap()
-        };
-        let mut tls = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
-        tls.push("src");
-        tls.push("transport");
-        tls.push("tls");
-        tls.push("testdata");
-
-        tls.push("foo-ns1-ca1.crt");
-        let cert = path_to_string(&tls);
-
-        tls.set_file_name("foo-ns1-ca1.p8");
-        let key = path_to_string(&tls);
-
-        tls.set_file_name("ca1.pem");
-        let trust_anchors = path_to_string(&tls);
-        (cert, key, trust_anchors)
-    };
-
-    let env = TestEnv::default();
-
-    // FIXME
-    //env.put(app::env::ENV_TLS_CERT, cert);
-    //env.put(app::env::ENV_TLS_PRIVATE_KEY, key);
-    //env.put(app::env::ENV_TLS_TRUST_ANCHORS, trust_anchors);
-    //env.put(
-    //    app::env::ENV_TLS_LOCAL_IDENTITY,
-    //    "foo.deployment.ns1.linkerd-managed.linkerd.svc.cluster.local".to_string(),
-    //);
-
-    test_inbound_server_speaks_first(env).await
-}
-
 #[tokio::test]
 async fn tcp_connections_close_if_client_closes() {
     let _trace = trace_init();