Class: Fluent::Plugin::FlattenOutput
- Inherits:
-
Output
- Object
- Output
- Fluent::Plugin::FlattenOutput
- Includes:
- HandleTagNameMixin
- Defined in:
- lib/fluent/plugin/out_flatten.rb
Defined Under Namespace
Classes: Error
Instance Method Summary collapse
Instance Method Details
#flatten(record) ⇒ Object
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 |
# File 'lib/fluent/plugin/out_flatten.rb', line 42 def flatten(record) flattened = {} if record.has_key?(key) && !record[key].empty? hash = nil begin if parse_json # XXX work-around # fluentd seems to escape json value excessively json = record[key].gsub(/\\"/, '"') hash = JSON.parse(json) else hash = record[key] end rescue JSON::ParserError return flattened end processor = lambda do |root, hash| flattened = {} return flattened unless hash.is_a?(Hash) hash.each do |path, value| keypath = [root, path].join('.') if value.is_a?(Hash) flattened = flattened.merge(processor.call(keypath, value)) else flattened[keypath] = { inner_key => value } end end flattened end flattened = processor.call(key, hash) end flattened end |
#multi_workers_ready? ⇒ Boolean
22 23 24 |
# File 'lib/fluent/plugin/out_flatten.rb', line 22 def multi_workers_ready? true end |
#process(tag, es) ⇒ Object
26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 |
# File 'lib/fluent/plugin/out_flatten.rb', line 26 def process(tag, es) es.each do |time, record| flattened = flatten(record) flattened.each do |keypath, value| tag_with_keypath = [tag.clone, keypath].join('.') filter_record(tag_with_keypath, time, value) if @replace_space_in_tag router.emit(tag_with_keypath.gsub(/\s+/, @replace_space_in_tag), time, value) else router.emit(tag_with_keypath, time, value) end end end end |