enable connection pooling and reuse for outbound wasi-http (#3229)

* enable connection pooling and reuse for outbound wasi-http

This updates `factor-outbound-http/wasi.rs` to use
`hyper_util::client::legacy::Client`, which does connection pooling by default.
`Client` requires a `tower_service::Service` implementation for creating new
connections, which in turn required moving code around to accomodate a different
flow of control.  It also forces us to work at a slightly higher level of
abstraction, so we don't have as much fine-grained control as before, notably:

- We can't make HTTP/2-specific request tweaks like we did before (e.g. removing
  or adding `host` headers, or tweaking the path-with-query).  I'm mainly
  leaving it up to `hyper-util` to do the right thing here :crossed_fingers:.

- We can't provide an `AbortOnDropJoinHandle` when constructing an
  `IncomingResponse` object, which may mean the connection continues to exist
  after the response is dropped, but that's kind of the point of connection
  pooling anyway.

Open questions: do we want to make this configurable, and if so, should it be
enabled or disabled by default?

Signed-off-by: Joel Dice <[email protected]>

* make outbound HTTP connection pooling runtime-configurable

It's enabled by default, but can be disabled via a runtime config file
containing:

```toml
[outbound_http]
connection_pooling = false
```

Signed-off-by: Joel Dice <[email protected]>

* use pinned Rust version for `all-integration-tests`

This will hopefully fix CI :crossed_fingers:

Signed-off-by: Joel Dice <[email protected]>

---------

Signed-off-by: Joel Dice <[email protected]>

Author: dicej

.github/workflows/build.yml

@@ -194,6 +194,11 @@ jobs:
     steps:
       - uses: actions/checkout@v3
 
+      - name: setup dependencies
+        uses: ./.github/actions/spin-ci-dependencies
+        with:
+          rust: true
+
       # Install all the toolchain dependencies
       - name: Install Rust wasm target
         run: rustup target add wasm32-wasip1 wasm32-unknown-unknown

Cargo.lock

@@ -164,6 +164,7 @@ tokio = "1"
 tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "tls12"] }
 toml = "0.8"
 toml_edit = "0.22"
+tower-service = "0.3.3"
 tracing = { version = "0.1.41", features = ["log"] }
 url = "2"
 walkdir = "2"
@@ -186,4 +187,4 @@ blocks_in_conditions = "allow"
 
 [[bin]]
 name = "spin"
-path = "src/bin/spin.rs"
+path = "src/bin/spin.rs"

Cargo.toml

@@ -13,12 +13,14 @@ hyper = { workspace = true }
 hyper-util = { workspace = true }
 reqwest = { workspace = true, features = ["gzip"] }
 rustls = { workspace = true }
+serde = { workspace = true }
 spin-factor-outbound-networking = { path = "../factor-outbound-networking" }
 spin-factors = { path = "../factors" }
 spin-telemetry = { path = "../telemetry" }
 spin-world = { path = "../world" }
 tokio = { workspace = true, features = ["macros", "rt", "net"] }
 tokio-rustls = { workspace = true }
+tower-service = { workspace = true }
 tracing = { workspace = true }
 wasmtime = { workspace = true }
 wasmtime-wasi = { workspace = true }
@@ -28,5 +30,10 @@ wasmtime-wasi-http = { workspace = true }
 spin-factor-variables = { path = "../factor-variables" }
 spin-factors-test = { path = "../factors-test" }
 
+[features]
+default = ["spin-cli"]
+# Includes the runtime configuration handling used by the Spin CLI
+spin-cli = []
+
 [lints]
 workspace = true

crates/factor-outbound-http/Cargo.toml

@@ -1,4 +1,5 @@
 pub mod intercept;
+pub mod runtime_config;
 mod spin;
 mod wasi;
 pub mod wasi_2023_10_18;
@@ -12,6 +13,7 @@ use http::{
     HeaderValue, Uri,
 };
 use intercept::OutboundHttpInterceptor;
