lib/fluent/plugin/output.rb
# metadata_and_data is a Hash of:
#  (standard format) metadata => event stream
#  (custom format)   metadata => array of formatted event
# For standard format, formatting should be done for whole event stream, but
#   "whole event stream" may be a split of "es" here when it's bigger than chunk_limit_size.
#   `@buffer.write` will do this splitting.
# For custom format, formatting will be done here. Custom formatting always requires
#   iteration of event stream, and it should be done just once even if total event stream size
#   is biggar than chunk_limit_size because of performance.
metadata_and_data is a Hash of:
(standard format) metadata => event stream
(custom format)   metadata => array of formatted event
For standard format, formatting should be done for whole event stream, but
"whole event stream" may be a split of "es" here when it's bigger than chunk_limit_size.
`@buffer.write` will do this splitting.
For custom format, formatting will be done here. Custom formatting always requires
iteration of event stream, and it should be done just once even if total event stream size
is biggar than chunk_limit_size because of performance.
metadata_and_data is a Hash of : ( standard format ) metadata = > event stream ( custom format ) metadata = > array of formatted event For standard format formatting should be done for whole event stream but " whole event stream " may be a split of " es " here when it ' s bigger than chunk_limit_size . ` @buffer . write ` will do this splitting . For custom format formatting will be done here . Custom formatting always requires iteration of event stream and it should be done just once even if total event stream size is biggar than chunk_limit_size because of performance .
def handle_stream_with_custom_format(tag, es, enqueue: false)
        meta_and_data = {}
        records = 0
        es.each do |time, record|
          meta = metadata(tag, time, record)
          meta_and_data[meta] ||= []
          meta_and_data[meta] << format(tag, time, record)
          records += 1
        end
        write_guard do
          @buffer.write(meta_and_data, enqueue: enqueue)
        end
        @counters_monitor.synchronize{ @emit_records += records }
        true
      end
def handle_stream_with_custom_format ( tag , es , enqueue : false ) meta_and_data = { } records = 0 es . each do | time , record | meta = metadata ( tag , time , record ) meta_and_data [ meta ] ||= [ ] meta_and_data [ meta ] << format ( tag , time , record ) records += 1 end write_guard do @buffer . write ( meta_and_data , enqueue : enqueue ) end @counters_monitor . synchronize { @emit_records += records } true end
metadata_and_data is a Hash of:
(standard format) metadata => event stream
(custom format)   metadata => array of formatted event
For standard format, formatting should be done for whole event stream, but
"whole event stream" may be a split of "es" here when it's bigger than chunk_limit_size.
metadata_and_data is a Hash of : ( standard format ) metadata = > event stream ( custom format ) metadata = > array of formatted event For standard format formatting should be done for whole event stream but " whole event stream " may be a split of " es " here when it ' s bigger than chunk_limit_size .
tag
<not_specific>
es
<not_specific>
enqueue
<not_specific>