Class: Fluent::Plugin::GraphiteOutput

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

Instance Method Summary collapse

Constructor Details

#initialize ⇒ GraphiteOutput

Returns a new instance of GraphiteOutput.



14
15
16
17
18
 
# File 'lib/fluent/plugin/out_graphite.rb', line 14

def initialize
  super
  require 'graphite-api'
  log.info("Initialize graphite plugin")
end

Instance Method Details

#configure(conf) ⇒ Object 



25
26
27
28
29
30
 
# File 'lib/fluent/plugin/out_graphite.rb', line 25

def configure(conf)
  super
  if @name_keys
    @name_keys = @name_keys.split(',')
  end
end

#connect_client! ⇒ Object 



91
92
93
94
 
# File 'lib/fluent/plugin/out_graphite.rb', line 91

def connect_client!
  @client = GraphiteAPI.new(graphite: "#{@host}:#{@port}")
  log.info("starting client")
end

#format_metrics(tag, record) ⇒ Object 



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/fluent/plugin/out_graphite.rb', line 44

def format_metrics(tag, record)
  filtered_record = if @name_keys
                      record.select { |k, v| @name_keys.include?(k.to_s) }
                    else
                      # defined @name_key_pattern
                      record.select { |k, v| @name_key_pattern.match(k.to_s) }
                    end

  return nil if filtered_record.empty?

  metrics = {}
  tag = tag.sub(/\.$/, '') # may include a dot at the end of the emit_tag fluent-mixin-rewrite-tag-name returns. remove it.
  filtered_record.each do |k, v|
    key = case @tag_for
          when 'ignore' then k.to_s
          when 'prefix' then "#{tag}.#{k}"
          when 'suffix' then "#{k}.#{tag}"
          end

    key = key.gsub(/(\s|\/)+/, '_') # cope with in the case of containing symbols or spaces in the key of the record like in_dstat.
    metrics[key] = v.to_f
  end
  metrics
end

#post(metrics, time) ⇒ Object 



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_graphite.rb', line 69

def post(metrics, time)
  trial ||= 1
  @client.metrics(metrics, time)
  log.warn("Sending metrics: #{metrics}")
rescue Errno::ETIMEDOUT
  # after long periods with nothing emitted, the connection will be closed and result in timeout
  if trial <= @max_retries
    log.warn "out_graphite: connection timeout to #{@host}:#{@port}. Reconnecting... "
    trial += 1
    connect_client!
    retry
  else
    log.error "out_graphite: ERROR: connection timeout to #{@host}:#{@port}. Exceeded max_retries #{@max_retries}"
  end
rescue Errno::ECONNREFUSED
  log.warn "out_graphite: connection refused by #{@host}:#{@port}"
rescue SocketError => se
  log.warn "out_graphite: socket error by #{@host}:#{@port} :#{se}"
rescue StandardError => e
  log.error "out_graphite: ERROR: #{e}"
end

#process(tag, es) ⇒ Object 



32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/fluent/plugin/out_graphite.rb', line 32

def process(tag, es)
  es.each do |time, record|
    emit_tag = tag.dup
    log.info("Emit graphite plugin: #{record}")
    next unless metrics = format_metrics(emit_tag, record)

    # implemented to immediate call post method in this loop, because graphite-api.gem has the buffers.
    post(metrics, time)
  end

end

#start ⇒ Object 



20
21
22
23
 
# File 'lib/fluent/plugin/out_graphite.rb', line 20

def start
  super
  connect_client!
end