class Fluent::TraceAccumulator
Buffers and groups log records if they contain exception stack traces.
Attributes
buffer_start_time[R]
Public Class Methods
new(message_field, languages, max_lines: 0, max_bytes: 0, &emit_callback)
click to toggle source
If message_field is nil, the instance is set up to accumulate records that are plain strings (i.e. the whole record is concatenated). Otherwise, the instance accepts records that are dictionaries (usually originating from structured JSON logs) and accumulates just the content of the given message field. message_field may contain the empty string. In this case, the TraceAccumulator
‘learns’ the field name from the first record by checking for some pre-defined common field names of text logs. The named parameters max_lines and max_bytes limit the maximum amount of data to be buffered. The default value 0 indicates ‘no limit’.
# File lib/fluent/plugin/exception_detector.rb, line 196 def initialize(message_field, languages, max_lines: 0, max_bytes: 0, &emit_callback) @exception_detector = Fluent::ExceptionDetector.new(*languages) @max_lines = max_lines @max_bytes = max_bytes @message_field = message_field @messages = [] @buffer_start_time = Time.now @buffer_size = 0 @first_record = nil @first_timestamp = nil @emit = emit_callback end
Public Instance Methods
flush()
click to toggle source
# File lib/fluent/plugin/exception_detector.rb, line 226 def flush case @messages.length when 0 return when 1 @emit.call(@first_timestamp, @first_record) else combined_message = @messages.join if @message_field.nil? output_record = combined_message else output_record = @first_record output_record[@message_field] = combined_message end @emit.call(@first_timestamp, output_record) end @messages = [] @first_record = nil @first_timestamp = nil @buffer_size = 0 end
force_flush()
click to toggle source
# File lib/fluent/plugin/exception_detector.rb, line 248 def force_flush flush @exception_detector.reset end
push(time_sec, record)
click to toggle source
# File lib/fluent/plugin/exception_detector.rb, line 210 def push(time_sec, record) message = extract_message(record) if message.nil? @exception_detector.reset detection_status = :no_trace else force_flush if @max_bytes > 0 && @buffer_size + message.length > @max_bytes detection_status = @exception_detector.update(message) end update_buffer(detection_status, time_sec, record, message) force_flush if @max_lines > 0 && @messages.length == @max_lines end
Private Instance Methods
add(time_sec, record, message)
click to toggle source
# File lib/fluent/plugin/exception_detector.rb, line 291 def add(time_sec, record, message) if @messages.empty? @first_record = record unless @message_field.nil? @first_timestamp = time_sec @buffer_start_time = Time.now end unless message.nil? @messages << message @buffer_size += message.length end end
extract_message(record)
click to toggle source
# File lib/fluent/plugin/exception_detector.rb, line 255 def extract_message(record) if !@message_field.nil? && @message_field.empty? ExceptionDetectorConfig::DEFAULT_FIELDS.each do |f| if record.key?(f) @message_field = f break end end end @message_field.nil? ? record : record[@message_field] end
update_buffer(detection_status, time_sec, record, message)
click to toggle source
# File lib/fluent/plugin/exception_detector.rb, line 267 def update_buffer(detection_status, time_sec, record, message) trigger_emit = detection_status == :no_trace || detection_status == :end_trace if @messages.empty? && trigger_emit @emit.call(time_sec, record) return end case detection_status when :inside_trace add(time_sec, record, message) when :end_trace add(time_sec, record, message) flush when :no_trace flush add(time_sec, record, message) flush when :start_trace flush add(time_sec, record, message) end end