class LogStash::Codecs::Multiline

Public Class Methods

new(*params) click to toggle source
Calls superclass method
# File lib/logstash/codecs/multiline.rb, line 151
def initialize(*params)
  super

  @original_field = ecs_select[disabled: nil, v1: '[event][original]']
end

Public Instance Methods

accept(listener) click to toggle source
# File lib/logstash/codecs/multiline.rb, line 200
def accept(listener)
  # memoize references to listener that holds upstream state
  @previous_listener = @last_seen_listener || listener
  @last_seen_listener = listener
  decode(listener.data) do |event|
    what_based_listener.process_event(event)
  end
end
auto_flush(listener = @last_seen_listener) click to toggle source
# File lib/logstash/codecs/multiline.rb, line 244
def auto_flush(listener = @last_seen_listener)
  return if listener.nil?

  flush do |event|
    listener.process_event(event)
  end
end
auto_flush_active?() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 308
def auto_flush_active?
  !@auto_flush_interval.nil?
end
auto_flush_runner() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 312
def auto_flush_runner
  @auto_flush_runner || AutoFlushUnset.new(nil, nil)
end
buffer(text) click to toggle source
# File lib/logstash/codecs/multiline.rb, line 222
def buffer(text)
  @buffer_bytes += text.bytesize
  @buffer.push(text)
end
buffer_over_limits?() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 295
def buffer_over_limits?
  over_maximum_lines? || over_maximum_bytes?
end
close() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 304
def close
  auto_flush_runner.stop
end
decode(text, &block) click to toggle source
# File lib/logstash/codecs/multiline.rb, line 209
def decode(text, &block)
  text = @converter.convert(text)
  text.split("\n").each do |line|
    match = @grok.match(line)
    @logger.debug? && @logger.debug("Multiline", :text => line, :pattern => @pattern,
                                    :match => (match != false), :negate => @negate)

    # Add negate option
    match = (match and !@negate) || (!match and @negate)
    @handler.call(line, match, &block)
  end
end
do_next(text, matched, &block) click to toggle source
# File lib/logstash/codecs/multiline.rb, line 275
def do_next(text, matched, &block)
  buffer(text)
  auto_flush_runner.start
  flush(&block) if !matched || buffer_over_limits?
end
do_previous(text, matched, &block) click to toggle source
# File lib/logstash/codecs/multiline.rb, line 281
def do_previous(text, matched, &block)
  flush(&block) if !matched || buffer_over_limits?
  auto_flush_runner.start
  buffer(text)
end
doing_previous?() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 267
def doing_previous?
  @what == "previous"
end
encode(event) click to toggle source
# File lib/logstash/codecs/multiline.rb, line 299
def encode(event)
  # Nothing to do.
  @on_event.call(event, event)
end
flush() { |event| ... } click to toggle source
# File lib/logstash/codecs/multiline.rb, line 227
def flush(&block)
  if block_given? && @buffer.any?
    no_error = true
    event = merge_events
    begin
      yield event
    rescue ::Exception => e
      # need to rescue everything
      # likliest cause: backpressure or timeout by exception
      # can't really do anything but leave the data in the buffer for next time if there is one
      @logger.error("Multiline: flush downstream error", :exception => e)
      no_error = false
    end
    reset_buffer if no_error
  end
end
initialize_copy(source) click to toggle source
Calls superclass method
# File lib/logstash/codecs/multiline.rb, line 316
def initialize_copy(source)
  super
  register
end
merge_events() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 252
def merge_events
  message = @buffer.join(NL)
  event = event_factory.new_event(LogStash::Event::TIMESTAMP => @time, "message" => message)
  event.set @original_field, message.dup.freeze if @original_field
  event.tag @multiline_tag if !@multiline_tag.empty? && @buffer.size > 1
  event.tag "multiline_codec_max_bytes_reached" if over_maximum_bytes?
  event.tag "multiline_codec_max_lines_reached" if over_maximum_lines?
  event
end
over_maximum_bytes?() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 291
def over_maximum_bytes?
  @buffer_bytes >= @max_bytes
end
over_maximum_lines?() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 287
def over_maximum_lines?
  @buffer.size > @max_lines
end
register() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 157
def register
  require "grok-pure" # rubygem 'jls-grok'
  require 'logstash/patterns/core'

  # Detect if we are running from a jarfile, pick the right path.
  patterns_path = []
  patterns_path += [LogStash::Patterns::Core.path]

  @grok = Grok.new

  @patterns_dir = patterns_path.to_a + @patterns_dir
  @patterns_dir.each do |path|
    if ::File.directory?(path)
      path = ::File.join(path, "*")
    end

    Dir.glob(path).each do |file|
      @logger.debug("Grok loading patterns from file", :path => file)
      @grok.add_patterns_from_file(file)
    end
  end

  @grok.compile(@pattern)
  @logger.trace("Registered multiline plugin", :type => @type, :config => @config)

  reset_buffer

  @handler = method("do_#{@what}".to_sym)

  @converter = LogStash::Util::Charset.new(@charset)
  @converter.logger = @logger
  if @auto_flush_interval
    # will start on first decode
    @auto_flush_runner = AutoFlush.new(self, @auto_flush_interval)
  end
end
reset_buffer() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 262
def reset_buffer
  @buffer = []
  @buffer_bytes = 0
end
use_mapper_auto_flush() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 194
def use_mapper_auto_flush
  return unless auto_flush_active?
  @auto_flush_runner = AutoFlushUnset.new(nil, nil)
  @auto_flush_interval = @auto_flush_interval.to_f
end
what_based_listener() click to toggle source
# File lib/logstash/codecs/multiline.rb, line 271
def what_based_listener
  doing_previous? ? @previous_listener : @last_seen_listener
end