diff --git a/README.md b/README.md
index f353f09..4e44e5f 100644
--- a/README.md
+++ b/README.md
@@ -266,6 +266,42 @@ You can access nested keys in records via dot or bracket notation (https://docs.
See Supported Metric Type and Labels for more configuration parameters.
+#### Limiting label expansion
+
+Label values come from records, so a metric can grow unboundedly when a label
+is bound to a field with many distinct values. Both plugins limit it:
+
+|parameter|description|default|
+|---|---|---|
+|max_label_value_length|The maximum length of a label value. A longer value is truncated. `0` means unlimited.|256|
+|max_series_per_metric|The maximum number of label sets a metric can hold. A label set beyond the limit is dropped, while the label sets already known keep being instrumented. `0` means unlimited.|10000|
+|ignore_error_log_interval|The interval in seconds to suppress the repeated warning about the dropped label sets. `0` logs every occurrence.|3600|
+
+```
+
+ @type prometheus
+ max_label_value_length 128
+ max_series_per_metric 1000
+
+ name message_foo_counter
+ type counter
+ desc The total number of foo in message.
+ key foo
+
+ path $.kubernetes.pod_name
+
+
+
+```
+
+Note that the number of label sets is counted per metric of each plugin
+instance. When two plugin instances instrument the same metric name, each of
+them has its own limit.
+
+A label set is counted only after the metric was instrumented successfully. A
+record which fails to be instrumented, for example when the value of `key` is
+not a number, does not consume `max_series_per_metric`.
+
## Supported Metric Types
For details of each metric type, see [Prometheus documentation](http://prometheus.io/docs/concepts/metric_types/). Also see [metric name guide](http://prometheus.io/docs/practices/naming/).
diff --git a/lib/fluent/plugin/filter_prometheus.rb b/lib/fluent/plugin/filter_prometheus.rb
index ccdfe78..eaf0963 100644
--- a/lib/fluent/plugin/filter_prometheus.rb
+++ b/lib/fluent/plugin/filter_prometheus.rb
@@ -19,7 +19,7 @@ def multi_workers_ready?
def configure(conf)
super
labels = parse_labels_elements(conf)
- @metrics = Fluent::Plugin::Prometheus.parse_metrics_elements(conf, @registry, labels)
+ @metrics = Fluent::Plugin::Prometheus.parse_metrics_elements(conf, @registry, labels, metric_options)
end
def filter(tag, time, record)
diff --git a/lib/fluent/plugin/in_prometheus.rb b/lib/fluent/plugin/in_prometheus.rb
index 82d4907..9c4f5fc 100644
--- a/lib/fluent/plugin/in_prometheus.rb
+++ b/lib/fluent/plugin/in_prometheus.rb
@@ -43,8 +43,7 @@ def initialize
super
@registry = ::Prometheus::Client.registry
@secure = nil
- @error_log_mutex = Mutex.new
- @last_error_logs = {} # scope => [logged_at, fingerprint, suppressed_count]
+ @error_log_throttle = nil
end
def configure(conf)
@@ -63,6 +62,8 @@ def configure(conf)
@base_port = @port
@port += fluentd_worker_id
+
+ @error_log_throttle = Fluent::Plugin::Prometheus::LogThrottle.new(@ignore_error_log_interval)
end
def multi_workers_ready?
@@ -281,22 +282,7 @@ def response(metrics)
def log_error_throttled(scope, message, error:)
fingerprint = [error.class, error.message]
- suppressed = 0
-
- emit = @error_log_mutex.synchronize do
- last = @last_error_logs[scope]
- now = Fluent::Clock.now
- if last.nil? ||
- last[1] != fingerprint ||
- (now - last[0]) >= @ignore_error_log_interval
- suppressed = last && last[1] == fingerprint ? last[2] : 0
- @last_error_logs[scope] = [now, fingerprint, 0]
- true
- else
- last[2] += 1
- false
- end
- end
+ emit, suppressed = @error_log_throttle.check(scope, fingerprint)
return unless emit
if suppressed > 0
diff --git a/lib/fluent/plugin/out_prometheus.rb b/lib/fluent/plugin/out_prometheus.rb
index cdaae4d..9c611e3 100644
--- a/lib/fluent/plugin/out_prometheus.rb
+++ b/lib/fluent/plugin/out_prometheus.rb
@@ -19,7 +19,7 @@ def multi_workers_ready?
def configure(conf)
super
labels = parse_labels_elements(conf)
- @metrics = Fluent::Plugin::Prometheus.parse_metrics_elements(conf, @registry, labels)
+ @metrics = Fluent::Plugin::Prometheus.parse_metrics_elements(conf, @registry, labels, metric_options)
end
def process(tag, es)
diff --git a/lib/fluent/plugin/prometheus.rb b/lib/fluent/plugin/prometheus.rb
index 5db615a..b62fdb5 100644
--- a/lib/fluent/plugin/prometheus.rb
+++ b/lib/fluent/plugin/prometheus.rb
@@ -1,5 +1,6 @@
require 'prometheus/client'
require 'prometheus/client/formats/text'
+require 'fluent/clock'
require 'fluent/plugin/prometheus/placeholder_expander'
module Fluent
@@ -31,6 +32,63 @@ def parse_labels_elements(conf)
module Prometheus
class AlreadyRegisteredError < StandardError; end
+ # raised when a metric is about to expand a label set beyond its limit
+ class LabelSetLimitError < StandardError; end
+
+ # 0 or less means unlimited
+ DEFAULT_MAX_LABEL_VALUE_LENGTH = 256
+ DEFAULT_MAX_SERIES_PER_METRIC = 10_000
+ DEFAULT_IGNORE_ERROR_LOG_INTERVAL = 3600
+
+ def self.included(klass)
+ klass.class_eval do
+ desc 'The maximum length of a label value. Longer values are truncated. 0 means unlimited.'
+ config_param :max_label_value_length, :integer, default: DEFAULT_MAX_LABEL_VALUE_LENGTH
+ desc 'The maximum number of label sets a metric can hold. Exceeding label sets are dropped. 0 means unlimited.'
+ config_param :max_series_per_metric, :integer, default: DEFAULT_MAX_SERIES_PER_METRIC
+ desc 'The interval to suppress the repeated same error log.'
+ config_param :ignore_error_log_interval, :time, default: DEFAULT_IGNORE_ERROR_LOG_INTERVAL
+ end
+ end
+
+ # Suppresses the repeated log for the same key within the interval.
+ # Shared by filter/out_prometheus (keyed by metric name) and in_prometheus
+ # (keyed by an error scope). Each plugin owns its own instance, since the
+ # lifetime differs; only the implementation is shared. The granularity is
+ # absorbed by the key, and an optional fingerprint lets a caller emit
+ # immediately when the content changes (e.g. a different error).
+ class LogThrottle
+ Entry = Struct.new(:time, :fingerprint, :suppressed)
+
+ def initialize(interval)
+ @interval = interval
+ @mutex = Mutex.new
+ # bounded by the number of keys (metrics / scopes), so it never grows
+ # unexpectedly
+ @entries = {}
+ end
+
+ # Returns [emit?, suppressed_count]. It emits (returns true) when the key
+ # is seen for the first time, when the fingerprint changes, or when the
+ # interval has elapsed. suppressed_count is how many logs were dropped
+ # for the same fingerprint since the last emission.
+ def check(key, fingerprint = nil)
+ return [true, 0] if @interval <= 0
+
+ @mutex.synchronize do
+ now = Fluent::Clock.now
+ last = @entries[key]
+ if last.nil? || last.fingerprint != fingerprint || (now - last.time) >= @interval
+ suppressed = (last && last.fingerprint == fingerprint) ? last.suppressed : 0
+ @entries[key] = Entry.new(now, fingerprint, 0)
+ [true, suppressed]
+ else
+ last.suppressed += 1
+ [false, 0]
+ end
+ end
+ end
+ end
def self.parse_labels_elements(conf)
labels = conf.elements.select { |e| e.name == 'labels' }
@@ -119,7 +177,7 @@ def self.parse_initlabels_elements(conf, base_labels)
base_initlabels
end
- def self.parse_metrics_elements(conf, registry, labels = {})
+ def self.parse_metrics_elements(conf, registry, labels = {}, opts = {})
metrics = []
conf.elements.select { |element|
element.name == 'metric'
@@ -130,13 +188,13 @@ def self.parse_metrics_elements(conf, registry, labels = {})
end
case element['type']
when 'summary'
- metrics << Fluent::Plugin::Prometheus::Summary.new(element, registry, labels)
+ metrics << Fluent::Plugin::Prometheus::Summary.new(element, registry, labels, opts)
when 'gauge'
- metrics << Fluent::Plugin::Prometheus::Gauge.new(element, registry, labels)
+ metrics << Fluent::Plugin::Prometheus::Gauge.new(element, registry, labels, opts)
when 'counter'
- metrics << Fluent::Plugin::Prometheus::Counter.new(element, registry, labels)
+ metrics << Fluent::Plugin::Prometheus::Counter.new(element, registry, labels, opts)
when 'histogram'
- metrics << Fluent::Plugin::Prometheus::Histogram.new(element, registry, labels)
+ metrics << Fluent::Plugin::Prometheus::Histogram.new(element, registry, labels, opts)
else
raise ConfigError, "type option must be 'counter', 'gauge', 'summary' or 'histogram'"
end
@@ -165,6 +223,28 @@ def configure(conf)
@placeholder_values = {}
@placeholder_expander_builder = Fluent::Plugin::Prometheus.placeholder_expander(log)
@hostname = Socket.gethostname
+ @label_set_limit_log_throttle = Fluent::Plugin::Prometheus::LogThrottle.new(@ignore_error_log_interval)
+ end
+
+ def metric_options
+ {
+ max_label_value_length: @max_label_value_length,
+ max_series_per_metric: @max_series_per_metric,
+ }
+ end
+
+ def warn_label_set_limit(metric)
+ emit, suppressed = @label_set_limit_log_throttle.check(metric.name)
+ return unless emit
+
+ if suppressed > 0
+ log.warn "prometheus: dropped a label set because the metric reached max_series_per_metric.",
+ name: metric.name, max_series_per_metric: metric.max_series_per_metric,
+ suppressed_log_count: suppressed
+ else
+ log.warn "prometheus: dropped a label set because the metric reached max_series_per_metric.",
+ name: metric.name, max_series_per_metric: metric.max_series_per_metric
+ end
end
def instrument_single(tag, time, record, metrics)
@@ -180,6 +260,9 @@ def instrument_single(tag, time, record, metrics)
metrics.each do |metric|
begin
metric.instrument(record, expander)
+ rescue Fluent::Plugin::Prometheus::LabelSetLimitError
+ # dropping the label set is intended, so it is not an error event
+ warn_label_set_limit(metric)
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)
@@ -201,6 +284,9 @@ def instrument(tag, es, metrics)
metrics.each do |metric|
begin
metric.instrument(record, expander)
+ rescue Fluent::Plugin::Prometheus::LabelSetLimitError
+ # dropping the label set is intended, so it is not an error event
+ warn_label_set_limit(metric)
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)
@@ -214,8 +300,10 @@ class Metric
attr_reader :name
attr_reader :key
attr_reader :desc
+ attr_reader :max_label_value_length
+ attr_reader :max_series_per_metric
- def initialize(element, registry, labels)
+ def initialize(element, registry, labels, opts = {})
['name', 'desc'].each do |key|
if element[key].nil?
raise ConfigError, "metric requires '#{key}' option"
@@ -230,8 +318,20 @@ def initialize(element, registry, labels)
@base_labels = Fluent::Plugin::Prometheus.parse_labels_elements(element)
@base_labels = labels.merge(@base_labels)
+ # can narrow down the limits given by the plugin
+ @max_label_value_length = metric_limit(element, 'max_label_value_length',
+ opts.fetch(:max_label_value_length, DEFAULT_MAX_LABEL_VALUE_LENGTH))
+ @max_series_per_metric = metric_limit(element, 'max_series_per_metric',
+ opts.fetch(:max_series_per_metric, DEFAULT_MAX_SERIES_PER_METRIC))
+ @series = {}
+ @series_mutex = Mutex.new
+
if @initialized
@base_initlabels = Fluent::Plugin::Prometheus.parse_initlabels_elements(element, @base_labels)
+ # the pre-initialized label sets consume the limit as well
+ @base_initlabels.each do |initlabels|
+ @series[normalize_label_set(initlabels)] = true
+ end
end
end
@@ -252,14 +352,26 @@ def labels(record, expander)
label = {}
@base_labels.each do |k, v|
if v.is_a?(String)
- label[k] = expander.expand(v)
+ label[k] = truncate_label_value(expander.expand(v))
else
- label[k] = v.call(record)
+ label[k] = truncate_label_value(v.call(record))
end
end
+ check_series_limit!(label)
label
end
+ # Instruments a record through the given block and counts its label set
+ # as a series only after the block succeeded. A record which fails to be
+ # instrumented (e.g. its value is not a number) must not consume
+ # max_series_per_metric, otherwise such records could exhaust the limit
+ # and make the following valid label sets dropped.
+ def with_label_set(record, expander)
+ label = labels(record, expander)
+ yield label
+ remember_series(label)
+ end
+
def self.get(registry, name, type, docstring)
metric = registry.get(name)
@@ -273,10 +385,68 @@ def self.get(registry, name, type, docstring)
metric
end
+
+ private
+
+ def metric_limit(element, name, default)
+ return default unless element.has_key?(name)
+
+ begin
+ # base 10 explicitly, so that a value like 08 is not an octal
+ Integer(element[name], 10)
+ rescue ArgumentError, TypeError
+ raise ConfigError, "#{name} in must be an integer: #{element[name]}"
+ end
+ end
+
+ def truncate_label_value(value)
+ # a RecordAccessor may return a value which is not a String
+ value = value.to_s unless value.is_a?(String)
+ return value if @max_label_value_length <= 0
+
+ value.length > @max_label_value_length ? value[0, @max_label_value_length] : value
+ end
+
+ def normalize_label_set(label)
+ label.each_with_object({}) do |(k, v), normalized|
+ normalized[k] = truncate_label_value(v)
+ end
+ end
+
+ # Keeps the cardinality of a metric bounded. Once the limit is reached,
+ # the already known label sets keep working and only a new one is refused.
+ # The label set is not counted here but by #remember_series, so that a
+ # failed instrumentation does not consume the limit.
+ def check_series_limit!(label)
+ return if @max_series_per_metric <= 0
+
+ @series_mutex.synchronize do
+ next if @series.key?(label)
+
+ if @series.size >= @max_series_per_metric
+ # the message must not contain the label set, it comes from a record
+ raise LabelSetLimitError, "#{@name} reached max_series_per_metric (#{@max_series_per_metric})"
+ end
+ end
+ end
+
+ def remember_series(label)
+ return if @max_series_per_metric <= 0
+
+ @series_mutex.synchronize do
+ next if @series.key?(label)
+ # a concurrent instrumentation may have filled the limit after
+ # check_series_limit! passed. The label set is already instrumented,
+ # but it is not counted so that the limit is never exceeded.
+ next if @series.size >= @max_series_per_metric
+
+ @series[label] = true
+ end
+ end
end
class Gauge < Metric
- def initialize(element, registry, labels)
+ def initialize(element, registry, labels, opts = {})
super
if @key.nil?
raise ConfigError, "gauge metric requires 'key' option"
@@ -300,13 +470,15 @@ def instrument(record, expander)
value = @key.call(record)
end
if value
- @gauge.set(value, labels: labels(record, expander))
+ with_label_set(record, expander) do |label|
+ @gauge.set(value, labels: label)
+ end
end
end
end
class Counter < Metric
- def initialize(element, registry, labels)
+ def initialize(element, registry, labels, opts = {})
super
begin
@counter = registry.counter(element['name'].to_sym, docstring: element['desc'], labels: @base_labels.keys)
@@ -332,12 +504,14 @@ def instrument(record, expander)
# ignore if record value is nil
return if value.nil?
- @counter.increment(by: value, labels: labels(record, expander))
+ with_label_set(record, expander) do |label|
+ @counter.increment(by: value, labels: label)
+ end
end
end
class Summary < Metric
- def initialize(element, registry, labels)
+ def initialize(element, registry, labels, opts = {})
super
if @key.nil?
raise ConfigError, "summary metric requires 'key' option"
@@ -361,13 +535,15 @@ def instrument(record, expander)
value = @key.call(record)
end
if value
- @summary.observe(value, labels: labels(record, expander))
+ with_label_set(record, expander) do |label|
+ @summary.observe(value, labels: label)
+ end
end
end
end
class Histogram < Metric
- def initialize(element, registry, labels)
+ def initialize(element, registry, labels, opts = {})
super
if @key.nil?
raise ConfigError, "histogram metric requires 'key' option"
@@ -398,7 +574,9 @@ def instrument(record, expander)
value = @key.call(record)
end
if value
- @histogram.observe(value, labels: labels(record, expander))
+ with_label_set(record, expander) do |label|
+ @histogram.observe(value, labels: label)
+ end
end
end
end
diff --git a/spec/fluent/plugin/filter_prometheus_spec.rb b/spec/fluent/plugin/filter_prometheus_spec.rb
index f98c8c6..345e85a 100644
--- a/spec/fluent/plugin/filter_prometheus_spec.rb
+++ b/spec/fluent/plugin/filter_prometheus_spec.rb
@@ -45,4 +45,99 @@
it_behaves_like 'instruments record'
end
+
+ describe 'max_series_per_metric' do
+ let(:config) {
+ BASE_CONFIG + %[
+ max_series_per_metric 1
+
+ name limited
+ type counter
+ desc Something foo.
+ key foo
+
+ path $.path
+
+
+ ]
+ }
+ let(:counter) { registry.get(:limited) }
+
+ def drop_logs
+ driver.logs.select { |log| log.include?('dropped a label set') }
+ end
+
+ it 'drops a new label set once the limit is reached' do
+ driver.run(default_tag: tag) do
+ driver.feed(event_time, {'foo' => 1, 'path' => '/a'})
+ driver.feed(event_time, {'foo' => 1, 'path' => '/b'})
+ end
+
+ expect(counter.values.keys).to eq([{path: '/a'}])
+ expect(drop_logs.size).to eq(1)
+ end
+
+ it 'keeps instrumenting a known label set after the limit is reached' do
+ driver.run(default_tag: tag) do
+ driver.feed(event_time, {'foo' => 1, 'path' => '/a'})
+ driver.feed(event_time, {'foo' => 1, 'path' => '/b'})
+ driver.feed(event_time, {'foo' => 1, 'path' => '/a'})
+ end
+
+ expect(counter.get(labels: {path: '/a'})).to eq(2)
+ end
+
+ it 'does not consume the limit by a label set which failed to be instrumented' do
+ driver.run(default_tag: tag) do
+ # a non numeric value makes Counter#increment raise, after the label set
+ # has been built
+ driver.feed(event_time, {'foo' => 'not a number', 'path' => '/a'})
+ driver.feed(event_time, {'foo' => 1, 'path' => '/b'})
+ end
+
+ expect(driver.error_events.size).to eq(1)
+ expect(counter.values.keys).to eq([{path: '/b'}])
+ expect(drop_logs).to be_empty
+ end
+ end
+
+ describe 'label set limit log throttling' do
+ let(:config) {
+ BASE_CONFIG + %[
+ ignore_error_log_interval 3600
+
+ name throttled
+ type counter
+ desc Something foo.
+ key foo
+
+ ]
+ }
+ # Fluent::Clock.now is monotonic, so a plain Hash is enough to drive it
+ let(:clock) { { now: 1000.0 } }
+ let(:metric) { double('metric', name: :throttled, max_series_per_metric: 5) }
+
+ before do
+ allow(Fluent::Clock).to receive(:now) { clock[:now] }
+ end
+
+ def drop_logs
+ driver.logs.select { |log| log.include?('dropped a label set') }
+ end
+
+ it 'warns only once within ignore_error_log_interval' do
+ 5.times { driver.instance.send(:warn_label_set_limit, metric) }
+ expect(drop_logs.size).to eq(1)
+ end
+
+ it 'reports how many warnings were suppressed in the meantime' do
+ 3.times { driver.instance.send(:warn_label_set_limit, metric) }
+ clock[:now] += driver.instance.ignore_error_log_interval
+ driver.instance.send(:warn_label_set_limit, metric)
+ logs = drop_logs
+ expect(logs.size).to eq(2)
+ expect(logs.first).not_to include('suppressed_log_count')
+ expect(logs.last).to include('suppressed_log_count=2')
+ end
+ end
end
diff --git a/spec/fluent/plugin/prometheus/log_throttle_spec.rb b/spec/fluent/plugin/prometheus/log_throttle_spec.rb
new file mode 100644
index 0000000..85e0c7f
--- /dev/null
+++ b/spec/fluent/plugin/prometheus/log_throttle_spec.rb
@@ -0,0 +1,107 @@
+require 'spec_helper'
+
+describe Fluent::Plugin::Prometheus::LogThrottle do
+ # Fluent::Clock.now is monotonic, so a plain Hash is enough to drive it
+ let(:clock) { { now: 1000.0 } }
+ let(:interval) { 3600 }
+ subject(:throttle) { described_class.new(interval) }
+
+ before do
+ allow(Fluent::Clock).to receive(:now) { clock[:now] }
+ end
+
+ describe '#check' do
+ it 'emits on the first occurrence of a key' do
+ emit, suppressed = throttle.check(:foo)
+ expect(emit).to be true
+ expect(suppressed).to eq(0)
+ end
+
+ it 'suppresses the same key within the interval' do
+ throttle.check(:foo)
+ clock[:now] += interval - 1
+ emit, _ = throttle.check(:foo)
+ expect(emit).to be false
+ end
+
+ it 'emits again once the interval has elapsed' do
+ throttle.check(:foo)
+ clock[:now] += interval
+ emit, _ = throttle.check(:foo)
+ expect(emit).to be true
+ end
+
+ it 'reports how many occurrences were suppressed in the meantime' do
+ throttle.check(:foo) # emits, suppressed=0
+ 2.times { throttle.check(:foo) } # suppressed 1, then 2
+ clock[:now] += interval
+ emit, suppressed = throttle.check(:foo)
+ expect(emit).to be true
+ expect(suppressed).to eq(2)
+ end
+
+ it 'resets the suppressed count after emitting' do
+ throttle.check(:foo)
+ 2.times { throttle.check(:foo) }
+ clock[:now] += interval
+ throttle.check(:foo) # emits with suppressed=2
+ clock[:now] += interval
+ _, suppressed = throttle.check(:foo)
+ expect(suppressed).to eq(0)
+ end
+
+ it 'keeps a separate slot per key' do
+ expect(throttle.check(:foo).first).to be true
+ expect(throttle.check(:bar).first).to be true
+ end
+
+ context 'with a fingerprint' do
+ it 'emits immediately when the fingerprint changes within the interval' do
+ expect(throttle.check(:foo, [RuntimeError, 'a']).first).to be true
+ expect(throttle.check(:foo, [RuntimeError, 'b']).first).to be true
+ end
+
+ # the caller builds a fresh fingerprint per event, so it must be compared
+ # by value, not by identity
+ it 'suppresses an equal fingerprint given as a different object' do
+ expect(throttle.check(:foo, [RuntimeError, 'a']).first).to be true
+ expect(throttle.check(:foo, [RuntimeError, 'a']).first).to be false
+ end
+
+ it 'does not carry the suppressed count across a fingerprint change' do
+ throttle.check(:foo, [RuntimeError, 'a'])
+ 2.times { throttle.check(:foo, [RuntimeError, 'a']) }
+ emit, suppressed = throttle.check(:foo, [RuntimeError, 'b'])
+ expect(emit).to be true
+ expect(suppressed).to eq(0)
+ end
+ end
+
+ context 'when interval is zero' do
+ let(:interval) { 0 }
+
+ it 'always emits without consulting the clock' do
+ expect(Fluent::Clock).not_to receive(:now)
+ 3.times do
+ emit, suppressed = throttle.check(:foo)
+ expect(emit).to be true
+ expect(suppressed).to eq(0)
+ end
+ end
+ end
+
+ context 'when interval is negative' do
+ let(:interval) { -1 }
+
+ it 'always emits' do
+ expect(throttle.check(:foo).first).to be true
+ expect(throttle.check(:foo).first).to be true
+ end
+ end
+
+ it 'serializes concurrent checks for the same key into a single emission' do
+ results = 10.times.map { Thread.new { throttle.check(:foo).first } }.map(&:value)
+ expect(results.count(true)).to eq(1)
+ end
+ end
+end