Class: LogStash::Filters::Elapsed

Inherits:
Base show all
Defined in:
lib/logstash/filters/elapsed.rb

Overview

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:

  • the tags “elapsed” and “elapsed.match”

  • the field “elapsed.time” with the difference, in seconds, between the two events timestamps

  • an ID filed with the task ID

  • the field “elapsed.timestamp_start” with the timestamp of the “start event”

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

  • the tags “elapsed” and “elapsed.expired_error”

  • a field called “elapsed.time” with the age, in seconds, of the “start event”

  • an ID filed with the task ID

  • the field “elapsed.timestamp_start” with the timestamp of the “start event”

Defined Under Namespace

Classes: Element

Constant Summary collapse

PREFIX = 
"elapsed."
ELAPSED_FIELD = 
PREFIX + "time"
TIMESTAMP_START_EVENT_FIELD = 
PREFIX + "timestamp_start"
HOST_FIELD = 
"host"
ELAPSED_TAG = 
"elapsed"
EXPIRED_ERROR_TAG = 
PREFIX + "expired_error"
END_WITHOUT_START_TAG = 
PREFIX + "end_wtihout_start"
MATCH_TAG = 
PREFIX + "match"

Constants inherited from Base

Base::RESERVED

Constants included from Config::Mixin

Config::Mixin::CONFIGSORT

Instance Attribute Summary

Attributes included from Config::Mixin

#config, #original_params

Attributes inherited from Plugin

#logger, #params

Instance Method Summary collapse

Methods inherited from Base

#execute, #initialize, #threadsafe?

Methods included from Config::Mixin

#config_init, included

Methods inherited from Plugin

#eql?, #finished, #finished?, #hash, #initialize, #inspect, lookup, #reload, #running?, #shutdown, #teardown, #terminating?, #to_s

Constructor Details

This class inherits a constructor from LogStash::Filters::Base

Instance Method Details

#filter(event) ⇒ Object 



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# 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 ⇒ Object

The method is invoked by LogStash every 5 seconds.



171
172
173
174
175
176
177
178
179
180
 
# 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 ⇒ Object 



116
117
118
119
120
121
122
123
 
# 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 ⇒ Object

Getter method used for the tests



126
127
128
 
# File 'lib/logstash/filters/elapsed.rb', line 126

def start_events
  @start_events
end