+use runtime_config::RuntimeConfig;
 use spin_factor_outbound_networking::{
     config::{allowed_hosts::OutboundAllowedHosts, blocked_networks::BlockedNetworks},
     ComponentTlsClientConfigs, OutboundNetworkingFactor,
@@ -34,8 +36,8 @@ pub struct OutboundHttpFactor {
 }
 
 impl Factor for OutboundHttpFactor {
-    type RuntimeConfig = ();
-    type AppState = ();
+    type RuntimeConfig = RuntimeConfig;
+    type AppState = AppState;
     type InstanceBuilder = InstanceState;
 
     fn init(&mut self, ctx: &mut impl spin_factors::InitContext<Self>) -> anyhow::Result<()> {
@@ -46,9 +48,14 @@ impl Factor for OutboundHttpFactor {
 
     fn configure_app<T: RuntimeFactors>(
         &self,
-        _ctx: ConfigureAppContext<T, Self>,
+        mut ctx: ConfigureAppContext<T, Self>,
     ) -> anyhow::Result<Self::AppState> {
-        Ok(())
+        Ok(AppState {
+            connection_pooling: ctx
+                .take_runtime_config()
+                .unwrap_or_default()
+                .connection_pooling,
+        })
     }
 
     fn prepare<T: RuntimeFactors>(
@@ -67,6 +74,8 @@ impl Factor for OutboundHttpFactor {
             self_request_origin: None,
             request_interceptor: None,
             spin_http_client: None,
+            wasi_http_clients: None,
+            connection_pooling: ctx.app_state().connection_pooling,
         })
     }
 }
@@ -80,6 +89,9 @@ pub struct InstanceState {
     request_interceptor: Option<Arc<dyn OutboundHttpInterceptor>>,
     // Connection-pooling client for 'fermyon:spin/http' interface
     spin_http_client: Option<reqwest::Client>,
+    // Connection pooling client for `wasi:http/outgoing-handler` interface
+    wasi_http_clients: Option<wasi::HttpClients>,
+    connection_pooling: bool,
 }
 
 impl InstanceState {
@@ -157,3 +169,7 @@ impl std::fmt::Display for SelfRequestOrigin {
         write!(f, "{}://{}", self.scheme, self.authority)
     }
 }
+
+pub struct AppState {
+    connection_pooling: bool,
+}

crates/factor-outbound-http/src/lib.rs

@@ -0,0 +1,17 @@
+#[cfg(feature = "spin-cli")]
+pub mod spin;
+
+/// Runtime configuration for outbound HTTP.
+#[derive(Debug)]
+pub struct RuntimeConfig {
+    /// If true, enable connection pooling and reuse.
+    pub connection_pooling: bool,
+}
+
+impl Default for RuntimeConfig {
+    fn default() -> Self {
+        Self {
+            connection_pooling: true,
+        }
+    }
+}

crates/factor-outbound-http/src/runtime_config.rs

@@ -0,0 +1,31 @@
+use serde::Deserialize;
+use spin_factors::runtime_config::toml::GetTomlValue;
+
+/// Get the runtime configuration for outbound HTTP from a TOML table.
+///
+/// Expects table to be in the format:
+/// ```toml
+/// [outbound_http]
+/// connection_pooling = true
+/// ```
+pub fn config_from_table(
+    table: &impl GetTomlValue,
+) -> anyhow::Result<Option<super::RuntimeConfig>> {
+    if let Some(outbound_http) = table.get("outbound_http") {
+        Ok(Some(super::RuntimeConfig {
+            connection_pooling: outbound_http
+                .clone()
+                .try_into::<OutboundHttpToml>()?
+                .connection_pooling,
+        }))
+    } else {
+        Ok(None)
+    }
+}
+
+#[derive(Debug, Default, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct OutboundHttpToml {
+    #[serde(default)]
+    connection_pooling: bool,
+}

crates/factor-outbound-http/src/runtime_config/spin.rs

@@ -90,7 +90,13 @@ impl spin_http::Host for crate::InstanceState {
 
         // Allow reuse of Client's internal connection pool for multiple requests
         // in a single component execution
-        let client = self.spin_http_client.get_or_insert_with(Default::default);
+        let client = self.spin_http_client.get_or_insert_with(|| {
+            let mut builder = reqwest::Client::builder();
+            if !self.connection_pooling {
+                builder = builder.pool_max_idle_per_host(0);
+            }
+            builder.build().unwrap()
+        });
 
         let resp = client.execute(req).await.map_err(log_reqwest_error)?;