Module: Fluent::Plugin::Prometheus

Included in:
PrometheusFilter, PrometheusOutput
Defined in:
lib/fluent/plugin/prometheus.rb,
lib/fluent/plugin/prometheus/placeholder_expander.rb

Defined Under Namespace

Classes: AlreadyRegisteredError, Counter, ExpandBuilder, Gauge, Histogram, Metric, Summary

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.parse_initlabels_elements(conf, base_labels) ⇒ Object 



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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/fluent/plugin/prometheus.rb', line 59

def self.parse_initlabels_elements(conf, base_labels)
  base_initlabels = []

  # We first treat the special case of RecordAccessors and Placeholders labels if any declared
  conf.elements.select { |e| e.name == 'initlabels' }.each { |block|
    initlabels = {}

    block.each do |key, value|
      if not base_labels.has_key? key.to_sym
        raise ConfigError, "Key #{key} in <initlabels> is non existent in <labels> for metric #{conf['name']}"
      end

      if value.start_with?('$.') || value.start_with?('$[') || value.start_with?('${')
        raise ConfigError, "Cannot use RecordAccessor or placeholder #{value} (key #{key}) in a <initlabels> in metric #{conf['name']}"
      end

      base_label_value = base_labels[key.to_sym]

      if !(base_label_value.class == Fluent::PluginHelper::RecordAccessor::Accessor) && ! (base_label_value.start_with?('${') )
        raise ConfigError, "Cannot set <initlabels> on non RecordAccessor/Placeholder key #{key} (value #{value}) in metric #{conf['name']}"
      end

      if base_label_value == '${worker_id}' || base_label_value == '${hostname}'
        raise ConfigError, "Cannot set <initlabels> on reserved placeholder #{base_label_value} for key #{key} in metric #{conf['name']}"
      end
      
      initlabels[key.to_sym] = value
    end

    # Now adding all labels that are not RecordAccessor nor Placeholder labels as is
    base_labels.each do |key, value|
      if base_labels[key.to_sym].class != Fluent::PluginHelper::RecordAccessor::Accessor
        if value == '${worker_id}'
          # We retrieve fluentd_worker_id this way to not overcomplicate the code
          initlabels[key.to_sym] = (ENV['SERVERENGINE_WORKER_ID'] || 0).to_i
        elsif value == '${hostname}'
          initlabels[key.to_sym] = Socket.gethostname
        elsif !(value.start_with?('${'))
          initlabels[key.to_sym] = value
        end
      end
    end

    base_initlabels << initlabels
  }

  # Testing for RecordAccessor/Placeholder labels missing a declaration in <initlabels> blocks
  base_labels.each do |key, value|
    if value.class == Fluent::PluginHelper::RecordAccessor::Accessor || value.start_with?('${')
      if not base_initlabels.map(&:keys).flatten.include? (key.to_sym)
          raise ConfigError, "RecordAccessor/Placeholder key #{key} with value #{value} has not been set in a <initlabels> for initialized metric #{conf['name']}"
      end
    end
  end

  if base_initlabels.length == 0
    # There were no RecordAccessor nor Placeholder labels, we blunty retrieve the static base_labels
    base_initlabels << base_labels
  end

  base_initlabels
end

.parse_labels_elements(conf) ⇒ Object 



35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/fluent/plugin/prometheus.rb', line 35

def self.parse_labels_elements(conf)
  labels = conf.elements.select { |e| e.name == 'labels' }
  if labels.size > 1
    raise ConfigError, "labels section must have at most 1"
  end

  base_labels = {}
  unless labels.empty?
    labels.first.each do |key, value|
      labels.first.has_key?(key)

      # use RecordAccessor only for $. and $[ syntax
      # otherwise use the value as is or expand the value by RecordTransformer for ${} syntax
      if value.start_with?('$.') || value.start_with?('$[')
        base_labels[key.to_sym] = PluginHelper::RecordAccessor::Accessor.new(value)
      else
        base_labels[key.to_sym] = value
      end
    end
  end

  base_labels
