in_forward: batch Forward mode worker ingress - #12163
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
🚧 Files skipped from review as they are similar to previous changes (3)
📝 WalkthroughWalkthroughForward-mode processing now encodes entries through a queue-aware batch pathway, preserves partial output on malformed entries, delegates complete arrays from frame parsing, and adds worker integration coverage for large batches, ingress queue limits, acknowledgements, ordering, and valid-prefix handling. ChangesForward-mode batch processing
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant fw_prot_process
participant fw_process_forward_mode
participant fw_encode_forward_mode_entry
participant fw_ingest_forward_mode_batch
fw_prot_process->>fw_process_forward_mode: pass complete Forward-mode array
fw_process_forward_mode->>fw_encode_forward_mode_entry: encode entries
fw_process_forward_mode->>fw_ingest_forward_mode_batch: submit encoded batch
fw_ingest_forward_mode_batch-->>fw_prot_process: return processing result
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 79fbda3930
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| (ins->ingress_queue_byte_limit == 0 || | ||
| length <= ins->ingress_queue_byte_limit)); |
There was a problem hiding this comment.
When listener workers are enabled and Mem_Buf_Limit or http_server.ingress_queue_byte_limit is smaller than the encoded Forward frame—even though every constituent record fits—flb_input_ingress_enqueue() always returns FLB_INPUT_INGRESS_BUSY because the event itself exceeds the limit; this condition then declines to retry, and the caller closes the connection without an ACK or ingesting any records. Before this change, the records were queued separately and the same valid frame succeeded, so oversized frames need to be split at record boundaries before submission.
AGENTS.md reference: AGENTS.md:L230-L236
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
tests/integration/scenarios/in_forward/tests/test_in_forward_001.py (1)
874-906: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winStrengthen the assertion to catch duplicate ingestion.
This test validates ordering/content of the first 256 records but never asserts the total count. Since this test exists specifically to validate the new batch-ingest path (and the batch path includes a busy-retry loop around ingestion), also assert
len(records) == total_recordsto catch any accidental double-ingestion.✅ Proposed addition
assert ack == {"ack": chunk} + assert len(records) == total_records assert [record["value"] for record in records[:total_records]] == list( range(total_records) )As per path instructions for
tests/**: "Add or update tests for behavior changes, especially protocol parsing and encoder/decoder paths."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/scenarios/in_forward/tests/test_in_forward_001.py` around lines 874 - 906, Strengthen test_in_forward_workers_forward_mode_batch by asserting that the total number of returned records equals total_records, in addition to the existing ordering/content assertion. Place the count assertion after records are collected so duplicate ingestion is detected.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@plugins/in_forward/fw_prot.c`:
- Around line 985-989: Update the warning in the result-checking block of the
forward-processing flow to avoid labeling every failure as an event decoder
failure. Use wording that accurately covers failures returned by both the
decoder and the encoder operations, while preserving the existing result value
and return behavior.
---
Nitpick comments:
In `@tests/integration/scenarios/in_forward/tests/test_in_forward_001.py`:
- Around line 874-906: Strengthen test_in_forward_workers_forward_mode_batch by
asserting that the total number of returned records equals total_records, in
addition to the existing ordering/content assertion. Place the count assertion
after records are collected so duplicate ingestion is detected.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 2d0d3dc8-c1be-4217-a551-242728178a03
📒 Files selected for processing (2)
plugins/in_forward/fw_prot.ctests/integration/scenarios/in_forward/tests/test_in_forward_001.py
79fbda3 to
ab25a72
Compare
ab25a72 to
97789a5
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tests/integration/scenarios/in_forward/tests/test_in_forward_001.py`:
- Around line 1050-1055: Update the test around service.wait_for_record_count
and the final assertion to synchronize processing after the malformed entry,
then verify the service’s final record collection contains exactly one record.
Preserve the existing assertion that the surviving record’s message is
"valid-prefix", ensuring the malformed [TEST_TS] entry cannot produce an
additional record.
- Around line 988-993: Update the shared _recv_msgpack_value() helper used by
both tests to accumulate TCP reads in a buffer and continue reading until a
complete MessagePack value can be decoded, rather than assuming one recv returns
the full ACK. Preserve the existing decoded-value return behavior and timeout
handling while correctly handling fragmented responses.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: abd86463-64cb-4ee2-ad0f-a07bb2b94baa
📒 Files selected for processing (3)
plugins/in_forward/fw_prot.ctests/integration/scenarios/in_forward/config/in_forward_workers_tiny_ingress_queue.yamltests/integration/scenarios/in_forward/tests/test_in_forward_001.py
🚧 Files skipped from review as they are similar to previous changes (1)
- plugins/in_forward/fw_prot.c
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>
6e8119e to
0a847d9
Compare
2 participants
Problem
Forward array mode processed and submitted every record independently. With listener workers, a 100-record Forward frame therefore produced 100 deferred-ingress allocations, payload copies, queue lock operations, and main-thread append calls. The listener threads were balanced, but the main ingestion thread saturated and four-worker throughput barely exceeded one worker.
Change
Message mode, PackedForward, gzip/zstd PackedForward, TLS, Unix sockets, tags, metadata, and secure-forward negotiation remain on their existing paths.
Performance
On the same 10-million-record, 16-client benchmark with 100 records per Forward frame:
That is a 5.25x throughput increase while using 63% less total CPU time for the workload.
Validation
cmake --build build -j8VALGRIND=1 VALGRIND_STRICT=1: 13 passed, zero Valgrind errorsGITHUB_EVENT_NAME=pull_request GITHUB_BASE_REF=master python .github/scripts/commit_prefix_check.pygit diff --checkSummary by CodeRabbit