From 243d3cc707ecb670bc604acdb5dcfae143a8a72c Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Wed, 5 Aug 2026 13:20:37 +0900 Subject: [PATCH 1/2] Introduce config parameter limit label expansion In the previous versions, there is no mechanism to limit label expansion. That causes a possibility of cardinality OOM DoS. To mitigate such situation, introduced the following parameters: * max_label_value_length: The maximum length of a label value. * max_series_per_metric: The maximum number of label sets a metric can hold. The above parameter is configurable for filter_prometheus and out_prometheus. For example, if about 8 million records are loaded without cardinality limitation, RSS increased from 64MB to 582MB. It might cause OOM DoS. In contrast to that case with cardinality limitation, RSS increased from 64MB to 84 MB in similar case. Then warning message is logged like this: 2026-08-06 14:06:04 +0900 [warn]: prometheus: dropped a label set because the metric reached max_series_per_metric. name="access_requests_total" max_series_per_metric=10000 Signed-off-by: Kentaro Hayashi --- README.md | 32 ++++ lib/fluent/plugin/filter_prometheus.rb | 2 +- lib/fluent/plugin/in_prometheus.rb | 22 +-- lib/fluent/plugin/out_prometheus.rb | 2 +- lib/fluent/plugin/prometheus.rb | 169 ++++++++++++++++-- spec/fluent/plugin/filter_prometheus_spec.rb | 40 +++++ .../plugin/prometheus/log_throttle_spec.rb | 107 +++++++++++ 7 files changed, 342 insertions(+), 32 deletions(-) create mode 100644 spec/fluent/plugin/prometheus/log_throttle_spec.rb diff --git a/README.md b/README.md index f353f09..253cc37 100644 --- a/README.md +++ b/README.md @@ -266,6 +266,38 @@ 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. + ## 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..1ddb9f5 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,11 +352,12 @@ 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 @@ -273,10 +374,54 @@ 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. + 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 + + @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" @@ -306,7 +451,7 @@ def instrument(record, expander) 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) @@ -337,7 +482,7 @@ def instrument(record, expander) 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" @@ -367,7 +512,7 @@ def instrument(record, expander) 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" diff --git a/spec/fluent/plugin/filter_prometheus_spec.rb b/spec/fluent/plugin/filter_prometheus_spec.rb index f98c8c6..46a8439 100644 --- a/spec/fluent/plugin/filter_prometheus_spec.rb +++ b/spec/fluent/plugin/filter_prometheus_spec.rb @@ -45,4 +45,44 @@ it_behaves_like 'instruments record' 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 From 39a9ae34a118132f4c81444a580dff9a0e6904a9 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Fri, 7 Aug 2026 07:12:46 +0000 Subject: [PATCH 2/2] Count a label set as a series only after instrumentation succeeded Metric#labels registered a label set into @series before the metric was actually instrumented. When the instrumentation failed afterwards, for example when the value of `key` is not a number and Counter#increment raises, the label set consumed max_series_per_metric even though nothing was recorded. Records which fail that way could therefore exhaust the limit and make the following valid label sets dropped. Split the responsibility: check_series_limit! only refuses an unknown label set once the limit is reached, and the new remember_series counts it after the instrumentation succeeded. Each metric type now goes through with_label_set, which calls them in that order. Signed-off-by: Kentaro Hayashi Co-Authored-By: Claude --- README.md | 4 ++ lib/fluent/plugin/prometheus.rb | 41 +++++++++++++-- spec/fluent/plugin/filter_prometheus_spec.rb | 55 ++++++++++++++++++++ 3 files changed, 96 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 253cc37..4e44e5f 100644 --- a/README.md +++ b/README.md @@ -298,6 +298,10 @@ 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/prometheus.rb b/lib/fluent/plugin/prometheus.rb index 1ddb9f5..b62fdb5 100644 --- a/lib/fluent/plugin/prometheus.rb +++ b/lib/fluent/plugin/prometheus.rb @@ -361,6 +361,17 @@ def labels(record, expander) 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) @@ -404,6 +415,8 @@ def normalize_label_set(label) # 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 @@ -414,6 +427,18 @@ def check_series_limit!(label) # 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 @@ -445,7 +470,9 @@ 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 @@ -477,7 +504,9 @@ 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 @@ -506,7 +535,9 @@ 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 @@ -543,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 46a8439..345e85a 100644 --- a/spec/fluent/plugin/filter_prometheus_spec.rb +++ b/spec/fluent/plugin/filter_prometheus_spec.rb @@ -46,6 +46,61 @@ 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 + %[