Fluent receiver protocol #242

mheffner · 2025-11-26T22:33:05Z

This is a minimal implementation of the Fluent (forwarder) protocol used by Fluentd, Fluent Bit, Docker's Fluent driver, etc. It is essentially MessagePack logs delivered over streaming TCP or Unix sockets.

Log messages can be sent individually in single message pack structures. Instead of populating single-log OTLP resource record structures, we introduce a small batching component to the fluent receiver. This will continue reading additional messages and construct a single resource record. While we perform batching at the pipeline level, this is after the processing step, so reducing the number of payloads reduces processing calls. This is hard coded at the moment, but we can make the configurable in the future.

Since this is early and experimental, I've feature flagged it initially. Depending on the stability, binary size increase, and usefulness we can move this to a default feature later.

Performance: Tested locally and I see about 125k logs/sec/core with Docker's fluent log driver.

Remaining pieces to follow on:

compressed forwarded message support (gzip)
delivery acknowledgement (requires sender to opt-in)

rjenkins · 2025-12-11T01:11:18Z

src/receivers/fluent/receiver.rs

+                let curr_batch = batch.take();
+
+                pending_encode = Some(tokio::task::spawn_blocking(move || {
+                    Some(convert_to_otlp_logs(curr_batch))


Do we need to consider partitioning of these batches into separate ResourceLogs and ScopeLogs where applicable? Particularly for the TCP connection endpoint? So for example if there is a tag that is "service.name" do we want to hoist that into a Resource attribute and partition these accordingly?

Great question, at the moment this is mostly following the OTEL collector implementation that serializes all tags to the log record attributes. In fluent bit you can assign tags to messages, but I think it's done at the input/output level. Meaning I don't know how often you would see a tag, like service.name, be different for messages received on a single socket connection.

OK, so it sounds like we're not messing up the OTel Resource/Scope hierarchy, specifically because we're not sticking anything in them anyway 🙃, feels like hoisting is something we could look at later as a nice to have. With hoisting though, we have to keep partitioning in mind so a bit of a pita. We'll see if anyone asks for it.

rjenkins · 2025-12-11T01:43:25Z

src/receivers/fluent/receiver.rs

+        let mut batch = Batch::new();
+
+        // Track single pending encoding task
+        let mut pending_encode: Option<tokio::task::JoinHandle<Option<ResourceLogs>>> = None;


Is there a good reason to use a single encoding task vs. an OrderedFutures similar to how we do this in the exporters?

So this receiver loop is operating per-connection, so if we ended up with 10 concurrent connections we'd have up to 10 concurrent encoding tasks. I guess there were two alternatives we could do, but neither really felt great to me:

Have a shared FuturesOrdered across connections (with some locking). Could this lead to starvation and unfair consumption across connections?

Have a FuturesOrdered per connection. This could lead to a large number of encoding tasks if there were many connections. We would likely want some global max pending futures across all connections to limit memory.

Hmm, OK so it's essentially unbounded, or bounded only by the number of connections. I think it's fine in the sense that we're not artificially constricting throughput but is there a likelihood this could eat up a lot of heap? IIRC the default max threads for spawn blocking is 512, so in terms of stack space it shouldn't be too much, but if there are 512 decoders running simultaneously I wonder if they'd eat up a lot more memory. When you ran your 125k logs/sec/core how many connections did you test?

Yeah, it looks like all of our http-based receivers could have an unbounded memory allocation explosion. Looking at the OTEL Collector they lean on the MaxConnsPerHost config from the Go Transport to limit concurrent connections. That's sort of a round-a-bout way to handle it, we could go the same way or ideally have a "receiver max memory" control that handles the limitation in a smarter fashion.

rjenkins · 2025-12-11T13:31:40Z

src/receivers/fluent/receiver.rs

+                                // Initiate async send without awaiting
+                                if let Some(logs_output) = &self.logs_output {
+                                    let payload_msg = payload::Message::new(None, vec![resource_logs]);
+                                    pending_send = Some(logs_output.send_async(payload_msg));


This is kind of cool because you essentially end up hoisting this select logic up to the main select! so you don't have to await here and also select on cancellation. However I find managing the state (setting and clearing pending_send, pending_encode) to be quite complex and potentially error prone on refactor. Hopefully we can find patterns to encapsulate this in the future.

Yeah, it would be nice to layer this into some abstractions so it's less messy and we can handle it similarly across the board.

src/receivers/fluent/receiver.rs

rjenkins · 2025-12-11T13:52:30Z

src/receivers/fluent/receiver.rs

+        let mut batch = Batch::new();
+
+        // Track single pending encoding task
+        let mut pending_encode: Option<tokio::task::JoinHandle<Option<ResourceLogs>>> = None;


Hmm, OK so it's essentially unbounded, or bounded only by the number of connections. I think it's fine in the sense that we're not artificially constricting throughput but is there a likelihood this could eat up a lot of heap? IIRC the default max threads for spawn blocking is 512, so in terms of stack space it shouldn't be too much, but if there are 512 decoders running simultaneously I wonder if they'd eat up a lot more memory. When you ran your 125k logs/sec/core how many connections did you test?

rjenkins · 2025-12-11T15:29:05Z

src/receivers/fluent/convert.rs

+                Some(any_value::Value::BytesValue(s.as_bytes().to_vec()))
+            }
+        }
+        Value::Binary(b) => Some(any_value::Value::BytesValue(b.clone())),


Unnecessary clone?

rjenkins · 2025-12-11T15:36:21Z

src/receivers/fluent/convert.rs

+        Value::Ext(_tag, data) => {
+            // Extension types are converted to bytes with a special attribute for the tag
+            // For now, just convert to bytes
+            Some(any_value::Value::BytesValue(data.clone()))


Unnecessary clone?

I pulled on this a bit more by switching to ownership over reference for the whole conversion. There's no need to maintain reference since we are fully converting here. Anyways, pushed a couple of commits that refactor to this and allow for removing the clones.

rjenkins

Left a couple small fixes/suggestions. Only other concern would be potentially unbounded memory growth for encoders but otherwise looking good.

Co-authored-by: Ray Jenkins <ray@streamfold.com>

mheffner added 25 commits November 26, 2025 11:34

WIP

4f9c948

framing

2c77e47

Move task spawning into fluent receiver

f0c651a

Push down cancellation token

6715b49

Some renames

00b64c8

Clean up initialization to just logs

838b934

WIP otlp convert

0e58205

Cleanups on conversion to OTLP

ec29571

Send async

b31a54a

Fix for single encoding future per conection

45af058

Fix arg parsing, remove printout

84f243b

Remove drop

8210bf3

Remove logging

027634c

Move to feature flag and fix spawning

1219c82

Fix config

f07ca3f

Switch to direct serde, fix timeout

60cac37

Cleanups and performance

11b56c9

Fix tests

ec0b929

Drain batch on exit

93eb678

Fix batching for Forward messages

d0c3c64

Rename to socket_path to better match

128f034

Update readme

ceed7b5

Comments

7a1eefa

Send full batch immediately after encoding

3f8a9b8

Lost biasing in refactor

c001c5c

mheffner force-pushed the fluent branch from a3519d6 to c001c5c Compare December 1, 2025 23:54

mheffner marked this pull request as ready for review December 1, 2025 23:57

mheffner requested a review from rjenkins December 1, 2025 23:57

mheffner added 2 commits December 2, 2025 16:54

Add counters

5672210

Fix conditional_wait

83aed54

rjenkins reviewed Dec 11, 2025

View reviewed changes

rjenkins approved these changes Dec 11, 2025

View reviewed changes

mheffner and others added 4 commits December 14, 2025 15:43

Merge from main.

656ad5c

Update src/receivers/fluent/receiver.rs

7f75944

Co-authored-by: Ray Jenkins <ray@streamfold.com>

Switch to ownership on record convert

3e34fb0

Remove additional clones

1218d49

rjenkins approved these changes Dec 16, 2025

View reviewed changes

mheffner merged commit ec30d9f into main Dec 16, 2025
4 checks passed

Fluent receiver protocol #242

Fluent receiver protocol #242

Uh oh!

Conversation

mheffner commented Nov 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rjenkins Dec 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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!

rjenkins left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mheffner commented Nov 26, 2025 •

edited

Loading

rjenkins Dec 11, 2025 •

edited

Loading