chore(http): split h2 client and server config

Hyper supports a variety of configuration options for for the HTTP/2 client and
servers. The proxy's http client and server types do not support modifying these
parameters. We want to support configuring HTTP/2 server and client keep-alives;
and if we are exposing these parameters to the control plane, it also makes
sense to include the others, especially the per-connection stream concurrency
limit and the initial window sizes.

This commit splits the h2 'Settings' type into 'ClientParams' and
'ServerParams', including the appropriate per-connection settings for each.
These parameters are wired into the hyper client and server builders.

These parameters are not yet exposed to the control plane via configuration
interfaces.

linkerd/app/admin/src/stack.rs

@@ -128,7 +128,7 @@ impl Config {
                 move |t: &Http| {
                     http::ServerParams {
                         version: t.version,
-                        h2: Default::default(),
+                        http2: Default::default(),
                         drain: drain.clone(),
                     }
                 }

linkerd/app/core/src/config.rs

@@ -10,16 +10,16 @@ use std::time::Duration;
 pub struct ServerConfig {
     pub addr: DualListenAddr,
     pub keepalive: Keepalive,
-    pub h2_settings: h2::Settings,
+    pub http2: h2::ServerParams,
 }
 
 #[derive(Clone, Debug)]
 pub struct ConnectConfig {
     pub backoff: ExponentialBackoff,
     pub timeout: Duration,
     pub keepalive: Keepalive,
-    pub h1_settings: h1::PoolSettings,
-    pub h2_settings: h2::Settings,
+    pub http1: h1::PoolSettings,
+    pub http2: h2::ClientParams,
 }
 
 #[derive(Clone, Debug)]

linkerd/app/core/src/control.rs

@@ -128,7 +128,7 @@ impl Config {
             .push(tls::Client::layer(identity))
             .push_connect_timeout(self.connect.timeout)
             .push_map_target(|(_version, target)| target)
-            .push(self::client::layer(self.connect.h2_settings))
+            .push(self::client::layer(self.connect.http2))
             .push_on_service(svc::MapErr::layer_boxed())
             .into_new_service();
 
@@ -330,7 +330,6 @@ mod client {
         svc, tls,
         transport::{Remote, ServerAddr},
     };
-    use linkerd_proxy_http::h2::Settings as H2Settings;
     use std::net::SocketAddr;
 
     #[derive(Clone, Hash, Debug, Eq, PartialEq)]
@@ -368,11 +367,11 @@ mod client {
     // === impl Layer ===
 
     pub fn layer<C, B>(
-        settings: H2Settings,
-    ) -> impl svc::Layer<C, Service = http::h2::Connect<C, B>> + Copy
+        h2: http::h2::ClientParams,
+    ) -> impl svc::Layer<C, Service = http::h2::Connect<C, B>> + Clone
     where
         http::h2::Connect<C, B>: tower::Service<Target>,
     {
-        svc::layer::mk(move |mk_conn| http::h2::Connect::new(mk_conn, settings))
+        svc::layer::mk(move |mk_conn| http::h2::Connect::new(mk_conn, h2.clone()))
     }
 }

linkerd/app/inbound/src/http/router.rs

@@ -94,8 +94,8 @@ impl<C> Inbound<C> {
                 .check_service::<Http>()
                 .push_map_target(|(_version, target)| target)
                 .push(http::client::layer(
-                    config.proxy.connect.h1_settings,
-                    config.proxy.connect.h2_settings,
+                    config.proxy.connect.http1,
+                    config.proxy.connect.http2.clone(),
                 ))
                 .check_service::<Http>()
                 .push_on_service(svc::MapErr::layer_boxed())

linkerd/app/inbound/src/http/server.rs

@@ -112,15 +112,15 @@ impl<H> Inbound<H> {
         HSvc::Future: Send,
     {
         self.map_stack(|config, rt, http| {
-            let h2 = config.proxy.server.h2_settings;
+            let h2 = config.proxy.server.http2.clone();
             let drain = rt.drain.clone();
 
             http.check_new_service::<T, http::Request<http::BoxBody>>()
                 .unlift_new()
                 .check_new_new_service::<T, http::ClientHandle, http::Request<_>>()
                 .push(http::NewServeHttp::layer(move |t: &T| http::ServerParams {
                     version: t.param(),
-                    h2,
+                    http2: h2.clone(),
                     drain: drain.clone(),
                 }))
                 .check_new_service::<T, I>()

linkerd/app/inbound/src/test_util.rs

@@ -59,7 +59,7 @@ pub fn default_config() -> Config {
             server: config::ServerConfig {
                 addr: DualListenAddr(([0, 0, 0, 0], 0).into(), None),
                 keepalive: Keepalive(None),
-                h2_settings: h2::Settings::default(),
+                http2: h2::ServerParams::default(),
             },
             connect: config::ConnectConfig {
                 keepalive: Keepalive(None),
@@ -70,11 +70,11 @@ pub fn default_config() -> Config {
                     0.1,
                 )
                 .unwrap(),
-                h1_settings: h1::PoolSettings {
+                http1: h1::PoolSettings {
                     max_idle: 1,
                     idle_timeout: Duration::from_secs(1),
                 },
-                h2_settings: h2::Settings::default(),
+                http2: h2::ClientParams::default(),
             },
             max_in_flight_requests: 10_000,
             detect_protocol_timeout: Duration::from_secs(10),

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

@@ -54,9 +54,7 @@ impl<C> Outbound<C> {
     {
         self.map_stack(|config, _, inner| {
             let config::ConnectConfig {
-                h1_settings,
-                h2_settings,
-                ..
+                http1, ref http2, ..
             } = config.proxy.connect;
 
             // Initiates an HTTP client on the underlying transport. Prior-knowledge HTTP/2
@@ -65,7 +63,7 @@ impl<C> Outbound<C> {
             svc::stack(inner.into_inner().into_service())
                 .check_service::<Connect<T>>()
                 .push_map_target(|(version, inner)| Connect { version, inner })
-                .push(http::client::layer(h1_settings, h2_settings))
+                .push(http::client::layer(http1, http2.clone()))
                 .push_on_service(svc::MapErr::layer_boxed())
                 .check_service::<T>()
                 .into_new_service()

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

@@ -9,7 +9,7 @@ pub(crate) struct ServerRescue {
 
 #[derive(Clone, Debug)]
 pub struct ExtractServerParams {
-    h2: http::h2::Settings,
+    h2: http::h2::ServerParams,
     drain: drain::Watch,
 }
 
@@ -86,7 +86,7 @@ impl<N> Outbound<N> {
             http.unlift_new()
                 .push(svc::ArcNewService::layer())
                 .push(http::NewServeHttp::layer(ExtractServerParams {
-                    h2: config.proxy.server.h2_settings,
+                    h2: config.proxy.server.http2.clone(),
                     drain: rt.drain.clone(),
                 }))
         })
@@ -194,7 +194,7 @@ where
     fn extract_param(&self, t: &T) -> http::ServerParams {
         http::ServerParams {
             version: t.param(),
-            h2: self.h2,
+            http2: self.h2.clone(),
             drain: self.drain.clone(),
         }
     }

linkerd/app/outbound/src/ingress.rs

@@ -301,11 +301,11 @@ impl<N> Outbound<N> {
             http.check_new_service::<Http<T>, http::Request<_>>()
                 .unlift_new()
                 .push(http::NewServeHttp::layer({
-                    let h2 = config.proxy.server.h2_settings;
+                    let h2 = config.proxy.server.http2.clone();
                     let drain = rt.drain.clone();
                     move |http: &Http<T>| http::ServerParams {
                             version: http.version,
-                            h2,
+                            http2: h2.clone(),
                             drain: drain.clone()
                     }
                 }))

linkerd/app/outbound/src/protocol.rs

@@ -47,13 +47,13 @@ impl<N> Outbound<N> {
         let opaq = self.clone().into_stack();
 
         let http = self.with_stack(http).map_stack(|config, rt, stk| {
-            let h2 = config.proxy.server.h2_settings;
+            let h2 = config.proxy.server.http2.clone();
             let drain = rt.drain.clone();
             stk.unlift_new()
                 .push(http::NewServeHttp::layer(move |t: &Http<T>| {
                     http::ServerParams {
                         version: t.version,
-                        h2,
+                        http2: h2.clone(),
                         drain: drain.clone(),
                     }
                 }))

linkerd/app/outbound/src/test_util.rs

@@ -26,7 +26,7 @@ pub(crate) fn default_config() -> Config {
             server: config::ServerConfig {
                 addr: DualListenAddr(([0, 0, 0, 0], 0).into(), None),
                 keepalive: Keepalive(None),
-                h2_settings: h2::Settings::default(),
+                http2: h2::ServerParams::default(),
             },
             connect: config::ConnectConfig {
                 keepalive: Keepalive(None),
@@ -37,11 +37,11 @@ pub(crate) fn default_config() -> Config {
                     0.1,
                 )
                 .unwrap(),
-                h1_settings: h1::PoolSettings {
+                http1: h1::PoolSettings {
                     max_idle: 1,
                     idle_timeout: Duration::from_secs(1),
                 },
-                h2_settings: h2::Settings::default(),
+                http2: h2::ClientParams::default(),
             },
             max_in_flight_requests: 10_000,
             detect_protocol_timeout: Duration::from_secs(3),

linkerd/app/src/env.rs

@@ -425,21 +425,13 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
     );
     let dst_profile_networks = parse(strings, ENV_DESTINATION_PROFILE_NETWORKS, parse_networks);
 
-    let initial_stream_window_size = parse(strings, ENV_INITIAL_STREAM_WINDOW_SIZE, parse_number);
-    let initial_connection_window_size =
-        parse(strings, ENV_INITIAL_CONNECTION_WINDOW_SIZE, parse_number);
-
     let tap = parse_tap_config(strings);
 
-    let h2_settings = h2::Settings {
-        initial_stream_window_size: Some(
-            initial_stream_window_size?.unwrap_or(DEFAULT_INITIAL_STREAM_WINDOW_SIZE),
-        ),
-        initial_connection_window_size: Some(
-            initial_connection_window_size?.unwrap_or(DEFAULT_INITIAL_CONNECTION_WINDOW_SIZE),
-        ),
-        ..Default::default()
-    };
+    let initial_stream_window_size = parse(strings, ENV_INITIAL_STREAM_WINDOW_SIZE, parse_number)?
+        .unwrap_or(DEFAULT_INITIAL_STREAM_WINDOW_SIZE);
+    let initial_connection_window_size =
+        parse(strings, ENV_INITIAL_CONNECTION_WINDOW_SIZE, parse_number)?
+            .unwrap_or(DEFAULT_INITIAL_CONNECTION_WINDOW_SIZE);
 
     let dst_profile_suffixes = dst_profile_suffixes?
         .unwrap_or_else(|| parse_dns_suffixes(DEFAULT_DESTINATION_PROFILE_SUFFIXES).unwrap());
@@ -481,7 +473,13 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
         let server = ServerConfig {
             addr,
             keepalive,
-            h2_settings,
+            http2: h2::ServerParams {
+                flow_control: Some(h2::FlowControl::Fixed {
+                    initial_stream_window_size,
+                    initial_connection_window_size,
+                }),
+                ..Default::default()
+            },
         };
         let discovery_idle_timeout =
             outbound_discovery_idle_timeout?.unwrap_or(DEFAULT_OUTBOUND_DISCOVERY_IDLE_TIMEOUT);
@@ -502,8 +500,14 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
                 OUTBOUND_CONNECT_BASE,
                 DEFAULT_OUTBOUND_CONNECT_BACKOFF,
             )?,
-            h2_settings,
-            h1_settings: h1::PoolSettings {
+            http2: h2::ClientParams {
+                flow_control: Some(h2::FlowControl::Fixed {
+                    initial_stream_window_size,
+                    initial_connection_window_size,
+                }),
+                ..Default::default()
+            },
+            http1: h1::PoolSettings {
                 max_idle,
                 idle_timeout: connection_pool_timeout
                     .unwrap_or(DEFAULT_OUTBOUND_HTTP1_CONNECTION_POOL_IDLE_TIMEOUT),
@@ -563,7 +567,13 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
         let server = ServerConfig {
             addr,
             keepalive,
-            h2_settings,
+            http2: h2::ServerParams {
+                flow_control: Some(h2::FlowControl::Fixed {
+                    initial_stream_window_size,
+                    initial_connection_window_size,
+                }),
+                ..Default::default()
+            },
         };
         let discovery_idle_timeout =
             inbound_discovery_idle_timeout?.unwrap_or(DEFAULT_INBOUND_DISCOVERY_IDLE_TIMEOUT);
@@ -584,8 +594,14 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
                 INBOUND_CONNECT_BASE,
                 DEFAULT_INBOUND_CONNECT_BACKOFF,
             )?,
-            h2_settings,
-            h1_settings: h1::PoolSettings {
+            http2: h2::ClientParams {
+                flow_control: Some(h2::FlowControl::Fixed {
+                    initial_stream_window_size,
+                    initial_connection_window_size,
+                }),
+                ..Default::default()
+            },
+            http1: h1::PoolSettings {
                 max_idle,
                 idle_timeout: connection_pool_timeout,
             },
@@ -758,7 +774,13 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
         server: ServerConfig {
             addr: DualListenAddr(admin_listener_addr, None),
             keepalive: inbound.proxy.server.keepalive,
-            h2_settings,
+            http2: h2::ServerParams {
+                flow_control: Some(h2::FlowControl::Fixed {
+                    initial_stream_window_size,
+                    initial_connection_window_size,
+                }),
+                ..Default::default()
+            },
         },
 
         // TODO(ver) Currently we always enable profiling when the pprof feature
@@ -817,7 +839,13 @@ pub fn parse_config<S: Strings>(strings: &S) -> Result<super::Config, EnvError>
             config: ServerConfig {
                 addr: DualListenAddr(addr, None),
                 keepalive: inbound.proxy.server.keepalive,
-                h2_settings,
+                http2: h2::ServerParams {
+                    flow_control: Some(h2::FlowControl::Fixed {
+                        initial_stream_window_size,
+                        initial_connection_window_size,
+                    }),
+                    ..Default::default()
+                },
             },
         })
         .unwrap_or(super::tap::Config::Disabled);

linkerd/app/src/tap.rs

@@ -13,6 +13,7 @@ use std::{collections::HashSet, pin::Pin};
 use tower::util::{service_fn, ServiceExt};
 
 #[derive(Clone, Debug)]
+#[allow(clippy::large_enum_variant)]
 pub enum Config {
     Disabled,
     Enabled {