perf(syslog): optimize the batch codec and add cooperative decoding by lquerel · Pull Request #4509 · open-telemetry/otel-arrow

lquerel · 2026-10-23T02:27:11Z

Change summary

Optimizes the syslog-batch codec and receiver path to reduce CPU, copying, and Arrow construction costs while keeping large-batch decoding responsive.

Cooperative decoding: owned incremental jobs yield between chunks without holding codec state across yields, then produce one complete OTAP batch.
Cheaper ingestion: borrowed TCP frames, amortized receipt-clock reads, and bounded, learned buffer reservations.
Faster Arrow construction: direct observed-timestamp appends, bulk parent-ID appends, and shared opt-in known-key dictionary caching. Syslog vocabulary remains in the receiver.
Small batches remain synchronous; cooperative decoding starts above 256 messages or 256 KiB. Supports batches up to 8,192 messages / 2 MiB; the receiver default stays 100.

Preserves decode-failure recovery and cancellation cleanup. Arrow finalization, encoding, and compression remain synchronous.

Related issue

Validation

cargo xtask check

Full-engine benchmarks

Pre-PR5 main (06cdc5979) -> PR5.2 (c2ded8be9). This comparison includes PR5, PR5.1, PR5.2, and intervening upstream changes, not async decoding alone.

TCP syslog -> DFE receiver / batch / zstd OTAP exporter -> DFE backend.
One idle SUT CPU held at verified 3 GHz; three interleaved pairs per case/pass,
168 accepted runs. Values are before -> after; CPU intervals are paired 95% CIs.

Scenario	CPU us/log	CPU change [95% paired CI]	Mean RSS MiB	HTTP p99 ms
RFC 3164, 100 -> 8K, fixed	6.831 -> 3.053	-55.3% [-55.7%, -54.9%]	60.5 -> 53.8	32.33 -> 4.68
RFC 5424, 100 -> 8K, fixed	4.919 -> 2.625	-46.6% [-48.2%, -45.0%]	61.0 -> 54.9	19.65 -> 4.91
RFC 3164, 8K -> 8K, fixed	5.093 -> 2.931	-42.4% [-44.7%, -40.2%]	51.5 -> 53.5	8.90 -> 4.56
RFC 5424, 8K -> 8K, fixed	2.995 -> 2.505	-16.3% [-16.9%, -15.8%]	52.1 -> 55.4	6.97 -> 4.83
RFC 3164, 8K -> 8K, high	5.017 -> 2.940	-41.4% [-42.7%, -40.1%]	50.8 -> 183.5	16.97 -> 6.87
RFC 5424, 8K -> 8K, high	2.927 -> 2.470	-15.6% [-17.7%, -13.5%]	52.1 -> 189.6	10.74 -> 7.54

Fixed load: 102,400 offered logs/s.
High load: 1M with TCP backpressure (finite overload tests, not maximum-throughput claims). CPU/RSS and latency use separate passes. HTTP p99 is the mean of per-run p99s for a concurrent OTLP/noop Ack probe targeting 100 requests/s, not syslog delivery latency.
The 100 -> 8K cases disable encoded fragment budgets to match output sizes.

Tradeoff: overload RSS increases substantially with the same 100-slot queues.

The chart sweeps matching receiver/output batch targets. Contributor notes summarize prior profiling and isolated comparisons, not per-change attribution of this campaign's total gains.

User-facing changes

More efficient syslog processing and support for larger batches with cooperative decoding. Existing configuration and the default batch size are unchanged.

Parse canonical timestamps directly and retain the generic Chrono fallback for legacy accepted forms. Preserve receipt-year and local timezone semantics without timestamp caching. Include parser regressions, Criterion coverage, and a dedicated changelog.

Use the UTC year when every supported local offset must share it, falling back to local conversion near New Year. Keep independent clock samples and the existing receipt-year, timezone, and DST behavior. Cover calendar boundaries, leap years, offset extremes, and clock and timezone changes without adding a timestamp or offset cache.

Add owned incremental decode jobs driven through effect handlers without holding codec locks across scheduling points. Preserve synchronous fast paths and exact failure recovery. Provide a feature-gated large-batch syslog decoder, bounded scheduling probes, and fidelity, recovery, cancellation, and view tests. Keep production admission limits unchanged. Validation: targeted codec, OTAP, and syslog tests, Clippy, formatting, and sanity checks passed. Full xtask check encountered unchanged filelog-checkpoint test failures; strict core documentation encountered existing broken links.

opentelemetry-pr-dashboard · 2026-10-23T02:27:46Z

Pull request dashboard status

Waiting on the author · refreshed 2026-10-26 05:12 UTC

Two things need attention:

Required checks are failing — investigate the failures.
7 review items — respond to each (e.g. link a commit, explain why not, ask a follow-up):
- Inline threads: 1, 2, 3, 4, 5, 6, 7

