Class: Fluent::BurrowPlugin

Inherits:
Output
  • Object
show all
Defined in:
lib/fluent/plugin/out_burrow.rb

Overview

Burrow Output Plugin This plugin allows to extract a single key from an existing event and re-parse it with a given format, and then re-emit a new event with the key’s value replaced, or with the whole record replaced.

Instance Method Summary collapse

Instance Method Details

#configure(conf) ⇒ Object

Parse config hash



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
 
# File 'lib/fluent/plugin/out_burrow.rb', line 32

def configure(conf)
  super

  # One of 'tag', 'remove_prefix' or 'add_prefix' must be specified
  if not @tag and not @remove_prefix and not @add_prefix
    raise Fluent::ConfigError, "One of 'tag', 'remove_prefix' or 'add_prefix' must be specified"
  end
  if @tag and (@remove_prefix or @add_prefix)
    raise Fluent::ConfigError, "Specifying both 'tag' and either 'remove_prefix' or 'add_prefix' is not supported"
  end

  # Prepare for tag modification if required
  if @remove_prefix
    @removed_prefix_string = @remove_prefix.chomp('.') + '.'
    @removed_length = @removed_prefix_string.length
  end
  if @add_prefix
    @added_prefix_string = @add_prefix.chomp('.') + '.'
  end

  # Validate action
  actions = ['replace','overlay','inplace']
  if not actions.include? @action
    raise Fluent::ConfigError, "Invalid 'action', must be one of #{actions.join(',')}"
  end

  # Validate action-based restrictions
  if @action == 'inplace' and @keep_key
    raise Fluent::ConfigError, "Specifying 'keep_key' with action 'inplace' is not supported"
  end

  # Prepare fluent's built-in parser
  @parser = Fluent::TextParser.new()
  @parser.configure(conf)
end

#emit(tag, es, chain) ⇒ Object

This method is called when an event reaches Fluentd.



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
 
# File 'lib/fluent/plugin/out_burrow.rb', line 78

def emit(tag, es, chain)

  # Figure out new event tag (either manually specified, or modified with add_prefix|remove_prefix)
  if @tag
    tag = @tag
  else
    if @remove_prefix and
        ( (tag.start_with?(@removed_prefix_string) and tag.length > @removed_length) or tag == @remove_prefix)
      tag = tag[@removed_length..-1]
    end
    if @add_prefix
      if tag and tag.length > 0
        tag = @added_prefix_string + tag
      else
        tag = @add_prefix
      end
    end
  end

  # Handle all currently available events in stream
  es.each do |time,record|
    # Extract raw key value
    raw_value = record[@key_name]

    # Remember original time key, or raw event time
    raw_time = record[@record_time_key]

    # Try to parse it according to 'format'
    t,values = raw_value ? @parser.parse(raw_value) : [nil, nil]

    # Set new event's time to current time unless new time key was found in the sub-event
    t ||= raw_time

    r = values;

    # Overlay new record on top of original record?
    case @action
    when 'inplace'
      r = record.merge({@key_name => r})
    when 'overlay'
      r = record.merge(r)
    when 'replace'
      # noop
    end

    if ['overlay','replace'].include? @action
      if not @keep_key
        r.delete(@key_name)
      end
    end

    # Preserve 'time' key?
    if @keep_time
      r[@record_time_key] = raw_time
    end

    # Emit event back to Fluent
    if r
      Fluent::Engine.emit(tag, t, r)
    end
  end

  chain.next
end

#shutdown ⇒ Object

This method is called when shutting down.



74
75
 
# File 'lib/fluent/plugin/out_burrow.rb', line 74

def shutdown
end

#start ⇒ Object

This method is called when starting.



69
70
71
 
# File 'lib/fluent/plugin/out_burrow.rb', line 69

def start
  super
end