Module: Fluent::Plugin::Prometheus

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

Defined Under Namespace

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

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.parse_labels_elements(conf) ⇒ Object 



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

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 



60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/fluent/plugin/prometheus.rb', line 60

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 



96
97
98
 
# File 'lib/fluent/plugin/prometheus.rb', line 96

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

.start_retention_threads(metrics, registry, thread_create, thread_running, log) ⇒ Object 



85
86
87
88
89
90
91
92
93
94
 
# File 'lib/fluent/plugin/prometheus.rb', line 85

def self.start_retention_threads(metrics, registry, thread_create, thread_running, log)
  metrics.select { |metric| metric.has_retention? }.each do |metric|
    thread_create.call("prometheus_retention_#{metric.name}".to_sym) do
      while thread_running.call()
        metric.remove_expired_metrics(registry, log)
        sleep(metric.retention_check_interval)
      end
    end
  end
end

Instance Method Details

#configure(conf) ⇒ Object 



117
118
119
120
121
122
 
# File 'lib/fluent/plugin/prometheus.rb', line 117

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

#initialize ⇒ Object 



112
113
114
115
 
# File 'lib/fluent/plugin/prometheus.rb', line 112

def initialize
  super
  ::Prometheus::Client.config.data_store = Fluent::Plugin::Prometheus::DataStore.new
end

#instrument(tag, es, metrics) ⇒ Object 



144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/fluent/plugin/prometheus.rb', line 144

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 



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

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 



100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/fluent/plugin/prometheus.rb', line 100

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