class LogStash::Filters::Elapsed

The elapsed filter tracks a pair of start/end events and uses their timestamps to calculate the elapsed time between them.

The filter has been developed to track the execution time of processes and other long tasks.

The configuration looks like this:

filter {
  elapsed {
    start_tag => "start event tag"
    end_tag => "end event tag"
    unique_id_field => "id field name"
    timeout => seconds
    new_event_on_match => true/false
  }
}

The events managed by this filter must have some particular properties. The event describing the start of the task (the “start event”) must contain a tag equal to 'start_tag'. On the other side, the event describing the end of the task (the “end event”) must contain a tag equal to 'end_tag'. Both these two kinds of event need to own an ID field which identify uniquely that particular task. The name of this field is stored in 'unique_id_field'.

You can use a Grok filter to prepare the events for the elapsed filter. An example of configuration can be:

filter {
  grok {
    match => ["message", "%{TIMESTAMP_ISO8601} START id: (?<task_id>.*)"]
    add_tag => [ "taskStarted" ]
  }

  grok {
    match => ["message", "%{TIMESTAMP_ISO8601} END id: (?<task_id>.*)"]
    add_tag => [ "taskTerminated"]
  }

  elapsed {
    start_tag => "taskStarted"
    end_tag => "taskTerminated"
    unique_id_field => "task_id"
  }
}

The elapsed filter collects all the “start events”. If two, or more, “start events” have the same ID, only the first one is recorded, the others are discarded.

When an “end event” matching a previously collected “start event” is received, there is a match. The configuration property 'new_event_on_match' tells where to insert the elapsed information: they can be added to the “end event” or a new “match event” can be created. Both events store the following information:

If the “end event” does not arrive before “timeout” seconds, the “start event” is discarded and an “expired event” is generated. This event contains:

Constants

ELAPSED_FIELD
ELAPSED_TAG
END_WITHOUT_START_TAG
EXPIRED_ERROR_TAG
HOST_FIELD
MATCH_TAG
PREFIX
TIMESTAMP_START_EVENT_FIELD

Public Instance Methods

filter(event) { |elapsed_event| ... } click to toggle source
# File lib/logstash/filters/elapsed.rb, line 130
def filter(event)
  return unless filter?(event)

  unique_id = event[@unique_id_field]
  return if unique_id.nil?

  if(start_event?(event))
    filter_matched(event)
    @logger.info("Elapsed, 'start event' received", start_tag: @start_tag, unique_id_field: @unique_id_field)

    @mutex.synchronize do
      unless(@start_events.has_key?(unique_id))
        @start_events[unique_id] = LogStash::Filters::Elapsed::Element.new(event)
      end
    end

  elsif(end_event?(event))
    filter_matched(event)
    @logger.info("Elapsed, 'end event' received", end_tag: @end_tag, unique_id_field: @unique_id_field)

    @mutex.lock
    if(@start_events.has_key?(unique_id))
      start_event = @start_events.delete(unique_id).event
      @mutex.unlock
      elapsed = event["@timestamp"] - start_event["@timestamp"]
      if(@new_event_on_match)
        elapsed_event = new_elapsed_event(elapsed, unique_id, start_event["@timestamp"])
        filter_matched(elapsed_event)
        yield elapsed_event if block_given?
      else
        return add_elapsed_info(event, elapsed, unique_id, start_event["@timestamp"])
      end
    else
      @mutex.unlock
      # The "start event" did not arrive.
      event.tag(END_WITHOUT_START_TAG)
    end
  end
end
flush() click to toggle source

The method is invoked by LogStash every 5 seconds.

# File lib/logstash/filters/elapsed.rb, line 171
def flush()
  expired_elements = []

  @mutex.synchronize do
    increment_age_by(5)
    expired_elements = remove_expired_elements()
  end

  return create_expired_events_from(expired_elements)
end
register() click to toggle source
# File lib/logstash/filters/elapsed.rb, line 116
def register
  @mutex = Mutex.new
  # This is the state of the filter. The keys are the "unique_id_field",
  # the values are couples of values: <start event, age>
  @start_events = {}

  @logger.info("Elapsed, timeout: #{@timeout} seconds")
end
start_events() click to toggle source

Getter method used for the tests

# File lib/logstash/filters/elapsed.rb, line 126
def start_events
  @start_events
end

Private Instance Methods

add_elapsed_info(event, elapsed_time, unique_id, timestamp_start_event) click to toggle source
# File lib/logstash/filters/elapsed.rb, line 237
def add_elapsed_info(event, elapsed_time, unique_id, timestamp_start_event)
    event.tag(ELAPSED_TAG)
    event.tag(MATCH_TAG)

    event[ELAPSED_FIELD] = elapsed_time
    event[@unique_id_field] = unique_id
    event[TIMESTAMP_START_EVENT_FIELD] = timestamp_start_event

    return event
end
create_expired_events_from(expired_elements) click to toggle source
# File lib/logstash/filters/elapsed.rb, line 204
def create_expired_events_from(expired_elements)
  events = []
  expired_elements.each do |element|
    error_event = LogStash::Event.new
    error_event.tag(ELAPSED_TAG)
    error_event.tag(EXPIRED_ERROR_TAG)

    error_event[HOST_FIELD] = Socket.gethostname
    error_event[@unique_id_field] = element.event[@unique_id_field]
    error_event[ELAPSED_FIELD] = element.age
    error_event[TIMESTAMP_START_EVENT_FIELD] = element.event["@timestamp"]

    events << error_event
    filter_matched(error_event)
  end

  return events
end
end_event?(event) click to toggle source
# File lib/logstash/filters/elapsed.rb, line 227
def end_event?(event)
  return (event["tags"] != nil && event["tags"].include?(@end_tag))
end
increment_age_by(seconds) click to toggle source
# File lib/logstash/filters/elapsed.rb, line 183
def increment_age_by(seconds)
  @start_events.each_pair do |key, element|
    element.age += seconds
  end
end
new_elapsed_event(elapsed_time, unique_id, timestamp_start_event) click to toggle source
# File lib/logstash/filters/elapsed.rb, line 231
def new_elapsed_event(elapsed_time, unique_id, timestamp_start_event)
    new_event = LogStash::Event.new
    new_event[HOST_FIELD] = Socket.gethostname
    return add_elapsed_info(new_event, elapsed_time, unique_id, timestamp_start_event)
end
remove_expired_elements() click to toggle source

Remove the expired “start events” from the internal buffer and return them.

# File lib/logstash/filters/elapsed.rb, line 191
def remove_expired_elements()
  expired = []
  @start_events.delete_if do |key, element|
    if(element.age >= @timeout)
      expired << element
      next true
    end
    next false
  end

  return expired
end
start_event?(event) click to toggle source
# File lib/logstash/filters/elapsed.rb, line 223
def start_event?(event)
  return (event["tags"] != nil && event["tags"].include?(@start_tag))
end