Repository navigation
fix: add gRPC length-prefixed framing to OTLP gRPC output - #628
Conversation
Agent-Logs-Url: https://github.com/strawgate/memagent/sessions/d70e44f8-32b2-4827-a899-848bdfe3cb19 Co-authored-by: strawgate <6384545+strawgate@users.noreply.github.com>
|
@coderabbitai full review |
✅ Actions performedFull review triggered. |
|
Important Review skippedBot user detected. To trigger a single review, invoke the ⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
WalkthroughThe 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@crates/logfwd-output/src/otlp_sink.rs`:
- Around line 741-746: The CI failure is due to rustfmt formatting drift in the
assert block using assert_eq! with msg_len, proto_payload and framed; fix by
running rustfmt (cargo fmt --all) to reformat
crates/logfwd-output/src/otlp_sink.rs so the two assert_eq! calls
(assert_eq!(msg_len as usize, proto_payload.len(), ...) and
assert_eq!(&framed[5..], &proto_payload, ...)) conform to project style, then
stage and commit the reformatted file(s).
- Around line 189-195: The send_batch hot path currently allocates a new Vec
each batch via grpc_buf = grpc_frame(payload); instead reuse a sink-owned buffer
(e.g., add a field self.grpc_buf: Vec<u8>) and write the framed bytes into it
instead of allocating: clear the buffer (self.grpc_buf.clear()), call grpc_frame
or an equivalent framing helper to append into self.grpc_buf, then set
payload_ref to &self.grpc_buf when self.protocol == OtlpProtocol::Grpc; replace
the local grpc_buf usage with self.grpc_buf to avoid per-batch allocations in
send_batch.
- Around line 185-195: The current gRPC path frames already-compressed messages
with grpc_frame(payload) using the wrong compression flag and relies on HTTP
Content-Encoding (which gRPC ignores); fast safe fix: when self.protocol ==
OtlpProtocol::Grpc, detect if the outgoing payload is already compressed and
reject or remap that path instead of sending an invalid gRPC message — e.g., in
the send path where grpc_frame(payload) is chosen, return an error if a
Content-Encoding (or caller-provided compression marker) is set, or refuse to
use OtlpProtocol::Grpc with compressed payloads; also avoid per-batch Vec<u8>
allocations by reusing a preallocated buffer (move the temporary grpc_buf into a
reusable buffer on the sink struct or use a shared bytes buffer) so grpc_frame
no longer allocates each batch. Ensure you modify the branch that checks
self.protocol == OtlpProtocol::Grpc and the grpc_frame usage accordingly.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 1386cb8f-8ac5-48ff-914b-a4a163caf47c
📒 Files selected for processing (1)
crates/logfwd-output/src/otlp_sink.rs
| // For gRPC, prepend the 5-byte length-prefixed frame header required by the | ||
| // gRPC wire protocol. Note: ureq uses HTTP/1.1; a true gRPC endpoint requires | ||
| // HTTP/2. Use a reverse proxy (e.g. Envoy) or `protocol: http` when the | ||
| // collector does not accept HTTP/1.1 upgrades. | ||
| let grpc_buf: Vec<u8>; | ||
| let payload: &[u8] = if self.protocol == OtlpProtocol::Grpc { | ||
| grpc_buf = grpc_frame(payload); | ||
| &grpc_buf | ||
| } else { | ||
| payload | ||
| }; |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
In gRPC over HTTP/2, when message payload is compressed, must the 1-byte gRPC frame compressed flag be set to 1 and grpc-encodingbe used (instead of HTTPContent-Encoding)?
💡 Result:
Yes, in gRPC over HTTP/2, when the message payload is compressed, the 1-byte gRPC frame compressed flag must be set to 1, and the grpc-encoding header (referred to as Message-Encoding in the protocol spec) must be used to declare the compression mechanism (e.g., "gzip"). The standard HTTP Content-Encoding header is not used for gRPC message compression; instead, gRPC defines its own grpc-encoding header for this purpose. If grpc-encoding is omitted (or set to "identity"), the compressed flag must be 0. Setting the flag to 1 without a corresponding non-identity grpc-encoding results in an INTERNAL error.
Citations:
- 1: https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md?plain=1
- 2: https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md
- 3: https://www.grpc.io/grpc/core/md_doc_compression.html
- 4: https://kreya.app/blog/grpc-deep-dive/
- 5: https://grpc.github.io/grpc/core/md_doc__p_r_o_t_o_c_o_l-_h_t_t_p2.html
- 6: https://chromium.googlesource.com/external/github.com/grpc/grpc/%2B/HEAD/doc/PROTOCOL-HTTP2.md
- 7: https://www.grpc.io/grpc/core/md_doc__p_r_o_t_o_c_o_l-_h_t_t_p2.html
- 8: https://fuchsia.googlesource.com/third_party/grpc/+/HEAD/doc/PROTOCOL-HTTP2.md
- 9: https://grpc.github.io/grpc/core/md_doc_compression.html
🏁 Script executed:
cd crates/logfwd-output && wc -l src/otlp_sink.rsRepository: strawgate/memagent
Length of output: 82
🏁 Script executed:
cd crates/logfwd-output && sed -n '160,210p' src/otlp_sink.rs | cat -nRepository: strawgate/memagent
Length of output: 2687
🏁 Script executed:
cd crates/logfwd-output && sed -n '555,580p' src/otlp_sink.rs | cat -nRepository: strawgate/memagent
Length of output: 1062
🏁 Script executed:
cd crates/logfwd-output && sed -n '185,200p' src/otlp_sink.rs | cat -nRepository: strawgate/memagent
Length of output: 1014
gRPC compression requires frame flag=1 and grpc-encoding header; current code breaks decompression.
Lines 185–195 frame already-compressed payload with flag 0x00 (line 565), but gRPC spec requires flag 0x01 and the grpc-encoding header when message is compressed. Current code sends Content-Encoding: zstd (line 210), which gRPC ignores for message compression. Receivers will fail to decompress with INTERNAL error.
Secondary: per-batch Vec<u8> allocation (line 190) in hot path should be reused or buffered.
Safe immediate fix: reject unsupported combination
fn send_batch(&mut self, batch: &RecordBatch, metadata: &BatchMetadata) -> io::Result<()> {
+ if self.protocol == OtlpProtocol::Grpc && self.compression != Compression::None {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidInput,
+ "OTLP gRPC compression requires grpc-encoding signaling and is not implemented",
+ ));
+ }
+
self.encode_batch(batch, metadata);🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd-output/src/otlp_sink.rs` around lines 185 - 195, The current
gRPC path frames already-compressed messages with grpc_frame(payload) using the
wrong compression flag and relies on HTTP Content-Encoding (which gRPC ignores);
fast safe fix: when self.protocol == OtlpProtocol::Grpc, detect if the outgoing
payload is already compressed and reject or remap that path instead of sending
an invalid gRPC message — e.g., in the send path where grpc_frame(payload) is
chosen, return an error if a Content-Encoding (or caller-provided compression
marker) is set, or refuse to use OtlpProtocol::Grpc with compressed payloads;
also avoid per-batch Vec<u8> allocations by reusing a preallocated buffer (move
the temporary grpc_buf into a reusable buffer on the sink struct or use a shared
bytes buffer) so grpc_frame no longer allocates each batch. Ensure you modify
the branch that checks self.protocol == OtlpProtocol::Grpc and the grpc_frame
usage accordingly.
| let grpc_buf: Vec<u8>; | ||
| let payload: &[u8] = if self.protocol == OtlpProtocol::Grpc { | ||
| grpc_buf = grpc_frame(payload); | ||
| &grpc_buf | ||
| } else { | ||
| payload | ||
| }; |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Avoid per-batch allocation for gRPC framing buffer in send_batch.
At Line 189-193, grpc_buf = grpc_frame(payload) allocates a new Vec<u8> each batch. Reuse a sink-owned buffer (self.grpc_buf) to reduce allocator churn in the output hot path.
♻️ Suggested refactor (reuse buffer)
pub struct OtlpSink {
name: String,
endpoint: String,
protocol: OtlpProtocol,
compression: Compression,
headers: Vec<(String, String)>,
pub(crate) encoder_buf: Vec<u8>,
compress_buf: Vec<u8>,
+ grpc_buf: Vec<u8>,
compressor: Option<ChunkCompressor>,
http_agent: ureq::Agent,
stats: Arc<ComponentStats>,
}
@@
OtlpSink {
@@
encoder_buf: Vec::with_capacity(64 * 1024),
compress_buf: Vec::with_capacity(64 * 1024),
+ grpc_buf: Vec::with_capacity(64 * 1024),
compressor,
http_agent,
stats,
}
@@
- let grpc_buf: Vec<u8>;
let payload: &[u8] = if self.protocol == OtlpProtocol::Grpc {
- grpc_buf = grpc_frame(payload);
- &grpc_buf
+ self.grpc_buf.clear();
+ self.grpc_buf.reserve(5 + payload.len());
+ self.grpc_buf.push(0x00);
+ self.grpc_buf
+ .extend_from_slice(&(payload.len() as u32).to_be_bytes());
+ self.grpc_buf.extend_from_slice(payload);
+ &self.grpc_buf
} else {
payload
};As per coding guidelines: "Do not introduce allocations in hot paths without benchmarking".
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd-output/src/otlp_sink.rs` around lines 189 - 195, The send_batch
hot path currently allocates a new Vec each batch via grpc_buf =
grpc_frame(payload); instead reuse a sink-owned buffer (e.g., add a field
self.grpc_buf: Vec<u8>) and write the framed bytes into it instead of
allocating: clear the buffer (self.grpc_buf.clear()), call grpc_frame or an
equivalent framing helper to append into self.grpc_buf, then set payload_ref to
&self.grpc_buf when self.protocol == OtlpProtocol::Grpc; replace the local
grpc_buf usage with self.grpc_buf to avoid per-batch allocations in send_batch.
| assert_eq!( | ||
| msg_len as usize, | ||
| proto_payload.len(), | ||
| "length field must match payload length" | ||
| ); | ||
| assert_eq!(&framed[5..], &proto_payload, "payload bytes must follow header"); |
There was a problem hiding this comment.
CI blocker: rustfmt mismatch in this assert block.
Pipeline already reports formatting drift around this assert (Line 741-746). Please run cargo fmt --all before merge.
🧰 Tools
🪛 GitHub Actions: CI
[error] 743-746: cargo fmt --check failed: formatting diff in assert_eq! message formatting. Run cargo fmt (or cargo fmt --all) to apply the expected rustfmt changes.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@crates/logfwd-output/src/otlp_sink.rs` around lines 741 - 746, The CI failure
is due to rustfmt formatting drift in the assert block using assert_eq! with
msg_len, proto_payload and framed; fix by running rustfmt (cargo fmt --all) to
reformat crates/logfwd-output/src/otlp_sink.rs so the two assert_eq! calls
(assert_eq!(msg_len as usize, proto_payload.len(), ...) and
assert_eq!(&framed[5..], &proto_payload, ...)) conform to project style, then
stage and commit the reformatted file(s).
Resolve conflict in otlp_sink.rs: keep both encode_key_value_bool (from master) and gRPC framing changes (from this branch). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
OtlpSinksent raw protobuf bytes for both HTTP and gRPC protocols — only theContent-Typeheader differed. gRPC requires every message to be wrapped in a 5-byte frame header before the protobuf payload; without it, any standard gRPC collector rejects the request.Changes
grpc_frame(payload)— new helper that prepends the gRPC wire frame:send_batch— wraps the payload throughgrpc_framewhenprotocol == OtlpProtocol::Grpcusing a maybe-initialized variable to avoid anOptionwrapper:ureqis HTTP/1.1 only; a real gRPC endpoint requires HTTP/2 (use Envoy sidecar orprotocol: httpas workaround).grpc_frame_prepends_five_byte_headerandgrpc_frame_empty_payloadverify the framing invariants.