-
Notifications
You must be signed in to change notification settings - Fork 35
/
elapsed.rb
263 lines (225 loc) · 8.66 KB
/
elapsed.rb
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
require "logstash/filters/base"
require "logstash/namespace"
require 'thread'
require 'socket'
# 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"
#
class LogStash::Filters::Elapsed < LogStash::Filters::Base
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_without_start"
MATCH_TAG = PREFIX + "match"
config_name "elapsed"
# The name of the tag identifying the "start event"
config :start_tag, :validate => :string, :required => true
# The name of the tag identifying the "end event"
config :end_tag, :validate => :string, :required => true
# The name of the field containing the task ID.
# This value must uniquely identify the task in the system, otherwise
# it's impossible to match the couple of events.
config :unique_id_field, :validate => :string, :required => true
# The amount of seconds after an "end event" can be considered lost.
# The corresponding "start event" is discarded and an "expired event"
# is generated. The default value is 30 minutes (1800 seconds).
config :timeout, :validate => :number, :required => false, :default => 1800
# This property manage what to do when an "end event" matches a "start event".
# If it's set to `false` (default value), the elapsed information are added
# to the "end event"; if it's set to `true` a new "match event" is created.
config :new_event_on_match, :validate => :boolean, :required => false, :default => false
# This property manage what to do when several "start events" were received
# before an "end event" for a concrete ID. If it's set to `first` (default
# value), the first "start event" will be used; if it's set to `last`,
# the last "start event" will be used.
config :keep_start_event, :validate => ['first', 'last'], :required => false, :default => 'first'
# This filter must have its flush function called periodically to be able to purge
# expired stored start events.
config :periodic_flush, :validate => :boolean, :default => true
public
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
def start_events
@start_events
end
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 # def filter
# The method is invoked by LogStash every 5 seconds.
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
private
def increment_age_by(seconds)
@start_events.each_pair do |key, element|
element.age += seconds
end
end
# Remove the expired "start events" from the internal
# buffer and return them.
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
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
def start_event?(event)
return (event.get("tags") != nil && event.get("tags").include?(@start_tag))
end
def end_event?(event)
return (event.get("tags") != nil && event.get("tags").include?(@end_tag))
end
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
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
end # class LogStash::Filters::Elapsed
class LogStash::Filters::Elapsed::Element
attr_accessor :event, :age
def initialize(event)
@event = event
@age = 0
end
end