out_forward plugin starts sending invalid packets

[ChristThePhone]

Describe the bug

fluentd 1.18

I'm using the out_forward plugin to let it forward incoming logs to my proprietary log server. Every once in a while it starts sending invalid messages consisting of 7 bytes : 93 a5 7a 2e 6c 6f 67

Once fluentd enters this state is keeps sending this message over and over again, incoming logs get buffered in files until they reach their max size and new logs are no longer accepted and therefore lost.

I enabled trace level logging for the out_forward plugin

To Reproduce

I cannot reproduce at will

Expected behavior

fluentd out_forward plugin should not send invalid packets and if it does it should not do so over and over again

Your Environment

- Fluentd version: 1.18
- Package version:
- Operating system: RHEL 9 Linux
- Kernel version:

Your Configuration

<match z.log>

     @type forward
     @log_level trace

     <server>
        name <myServerName>
        host 127.0.0.1
        port 24225
        ignore_network_errors_at_startup true
        verify_connection_at_startup false
     </server>

     <buffer>
        @type file
        path /<path to buffers>/zlog.*.buf
        flush_mode interval
        flush_interval 1s
        overflow_action throw_exception
        total_limit_size 1GB
        queued_chunks_limit_size 1024
        compress gzip
        retry_forever true
        retry_type periodic
        retry_wait 1s
     </buffer>

</match>

Your Error Log