end

.parse_metrics_elements(conf, registry, labels = {}) ⇒ Object 



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/fluent/plugin/prometheus.rb', line 122

def self.parse_metrics_elements(conf, registry, labels = {})
  metrics = []
  conf.elements.select { |element|
    element.name == 'metric'
  }.each { |element|
    if element.has_key?('key') && (element['key'].start_with?('$.') || element['key'].start_with?('$['))
      value = element['key']
      element['key'] = PluginHelper::RecordAccessor::Accessor.new(value)
    end
    case element['type']
    when 'summary'
      metrics << Fluent::Plugin::Prometheus::Summary.new(element, registry, labels)
    when 'gauge'
      metrics << Fluent::Plugin::Prometheus::Gauge.new(element, registry, labels)
    when 'counter'
      metrics << Fluent::Plugin::Prometheus::Counter.new(element, registry, labels)
    when 'histogram'
      metrics << Fluent::Plugin::Prometheus::Histogram.new(element, registry, labels)
    else
      raise ConfigError, "type option must be 'counter', 'gauge', 'summary' or 'histogram'"
    end
  }
  metrics
end

.placeholder_expander(log) ⇒ Object 



147
148
149
 
# File 'lib/fluent/plugin/prometheus.rb', line 147

def self.placeholder_expander(log)
  Fluent::Plugin::Prometheus::ExpandBuilder.new(log: log)
end

Instance Method Details

#configure(conf) ⇒ Object 



163
164
165
166
167
168
 
# File 'lib/fluent/plugin/prometheus.rb', line 163

def configure(conf)
  super
  @placeholder_values = {}
  @placeholder_expander_builder = Fluent::Plugin::Prometheus.placeholder_expander(log)
  @hostname = Socket.gethostname
end

#instrument(tag, es, metrics) ⇒ Object 



190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/fluent/plugin/prometheus.rb', line 190

def instrument(tag, es, metrics)
  placeholder_values = {
    'tag' => tag,
    'hostname' => @hostname,
    'worker_id' => fluentd_worker_id,
  }

  es.each do |time, record|
    record = stringify_keys(record)
    placeholders = record.merge(placeholder_values)
    expander = @placeholder_expander_builder.build(placeholders)
    metrics.each do |metric|
      begin
        metric.instrument(record, expander)
      rescue => e
        log.warn "prometheus: failed to instrument a metric.", error_class: e.class, error: e, tag: tag, name: metric.name
        router.emit_error_event(tag, time, record, e)
      end
    end
  end
end

#instrument_single(tag, time, record, metrics) ⇒ Object 



170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/fluent/plugin/prometheus.rb', line 170

def instrument_single(tag, time, record, metrics)
  @placeholder_values[tag] ||= {
    'tag' => tag,
    'hostname' => @hostname,
    'worker_id' => fluentd_worker_id,
  }

  record = stringify_keys(record)
  placeholders = record.merge(@placeholder_values[tag])
  expander = @placeholder_expander_builder.build(placeholders)
  metrics.each do |metric|
    begin
      metric.instrument(record, expander)
    rescue => e
      log.warn "prometheus: failed to instrument a metric.", error_class: e.class, error: e, tag: tag, name: metric.name
      router.emit_error_event(tag, time, record, e)
    end
  end
end

#stringify_keys(hash_to_stringify) ⇒ Object 



151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/fluent/plugin/prometheus.rb', line 151

def stringify_keys(hash_to_stringify)
  # Adapted from: https://www.jvt.me/posts/2019/09/07/ruby-hash-keys-string-symbol/
  hash_to_stringify.map do |k,v|
    value_or_hash = if v.instance_of? Hash
                      stringify_keys(v)
                    else
                      v
                    end
    [k.to_s, value_or_hash]
  end.to_h
end