Class: Fluent::GroupCounterOutput
- Inherits:
-
Output
- Object
- Output
- Fluent::GroupCounterOutput
- Defined in:
- lib/fluent/plugin/out_groupcounter.rb
Defined Under Namespace
Classes: UndefOpenStruct
Constant Summary collapse
- PATTERN_MAX_NUM =
20
Instance Attribute Summary collapse
-
#count_interval ⇒ Object
Returns the value of attribute count_interval.
-
#counts ⇒ Object
Returns the value of attribute counts.
-
#last_checked ⇒ Object
Returns the value of attribute last_checked.
-
#saved_at ⇒ Object
Returns the value of attribute saved_at.
-
#saved_duration ⇒ Object
Returns the value of attribute saved_duration.
Instance Method Summary collapse
- #configure(conf) ⇒ Object
- #count_initialized ⇒ Object
- #countup(counts, count) ⇒ Object
-
#emit(tag, es, chain) ⇒ Object
recieve messages at here.
- #flush ⇒ Object
-
#flush_emit ⇒ Object
this method emits messages (periodically called).
- #generate_fields(counts_per_tag, output = {}, key_prefix = '') ⇒ Object
- #generate_output(counts) ⇒ Object
-
#group_key(tag, time, record) ⇒ Object
Expand record with @group_by_keys, and get a value to be a group_key.
-
#initialize ⇒ GroupCounterOutput
constructor
A new instance of GroupCounterOutput.
-
#load_status(file_path, count_interval) ⇒ Object
Load internal status from a file.
- #max(a, b) ⇒ Object
- #min(a, b) ⇒ Object
-
#save_status(file_path) ⇒ Object
Store internal status into a file.
- #shutdown ⇒ Object
- #start ⇒ Object
- #start_watch ⇒ Object
- #stripped_tag(tag) ⇒ Object
- #sum(a, b) ⇒ Object
-
#summarize_counts(tag, group_counts) ⇒ Object
Summarize counts for each tag.
- #watch ⇒ Object
Constructor Details
#initialize ⇒ GroupCounterOutput
Returns a new instance of GroupCounterOutput.
6 7 8 9 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 6 def initialize super require 'pathname' end |
Instance Attribute Details
#count_interval ⇒ Object
Returns the value of attribute count_interval.
33 34 35 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 33 def count_interval @count_interval end |
#counts ⇒ Object
Returns the value of attribute counts.
34 35 36 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 34 def counts @counts end |
#last_checked ⇒ Object
Returns the value of attribute last_checked.
37 38 39 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 37 def last_checked @last_checked end |
#saved_at ⇒ Object
Returns the value of attribute saved_at.
36 37 38 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 36 def saved_at @saved_at end |
#saved_duration ⇒ Object
Returns the value of attribute saved_duration.
35 36 37 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 35 def saved_duration @saved_duration end |
Instance Method Details
#configure(conf) ⇒ Object
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 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 39 def configure(conf) super if @count_interval @count_interval = @count_interval.to_i else @count_interval = case @unit when 'minute' then 60 when 'hour' then 3600 when 'day' then 86400 else raise RuntimeError, "@unit must be one of minute/hour/day" end end @aggregate = case @aggregate when 'tag' then :tag when 'all' then :all else raise Fluent::ConfigError, "groupcounter aggregate allows tag/all" end @add_tag_prefix ||= @tag_prefix @remove_tag_prefix ||= @input_tag_remove_prefix if @output_per_tag raise Fluent::ConfigError, "add_tag_prefix must be specified with output_per_tag" unless @add_tag_prefix end if @add_tag_prefix @tag_prefix_string = @add_tag_prefix + '.' else @tag_prefix_string = '' end if @remove_tag_prefix @removed_prefix_string = @remove_tag_prefix + '.' @removed_length = @removed_prefix_string.length end @group_by_keys = @group_by_keys.split(',') if @group_by_keys @pattern = {} (1..PATTERN_MAX_NUM).each do |i| next unless conf["pattern#{i}"] replace, regexp = conf["pattern#{i}"].split(/ +/, 2) raise Fluent::ConfigError, "pattern#{i} does not contain 2 parameters" unless regexp @pattern[replace] = Regexp.compile(regexp) end if @store_file f = Pathname.new(@store_file) if (f.exist? && !f.writable_real?) || (!f.exist? && !f.parent.writable_real?) raise Fluent::ConfigError, "#{@store_file} is not writable" end end @counts = count_initialized @hostname = Socket.gethostname @mutex = Mutex.new end |
#count_initialized ⇒ Object
111 112 113 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 111 def count_initialized {} end |
#countup(counts, count) ⇒ Object
232 233 234 235 236 237 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 232 def countup(counts, count) counts[:count] = sum(counts[:count], count[:count]) counts[:sum] = sum(counts[:sum], count[:sum]) if @avg_key and count[:sum] counts[:max] = max(counts[:max], count[:max]) if @max_key and count[:max] counts[:min] = min(counts[:min], count[:min]) if @min_key and count[:min] end |
#emit(tag, es, chain) ⇒ Object
recieve messages at here
193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 193 def emit(tag, es, chain) group_counts = {} = tag.split('.') es.each do |time, record| count = {} count[:count] = 1 count[:sum] = record[@avg_key].to_f if @avg_key and record[@avg_key] count[:max] = record[@max_key].to_f if @max_key and record[@max_key] count[:min] = record[@min_key].to_f if @min_key and record[@min_key] group_key = group_key(tag, time, record) group_counts[group_key] ||= {} countup(group_counts[group_key], count) end summarize_counts(tag, group_counts) chain.next rescue => e $log.warn "#{e.class} #{e.} #{e.backtrace.first}" end |
#flush ⇒ Object
152 153 154 155 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 152 def flush flushed, @counts = @counts, count_initialized() generate_output(flushed) end |
#flush_emit ⇒ Object
this method emits messages (periodically called)
158 159 160 161 162 163 164 165 166 167 168 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 158 def flush_emit time = Fluent::Engine.now if @output_per_tag flush.each do |tag, | Fluent::Engine.emit("#{@tag_prefix_string}#{tag}", time, ) end else = flush Fluent::Engine.emit(@tag, time, ) unless .empty? end end |
#generate_fields(counts_per_tag, output = {}, key_prefix = '') ⇒ Object
115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 115 def generate_fields(counts_per_tag, output = {}, key_prefix = '') return {} unless counts_per_tag # total_count = counts_per_tag.delete('__total_count') counts_per_tag.each do |group_key, count| group_key_with = group_key.empty? ? "" : group_key + @delimiter output[key_prefix + group_key + @count_suffix] = count[:count] if count[:count] output[key_prefix + group_key_with + "#{@min_key}#{@min_suffix}"] = count[:min] if count[:min] output[key_prefix + group_key_with + "#{@max_key}#{@max_suffix}"] = count[:max] if count[:max] output[key_prefix + group_key_with + "#{@avg_key}#{@avg_suffix}"] = count[:sum] / (count[:count] * 1.0) if count[:sum] and count[:count] > 0 # output[key_prefix + group_key_with + "rate"] = ((count[:count] * 100.0) / (1.00 * step)).floor / 100.0 # output[key_prefix + group_key_with + "percentage"] = count[:count] * 100.0 / (1.00 * total_count) if total_count > 0 end output end |
#generate_output(counts) ⇒ Object
132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 132 def generate_output(counts) if @output_per_tag # tag => output return {'all' => generate_fields(counts['all'])} if @aggregate == :all output_pairs = {} counts.keys.each do |tag| output_pairs[stripped_tag(tag)] = generate_fields(counts[tag]) end output_pairs else return generate_fields(counts['all']) if @aggregate == :all output = {} counts.keys.each do |tag| generate_fields(counts[tag], output, stripped_tag(tag) + @delimiter) end output end end |
#group_key(tag, time, record) ⇒ Object
Expand record with @group_by_keys, and get a value to be a group_key
240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 240 def group_key(tag, time, record) if @group_by_expression = tag.split('.') group_key = (@group_by_expression, record, tag, , Time.at(time)) elsif @group_by_keys values = @group_by_keys.map {|key| record[key] || 'undef'} group_key = values.join(@delimiter) else return "" end group_key = group_key.to_s.force_encoding('ASCII-8BIT') @pattern.each {|replace, regexp| break if group_key.gsub!(regexp, replace) } group_key end |
#load_status(file_path, count_interval) ⇒ Object
Load internal status from a file
306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 306 def load_status(file_path, count_interval) return unless (f = Pathname.new(file_path)).exist? begin f.open('rb') do |f| stored = Marshal.load(f) if stored[:aggregate] == @aggregate and stored[:group_by_keys] == @group_by_keys and if Fluent::Engine.now <= stored[:saved_at] + count_interval @counts = stored[:counts] @saved_at = stored[:saved_at] @saved_duration = stored[:saved_duration] # skip the saved duration to continue counting @last_checked = Fluent::Engine.now - @saved_duration else $log.warn "out_groupcounter: stored data is outdated. ignore stored data" end else $log.warn "out_groupcounter: configuration param was changed. ignore stored data" end end rescue => e $log.warn "out_groupcounter: Can't load store_file #{e.class} #{e.}" end end |
#max(a, b) ⇒ Object
262 263 264 265 266 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 262 def max(a, b) return b if a.nil? return a if b.nil? a > b ? a : b end |
#min(a, b) ⇒ Object
268 269 270 271 272 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 268 def min(a, b) return b if a.nil? return a if b.nil? a > b ? b : a end |
#save_status(file_path) ⇒ Object
Store internal status into a file
283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 283 def save_status(file_path) begin Pathname.new(file_path).open('wb') do |f| @saved_at = Fluent::Engine.now @saved_duration = @saved_at - @last_checked Marshal.dump({ :counts => @counts, :saved_at => @saved_at, :saved_duration => @saved_duration, :aggregate => @aggregate, :group_by_keys => @group_by_keys, }, f) end rescue => e $log.warn "out_groupcounter: Can't write store_file #{e.class} #{e.}" end end |
#shutdown ⇒ Object
104 105 106 107 108 109 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 104 def shutdown super @watcher.terminate @watcher.join save_status(@store_file) if @store_file end |
#start ⇒ Object
98 99 100 101 102 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 98 def start super load_status(@store_file, @count_interval) if @store_file start_watch end |
#start_watch ⇒ Object
170 171 172 173 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 170 def start_watch # for internal, or tests only @watcher = Thread.new(&method(:watch)) end |
#stripped_tag(tag) ⇒ Object
274 275 276 277 278 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 274 def stripped_tag(tag) return tag unless @remove_tag_prefix return tag[@removed_length..-1] if tag.start_with?(@removed_prefix_string) and tag.length > @removed_length tag end |
#sum(a, b) ⇒ Object
256 257 258 259 260 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 256 def sum(a, b) a ||= 0 b ||= 0 a + b end |
#summarize_counts(tag, group_counts) ⇒ Object
Summarize counts for each tag
217 218 219 220 221 222 223 224 225 226 227 228 229 230 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 217 def summarize_counts(tag, group_counts) tag = 'all' if @aggregate == :all @counts[tag] ||= {} @mutex.synchronize { group_counts.each do |group_key, count| @counts[tag][group_key] ||= {} countup(@counts[tag][group_key], count) end # total_count = group_counts.map {|group_key, count| count[:count] }.inject(:+) # @counts[tag]['__total_count'] = sum(@counts[tag]['__total_count'], total_count) } end |
#watch ⇒ Object
175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 175 def watch # instance variable, and public accessable, for test @last_checked ||= Fluent::Engine.now while true sleep 0.5 begin if Fluent::Engine.now - @last_checked >= @count_interval now = Fluent::Engine.now flush_emit @last_checked = now end rescue => e $log.warn "#{e.class} #{e.} #{e.backtrace.first}" end end end |