2025-11-24 15:09:40 +0100 [trace]: dequeueing a chunk instance=2260
2025-11-24 15:09:40 +0100 [trace]: chunk dequeued instance=2260 metadata=#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag="z.log", variables=nil, seq=0>
2025-11-24 15:09:40 +0100 [trace]: trying flush for a chunk chunk="64457b6c5c3bd14110513c43a4d156cd"
2025-11-24 15:09:40 +0100 [trace]: adding write count instance=2240
2025-11-24 15:09:40 +0100 [trace]: executing sync write chunk="64457b6c5c3bd14110513c43a4d156cd"
2025-11-24 15:09:40 +0100 [debug]: connect new socket
2025-11-24 15:09:40 +0100 [trace]: write operation done, committing chunk="64457b6c5c3bd14110513c43a4d156cd"
2025-11-24 15:09:40 +0100 [trace]: committing write operation to a chunk chunk="64457b6c5c3bd14110513c43a4d156cd" delayed=false
2025-11-24 15:09:40 +0100 [trace]: purging a chunk instance=2260 chunk_id="64457b6c5c3bd14110513c43a4d156cd" metadata=#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag="z.log", variables=nil, seq=0>
2025-11-24 15:09:40 +0100 [trace]: chunk purged instance=2260 chunk_id="64457b6c5c3bd14110513c43a4d156cd" metadata=#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag="z.log", variables=nil, seq=0>
2025-11-24 15:09:40 +0100 [trace]: done to commit a chunk chunk="64457b6c5c3bd14110513c43a4d156cd"
2025-11-24 15:09:40.299441051 +0100 fluent.trace: {"instance":2260,"message":"dequeueing a chunk instance=2260"}
2025-11-24 15:09:40.299556774 +0100 fluent.trace: {"instance":2260,"metadata":"#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag=\"z.log\", variables=nil, seq=0>","message":"chunk dequeued instance=2260 metadata=#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag=\"z.log\", variables=nil, seq=0>"}
2025-11-24 15:09:40.299602102 +0100 fluent.trace: {"chunk":"64457b6c5c3bd14110513c43a4d156cd","message":"trying flush for a chunk chunk=\"64457b6c5c3bd14110513c43a4d156cd\""}
2025-11-24 15:09:40.299639559 +0100 fluent.trace: {"instance":2240,"message":"adding write count instance=2240"}
2025-11-24 15:09:40.299662757 +0100 fluent.trace: {"chunk":"64457b6c5c3bd14110513c43a4d156cd","message":"executing sync write chunk=\"64457b6c5c3bd14110513c43a4d156cd\""}
2025-11-24 15:09:40.299701023 +0100 fluent.debug: {"message":"connect new socket"}
2025-11-24 15:09:40.302766920 +0100 fluent.trace: {"chunk":"64457b6c5c3bd14110513c43a4d156cd","message":"write operation done, committing chunk=\"64457b6c5c3bd14110513c43a4d156cd\""}
2025-11-24 15:09:40.302835320 +0100 fluent.trace: {"chunk":"64457b6c5c3bd14110513c43a4d156cd","delayed":false,"message":"committing write operation to a chunk chunk=\"64457b6c5c3bd14110513c43a4d156cd\" delayed=false"}
2025-11-24 15:09:40.302873690 +0100 fluent.trace: {"instance":2260,"chunk_id":"64457b6c5c3bd14110513c43a4d156cd","metadata":"#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag=\"z.log\", variables=nil, seq=0>","message":"purging a chunk instance=2260 chunk_id=\"64457b6c5c3bd14110513c43a4d156cd\" metadata=#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag=\"z.log\", variables=nil, seq=0>"}
2025-11-24 15:09:40.302983926 +0100 fluent.trace: {"instance":2260,"chunk_id":"64457b6c5c3bd14110513c43a4d156cd","metadata":"#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag=\"z.log\", variables=nil, seq=0>","message":"chunk purged instance=2260 chunk_id=\"64457b6c5c3bd14110513c43a4d156cd\" metadata=#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag=\"z.log\", variables=nil, seq=0>"}
2025-11-24 15:09:40.303023355 +0100 fluent.trace: {"chunk":"64457b6c5c3bd14110513c43a4d156cd","message":"done to commit a chunk chunk=\"64457b6c5c3bd14110513c43a4d156cd\""}
2025-11-24 15:09:40 +0100 [trace]: writing events into buffer instance=2260 metadata_size=1
2025-11-24 15:09:40 +0100 [debug]: Created new chunk chunk_id="64457b6d67cb7d7c1d004aa1758ad9b9" metadata=#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag="z.log", variables=nil, seq=0>
2025-11-24 15:09:40 +0100 [trace]: chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 159
2025-11-24 15:09:40.617262760 +0100 fluent.trace: {"instance":2260,"metadata_size":1,"message":"writing events into buffer instance=2260 metadata_size=1"}
2025-11-24 15:09:40.617614515 +0100 fluent.debug: {"chunk_id":"64457b6d67cb7d7c1d004aa1758ad9b9","metadata":"#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag=\"z.log\", variables=nil, seq=0>","message":"Created new chunk chunk_id=\"64457b6d67cb7d7c1d004aa1758ad9b9\" metadata=#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag=\"z.log\", variables=nil, seq=0>"}
2025-11-24 15:09:40.617796147 +0100 fluent.trace: {"message":"chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 159"}
2025-11-24 15:09:40 +0100 [trace]: writing events into buffer instance=2260 metadata_size=1
2025-11-24 15:09:40 +0100 [trace]: chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 318
2025-11-24 15:09:40.618903549 +0100 fluent.trace: {"instance":2260,"metadata_size":1,"message":"writing events into buffer instance=2260 metadata_size=1"}
2025-11-24 15:09:40.619042316 +0100 fluent.trace: {"message":"chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 318"}
2025-11-24 15:09:40 +0100 [trace]: sending heartbeat host="127.0.0.1" port=24225 heartbeat_type=:transport
2025-11-24 15:09:40 +0100 [debug]: connect new socket
2025-11-24 15:09:40.983441674 +0100 fluent.trace: {"host":"127.0.0.1","port":24225,"heartbeat_type":"transport","message":"sending heartbeat host=\"127.0.0.1\" port=24225 heartbeat_type=:transport"}
2025-11-24 15:09:40.983567212 +0100 fluent.debug: {"message":"connect new socket"}
2025-11-24 15:09:41 +0100 [trace]: writing events into buffer instance=2260 metadata_size=1
2025-11-24 15:09:41 +0100 [trace]: chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 477
2025-11-24 15:09:41.122298570 +0100 fluent.trace: {"instance":2260,"metadata_size":1,"message":"writing events into buffer instance=2260 metadata_size=1"}
2025-11-24 15:09:41.122606090 +0100 fluent.trace: {"message":"chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 477"}
2025-11-24 15:09:41 +0100 [trace]: writing events into buffer instance=2260 metadata_size=1
2025-11-24 15:09:41 +0100 [trace]: chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 636
2025-11-24 15:09:41.129487541 +0100 fluent.trace: {"instance":2260,"metadata_size":1,"message":"writing events into buffer instance=2260 metadata_size=1"}
2025-11-24 15:09:41.129718161 +0100 fluent.trace: {"message":"chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 636"}
2025-11-24 15:09:41 +0100 [trace]: writing events into buffer instance=2260 metadata_size=1
2025-11-24 15:09:41.132771965 +0100 fluent.trace: {"instance":2260,"metadata_size":1,"message":"writing events into buffer instance=2260 metadata_size=1"}
2025-11-24 15:09:41 +0100 [trace]: chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 7395 new_size: 8031
2025-11-24 15:09:41.133715266 +0100 fluent.trace: {"message":"chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 7395 new_size: 8031"}
2025-11-24 15:09:41 +0100 [trace]: writing events into buffer instance=2260 metadata_size=1
2025-11-24 15:09:41.182099288 +0100 fluent.trace: {"instance":2260,"metadata_size":1,"message":"writing events into buffer instance=2260 metadata_size=1"}
2025-11-24 15:09:41 +0100 [trace]: chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 8190
2025-11-24 15:09:41.182447693 +0100 fluent.trace: {"message":"chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 8190"}
2025-11-24 15:09:41 +0100 [trace]: writing events into buffer instance=2260 metadata_size=1
2025-11-24 15:09:41.184271628 +0100 fluent.trace: {"instance":2260,"metadata_size":1,"message":"writing events into buffer instance=2260 metadata_size=1"}
2025-11-24 15:09:41 +0100 [trace]: chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 8349
2025-11-24 15:09:41.184573445 +0100 fluent.trace: {"message":"chunk /ffdc/fluent/buffers/zlog.b64457b6d67cb7d7c1d004aa1758ad9b9.buf size_added: 159 new_size: 8349"}
2025-11-24 15:09:41 +0100 [trace]: enqueueing all chunks in buffer instance=2260
2025-11-24 15:09:41 +0100 [trace]: enqueueing chunk instance=2260 metadata=#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag="z.log", variables=nil, seq=0>
2025-11-24 15:09:41.224905148 +0100 fluent.trace: {"instance":2260,"message":"enqueueing all chunks in buffer instance=2260"}
2025-11-24 15:09:41.225123287 +0100 fluent.trace: {"instance":2260,"metadata":"#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag=\"z.log\", variables=nil, seq=0>","message":"enqueueing chunk instance=2260 metadata=#<struct Fluent::Plugin::Buffer::Metadata timekey=nil, tag=\"z.log\", variables=nil, seq=0>"}

