diff --git a/crates/logfwd-output/src/json_lines.rs b/crates/logfwd-output/src/json_lines.rs index 249cad218..91bb312bc 100644 --- a/crates/logfwd-output/src/json_lines.rs +++ b/crates/logfwd-output/src/json_lines.rs @@ -1,3 +1,4 @@ +use std::future::Future; use std::io; use std::sync::Arc; @@ -6,10 +7,9 @@ use arrow::record_batch::RecordBatch; use logfwd_types::diagnostics::ComponentStats; -#[allow(deprecated)] use super::{ - BatchMetadata, Compression, HTTP_MAX_RETRIES, HTTP_RETRY_INITIAL_DELAY_MS, OutputSink, - build_col_infos, is_transient_error, str_value, write_row_json, + BatchMetadata, Compression, HTTP_MAX_RETRIES, HTTP_RETRY_INITIAL_DELAY_MS, build_col_infos, + str_value, write_row_json, }; // --------------------------------------------------------------------------- @@ -24,22 +24,23 @@ pub struct JsonLinesSink { pub(crate) batch_buf: Vec, compress_buf: Vec, compression: Compression, - http_agent: ureq::Agent, + client: reqwest::Client, stats: Arc, } impl JsonLinesSink { + /// Create a new JSON-lines sink. + /// + /// The caller supplies a shared `reqwest::Client` (typically cloned from + /// the factory) so all workers reuse the same connection pool. pub fn new( name: String, url: String, headers: Vec<(String, String)>, compression: Compression, + client: reqwest::Client, stats: Arc, ) -> Self { - let http_agent = ureq::config::Config::builder() - .timeout_global(Some(std::time::Duration::from_secs(30))) - .build() - .new_agent(); JsonLinesSink { name, url, @@ -47,7 +48,7 @@ impl JsonLinesSink { batch_buf: Vec::with_capacity(64 * 1024), compress_buf: Vec::new(), compression, - http_agent, + client, stats, } } @@ -118,21 +119,23 @@ impl JsonLinesSink { } Ok(()) } -} -#[allow(deprecated)] -impl OutputSink for JsonLinesSink { - fn send_batch(&mut self, batch: &RecordBatch, _metadata: &BatchMetadata) -> io::Result<()> { + /// Serialize, compress (if configured), and POST the batch over HTTP. + /// + /// Retries transient failures (5xx, timeout, and connection errors) with + /// exponential backoff. Returns `SendResult::RetryAfter` on HTTP 429 and + /// `SendResult::Rejected` on non-429 4xx client errors. + async fn send_batch_inner( + &mut self, + batch: &RecordBatch, + ) -> io::Result { self.serialize_batch(batch)?; if self.batch_buf.is_empty() { - return Ok(()); + return Ok(super::sink::SendResult::Ok); } - // Retry with exponential backoff for transient failures. - // 1 initial attempt + up to HTTP_MAX_RETRIES retries; delays: 100ms → 200ms → 400ms. - // Note: `self.batch_buf` is re-sent as `&[u8]` on each attempt — no - // allocation, but the full NDJSON payload is retransmitted each time. - // This is acceptable as a temporary measure until SinkDriver (#319). + // Compress if configured. The compressed payload is stored in + // `self.compress_buf` and lives until the next call. let payload: &[u8] = match self.compression { Compression::Gzip => { use flate2::Compression as GzLevel; @@ -152,38 +155,169 @@ impl OutputSink for JsonLinesSink { unreachable!("zstd is not supported for json_lines; rejected at config validation") } }; - let build_req = || { - let mut req = self.http_agent.post(&self.url); - for (k, v) in &self.headers { - req = req.header(k.as_str(), v.as_str()); - } - req = req.header("Content-Type", "application/x-ndjson"); - if self.compression == Compression::Gzip { - req = req.header("Content-Encoding", "gzip"); - } - req - }; + + // Retry with exponential backoff for transient failures. + // 1 initial attempt + up to HTTP_MAX_RETRIES retries; delays: 100ms → 200ms → 400ms. let mut delay_ms: u64 = HTTP_RETRY_INITIAL_DELAY_MS; let mut attempt: u32 = 0; + + // Own the payload once; on retries we call `.clone()` which allocates + // but is acceptable for a path limited to HTTP_MAX_RETRIES iterations. + let body = payload.to_vec(); + loop { - match build_req().send(payload) { - Ok(_) => { - self.stats.inc_lines(batch.num_rows() as u64); - self.stats.inc_bytes(payload.len() as u64); - return Ok(()); + let mut req = self + .client + .post(&self.url) + .header("Content-Type", "application/x-ndjson"); + if self.compression == Compression::Gzip { + req = req.header("Content-Encoding", "gzip"); + } + for (k, v) in &self.headers { + req = req.header(k.as_str(), v.as_str()); + } + let req = req.body(body.clone()); + + match req.send().await { + Ok(response) => { + let status = response.status(); + + if status == reqwest::StatusCode::TOO_MANY_REQUESTS { + // Parse Retry-After header (seconds integer) or use default. + let retry_after = response + .headers() + .get("Retry-After") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse::().ok()) + .unwrap_or(5); + return Ok(super::sink::SendResult::RetryAfter( + std::time::Duration::from_secs(retry_after), + )); + } + + if status.is_client_error() { + return Ok(super::sink::SendResult::Rejected(format!( + "HTTP endpoint rejected batch with HTTP {status}" + ))); + } + + if status.is_server_error() { + if attempt < HTTP_MAX_RETRIES { + tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; + delay_ms *= 2; + attempt += 1; + continue; + } + return Err(io::Error::other(format!( + "HTTP endpoint returned {status} after {attempt} retries" + ))); + } + + if status.is_success() { + self.stats.inc_lines(batch.num_rows() as u64); + self.stats.inc_bytes(payload.len() as u64); + return Ok(super::sink::SendResult::Ok); + } + + return Err(io::Error::other(format!( + "HTTP endpoint returned non-success status {status}" + ))); } - Err(e) if attempt < HTTP_MAX_RETRIES && is_transient_error(&e) => { - std::thread::sleep(std::time::Duration::from_millis(delay_ms)); + Err(e) if attempt < HTTP_MAX_RETRIES && (e.is_timeout() || e.is_connect()) => { + // Retry transient transport errors only. + tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await; delay_ms *= 2; attempt += 1; } - Err(e) => return Err(io::Error::other(e.to_string())), + Err(e) => { + return Err(io::Error::other(format!( + "HTTP request failed after {attempt} retries: {e}" + ))); + } } } } +} - fn flush(&mut self) -> io::Result<()> { - Ok(()) +impl super::sink::Sink for JsonLinesSink { + fn send_batch<'a>( + &'a mut self, + batch: &'a RecordBatch, + _metadata: &'a BatchMetadata, + ) -> std::pin::Pin> + Send + 'a>> + { + Box::pin(async move { self.send_batch_inner(batch).await }) + } + + fn flush(&mut self) -> std::pin::Pin> + Send + '_>> { + Box::pin(async { Ok(()) }) + } + + fn name(&self) -> &str { + &self.name + } + + fn shutdown(&mut self) -> std::pin::Pin> + Send + '_>> { + Box::pin(async { Ok(()) }) + } +} + +// --------------------------------------------------------------------------- +// JsonLinesSinkFactory +// --------------------------------------------------------------------------- + +/// Creates [`JsonLinesSink`] instances for the output worker pool. +/// +/// All workers share a single `reqwest::Client` (which is internally +/// `Arc`-wrapped) so they reuse the same connection pool, TLS sessions, +/// and DNS cache. +pub struct JsonLinesSinkFactory { + name: String, + url: String, + headers: Vec<(String, String)>, + compression: Compression, + client: reqwest::Client, + stats: Arc, +} + +impl JsonLinesSinkFactory { + /// Create a new factory. + /// + /// Builds a shared `reqwest::Client` with a 30-second timeout and a + /// generous idle-connection pool. + pub fn new( + name: String, + url: String, + headers: Vec<(String, String)>, + compression: Compression, + stats: Arc, + ) -> io::Result { + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .pool_max_idle_per_host(64) + .build() + .map_err(io::Error::other)?; + Ok(JsonLinesSinkFactory { + name, + url, + headers, + compression, + client, + stats, + }) + } +} + +impl super::sink::SinkFactory for JsonLinesSinkFactory { + fn create(&self) -> io::Result> { + Ok(Box::new(JsonLinesSink::new( + self.name.clone(), + self.url.clone(), + self.headers.clone(), + self.compression, + self.client.clone(), // reqwest::Client is Arc-wrapped, clone is cheap + Arc::clone(&self.stats), + ))) } fn name(&self) -> &str { @@ -203,11 +337,13 @@ mod tests { use logfwd_types::diagnostics::ComponentStats; fn make_sink() -> JsonLinesSink { + let client = reqwest::Client::new(); JsonLinesSink::new( "test".to_string(), "http://localhost:1".to_string(), // unreachable — tests only call serialize_batch vec![], Compression::None, + client, Arc::new(ComponentStats::default()), ) } diff --git a/crates/logfwd-output/src/lib.rs b/crates/logfwd-output/src/lib.rs index e7df4822d..a464b5b61 100644 --- a/crates/logfwd-output/src/lib.rs +++ b/crates/logfwd-output/src/lib.rs @@ -21,7 +21,7 @@ mod parquet; pub use arrow_ipc_sink::{ArrowIpcSinkFactory, deserialize_ipc, serialize_ipc}; pub use elasticsearch::{ElasticsearchRequestMode, ElasticsearchSink, ElasticsearchSinkFactory}; pub use fanout::{FanoutSink, FanoutSinkError}; -pub use json_lines::JsonLinesSink; +pub use json_lines::{JsonLinesSink, JsonLinesSinkFactory}; pub use loki::{LokiSink, LokiSinkFactory}; pub use null::NullSink; pub use otap_sink::{ @@ -59,6 +59,7 @@ pub(crate) const HTTP_RETRY_INITIAL_DELAY_MS: u64 = 100; /// /// Transient errors are: HTTP 429 Too Many Requests, 5xx server errors, and /// network/transport failures (I/O, host not found, connection failed, timeout). +#[allow(dead_code)] // Still used by OTLP ureq sink; will be removed when OTLP migrates to reqwest pub(crate) fn is_transient_error(e: &ureq::Error) -> bool { match e { ureq::Error::StatusCode(status) => *status == 429 || *status >= 500, @@ -537,7 +538,7 @@ pub fn build_output_sink( cfg: &OutputConfig, stats: Arc, ) -> Result, String> { - let auth_headers = build_auth_headers(cfg.auth.as_ref()); + let _auth_headers = build_auth_headers(cfg.auth.as_ref()); match cfg.output_type { OutputType::Stdout => Err(format!( "output '{name}': stdout requires the async pipeline — use build_sink_factory() instead" @@ -545,28 +546,9 @@ pub fn build_output_sink( OutputType::Otlp => Err(format!( "output '{name}': OTLP requires the async pipeline — use build_sink_factory() instead" )), - OutputType::Http => { - let endpoint = cfg - .endpoint - .as_ref() - .ok_or_else(|| format!("output '{name}': HTTP requires 'endpoint'"))?; - let compression = match cfg.compression.as_deref() { - Some("gzip") => Compression::Gzip, - Some(other) => { - return Err(format!( - "output '{name}': json_lines HTTP does not support '{other}' compression (use 'gzip')" - )); - } - None => Compression::None, - }; - Ok(Box::new(JsonLinesSink::new( - name.to_string(), - endpoint.clone(), - auth_headers, - compression, - stats, - ))) - } + OutputType::Http => Err(format!( + "output '{name}': http requires the async pipeline — use build_sink_factory() instead" + )), OutputType::Null => Ok(Box::new(NullSink::new(name.to_string(), stats))), OutputType::Tcp => Err(format!( "output '{name}': tcp requires the async pipeline — use build_sink_factory() instead" @@ -789,6 +771,30 @@ pub fn build_sink_factory( stats, ))) } + OutputType::Http => { + let endpoint = cfg + .endpoint + .as_ref() + .ok_or_else(|| format!("output '{name}': HTTP requires 'endpoint'"))?; + let compression = match cfg.compression.as_deref() { + Some("gzip") => Compression::Gzip, + Some(other) => { + return Err(format!( + "output '{name}': json_lines HTTP does not support '{other}' compression (use 'gzip')" + )); + } + None => Compression::None, + }; + let factory = JsonLinesSinkFactory::new( + name.to_string(), + endpoint.clone(), + auth_headers, + compression, + stats, + ) + .map_err(|e| format!("output '{name}': json_lines factory: {e}"))?; + Ok(Arc::new(factory)) + } _ => { // Sync sink — build it once, wrap in OnceFactory (max_workers=1). let sink = build_output_sink(name, cfg, stats)?; @@ -1071,6 +1077,7 @@ mod tests { "http://localhost:9200".to_string(), vec![], Compression::None, + reqwest::Client::new(), Arc::new(ComponentStats::new()), ); sink.serialize_batch(&batch).unwrap(); @@ -1224,7 +1231,7 @@ mod tests { } #[test] - fn test_build_output_sink_http() { + fn test_build_sink_factory_http() { let cfg = OutputConfig { name: Some("es".to_string()), output_type: OutputType::Http, @@ -1237,12 +1244,12 @@ mod tests { index: None, auth: None, }; - let sink = build_output_sink("es", &cfg, Arc::new(ComponentStats::new())).unwrap(); - assert_eq!(sink.name(), "es"); + let factory = build_sink_factory("es", &cfg, Arc::new(ComponentStats::new())).unwrap(); + assert_eq!(factory.name(), "es"); } #[test] - fn test_build_output_sink_http_with_gzip_compression() { + fn test_build_sink_factory_http_with_gzip_compression() { let cfg = OutputConfig { name: Some("http-gz".to_string()), output_type: OutputType::Http, @@ -1255,12 +1262,12 @@ mod tests { index: None, auth: None, }; - let sink = build_output_sink("http-gz", &cfg, Arc::new(ComponentStats::new())).unwrap(); - assert_eq!(sink.name(), "http-gz"); + let factory = build_sink_factory("http-gz", &cfg, Arc::new(ComponentStats::new())).unwrap(); + assert_eq!(factory.name(), "http-gz"); } #[test] - fn test_build_output_sink_http_rejects_unknown_compression() { + fn test_build_sink_factory_http_rejects_unknown_compression() { let cfg = OutputConfig { name: Some("http-bad".to_string()), output_type: OutputType::Http, @@ -1273,7 +1280,7 @@ mod tests { index: None, auth: None, }; - let result = build_output_sink("http-bad", &cfg, Arc::new(ComponentStats::new())); + let result = build_sink_factory("http-bad", &cfg, Arc::new(ComponentStats::new())); assert!( result.is_err(), "unsupported compression should be rejected" @@ -1361,7 +1368,7 @@ mod tests { } #[test] - fn test_build_output_sink_http_with_bearer_auth() { + fn test_build_sink_factory_http_with_bearer_auth() { use logfwd_config::AuthConfig; let cfg = OutputConfig { name: Some("auth-sink".to_string()), @@ -1378,8 +1385,9 @@ mod tests { headers: std::collections::HashMap::new(), }), }; - let sink = build_output_sink("auth-sink", &cfg, Arc::new(ComponentStats::new())).unwrap(); - assert_eq!(sink.name(), "auth-sink"); + let factory = + build_sink_factory("auth-sink", &cfg, Arc::new(ComponentStats::new())).unwrap(); + assert_eq!(factory.name(), "auth-sink"); } // ----------------------------------------------------------------------- @@ -1459,6 +1467,7 @@ mod tests { "http://localhost:9200".to_string(), vec![], Compression::None, + reqwest::Client::new(), Arc::new(ComponentStats::new()), ); // Must not panic.