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:
- source,ruby
-
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:
- source,ruby
-
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”
Constants
- ELAPSED_FIELD
- ELAPSED_TAG
- END_WITHOUT_START_TAG
- EXPIRED_ERROR_TAG
- HOST_FIELD
- MATCH_TAG
- PREFIX
- TIMESTAMP_START_EVENT_FIELD
Public Instance Methods
# File lib/logstash/filters/elapsed.rb, line 137 def filter(event) unique_id = event.get(@unique_id_field) return if unique_id.nil? if(start_event?(event)) filter_matched(event) @logger.debug("Elapsed, 'start event' received", start_tag: @start_tag, unique_id_field: @unique_id_field) @mutex.synchronize do unless(@keep_start_event == 'first' && @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.debug("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.get("@timestamp") - start_event.get("@timestamp") if(@new_event_on_match) elapsed_event = new_elapsed_event(elapsed, unique_id, start_event.get("@timestamp")) filter_matched(elapsed_event) yield elapsed_event if block_given? else return add_elapsed_info(event, elapsed, unique_id, start_event.get("@timestamp")) end else @mutex.unlock # The "start event" did not arrive. event.tag(END_WITHOUT_START_TAG) end end end
The method is invoked by LogStash
every 5 seconds.
# File lib/logstash/filters/elapsed.rb, line 178 def flush(options = {}) expired_elements = [] @mutex.synchronize do increment_age_by(5) expired_elements = remove_expired_elements() end return create_expired_events_from(expired_elements) end
# File lib/logstash/filters/elapsed.rb, line 123 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
Getter method used for the tests
# File lib/logstash/filters/elapsed.rb, line 133 def start_events @start_events end
Private Instance Methods
# File lib/logstash/filters/elapsed.rb, line 244 def add_elapsed_info(event, elapsed_time, unique_id, timestamp_start_event) event.tag(ELAPSED_TAG) event.tag(MATCH_TAG) event.set(ELAPSED_FIELD, elapsed_time) event.set(@unique_id_field, unique_id) event.set(TIMESTAMP_START_EVENT_FIELD, timestamp_start_event) return event end
# File lib/logstash/filters/elapsed.rb, line 211 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.set(HOST_FIELD, Socket.gethostname) error_event.set(@unique_id_field, element.event.get(@unique_id_field) ) error_event.set(ELAPSED_FIELD, element.age) error_event.set(TIMESTAMP_START_EVENT_FIELD, element.event.get("@timestamp") ) events << error_event filter_matched(error_event) end return events end
# File lib/logstash/filters/elapsed.rb, line 234 def end_event?(event) return (event.get("tags") != nil && event.get("tags").include?(@end_tag)) end
# File lib/logstash/filters/elapsed.rb, line 190 def increment_age_by(seconds) @start_events.each_pair do |key, element| element.age += seconds end end
# File lib/logstash/filters/elapsed.rb, line 238 def new_elapsed_event(elapsed_time, unique_id, timestamp_start_event) new_event = LogStash::Event.new new_event.set(HOST_FIELD, Socket.gethostname) return add_elapsed_info(new_event, elapsed_time, unique_id, timestamp_start_event) end
Remove the expired “start events” from the internal buffer and return them.
# File lib/logstash/filters/elapsed.rb, line 198 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
# File lib/logstash/filters/elapsed.rb, line 230 def start_event?(event) return (event.get("tags") != nil && event.get("tags").include?(@start_tag)) end