Status above doesn't look right?

Just replied or pushed? Anything around or after the refresh time above may not be picked up yet — give it a few minutes.
Should this be with reviewers? Comment /dashboard route:reviewers to route it to them.
Anything wrong — including the routing? Report it with what you expected; it helps us improve the dashboard.

codecov · 2026-10-23T02:54:54Z

Codecov Report

❌ Patch coverage is 94.08602% with 77 lines in your changes missing coverage. Please review.
✅ Project coverage is 89.38%. Comparing base (4431586) to head (a7a5c1b).

Additional details and impacted files

@@            Coverage Diff            @@
##             main    #4509     +/-   ##
=========================================
  Coverage   89.37%   89.38%             
=========================================
  Files        1060     1062      +2     
  Lines      437314   438516   +1202     
=========================================
+ Hits       390853   391968   +1115     
- Misses      46461    46548     +87

Components	Coverage Δ
otap-dataflow	`90.26% <94.08%> (+<0.01%)`	⬆️
data_engine	`88.16% <ø> (ø)`
otel-arrow-go	`57.57% <ø> (ø)`
quiver	`92.28% <ø> (ø)`

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

…05-2-cooperative-syslog

Append repeated attribute parent IDs once per log instead of once per attribute. Cover zero-length runs, buffer growth, and dictionary key upgrades against scalar appends.

Move fixed-vocabulary dictionary caching into shared pdata attribute builders while keeping the syslog vocabulary in the receiver. Reuse common record assembly and preserve dynamic-key collisions, ordering, widening, fallback, and batch-local cache reset behavior. Preserve and extend equivalence and reuse tests. Pdata, codec, core-node, and engine tests pass, together with all-target Clippy, formatting, and sanity checks. The workspace check encounters three unrelated filelog-checkpoint Locked failures; Rustdoc encounters existing broken links elsewhere. Compare 58 controlled full-engine runs against the preceding implementation: small RFC 3164 CPU increase, RFC 5424 and dense CEF close to baseline, lower RSS, and nearly unchanged high-load HTTP p99 latency.

Copilot

🔵 Needs a closer look

Cross-layer scheduling and ownership changes, together with substantially higher overload memory use, warrant final human validation.

1 open issue

Document shared-engine rationale for PdataDecodeJob: Send · New

What changed in this PR

This PR advances pluggable pdata codecs with cooperative syslog decoding, while reducing ingestion and Arrow construction costs.

Changes:

Adds owned incremental decode jobs through codec services and effect handlers, retaining synchronous decoding for small batches.
Supports batches up to 8,192 messages or 2 MiB; the receiver default remains 100 messages.
Optimizes TCP framing, buffer reservations, and attribute construction, with tests and documentation for lifecycle and memory tradeoffs.

File	Description
`rust/otap-dataflow/docs/async-best-practices.md`	Documents cooperative work, bounded offloading, and responsiveness measurement.
`rust/otap-dataflow/docs/ai/ai-assisted-pr-review.md`	Links runtime responsiveness review guidance.
`rust/otap-dataflow/crates/pdata/src/encode/record/attributes/known_keys.rs`	Adds opt-in key dictionary caching and equivalence tests.
`rust/otap-dataflow/crates/pdata/src/encode/record/attributes.rs`	Supports configurable key builders and bulk parent-ID appends.
`rust/otap-dataflow/crates/pdata/README.md`	Documents known-key attribute builders.
`rust/otap-dataflow/crates/pdata-codec/src/runtime.rs`	Drives incremental jobs outside the service lock.
`rust/otap-dataflow/crates/pdata-codec/src/payload.rs`	Adds cooperative payload conversion and output preparation.
`rust/otap-dataflow/crates/pdata-codec/src/lib.rs`	Exports incremental decoding interfaces.
`rust/otap-dataflow/crates/pdata-codec/src/decoder.rs`	Defines decode jobs, tasks, and work budgets.
`rust/otap-dataflow/crates/pdata-codec/src/batching.rs`	Adds cooperative batch preparation and finishing.
`rust/otap-dataflow/crates/pdata-codec/README.md`	Explains execution guarantees and remaining synchronous work.
`rust/otap-dataflow/crates/otap/src/pdata.rs`	Routes effect handlers through cooperative conversions.
`rust/otap-dataflow/crates/core-nodes/src/receivers/syslog_cef_receiver/README.md`	Documents larger limits, receipt context, and memory implications.
`rust/otap-dataflow/crates/core-nodes/src/receivers/syslog_cef_receiver/parser/parsed_message.rs`	Uses cached vocabulary slots for fixed attributes.
`rust/otap-dataflow/crates/core-nodes/src/receivers/syslog_cef_receiver/mod.rs`	Borrows buffered TCP frames and samples receipt years per refill.
`rust/otap-dataflow/crates/core-nodes/src/receivers/syslog_cef_receiver/codec.rs`	Implements cooperative decoding, larger limits, and learned reservations.
`rust/otap-dataflow/crates/core-nodes/src/receivers/syslog_cef_receiver/arrow_records_encoder/attributes.rs`	Defines receiver-owned syslog and CEF vocabulary.
`rust/otap-dataflow/crates/core-nodes/src/receivers/syslog_cef_receiver/arrow_records_encoder.rs`	Uses cached keys and bulk parent-ID appends.
`rust/otap-dataflow/crates/core-nodes/Cargo.toml`	Adds `memchr` for newline scanning.
`rust/otap-dataflow/CONTRIBUTING.md`	Adds runtime responsiveness guidance for contributors.
`rust/otap-dataflow/Cargo.lock`	Records the core-nodes `memchr` dependency.
`rust/otap-dataflow/.chloggen/cooperative-syslog-decoding.yaml`	Announces cooperative decoding and larger syslog batches.

