diff --git a/rust/crates/adc-backend-apisix-standalone/src/backend.rs b/rust/crates/adc-backend-apisix-standalone/src/backend.rs index 8de7e000..b42dd29c 100644 --- a/rust/crates/adc-backend-apisix-standalone/src/backend.rs +++ b/rust/crates/adc-backend-apisix-standalone/src/backend.rs @@ -49,6 +49,9 @@ pub struct BackendOptions { pub bypass_cache: bool, pub timeout: Option, pub tls: TlsConfig, + /// A caller-owned, pooled client shared by every server. `None` builds + /// one client per server from `tls`. + pub shared_client: Option, } /// Stands in for a version that couldn't be determined (missing/unparseable @@ -111,12 +114,17 @@ impl Backend { let token = if paired_tokens { opts.tokens.get(index) } else { opts.tokens.first() } .cloned() .ok_or_else(|| BackendError::Other("apisix-standalone backend requires at least one token".into()))?; - let client = HttpClient::new(HttpClientConfig { - server: server.clone(), - token, - timeout: opts.timeout, - tls: opts.tls.clone(), - })? + let client = match &opts.shared_client { + Some(shared) => { + HttpClient::with_shared_client(shared.clone(), server.clone(), token, opts.timeout)? + } + None => HttpClient::new(HttpClientConfig { + server: server.clone(), + token, + timeout: opts.timeout, + tls: opts.tls.clone(), + })?, + } .with_log_scope(vec!["APISIX".to_string()]); Ok(StandaloneServer { server, client }) }) diff --git a/rust/crates/adc-backend-apisix-standalone/tests/common/mod.rs b/rust/crates/adc-backend-apisix-standalone/tests/common/mod.rs index e5fe66ea..61048a04 100644 --- a/rust/crates/adc-backend-apisix-standalone/tests/common/mod.rs +++ b/rust/crates/adc-backend-apisix-standalone/tests/common/mod.rs @@ -34,6 +34,7 @@ pub fn backend_options(servers: Vec, cache_key: &str) -> adc_backend_api bypass_cache: false, timeout: Some(Duration::from_secs(10)), tls: tls(), + shared_client: None, } } diff --git a/rust/crates/adc-backend-core/src/lib.rs b/rust/crates/adc-backend-core/src/lib.rs index 9d896ba8..37087d5e 100644 --- a/rust/crates/adc-backend-core/src/lib.rs +++ b/rust/crates/adc-backend-core/src/lib.rs @@ -21,4 +21,4 @@ pub use resource_path::resource_type_collection_name; pub use retry::RetryPolicy; pub use tls::TlsConfig; -pub use reqwest::{Method, RequestBuilder, Response}; +pub use reqwest::{Client, Method, RequestBuilder, Response}; diff --git a/rust/crates/adc-cli/src/pipeline.rs b/rust/crates/adc-cli/src/pipeline.rs index e1a0eafc..cd3d929d 100644 --- a/rust/crates/adc-cli/src/pipeline.rs +++ b/rust/crates/adc-cli/src/pipeline.rs @@ -74,8 +74,7 @@ impl TryFrom<&BackendArgs> for BackendSpec { /// `shared_client`: `None` builds a fresh `HttpClient` from `spec.tls` (the /// one-shot CLI); `Some` reuses an already-built, pooled `reqwest::Client` -/// (the ingress-server daemon). Only `apisix`/`api7ee` look at it — -/// `apisix-standalone` always builds its own clients from `spec.tls`. +/// (the ingress-server daemon), shared by every server of the backend. pub fn init_backend( spec: BackendSpec, shared_client: Option, @@ -103,6 +102,7 @@ pub fn init_backend( bypass_cache: spec.bypass_cache, timeout: spec.timeout, tls: spec.tls, + shared_client, }, )?)), "" | "apisix" => { diff --git a/rust/crates/adc-cli/src/server/backend.rs b/rust/crates/adc-cli/src/server/backend.rs index f542fc33..2e127ad5 100644 --- a/rust/crates/adc-cli/src/server/backend.rs +++ b/rust/crates/adc-cli/src/server/backend.rs @@ -146,6 +146,56 @@ mod tests { task.abort(); } + /// A plain HTTP stub that answers every request with an empty `200` and + /// counts accepted TCP connections. + async fn spawn_counting_http_stub() -> (std::net::SocketAddr, Arc) { + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let accepted = Arc::new(AtomicUsize::new(0)); + let counter = accepted.clone(); + tokio::spawn(async move { + loop { + let (mut stream, _) = listener.accept().await.unwrap(); + counter.fetch_add(1, Ordering::SeqCst); + tokio::spawn(async move { + let response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"; + let mut pending = Vec::new(); + let mut buf = [0u8; 1024]; + while stream.read(&mut buf).await.is_ok_and(|n| { + pending.extend_from_slice(&buf[..n]); + n > 0 + }) { + // One response per complete request head, however the + // bytes were split across reads. + while let Some(end) = pending.windows(4).position(|w| w == b"\r\n\r\n") { + pending.drain(..end + 4); + if stream.write_all(response).await.is_err() { + return; + } + } + } + }); + } + }); + (addr, accepted) + } + + #[tokio::test] + async fn apisix_standalone_backends_reuse_one_connection_across_builds() { + let (addr, accepted) = spawn_counting_http_stub().await; + let mut o = opts(&format!("http://{addr}"), None); + o.backend = "apisix-standalone".to_string(); + o.cache_key = "test-standalone-reuse".to_string(); + + for _ in 0..3 { + build_backend(&o).unwrap().ping().await.unwrap(); + } + assert_eq!(accepted.load(std::sync::atomic::Ordering::SeqCst), 1); + } + #[test] fn build_backend_constructs_an_api7ee_backend() { let mut o = opts("http://127.0.0.1:9180", None);