Class: Fluent::TraceAccumulator

Inherits:
Object
  • Object
show all
Defined in:
lib/fluent/plugin/exception_detector.rb

Overview

Buffers and groups log records if they contain exception stack traces.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(message_field, languages, &emit_callback) ⇒ TraceAccumulator

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.



193
194
195
196
197
198
199
200
201
 
# File 'lib/fluent/plugin/exception_detector.rb', line 193

def initialize(message_field, languages, &emit_callback)
  @exception_detector = Fluent::ExceptionDetector.new(*languages)
  @message_field = message_field
  @messages = []
  @buffer_start_time = Time.now
  @first_record = nil
  @first_timestamp = nil
  @emit = emit_callback
end

Instance Attribute Details

#buffer_start_time ⇒ Object (readonly)

Returns the value of attribute buffer_start_time.



183
184
185
 
# File 'lib/fluent/plugin/exception_detector.rb', line 183

def buffer_start_time
  @buffer_start_time
end

Instance Method Details

#flush ⇒ Object 



223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
 
# File 'lib/fluent/plugin/exception_detector.rb', line 223

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
end

#force_flush ⇒ Object 



244
245
246
247
 
# File 'lib/fluent/plugin/exception_detector.rb', line 244

def force_flush
  flush
  @exception_detector.reset
end

#length ⇒ Object 



249
250
251
 
# File 'lib/fluent/plugin/exception_detector.rb', line 249

def length
  @messages.length
end

#push(time_sec, record) ⇒ Object 



203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
 
# File 'lib/fluent/plugin/exception_detector.rb', line 203

def push(time_sec, 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 = @message_field.nil? ? record : record[@message_field]
  if message.nil?
    @exception_detector.reset
    detection_status = :no_trace
  else
    detection_status = @exception_detector.update(message)
  end

  update_buffer(detection_status, time_sec, record, message)
end