Fluentd emits invalid msgpacked chunks to output when chunks over the size limit are consumed #2084
Comments
|
Actually I think this has to be fixed in fluentd, as that's what provides the ChunkMessagePackEventStreamer mixin, which directly uses the chunk's io. So there's no way to handle this in the plugin when using |
|
Is this intentional that you use I've migrated your code to v1 API like followings. require 'fluent/plugin/output'
require 'fluent/test'
require 'fluent/test/driver/output'
module Fluent
module Plugin
class TestOutput < Fluent::Plugin::Output
Fluent::Plugin.register_output('out_test', self)
helpers :formatter
def configure(conf)
super
@formatter = formatter_create
end
def format(tag, time, record)
[@formatter.format(tag, time, record).chomp.b].to_msgpack
end
def write(chunk)
chunk.open do |io|
begin
data = io.read
records = Fluent::Engine.msgpack_unpacker(StringIO.new(data)).to_enum.to_a
puts "data #{data.size / 1024} KB - #{records.size} records"
puts records.first
puts "^^ this should not happen - msgpack parsing error" unless records.first.is_a? Array
end
end
end
def multi_workers_ready?
true
end
end
end
end
$driver = Fluent::Test::Driver::Output.new(Fluent::Plugin::TestOutput).configure <<~CONF
log_level debug
<format>
@type json
</format>
<buffer>
chunk_limit_size "1m"
</buffer>
CONF
$driver.run(force_flush_retry: true) do
5.times do
time = Fluent::EventTime.now
events = Array.new(Kernel.rand(4000..6000)).enum_for(:each_with_index).map { |_,i| [time, { msg: ('a'.ord + i % 26).chr * 256 }] }
puts "emitting #{events.size} events"
$driver.feed("my.tag", events)
end
end
puts $driver.logs |
|
It is intentional as I'm debugging an issue i'm seeing in the https://github.com/awslabs/aws-fluent-plugin-kinesis plugin, which inherits from the BufferedOutput class so that it can be compatible with both Fluentd v0.12 and v1.0. As shown in the example, the I believe I tried upgrading the plugin to use the new API and remove the mixins, however I got other errors relating to not being able to iterate the chunk in the same way. |
|
@adammw I've sent the PR to remove previous tricky msgpack unpacker code. Please check it? |
|
I'm poking around this issue a bit. The root cause seems to be in the The question to ask here is: what happens if a plugin pass a binary Read Fluent::Compat::BufferedOutput on how it really occurs This situation is not really handled well in the current code. The special My current idea is to add something like below: def write_once(metadata, data, format: nil, size: nil, &block)
...
# We cannot split the data because it's an opaque binary blob.
# This occurs only for v0.12 plugins with a custom `format_stream()`
if not stored and data.is_a?(String)
raise BufferChunkOverflowError, "The formatted data is #{data.size} bytes (> chunk_limit_size)"
endIt does not fully fix the issue, but it should (at least) prevent Fluentd @cosmo0920 @okkez |
Hmm, it's hard to be well-handled.
But, @fujimoto's idea sounds good to me. |
|
I've been hitting this issue every so often. Have several forwarders which push logs into one centralised td-agent server (which then pushes logs to ElasticSearch). This morning the central server fell over with a series of I've restarted td-agent several times but it keeps getting stuck. Let me know if there's any steps for debugging which might help. |
|
I have noticed that same issue in our system as well. I can reproduce the error by doing the following:
How it's possible that fluentd squeeze in more data in a chunk buffer (209002 bytes ~ 204k) than I configured ie 170k? My configuration Versions: |
|
This issue has been automatically marked as stale because it has been open 90 days with no activity. Remove stale label or comment or this issue will be closed in 30 days |
|
This issue was automatically closed because of stale in 30 days |
Fluentd emits invalid msgpacked chunks to output plugins when chunks are over the size limit set with
chunk_size_limit.We are seeing this in the out_kinesis plugin: awslabs/aws-fluent-plugin-kinesis#133. Unclear whether or not this is expected behaviour and the plugin needs to handle chunk fragments (or if this is even possible with the streaming msgpack parser), or if it's a bug within fluentd (that it should only provide parsable chunks).
Fluentd version: 1.2.3
Ruby version: 2.3.3
OS: Mac OS / Linux
/cc @repeatedly @tagomoris @riywo @gurunathj @grosser
Test Script
Expected behaviour
Each chunk should be a JSON-encoded string.
Actual behaviour / investigation
When a chunk is larger than the chunk size, "chunk bytes limit exceeds for an emitted event stream" log line is emitted and the code uses a different code path to split the chunks into smaller chunks, however it does not appear to respect the boundaries of the msgpack encoding. Therefore parsing this split chunk results in the last element being cut off (or unavailable), and further chunks being parsed incorrectly as an Integer rather than an array of strings.
The text was updated successfully, but these errors were encountered: