Repository navigation
Conversation
ApprovabilityVerdict: Needs human review This PR migrates HTTP transport from ureq (sync) to reqwest (async) with significant runtime behavior changes including new retry logic, status code handling, and breaking API changes. Multiple unresolved review comments identify substantive issues: incorrect redirect handling, potential breaking call sites, missing test coverage, and performance regressions. You can customize Macroscope's approvability policy. Learn more. |
There was a problem hiding this comment.
Pull request overview
This PR migrates the JsonLinesSink HTTP output from a deprecated sync (OutputSink + SyncSinkAdapter) implementation to an async-native Sink implementation using reqwest, aligning the HTTP output with the existing async sink patterns (e.g., Elasticsearch).
Changes:
- Converted
JsonLinesSinkto implement the asyncSinktrait and switched HTTP transport fromureqtoreqwest. - Added
JsonLinesSinkFactoryto build and share a configuredreqwest::Clientacross worker sinks. - Updated routing so
OutputType::Httpis constructed viabuild_sink_factory()(and errors viabuild_output_sink()).
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 6 comments.
| File | Description |
|---|---|
| crates/logfwd-output/src/lib.rs | Re-exports JsonLinesSinkFactory, routes OutputType::Http to build_sink_factory(), and updates related tests. |
| crates/logfwd-output/src/json_lines.rs | Replaces sync HTTP send loop with async reqwest send + retry logic; introduces JsonLinesSinkFactory. |
| if status.is_client_error() { | ||
| return Ok(super::sink::SendResult::Rejected(format!( | ||
| "HTTP endpoint rejected batch with HTTP {status}" | ||
| ))); | ||
| } |
There was a problem hiding this comment.
send_batch_inner treats any status that is neither 4xx nor 5xx as success. That means 3xx redirects (and other non-success codes) will incorrectly be counted as delivered. Consider explicitly checking status.is_success() and handling non-success/non-4xx/non-5xx statuses as an error or rejection.
| Err(_e) if attempt < HTTP_MAX_RETRIES => { | ||
| // Transport-level error (connection refused, timeout, DNS, etc.) |
There was a problem hiding this comment.
The retry loop currently retries all reqwest::Error values (including non-transient request construction errors like invalid URLs). This can hide configuration problems behind backoff delays. Consider retrying only transient errors (e.g., e.is_timeout() / e.is_connect()), and failing fast for e.is_request() and similar non-retryable cases.
| Err(_e) if attempt < HTTP_MAX_RETRIES => { | |
| // Transport-level error (connection refused, timeout, DNS, etc.) | |
| Err(e) if attempt < HTTP_MAX_RETRIES && (e.is_timeout() || e.is_connect()) => { | |
| // Retry only transient transport-level errors. |
| // 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 { |
There was a problem hiding this comment.
let body = payload.to_vec(); plus body.clone() on every attempt copies the full (potentially large) payload per retry and adds an extra allocation even for the first attempt. To avoid this regression vs the old slice-based resend, consider using a cheap-clone buffer type (e.g., bytes::Bytes) or otherwise sharing the underlying payload across retries without re-copying.
| /// 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, |
There was a problem hiding this comment.
JsonLinesSink::new now requires an explicit reqwest::Client, which is a breaking API change. There are existing call sites in this repo that still construct JsonLinesSink directly (e.g., bench + fuzz targets) and will fail to compile unless they are updated or a convenience constructor is kept.
| /// 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, | |
| /// This convenience constructor preserves the historical API by creating | |
| /// a default `reqwest::Client` for the sink. | |
| pub fn new( | |
| name: String, | |
| url: String, | |
| headers: Vec<(String, String)>, | |
| compression: Compression, | |
| stats: Arc<ComponentStats>, | |
| ) -> Self { | |
| Self::with_client( | |
| name, | |
| url, | |
| headers, | |
| compression, | |
| reqwest::Client::new(), | |
| stats, | |
| ) | |
| } | |
| /// Create a new JSON-lines sink using a shared `reqwest::Client`. | |
| /// | |
| /// Callers that manage sink creation centrally should prefer this so all | |
| /// workers reuse the same connection pool. | |
| pub fn with_client( | |
| name: String, | |
| url: String, | |
| headers: Vec<(String, String)>, | |
| compression: Compression, |
| OutputType::Http => Err(format!( | ||
| "output '{name}': http requires the async pipeline — use build_sink_factory() instead" | ||
| )), |
There was a problem hiding this comment.
OutputType::Http now hard-errors in build_output_sink, but there are still in-repo call sites/tests that build an HTTP sink via build_output_sink (e.g., crates/logfwd/src/pipeline.rs and its unit test). Those will start failing; update them to use the async pipeline (build_sink_factory) or adjust the legacy pipeline to reject HTTP earlier with a clear error.
| /// Serialize, compress (if configured), and POST the batch over HTTP. | ||
| /// | ||
| /// Retries transient failures (5xx, timeouts, 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<super::sink::SendResult> { |
There was a problem hiding this comment.
This PR introduces substantial new HTTP behavior (reqwest transport, retry/backoff, 429 → RetryAfter, 4xx → Rejected). The current unit tests in this module cover serialization but not the send/retry/status-handling logic; adding async tests (e.g., with a lightweight local HTTP server) would help prevent regressions around status classification and Retry-After parsing.
b492552 to
9ad2a6a
Compare
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
There was a problem hiding this comment.
🟢 Low
In send_batch_inner, self.stats.inc_lines(batch.num_rows() as u64) counts all rows regardless of whether they were actually written. In raw passthrough mode, rows where _raw is null are skipped during serialization (lines 106-112) but still counted in the stats, inflating the line count when _raw contains null values. Consider counting only rows that were actually written to the output buffer.
- for row in 0..num_rows {
- if !col.is_null(row) {
- self.batch_buf
- .extend_from_slice(str_value(col, row).as_bytes());
- self.batch_buf.push(b'\n');
- }
- }
+ let mut lines_written: u64 = 0;
+ for row in 0..num_rows {
+ if !col.is_null(row) {
+ self.batch_buf
+ .extend_from_slice(str_value(col, row).as_bytes());
+ self.batch_buf.push(b'\n');
+ lines_written += 1;
+ }
+ }
+ // Update stats with actual lines written, not total rows
+ self.stats.inc_lines(lines_written);🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file crates/logfwd-output/src/json_lines.rs around lines 103-118:
In `send_batch_inner`, `self.stats.inc_lines(batch.num_rows() as u64)` counts all rows regardless of whether they were actually written. In raw passthrough mode, rows where `_raw` is null are skipped during serialization (lines 106-112) but still counted in the stats, inflating the line count when `_raw` contains null values. Consider counting only rows that were actually written to the output buffer.
Evidence trail:
crates/logfwd-output/src/json_lines.rs line 217: `self.stats.inc_lines(batch.num_rows() as u64)` counts all rows. crates/logfwd-output/src/json_lines.rs lines 107-112: `if !col.is_null(row)` shows null `_raw` rows are skipped in raw passthrough mode. git_grep confirmed `inc_lines` is at line 217, `fn send_batch_inner` starts at line 128.
9ad2a6a to
a723f31
Compare
strawgate
left a comment
strawgate
left a comment
There was a problem hiding this comment.
Solid migration from ureq to reqwest. The async Sink impl, SinkFactory, and routing changes all follow the established pattern. Two issues to fix:
1. 3xx redirects silently counted as success (blocking)
In send_batch_inner, after checking for 429, 4xx, and 5xx, the fallthrough treats everything else (including 3xx) as success. reqwest follows redirects by default (up to 10), so in practice 3xx won't reach this code -- but if redirect following is ever disabled or the limit is hit, a 3xx would be silently counted as delivered. Tighten the success check:
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);
}
// Unexpected status (1xx, 3xx that wasn't followed, etc.)
return Err(io::Error::other(format!(
"HTTP endpoint returned unexpected status {status}"
)));2. Retry loop retries non-transient errors (blocking)
The Err(_e) if attempt < HTTP_MAX_RETRIES arm retries all reqwest errors including request-construction failures (bad URL, invalid headers). These will never succeed on retry and just burn through the backoff delay. Filter to transient errors only:
Err(e) if attempt < HTTP_MAX_RETRIES && (e.is_timeout() || e.is_connect()) => {Minor observations (non-blocking)
body.clone()on each retry: allocates a full copy of the payload per attempt. Abytes::Byteswould make retries zero-copy. Fine for now since HTTP_MAX_RETRIES is small, but worth a follow-up.- HTTP 429 handling with
Retry-Afterheader parsing andSendResult::RetryAfter-- nice improvement over the old ureq path which treated 429 as a generic transient error. JsonLinesSinkFactoryshares a singlereqwest::Client(Arc-wrapped) withpool_max_idle_per_host(64)-- good for connection reuse across workers.- Compression logic (gzip/zstd rejection) preserved unchanged.
- Tests updated to pass
reqwest::Client::new()toJsonLinesSink::new-- correct. is_transient_errorimport removed from json_lines.rs since retry classification is now inline -- the function itself still exists forotlp_sink.rswhich still uses ureq. Clean separation.
strawgate
left a comment
strawgate
left a comment
There was a problem hiding this comment.
Shared reqwest::Client, proper Retry-After parsing, compression preserved. Two blocking issues:
-
Success fallthrough:
send_batch_innertreats any non-4xx/non-5xx status as success, including 3xx redirects. Should explicitly checkstatus.is_success()instead of falling through. -
Retry on non-transient errors: The retry loop retries ALL
reqwest::Errorvariants including request-construction errors (bad URL, invalid headers). Filter toe.is_timeout() || e.is_connect()for retry; other errors should be immediate failures.
Addressed in 0a61dba, with a small follow-up wording cleanup in cb919d7. Only 2xx responses are now counted as delivered, and reqwest retries are limited to timeout/connect transport errors while 5xx responses continue through the existing backoff path. |
Already addressed in 0a61dba, with the follow-up wording cleanup in cb919d7. The current code only counts 2xx responses as delivered, and retries are limited to timeout/connect reqwest errors. |
Replace the synchronous ureq-based HTTP client with async reqwest in JsonLinesSink, following the same pattern as ElasticsearchSink: - Replace http_agent: ureq::Agent with client: reqwest::Client - Remove OutputSink impl, add async Sink trait impl - Add JsonLinesSinkFactory for multi-worker support - Handle 429 with SendResult::RetryAfter, 4xx with SendResult::Rejected - Use tokio::time::sleep for async retry backoff - Update build_sink_factory() with OutputType::Http arm - Update build_output_sink() to redirect Http to async pipeline - Update all tests to use build_sink_factory and pass reqwest::Client Co-authored-by: strawgate <6384545+strawgate@users.noreply.github.com>
Agent-Logs-Url: https://github.com/strawgate/memagent/sessions/c664a69c-1bfb-400d-987c-b5f4fabb45c0 Co-authored-by: strawgate <6384545+strawgate@users.noreply.github.com>
Agent-Logs-Url: https://github.com/strawgate/memagent/sessions/c664a69c-1bfb-400d-987c-b5f4fabb45c0 Co-authored-by: strawgate <6384545+strawgate@users.noreply.github.com>
…eaders Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
cb919d7 to
dd9680e
Compare
TriageStatus: Blocked — 2 correctness issues + CI failing Blocking issues (from @strawgate's review)1. 3xx redirects counted as success
if !status.is_success() {
return Ok(SendResult::Rejected(format!("HTTP {status}: unexpected response")));
}
Ok(SendResult::Ok)2. Retry-After parsing / retry logic (second review round — need to see the specific change requested) CI failuresTests failing on Linux and macOS. Need to inspect the test output — likely related to one of the above correctness issues causing a test assertion to fail. What's in good shape
Next stepFix the 3xx fallthrough (one-line change), check the second reviewer comment for the retry issue, re-run CI. |
|
Closing — persistent merge conflicts with #1003 (thiserror) and other recent PRs. The JsonLinesSink async conversion needs to be redone from current master with the OutputError type. Reassigning to Copilot. |
The Http async sink was planned in PR #1000 but that PR was closed without merging. Three tests were written assuming Http would be in build_sink_factory, causing them to fail on master since the implementation never landed: - test_build_sink_factory_http_with_bearer_auth → unwrap() panics - test_build_sink_factory_http_rejects_unknown_compression → wrong error msg - test_build_output_sink_http → unwrap() panics Update all three to document the current behavior (Http returns "async pipeline" error) and add TODO comments for when the Http async sink is implemented. Fixes #1146 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
test_http_output_sends_to_server spins up a pipeline with type: http output, which fails at startup because the Http async sink was never implemented (closed PR #1000). Mark it ignored with a reference to #1146 so it doesn't block unrelated CI. Also commit the cargo fmt changes to streaming_builder.rs that the pre-commit hook required but weren't staged in the previous commit. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Note
Migrate
JsonLinesSinkHTTP posting from ureq to async reqwestOutputSinkimplementation with an asyncSinktrait impl backed byreqwest::Client, enabling use in the async pipeline.JsonLinesSinkFactorywhich holds a sharedreqwest::Client(30s timeout, 64 max idle connections) and creates per-workerJsonLinesSinkinstances.send_batch_innerretries on 5xx and transport errors with exponential backoff, maps HTTP 429 toRetryAfter, and maps other 4xx responses toRejected.build_output_sinknow returns an error forOutputType::Http; HTTP sinks must be created via the newbuild_sink_factorypath.build_output_sinkfor HTTP will receive an error and must migrate tobuild_sink_factory.Macroscope summarized dd9680e.