_{Review effort: Balanced}

+/// Jobs cannot borrow the decoder or the service's runtime lock. Every pending
+/// step must make bounded CPU progress. This is not an external-I/O readiness
+/// interface: returning pending while waiting for I/O would cause a busy loop.
+pub trait PdataDecodeJob: Send {


jmacd

Very cool, this makes sense. 🚀

jmacd · 2026-10-24T01:37:46Z

+  including when surrounding awaits are immediately ready
+- [ ] Finalization and other indivisible work are assessed separately
+- [ ] Benchmarks assess CPU, memory, and responsiveness together
+- [ ] No synchronous locks or mutable codec borrows held across `.await`


jmacd · 2026-10-24T01:41:26Z

+    ///
+    /// Decode failure or cancellation preserves encoded bytes, identity, and caches.
+    /// Matching formats stay untouched; counting and encoding remain synchronous.
+    pub async fn prepare_cooperative<Y: Future<Output = ()>>(


This is wild. 💯

lalitb · 2026-10-24T02:25:42Z

                let (context, payload) = pdata.into_parts();
-                match payload.try_into_otap(self.codec_service()) {
+                match payload
+                    .try_into_otap_cooperative(self.codec_service(), tokio::task::yield_now)


Non-blocking: does the flow timer keep running during these yields? It looks like time spent running other tasks would then count toward flow.compute.duration. Other awaits already have this limitation, so I’m happy for this to be documented or handled in a follow-up.

lalitb

LGTM. The owned decode jobs and releasing codec state before yielding look good to me. Left one non-blocking question about flow timing.

utpilla · 2026-10-26T02:55:38Z

+/// One bounded, independently owned batch conversion.
+///
+/// Returning `None` requests a scheduling opportunity. Return records only after
+/// the complete batch is validated; partial output must never escape. Dropping


This preserves a batch-wide failure boundary: an error from any decode step fails materialization of the entire logical batch, including records processed successfully in earlier steps.

This is not an issue for Syslog since empty messages are rejected before batching, and the permissive parser accepts unrecognized nonempty input rather than failing conversion.

However, for a future codec that discovers invalid individual records during deferred parsing, one bad record could fail conversion of otherwise-valid records in the same batch. Preserving per-record rejection while keeping decoding lazy would then need explicit partial-success reporting with rejection accounting.

Could we leave a TODO here to revisit per-record failure isolation if a future codec needs it, including partial-success reporting and rejection accounting?

utpilla · 2026-10-26T03:09:32Z

+impl From<KnownAttributeKey> for KnownKey<{ KnownAttributeKey::COUNT }> {
+    fn from(key: KnownAttributeKey) -> Self {
+        // Every enum discriminant indexes the corresponding vocabulary entry.
+        SLOTS[key as usize]


This lookup depends on the enum discriminant order matching the separately maintained vocabulary array. The mapping is correct today, but reordering enum variants without updating the array could still compile and silently assign the wrong attribute names.

I think we could do better here. I think we could derive these from one declaration, for example, with a small local macro or add an exhaustive enum-to-name mapping test.

utpilla · 2026-10-26T03:10:53Z

+                    .map_err(|error| operation_error(CodecOperation::Decode, error));
+            };
+            let frame = frame.map_err(|error| operation_error(CodecOperation::Decode, error))?;
+            let parsed = parser::parse(frame.message)


The synchronous and incremental paths repeat the same message parsing, error translation, and append-with-receipt-context operation. Could we extract that into one private helper so future parsing or timestamp changes do not drift between execution modes?

The execution loops can remain separate, with the incremental loop retaining its budget checks and the synchronous path avoiding job allocation.

utpilla · 2026-10-26T03:18:33Z

+struct ReceiptReader<R, C> {
+    inner: R,
+    clock: C,
+    remaining: usize,


All current construction sites wrap a Tokio buffered reader. Could we make the inner field BufReader<R>, while keeping the underlying reader and clock generic, and detect refills from its existing buffer state? That would remove this mirrored remaining-byte count and its consumption bookkeeping.

The check would inspect the buffer before polling and sample only after a nonempty successful read. We should retain the clock-aware read path and the cancellation/refill/EOF tests: pending reads and empty EOF must not resample, and messages sharing buffered bytes must retain their original receipt context.

utpilla · 2026-10-26T03:23:43Z

+                    // Exercise the same scheduler-aware boundary as effect handlers,
+                    // including the final synchronous Arrow finalization.
+                    let records = payload
+                        .try_into_otap_cooperative(&service, tokio::task::yield_now)


The decode stopwatch includes finalization, but the responsiveness probe does not necessarily cover it. The timer fires only once per conversion; if it completes during an early yield, later work is no longer monitored. A later 20 ms uninterrupted pause could still fit within the 50 ms total-decode limit while the reported timer lateness remains small.

Could we use a periodic probe that continues through the whole conversion, including finalization? That would measure responsiveness throughout the batch, rather than only how promptly the first timer was serviced.

lquerel added 5 commits October 22, 2026 13:23

fix(syslog): preserve timestamp compatibility and benchmark limits

b69941a

feat(pdata): enable cooperative syslog decoding by default

3d61aee

github-project-automation Bot added this to OTel-Arrow Oct 23, 2026

github-actions Bot added area:pipeline Rust Pipeline Related Tasks lang:rust Pull requests that update Rust code area:receiver Core and Contrib receiver nodes area:syslog Work involving the syslog protocol labels Oct 23, 2026

opentelemetry-pr-dashboard Bot mentioned this pull request Oct 23, 2026

Pull Request Dashboard #3493

Open

lquerel added 8 commits October 22, 2026 22:19

Merge remote-tracking branch 'upstream/main' into pdata-codec-series-…

fd5e2f7

…05-2-cooperative-syslog

fix(pdata): cooperatively convert batch-plan inputs and outputs

6feca88

perf(syslog): learn bounded batch buffer reservations

b5c4e02

perf(syslog): amortize receipt clocks and borrow TCP frames

2e7913e

perf(syslog): bulk append Arrow attribute parent IDs

1915f03

Append repeated attribute parent IDs once per log instead of once per attribute. Cover zero-length runs, buffer growth, and dictionary key upgrades against scalar appends.

perf(syslog): reuse fixed attribute key dictionary IDs

c2ded8b

docs(pdata): add annotated syslog batch-size benchmarks

7e67152

docs: document cooperative runtime responsiveness

e98864e

github-actions Bot added the documentation Improvements or additions to documentation label Oct 23, 2026

lquerel added 2 commits October 23, 2026 14:25

docs(syslog): explain attribute key builder optimization

14f66e9

lquerel changed the title ~~feat(pdata): cooperatively decode large syslog batches~~ perf(syslog): optimize the batch codec and add cooperative decoding Oct 23, 2026

lquerel marked this pull request as ready for review October 23, 2026 23:08

lquerel requested review from a team and a balanced review from Copilot October 23, 2026 23:08

lquerel requested review from albertlockett and jmacd as code owners October 23, 2026 23:08

lquerel requested review from a team and JakeDern as code owners October 23, 2026 23:08

github-actions Bot requested a review from utpilla October 23, 2026 23:08

github-actions Bot added the size/XL label Oct 23, 2026

Copilot started reviewing on behalf of lquerel October 23, 2026 23:08

View session

Copilot AI reviewed Oct 23, 2026

View reviewed changes

jmacd approved these changes Oct 24, 2026

View reviewed changes

lalitb reviewed Oct 24, 2026

View reviewed changes

lalitb approved these changes Oct 24, 2026

View reviewed changes

lquerel mentioned this pull request Oct 25, 2026

feat(pdata): add codec bindings and named configurations #4480

Draft

utpilla reviewed Oct 26, 2026

View reviewed changes

utpilla approved these changes Oct 26, 2026

View reviewed changes

Conversation

lquerel commented Oct 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Change summary

Related issue

Validation

Full-engine benchmarks

User-facing changes

Uh oh!

opentelemetry-pr-dashboard Bot commented Oct 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Pull request dashboard status

Uh oh!

codecov Bot commented Oct 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🔵 Needs a closer look

Uh oh!

jmacd left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lalitb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

lquerel commented Oct 23, 2026 •

edited

Loading

opentelemetry-pr-dashboard Bot commented Oct 23, 2026 •

edited

Loading

codecov Bot commented Oct 23, 2026 •

edited

Loading