diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 1e9bfb5e..9ec3fe08 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -2,6 +2,18 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adc-backend-core" +version = "0.1.0" +dependencies = [ + "adc-sdk", + "axum", + "futures", + "reqwest", + "serde_json", + "tokio", +] + [[package]] name = "adc-differ" version = "0.1.0" @@ -241,18 +253,18 @@ dependencies = [ [[package]] name = "clap" -version = "4.6.4" +version = "4.6.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d91e0c145792ef73a6ad36d27c75ac09f1832222a3c209689d90f534685ee5b7" +checksum = "301b56658598e48f3648647ac6fc887be7e7108eddfa4e9b63fcf3ec58c0cadf" dependencies = [ "clap_builder", ] [[package]] name = "clap_builder" -version = "4.6.2" +version = "4.6.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f09628afdcc538b57f3c6341e9c8e9970f18e4a481690a64974d7023bd33548b" +checksum = "94a65403d1a1bd28f7dc68eb8506e8874808ee5eecb59298de588e2e1407a078" dependencies = [ "anstyle", "clap_lex", diff --git a/rust/Cargo.toml b/rust/Cargo.toml index cbd5e78e..0f3d35e4 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -3,6 +3,7 @@ resolver = "3" members = [ "crates/adc-sdk", "crates/adc-differ", + "crates/adc-backend-core", "crates/adc-sync-bench", "crates/adc-mock-server", ] diff --git a/rust/crates/adc-backend-core/Cargo.toml b/rust/crates/adc-backend-core/Cargo.toml new file mode 100644 index 00000000..98bf7352 --- /dev/null +++ b/rust/crates/adc-backend-core/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "adc-backend-core" +version = "0.1.0" +edition.workspace = true +publish.workspace = true +rust-version.workspace = true + +[dependencies] +adc-sdk = { path = "../adc-sdk" } +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"] } +futures = "0.3" +tokio = { version = "1", features = ["time"] } + +[dev-dependencies] +tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "time"] } +axum = "0.8" +serde_json = { workspace = true } diff --git a/rust/crates/adc-backend-core/src/client.rs b/rust/crates/adc-backend-core/src/client.rs new file mode 100644 index 00000000..153642a6 --- /dev/null +++ b/rust/crates/adc-backend-core/src/client.rs @@ -0,0 +1,100 @@ +use std::time::Duration; + +use adc_sdk::BackendError; +use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderValue}; +use reqwest::{Method, RequestBuilder, Response, Url}; + +use crate::tls::TlsConfig; + +/// Everything needed to stand up a backend's HTTP client: where it lives, +/// how to authenticate, and how to connect. Doesn't include backend-specific +/// concerns like a gateway group name — those live in each backend crate. +#[derive(Debug, Clone)] +pub struct HttpClientConfig { + pub server: String, + pub token: String, + pub timeout: Option, + pub tls: TlsConfig, +} + +/// A backend's HTTP client: a `reqwest::Client` pre-configured with the +/// `X-API-KEY` auth header apisix/api7 both expect, TLS settings, and a base +/// URL, plus request/response handling that classifies failures into +/// `BackendError` uniformly. Connection pooling (the Node `agentkeepalive` +/// equivalent) comes for free from `reqwest::Client`'s own pool — nothing to +/// configure for that. +pub struct HttpClient { + inner: reqwest::Client, + base_url: Url, +} + +impl HttpClient { + pub fn new(config: HttpClientConfig) -> Result { + let base_url = Url::parse(&config.server) + .map_err(|e| BackendError::Other(format!("invalid server URL {:?}: {e}", config.server).into()))?; + + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + let mut token = HeaderValue::from_str(&config.token) + .map_err(|e| BackendError::Other(format!("invalid token: {e}").into()))?; + token.set_sensitive(true); + headers.insert("X-API-KEY", token); + + let mut builder = reqwest::Client::builder().default_headers(headers); + if let Some(timeout) = config.timeout { + builder = builder.timeout(timeout); + } + builder = config.tls.apply(builder)?; + + let inner = builder + .build() + .map_err(|e| BackendError::Other(format!("failed to build HTTP client: {e}").into()))?; + + Ok(Self { inner, base_url }) + } + + /// Starts a request against `path`, resolved relative to the configured + /// server URL (e.g. `/apisix/admin/routes`). + pub fn request(&self, method: Method, path: &str) -> Result { + let url = self + .base_url + .join(path) + .map_err(|e| BackendError::Other(format!("invalid request path {path:?}: {e}").into()))?; + Ok(self.inner.request(method, url)) + } + + /// Sends a request built via [`HttpClient::request`], classifying + /// transport failures and non-2xx responses into `BackendError`. Callers + /// decode the body themselves (`.json()`, `.text()`, or just read + /// headers) since the right shape depends on the endpoint. + pub async fn send(&self, builder: RequestBuilder) -> Result { + let (client, request) = builder.build_split(); + let request = request.map_err(|e| BackendError::Other(format!("failed to build request: {e}").into()))?; + let method = request.method().clone(); + let url = request.url().clone(); + + let response = client.execute(request).await.map_err(|e| classify_transport_error(&e, &method, &url))?; + + let status = response.status(); + if status.is_success() { + return Ok(response); + } + + let message = response.text().await.unwrap_or_default(); + Err(match status.as_u16() { + 401 | 403 => BackendError::Auth(message), + 404 => BackendError::NotFound(format!("{method} {url}")), + _ => BackendError::Api { status: status.as_u16(), message }, + }) + } +} + +fn classify_transport_error(error: &reqwest::Error, method: &Method, url: &Url) -> BackendError { + if error.is_timeout() { + BackendError::Transport(format!( + "request \"{method} {url}\" timed out. Consider increasing the timeout." + )) + } else { + BackendError::Transport(format!("request \"{method} {url}\" failed: {error}")) + } +} diff --git a/rust/crates/adc-backend-core/src/concurrency.rs b/rust/crates/adc-backend-core/src/concurrency.rs new file mode 100644 index 00000000..6926dd23 --- /dev/null +++ b/rust/crates/adc-backend-core/src/concurrency.rs @@ -0,0 +1,20 @@ +use std::future::Future; + +use futures::stream::{self, StreamExt}; + +/// Runs `f` over `items` with at most `concurrency` in flight at once. +/// `None` means unbounded — every item starts immediately, matching RxJS +/// `mergeMap(fn)` called with no concurrency argument. +/// +/// Results come back in completion order, not input order (same as +/// `mergeMap`'s emission order): fine for callers like `Backend::sync`, +/// where each output already carries its own identity (the `Event` it came +/// from) rather than relying on positional correlation with `items`. +pub async fn concurrent_map(items: Vec, concurrency: Option, f: F) -> Vec +where + F: FnMut(T) -> Fut, + Fut: Future, +{ + let concurrency = concurrency.unwrap_or(items.len()).max(1); + stream::iter(items).map(f).buffer_unordered(concurrency).collect().await +} diff --git a/rust/crates/adc-backend-core/src/lib.rs b/rust/crates/adc-backend-core/src/lib.rs new file mode 100644 index 00000000..213e4b91 --- /dev/null +++ b/rust/crates/adc-backend-core/src/lib.rs @@ -0,0 +1,18 @@ +//! Shared HTTP client plumbing for the backend integrations +//! (`adc-backend-apisix`, `adc-backend-api7`, `adc-backend-apisix-standalone`): +//! building a `reqwest::Client` with auth headers and TLS settings, +//! classifying request failures into `adc_sdk::BackendError` uniformly, and +//! the retry/concurrency helpers each backend's operator/fetcher builds +//! request fan-out on top of. + +mod client; +mod concurrency; +mod retry; +mod tls; + +pub use client::{HttpClient, HttpClientConfig}; +pub use concurrency::concurrent_map; +pub use retry::RetryPolicy; +pub use tls::TlsConfig; + +pub use reqwest::{Method, Response}; diff --git a/rust/crates/adc-backend-core/src/retry.rs b/rust/crates/adc-backend-core/src/retry.rs new file mode 100644 index 00000000..5b81dff5 --- /dev/null +++ b/rust/crates/adc-backend-core/src/retry.rs @@ -0,0 +1,38 @@ +use std::future::Future; +use std::time::Duration; + +/// A fixed-delay retry policy: on failure, wait `delay` and try again, up to +/// `retries` additional attempts (so `retries + 1` attempts total). +#[derive(Debug, Clone, Copy)] +pub struct RetryPolicy { + pub retries: usize, + pub delay: Duration, +} + +impl Default for RetryPolicy { + /// Matches the apisix operator's hardcoded `retry({ count: 3, delay: 100 })` + /// for mutating requests (PUT/DELETE against `/apisix/admin/*`). + fn default() -> Self { + Self { retries: 3, delay: Duration::from_millis(100) } + } +} + +impl RetryPolicy { + pub async fn run(&self, mut f: F) -> Result + where + F: FnMut() -> Fut, + Fut: Future>, + { + let mut attempt = 0; + loop { + match f().await { + Ok(value) => return Ok(value), + Err(_) if attempt < self.retries => { + attempt += 1; + tokio::time::sleep(self.delay).await; + } + Err(err) => return Err(err), + } + } + } +} diff --git a/rust/crates/adc-backend-core/src/tls.rs b/rust/crates/adc-backend-core/src/tls.rs new file mode 100644 index 00000000..3767fc48 --- /dev/null +++ b/rust/crates/adc-backend-core/src/tls.rs @@ -0,0 +1,42 @@ +use adc_sdk::BackendError; + +/// TLS options for connecting to a backend's admin API. Mirrors what the CLI +/// previously built into per-request Node `http(s).Agent` instances (custom +/// CA, mTLS client identity, skip-verify) — here they're baked directly into +/// the `reqwest::Client` at construction time instead. +/// +/// Takes PEM bytes directly rather than file paths: reading config off disk +/// is the CLI/config layer's job, not this HTTP client's — keeps this crate +/// free of filesystem I/O and lets callers source certs from anywhere +/// (files, env vars, a secrets manager). +#[derive(Debug, Clone, Default)] +pub struct TlsConfig { + pub ca_cert_pem: Option>, + pub client_cert_pem: Option>, + pub client_key_pem: Option>, + pub skip_verify: bool, +} + +impl TlsConfig { + pub(crate) fn apply(&self, mut builder: reqwest::ClientBuilder) -> Result { + if self.skip_verify { + builder = builder.danger_accept_invalid_certs(true); + } + + if let Some(pem) = &self.ca_cert_pem { + let cert = reqwest::Certificate::from_pem(pem) + .map_err(|e| BackendError::Other(format!("invalid CA certificate: {e}").into()))?; + builder = builder.add_root_certificate(cert); + } + + if let (Some(cert_pem), Some(key_pem)) = (&self.client_cert_pem, &self.client_key_pem) { + let mut pem = cert_pem.clone(); + pem.extend(key_pem); + let identity = reqwest::Identity::from_pem(&pem) + .map_err(|e| BackendError::Other(format!("invalid client certificate/key: {e}").into()))?; + builder = builder.identity(identity); + } + + Ok(builder) + } +} diff --git a/rust/crates/adc-backend-core/tests/concurrency.rs b/rust/crates/adc-backend-core/tests/concurrency.rs new file mode 100644 index 00000000..98f2c96c --- /dev/null +++ b/rust/crates/adc-backend-core/tests/concurrency.rs @@ -0,0 +1,54 @@ +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +use adc_backend_core::concurrent_map; + +#[tokio::test] +async fn respects_the_concurrency_bound() { + let current = AtomicUsize::new(0); + let peak = AtomicUsize::new(0); + + let items: Vec = (0..20).collect(); + let results = concurrent_map(items.clone(), Some(4), |item| { + let current = ¤t; + let peak = &peak; + async move { + let now = current.fetch_add(1, Ordering::SeqCst) + 1; + peak.fetch_max(now, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(5)).await; + current.fetch_sub(1, Ordering::SeqCst); + item + } + }) + .await; + + assert_eq!(results.len(), items.len()); + assert!(peak.load(Ordering::SeqCst) <= 4, "peak concurrency was {}", peak.load(Ordering::SeqCst)); + + let mut sorted = results; + sorted.sort(); + assert_eq!(sorted, items); +} + +#[tokio::test] +async fn unbounded_when_concurrency_is_none() { + let current = AtomicUsize::new(0); + let peak = AtomicUsize::new(0); + + let items: Vec = (0..10).collect(); + let results = concurrent_map(items.clone(), None, |item| { + let current = ¤t; + let peak = &peak; + async move { + let now = current.fetch_add(1, Ordering::SeqCst) + 1; + peak.fetch_max(now, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(5)).await; + current.fetch_sub(1, Ordering::SeqCst); + item + } + }) + .await; + + assert_eq!(results.len(), items.len()); + assert_eq!(peak.load(Ordering::SeqCst), items.len()); +} diff --git a/rust/crates/adc-backend-core/tests/http_client.rs b/rust/crates/adc-backend-core/tests/http_client.rs new file mode 100644 index 00000000..4b42f442 --- /dev/null +++ b/rust/crates/adc-backend-core/tests/http_client.rs @@ -0,0 +1,90 @@ +use std::time::Duration; + +use adc_backend_core::{HttpClient, HttpClientConfig, Method, TlsConfig}; +use adc_sdk::BackendError; +use axum::Router; +use axum::extract::State; +use axum::http::HeaderMap; +use axum::response::Json; +use axum::routing::get; +use serde_json::{Value, json}; +use tokio::net::TcpListener; + +async fn spawn_server() -> String { + let router = Router::new() + .route( + "/apisix/admin/routes", + get( + |State(token): State<&'static str>, headers: HeaderMap| async move { + assert_eq!(headers.get("X-API-KEY").unwrap().to_str().unwrap(), token); + assert_eq!( + headers.get("Content-Type").unwrap().to_str().unwrap(), + "application/json" + ); + Json(json!({ "list": [] })) + }, + ), + ) + .route( + "/slow", + get(|| async { + tokio::time::sleep(Duration::from_secs(5)).await; + Json(json!({})) + }), + ) + .with_state("test-token"); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, router).await.unwrap(); + }); + format!("http://{addr}") +} + +fn client(server: String, timeout: Option) -> HttpClient { + HttpClient::new(HttpClientConfig { + server, + token: "test-token".into(), + timeout, + tls: TlsConfig::default(), + }) + .unwrap() +} + +#[tokio::test] +async fn injects_auth_header_and_decodes_success_response() { + let server = spawn_server().await; + let client = client(server, None); + + let req = client.request(Method::GET, "/apisix/admin/routes").unwrap(); + let resp = client.send(req).await.unwrap(); + let body: Value = resp.json().await.unwrap(); + assert_eq!(body, json!({ "list": [] })); +} + +#[tokio::test] +async fn classifies_404_as_not_found() { + let server = spawn_server().await; + let client = client(server, None); + + let req = client.request(Method::GET, "/does-not-exist").unwrap(); + let err = client.send(req).await.unwrap_err(); + assert!( + matches!(err, BackendError::NotFound(_)), + "expected NotFound, got {err:?}" + ); +} + +#[tokio::test] +async fn classifies_slow_response_as_transport_timeout() { + let server = spawn_server().await; + let client = client(server, Some(Duration::from_millis(50))); + + let req = client.request(Method::GET, "/slow").unwrap(); + let err = client.send(req).await.unwrap_err(); + let BackendError::Transport(message) = &err else { + panic!("expected Transport, got {err:?}"); + }; + assert!(message.contains("timed out"), "message: {message}"); +} diff --git a/rust/crates/adc-backend-core/tests/retry.rs b/rust/crates/adc-backend-core/tests/retry.rs new file mode 100644 index 00000000..3d3f4e87 --- /dev/null +++ b/rust/crates/adc-backend-core/tests/retry.rs @@ -0,0 +1,40 @@ +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +use adc_backend_core::RetryPolicy; + +#[tokio::test] +async fn succeeds_after_transient_failures_within_budget() { + let attempts = AtomicUsize::new(0); + let policy = RetryPolicy { retries: 3, delay: Duration::from_millis(1) }; + + let result: Result<&str, &str> = policy + .run(|| async { + if attempts.fetch_add(1, Ordering::SeqCst) < 2 { + Err("not yet") + } else { + Ok("done") + } + }) + .await; + + assert_eq!(result, Ok("done")); + assert_eq!(attempts.load(Ordering::SeqCst), 3); +} + +#[tokio::test] +async fn gives_up_after_exhausting_retries() { + let attempts = AtomicUsize::new(0); + let policy = RetryPolicy { retries: 2, delay: Duration::from_millis(1) }; + + let result: Result<&str, &str> = policy + .run(|| async { + attempts.fetch_add(1, Ordering::SeqCst); + Err("always fails") + }) + .await; + + assert_eq!(result, Err("always fails")); + // 1 initial attempt + 2 retries = 3. + assert_eq!(attempts.load(Ordering::SeqCst), 3); +}