Additional context

No response

[Watson1978]

Hmmm

Looking at the byte sequence 93 a5 7a 2e 6c 6f 67 as MessagePack, it looks like the data is cut off after ["z.log" (a 3-element array header + the string "z.log"). It seems the packet is being sent partially or is truncated.

[Watson1978]

Could you please clarify the nature of your 'proprietary log server'?

Is it another Fluentd instance running the in_forward plugin,
or is it a custom-developed application?

[ChristThePhone]

It is a custom developed socket server written in C++ to service the forward protocol

[Watson1978]

Thank you for clarifying that your log server is a custom application implemented in C++.

out_forward uses IO.copy_stream to flush a log chunk into socket as stream,
the TCP protocol does not guarantee message boundaries.

The 7-byte sequence you reported (93 a5 7a 2e 6c 6f 67) looks like a valid fragment of a MessagePack message.
What data are you receiving after this?
I think it may receive the remaining data during the next time...

If your C++ server attempts to decode a MessagePack object immediately upon receiving data,
without implementing a proper stream parser or buffering layer,
it would treat this fragment as a fatal error (e.g., "invalid message").

in_forward will decode the data with buffering in:

fluentd/lib/fluent/plugin/in_forward.rb

Lines 262 to 265 in b4dc14d

	parser.feed_each(d){|obj|
	block.call(obj, bytes, serializer)
	bytes = 0
	}

Your C++ server should also require similar decoding processing.

[ChristThePhone]

Thanks for the link to the in_forward plugin implementation, I will make sure my code works in a similar way, it currently reads data from the socket until the fluentd side closes it and then decodes the received data via msgpack. Unfortunately the out_forward plugin keeps repeating this invalid 7-byte sequence (opening the connection, sending these 7 bytes, then closing the socket) over and over again and the only way out is to stop fluentd, delete all buffered events, and restart fluentd.

[Watson1978]

If possible, can you capture and attach a series of TCP packets?
Then, we may find the way to reproduce your issue.

[ChristThePhone]

If possible, can you capture and attach a series of TCP packets? Then, we may find the way to reproduce your issue.

Unfortunately not, I cannot have customer systems run a tcp trace and I only get to know after the facts.

[Watson1978]

out_forward sends the file contents as-is.
If the buffer file handled by Fluentd becomes corrupted due to system failure (etc.), corrupted data will be sent.

in_forward performs a simple check to ensure the data is not corrupted.

fluentd/lib/fluent/plugin/in_forward.rb

Lines 367 to 381 in b4dc14d

	def invalid_event?(tag, time, record)
	!((time.is_a?(Integer) || time.is_a?(::Fluent::EventTime)) && record.is_a?(Hash) && tag.is_a?(String))
	end
	def check_and_skip_invalid_event(tag, es, remote_host)
	new_es = Fluent::MultiEventStream.new
	es.each { |time, record|
	if invalid_event?(tag, time, record)
	log.warn "skip invalid event:", host: remote_host, tag: tag, time: time, record: record
	next
	end
	new_es.add(time, record)
	}
	new_es
	end

[ChristThePhone]

Then the out_forward buffering code should check the validity of the files ? I have no indication of a problem writing files so I would not exclude the possibility that out_forward is already generating these invalid files.

[ChristThePhone]

I will try to get my hands on a system in the failing state to see what the buffer files look like. If your theory of corrupted files is correct then grabbing these files and putting them on a test system should reproduce the issue, correct ?