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.
31 32 33 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 31 def count_interval @count_interval end |
#counts ⇒ Object
Returns the value of attribute counts.
32 33 34 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 32 def counts @counts end |
#last_checked ⇒ Object
Returns the value of attribute last_checked.
35 36 37 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 35 def last_checked @last_checked end |
#saved_at ⇒ Object
Returns the value of attribute saved_at.
34 35 36 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 34 def saved_at @saved_at end |
#saved_duration ⇒ Object
Returns the value of attribute saved_duration.
33 34 35 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 33 def saved_duration @saved_duration end |
Instance Method Details
#configure(conf) ⇒ Object
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 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 37 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 if @output_per_tag raise Fluent::ConfigError, "tag_prefix must be specified with output_per_tag" unless @tag_prefix @tag_prefix_string = @tag_prefix + '.' end if @input_tag_remove_prefix @removed_prefix_string = @input_tag_remove_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
104 105 106 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 104 def count_initialized {} end |
#countup(counts, count) ⇒ Object
225 226 227 228 229 230 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 225 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
186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 186 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.message} #{e.backtrace.first}" end |
#flush ⇒ Object
145 146 147 148 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 145 def flush flushed, @counts = @counts, count_initialized() generate_output(flushed) end |
#flush_emit ⇒ Object
this method emits messages (periodically called)
151 152 153 154 155 156 157 158 159 160 161 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 151 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
108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 108 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
125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 125 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
233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 233 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
300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 300 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.message}" end end |
#max(a, b) ⇒ Object
255 256 257 258 259 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 255 def max(a, b) return b if a.nil? return a if b.nil? a > b ? a : b end |
#min(a, b) ⇒ Object
261 262 263 264 265 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 261 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
277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 277 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.message}" end end |
#shutdown ⇒ Object
97 98 99 100 101 102 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 97 def shutdown super @watcher.terminate @watcher.join save_status(@store_file) if @store_file end |
#start ⇒ Object
91 92 93 94 95 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 91 def start super load_status(@store_file, @count_interval) if @store_file start_watch end |
#start_watch ⇒ Object
163 164 165 166 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 163 def start_watch # for internal, or tests only @watcher = Thread.new(&method(:watch)) end |
#stripped_tag(tag) ⇒ Object
267 268 269 270 271 272 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 267 def stripped_tag(tag) return tag unless @input_tag_remove_prefix return tag[@removed_length..-1] if tag.start_with?(@removed_prefix_string) and tag.length > @removed_length return tag[@removed_length..-1] if tag == @input_tag_remove_prefix tag end |
#sum(a, b) ⇒ Object
249 250 251 252 253 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 249 def sum(a, b) a ||= 0 b ||= 0 a + b end |
#summarize_counts(tag, group_counts) ⇒ Object
Summarize counts for each tag
210 211 212 213 214 215 216 217 218 219 220 221 222 223 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 210 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
168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 |
# File 'lib/fluent/plugin/out_groupcounter.rb', line 168 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.message} #{e.backtrace.first}" end end end |