From 1bc2fd662750c6b59ab67f769a02cc78407a0a85 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Wed, 26 Aug 2026 02:33:39 +0000 Subject: [PATCH 1/3] Introduce max_series_per_metric Before: a metric held a series for every distinct label set a record expanded to. In one experiment, about 8 million records took the RSS from 64MB to 582MB, which is an OOM DoS a client can drive. After: max_series_per_metric drops a record which brings a new label set once the metric holds that many, and the same experiment ends at 84MB instead. Set it on the plugin, or per . It defaults to 0 (no limit), because a drop loses the record for good. The label sets are counted per client metric, so sections with the same name share one count. A slot is taken before instrumenting and given back on failure, so that concurrent records cannot both pass the limit and a record which fails to be instrumented does not consume it. A drop gets a throttled warning and is counted in fluentd_prometheus_dropped_label_sets_total. Signed-off-by: Kentaro Hayashi Co-Authored-By: Claude --- README.md | 65 ++++ lib/fluent/plugin/filter_prometheus.rb | 2 +- lib/fluent/plugin/out_prometheus.rb | 2 +- lib/fluent/plugin/prometheus.rb | 252 ++++++++++++++- lib/fluent/plugin/prometheus/log_throttle.rb | 12 +- spec/fluent/plugin/filter_prometheus_spec.rb | 72 +++++ spec/fluent/plugin/out_prometheus_spec.rb | 4 + .../plugin/prometheus/log_throttle_spec.rb | 19 ++ .../plugin/prometheus/series_limit_spec.rb | 289 ++++++++++++++++++ spec/fluent/plugin/shared.rb | 108 +++++++ 10 files changed, 803 insertions(+), 22 deletions(-) create mode 100644 spec/fluent/plugin/prometheus/series_limit_spec.rb diff --git a/README.md b/README.md index f353f09..cd1f672 100644 --- a/README.md +++ b/README.md @@ -266,6 +266,71 @@ 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 grows unboundedly when a label is +bound to a field with many distinct values. Both plugins can bound it with +`max_series_per_metric`: once a metric holds that many label sets, a record +which brings a new one is dropped, while the label sets already known keep +being instrumented. + +|parameter|description|default| +|---|---|---| +|max_series_per_metric|The maximum number of label sets a metric can hold. `0` means unlimited.|0| +|ignore_error_log_interval|The interval in seconds to suppress the repeated warning about the drops. `0` logs every occurrence.|3600| + +**The limit is disabled by default and must be enabled explicitly**, since a +dropped record is lost and cannot be recovered. A `` section overrides +the value given to the plugin, so that a metric which expands faster than the +others is bound on its own, while one whose labels are known to be bounded +stays unlimited with `0`: + +``` + + @type prometheus + max_series_per_metric 1000 + + name message_foo_counter + type counter + desc The total number of foo in message. + key foo + max_series_per_metric 10 + + path $.kubernetes.pod_name + + + +``` + +The label sets are counted per metric name, not per `` section: sections +with the same `name`, in one plugin or in two, share one count. Each of them +refuses a new label set once that shared count reaches its own limit, so a +section which stays at `0` adds label sets without counting them. + +A record which fails to be instrumented, for example when the value of `key` is +not a number, does not consume the limit. A pre-initialized label set +(`initialized` and ``) consumes it from the start, since the metric +holds it before any record arrives. + +##### Observing what the limit leaves out + +A dropped label set is not routed to `@ERROR`, because it is what the +configuration asks for. It is reported in two ways instead: + +* a warning in the Fluentd log, throttled per metric: it is suppressed for + `ignore_error_log_interval` seconds and reports how many warnings were + suppressed in the meantime. +* `fluentd_prometheus_dropped_label_sets_total{name}`, which counts the records + the metric `name` did not instrument. It is registered on the first drop, so + it does not show up as long as nothing is dropped, and its label comes from + the configuration and not from a record, so it cannot expand on its own. + +Alert on the counter to notice that a metric is losing records: + +``` +rate(fluentd_prometheus_dropped_label_sets_total[5m]) > 0 +``` + ## 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/out_prometheus.rb b/lib/fluent/plugin/out_prometheus.rb index 245b666..3225fec 100644 --- a/lib/fluent/plugin/out_prometheus.rb +++ b/lib/fluent/plugin/out_prometheus.rb @@ -21,7 +21,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 112fd31..2df90ed 100644 --- a/lib/fluent/plugin/prometheus.rb +++ b/lib/fluent/plugin/prometheus.rb @@ -32,6 +32,85 @@ def parse_labels_elements(conf) module Prometheus class AlreadyRegisteredError < StandardError; end + class LabelSetLimitError < StandardError; end + + # 0 or less means unlimited. The limit is unlimited by default, because + # enabling it changes the existing metrics silently: dropping a label set + # loses the record without any way to recover it. An operator who needs + # to bound the cardinality has to opt in explicitly. + DEFAULT_MAX_SERIES_PER_METRIC = 0 + DEFAULT_IGNORE_ERROR_LOG_INTERVAL = 3600 + + # the drops are visible in Prometheus, not only in the Fluentd log + DROPPED_LABEL_SETS_METRIC_NAME = :fluentd_prometheus_dropped_label_sets_total + DROPPED_LABEL_SETS_METRIC_DESC = 'The total number of records dropped because the metric reached max_series_per_metric.' + + def self.included(klass) + klass.class_eval do + desc 'The maximum number of label sets a metric can hold. Exceeding label sets are dropped. 0 (default) means unlimited.' + config_param :max_series_per_metric, :integer, default: DEFAULT_MAX_SERIES_PER_METRIC + desc 'The interval to suppress the repeated warning about the drops.' + config_param :ignore_error_log_interval, :time, default: DEFAULT_IGNORE_ERROR_LOG_INTERVAL + end + end + + # The label sets a client metric holds. The client registry keys its + # metrics by name alone, so every section with the same name + # instruments the same client metric and shares this set, instead of + # holding max_series_per_metric label sets of its own. + class SeriesSet + # The set is kept on the client metric, so that it is found again by + # every section and goes away with it. Metrics are built at + # configuration time, which is single threaded, so no lock is needed. + IVAR = :@fluent_plugin_prometheus_series_set + + def self.of(client_metric) + client_metric.instance_variable_get(IVAR) || + client_metric.instance_variable_set(IVAR, new) + end + + def initialize + @series = {} + @mutex = Mutex.new + end + + def size + @mutex.synchronize { @series.size } + end + + # Checking the limit and taking the slot happen under the same lock, so + # that concurrent calls cannot both take the last one. The slot stays + # :reserved until the instrumentation confirms it, so that a failing + # call can tell an in-flight reservation from a series the client holds. + def reserve(label, limit, name) + @mutex.synchronize do + next false if @series.key?(label) + + if @series.size >= limit + raise LabelSetLimitError, "#{name} reached max_series_per_metric (#{limit})" + end + + @series[label] = :reserved + next true + end + end + + # Marks a label set as established, once the client actually holds it. + # The limit is not checked on purpose: the series exists on the client + # side already, so it has to be accounted for even when a concurrent + # failure gave the reservation back in the meantime. + def confirm(label) + @mutex.synchronize { @series[label] = :confirmed } + end + + # Gives a reserved slot back when the instrumentation failed, so that a + # record which never reached the client does not consume the limit. One + # which a concurrent call confirmed in the meantime is kept: the client + # holds that series. + def release(label) + @mutex.synchronize { @series.delete(label) if @series[label] == :reserved } + end + end def self.parse_labels_elements(conf) labels = conf.elements.select { |e| e.name == 'labels' } @@ -120,7 +199,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' @@ -131,13 +210,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 @@ -166,6 +245,49 @@ 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) + @dropped_label_sets_counter = nil + end + + def metric_options + { + max_series_per_metric: @max_series_per_metric, + } + end + + # Registered on the first occurrence only, so that a plugin which never + # drops anything does not expose a counter which stays 0 forever. Its + # labels come from the configuration, so they cannot blow up on their own. + def limit_counter(name, docstring, labels) + @registry.counter(name, docstring: docstring, labels: labels) + rescue ::Prometheus::Client::Registry::AlreadyRegisteredError + # another plugin instance shares the registry and registered it first + Fluent::Plugin::Prometheus::Metric.get(@registry, name, :counter, docstring) + end + + def dropped_label_sets_counter + @dropped_label_sets_counter ||= + limit_counter(DROPPED_LABEL_SETS_METRIC_NAME, DROPPED_LABEL_SETS_METRIC_DESC, [:name]) + end + + def warn_label_set_limit(metric) + # the drop is always counted, while the log below is throttled + dropped_label_sets_counter.increment(labels: { name: metric.name.to_s }) + + warn_throttled(@label_set_limit_log_throttle, metric.name, + "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 + + # The counter above is never throttled, only the log which comes with it: + # one line per record would flood the Fluentd log, and the count is in + # Prometheus already. + def warn_throttled(throttle, key, message, **details) + emit, suppressed = throttle.check(key) + return unless emit + + details = details.merge(suppressed_log_count: suppressed) if suppressed > 0 + log.warn(message, details) end def instrument_single(tag, time, record, metrics) @@ -181,6 +303,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) @@ -202,6 +327,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) @@ -215,8 +343,9 @@ class Metric attr_reader :name attr_reader :key attr_reader :desc + 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" @@ -231,6 +360,10 @@ def initialize(element, registry, labels) @base_labels = Fluent::Plugin::Prometheus.parse_labels_elements(element) @base_labels = labels.merge(@base_labels) + # overrides the limit given by the plugin + @max_series_per_metric = metric_limit(element, 'max_series_per_metric', + opts.fetch(:max_series_per_metric, DEFAULT_MAX_SERIES_PER_METRIC)) + if @initialized @base_initlabels = Fluent::Plugin::Prometheus.parse_initlabels_elements(element, @base_labels) end @@ -255,12 +388,30 @@ def labels(record, expander) if v.is_a?(String) label[k] = expander.expand(v) else - label[k] = v.call(record) + label[k] = normalize_label_value(v.call(record)) end end label end + # Instruments a record through the given block and keeps its label set + # as a series once the client holds it. The slot is taken before + # instrumenting and given back on failure, so that a record which never + # reached the client does not exhaust max_series_per_metric. + def with_label_set(record, expander) + label = labels(record, expander) + reserved = reserve_series!(label) + begin + result = yield label + rescue + # a call which joined another's reservation has nothing to release + release_series(label) if reserved + raise + end + confirm_series(label) + result + end + def self.get(registry, name, type, docstring) metric = registry.get(name) @@ -274,10 +425,69 @@ 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 + + # Called by a subclass, once it has its client metric. + def bind_series_set(client_metric) + @series_set = SeriesSet.of(client_metric) + + if @initialized + # the client is given them at startup, so they take their slots now + @base_initlabels.each do |initlabels| + confirm_series(normalize_label_set(initlabels)) + end + end + end + + # The SeriesSet keys a label set by its values, so the same value has to + # look the same whether a RecordAccessor or produced it. + def normalize_label_value(value) + value.is_a?(String) ? value : value.to_s + end + + def normalize_label_set(label) + label.each_with_object({}) do |(k, v), normalized| + normalized[k] = normalize_label_value(v) + end + end + + # Returns true when this call took the slot, which tells #with_label_set + # whether it has something to give back on failure. + def reserve_series!(label) + # nothing is counted with the limit off, otherwise the set would grow + # with every label set and leak what the limit is there to prevent + return false if @max_series_per_metric <= 0 + + @series_set.reserve(label, @max_series_per_metric, @name) + end + + def confirm_series(label) + return if @max_series_per_metric <= 0 + + @series_set.confirm(label) + end + + def release_series(label) + return if @max_series_per_metric <= 0 + + @series_set.release(label) + 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" @@ -288,6 +498,7 @@ def initialize(element, registry, labels) rescue ::Prometheus::Client::Registry::AlreadyRegisteredError @gauge = Fluent::Plugin::Prometheus::Metric.get(registry, element['name'].to_sym, :gauge, element['desc']) end + bind_series_set(@gauge) if @initialized Fluent::Plugin::Prometheus::Metric.init_label_set(@gauge, @base_initlabels, @base_labels) @@ -301,19 +512,22 @@ 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) rescue ::Prometheus::Client::Registry::AlreadyRegisteredError @counter = Fluent::Plugin::Prometheus::Metric.get(registry, element['name'].to_sym, :counter, element['desc']) end + bind_series_set(@counter) if @initialized Fluent::Plugin::Prometheus::Metric.init_label_set(@counter, @base_initlabels, @base_labels) @@ -333,12 +547,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" @@ -349,6 +565,7 @@ def initialize(element, registry, labels) rescue ::Prometheus::Client::Registry::AlreadyRegisteredError @summary = Fluent::Plugin::Prometheus::Metric.get(registry, element['name'].to_sym, :summary, element['desc']) end + bind_series_set(@summary) if @initialized Fluent::Plugin::Prometheus::Metric.init_label_set(@summary, @base_initlabels, @base_labels) @@ -362,13 +579,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" @@ -386,6 +605,7 @@ def initialize(element, registry, labels) rescue ::Prometheus::Client::Registry::AlreadyRegisteredError @histogram = Fluent::Plugin::Prometheus::Metric.get(registry, element['name'].to_sym, :histogram, element['desc']) end + bind_series_set(@histogram) if @initialized Fluent::Plugin::Prometheus::Metric.init_label_set(@histogram, @base_initlabels, @base_labels) @@ -399,7 +619,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/lib/fluent/plugin/prometheus/log_throttle.rb b/lib/fluent/plugin/prometheus/log_throttle.rb index 0df9956..198d113 100644 --- a/lib/fluent/plugin/prometheus/log_throttle.rb +++ b/lib/fluent/plugin/prometheus/log_throttle.rb @@ -4,9 +4,10 @@ module Fluent module Plugin module Prometheus # Suppresses the repeated log for the same key within the interval. - # in_prometheus uses it, with an instance of its own. The key decides - # what is throttled, an error scope for now. The fingerprint tells the - # logs of a key apart: one which differs from the last is not suppressed. + # in_prometheus and filter/out_prometheus use it, each with its own + # instance. The key decides what is throttled: an error scope or a + # metric. When a fingerprint is given, a log whose fingerprint differs + # from the last one is not suppressed. class LogThrottle Entry = Struct.new(:time, :fingerprint, :suppressed) @@ -20,8 +21,9 @@ def initialize(interval) # Returns [emit, suppressed_count]. emit is true for the first log of a # key, for a new fingerprint, and after the interval has passed. # suppressed_count is how many logs were suppressed since the last one - # was emitted. - def check(key, fingerprint) + # was emitted. Without a fingerprint, a key is throttled by the + # interval alone. + def check(key, fingerprint = nil) return [true, 0] if @interval <= 0 @mutex.synchronize do diff --git a/spec/fluent/plugin/filter_prometheus_spec.rb b/spec/fluent/plugin/filter_prometheus_spec.rb index c22c884..82e8ba9 100644 --- a/spec/fluent/plugin/filter_prometheus_spec.rb +++ b/spec/fluent/plugin/filter_prometheus_spec.rb @@ -114,4 +114,76 @@ ) end end + + describe 'limiting label expansion' do + it_behaves_like 'limits label expansion' + end + + # the throttling itself is covered by the LogThrottle spec; what is left here + # is the warning warn_label_set_limit builds out of it, and the interval it + # takes from the configuration + describe 'drop 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) } + let(:text) { 'dropped a label set' } + + before do + allow(Fluent::Clock).to receive(:now) { clock[:now] } + end + + def logs_about(text) + driver.logs.select { |log| log.include?(text) } + end + + def drop + driver.instance.send(:warn_label_set_limit, metric) + end + + it 'warns only once within ignore_error_log_interval' do + 5.times { drop } + expect(logs_about(text).size).to eq(1) + end + + it 'reports how many warnings were suppressed in the meantime' do + 3.times { drop } + clock[:now] += driver.instance.ignore_error_log_interval + drop + logs = logs_about(text) + 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 + + # the only example which takes the interval from the configuration + context 'with ignore_error_log_interval 0' do + let(:config) { + BASE_CONFIG + %[ + ignore_error_log_interval 0 + + name throttled + type counter + desc Something foo. + key foo + + ] + } + + it 'warns about every drop' do + 3.times { drop } + expect(logs_about(text).size).to eq(3) + end + end + end end diff --git a/spec/fluent/plugin/out_prometheus_spec.rb b/spec/fluent/plugin/out_prometheus_spec.rb index d59890f..339b738 100644 --- a/spec/fluent/plugin/out_prometheus_spec.rb +++ b/spec/fluent/plugin/out_prometheus_spec.rb @@ -20,6 +20,10 @@ it_behaves_like 'initalized metrics' end + describe 'limiting label expansion' do + it_behaves_like 'limits label expansion' + end + # filter_prometheus routes such a record to @ERROR already. The output has to # do the same, instead of failing on the router itself. describe 'a record which cannot be instrumented' do diff --git a/spec/fluent/plugin/prometheus/log_throttle_spec.rb b/spec/fluent/plugin/prometheus/log_throttle_spec.rb index 99057b4..5203b43 100644 --- a/spec/fluent/plugin/prometheus/log_throttle_spec.rb +++ b/spec/fluent/plugin/prometheus/log_throttle_spec.rb @@ -58,6 +58,25 @@ expect(throttle.check(:bar, fingerprint).first).to be true end + # filter/out_prometheus throttles on the metric alone, since every drop of + # a metric reads the same + context 'without a fingerprint' do + it 'throttles on the key alone' do + expect(throttle.check(:foo).first).to be true + expect(throttle.check(:foo).first).to be false + expect(throttle.check(:bar).first).to be true + end + + it 'reports how many occurrences were suppressed in the meantime' do + throttle.check(:foo) + 2.times { throttle.check(:foo) } + clock[:now] += interval + emit, suppressed = throttle.check(:foo) + expect(emit).to be true + expect(suppressed).to eq(2) + end + end + 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 diff --git a/spec/fluent/plugin/prometheus/series_limit_spec.rb b/spec/fluent/plugin/prometheus/series_limit_spec.rb new file mode 100644 index 0000000..a135c7c --- /dev/null +++ b/spec/fluent/plugin/prometheus/series_limit_spec.rb @@ -0,0 +1,289 @@ +require 'spec_helper' + +# The limit is exercised through the plugins as well, by the 'limits label +# expansion' shared examples. These examples stay at the Metric level, where a +# slot can be observed while an instrumentation is still running. +describe Fluent::Plugin::Prometheus::Metric do + let(:registry) { ::Prometheus::Client::Registry.new } + let(:max_series_per_metric) { 1 } + let(:element) do + Fluent::Config::Element.new( + 'metric', '', + { + 'name' => 'limited', + 'type' => 'counter', + 'desc' => 'Something foo.', + 'key' => 'foo', + 'max_series_per_metric' => max_series_per_metric.to_s, + }, + [Fluent::Config::Element.new('labels', '', {'path' => '$.path'}, [])] + ) + end + # the label is a RecordAccessor, so no placeholder is expanded here + let(:expander) { double('expander') } + let(:metric) { Fluent::Plugin::Prometheus::Counter.new(element, registry, {}, {}) } + # the client metric is registered by the Metric, so it has to be built before + # the registry is asked for it + let(:client_counter) do + metric + registry.get(:limited) + end + + def instrument(path, value = 1) + metric.instrument({'foo' => value, 'path' => path}, expander) + end + + describe 'max_series_per_metric' do + it 'refuses a new label set once the limit is reached' do + instrument('/a') + + expect { instrument('/b') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError) + expect(client_counter.values.keys).to eq([{path: '/a'}]) + end + + it 'gives the slot back when the instrumentation failed' do + # a non numeric value makes Counter#increment raise, after the label set + # has been reserved + expect { instrument('/a', 'not a number') }.to raise_error(ArgumentError) + + expect { instrument('/b') }.not_to raise_error + expect(client_counter.values.keys).to eq([{path: '/b'}]) + end + + it 'takes the slot before instrumenting, so that concurrent calls cannot both pass' do + # the slot has to be taken under the same lock as the check: taking it + # after the client call would let both label sets through and expand the + # metric beyond max_series_per_metric + instrumenting = Queue.new + resume = Queue.new + allow(client_counter).to receive(:increment).and_wrap_original do |original, *args, **kwargs| + instrumenting << true + resume.pop + original.call(*args, **kwargs) + end + + first = Thread.new { instrument('/a') } + instrumenting.pop # '/a' is inside the client call and holds the only slot + + expect { instrument('/b') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError) + + resume << true + first.join + + expect(client_counter.values.keys).to eq([{path: '/a'}]) + end + + # Two records which expand to the very same label set may be instrumented + # at the same time: only one of them reserves the slot, the other one joins + # that reservation. Giving the slot back on failure must then not drop a + # label set the client already holds, otherwise a new one would take its + # place and the metric would grow past max_series_per_metric. + context 'when concurrent instrumentations share a label set' do + # stalls the very first client call, so that a second instrumentation can + # be run while the first one is still in flight + def stall_first_instrumentation(entered, resume) + stalled = false + allow(client_counter).to receive(:increment).and_wrap_original do |original, *args, **kwargs| + unless stalled + stalled = true + entered << true + resume.pop + end + original.call(*args, **kwargs) + end + end + + it 'keeps the slot when the call which reserved it fails after another one succeeded' do + entered = Queue.new + resume = Queue.new + stall_first_instrumentation(entered, resume) + + failing = Thread.new do + expect { instrument('/a', 'not a number') }.to raise_error(ArgumentError) + end + entered.pop # {path: '/a'} is reserved by the record which is about to fail + + # joins that reservation and does give the label set to the client + instrument('/a') + + resume << true + failing.join + + # the client holds {path: '/a'}, so its slot must stay taken + expect { instrument('/b') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError) + expect(client_counter.values.keys).to eq([{path: '/a'}]) + end + + it 'keeps the slot when a call which joined a reservation fails' do + entered = Queue.new + resume = Queue.new + stall_first_instrumentation(entered, resume) + + pending_call = Thread.new { instrument('/a') } + entered.pop # {path: '/a'} is reserved and being instrumented + + # joins that reservation and fails, without owning the slot + expect { instrument('/a', 'not a number') }.to raise_error(ArgumentError) + + resume << true + pending_call.join + + expect { instrument('/b') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError) + expect(client_counter.values.keys).to eq([{path: '/a'}]) + end + + it 'takes the slot back when a joined call succeeds after the reservation was released' do + failing_entered = Queue.new + failing_resume = Queue.new + succeeding_entered = Queue.new + succeeding_resume = Queue.new + allow(client_counter).to receive(:increment).and_wrap_original do |original, *args, **kwargs| + case kwargs[:by] + when 'not a number' + failing_entered << true + failing_resume.pop + when 2 + succeeding_entered << true + succeeding_resume.pop + end + original.call(*args, **kwargs) + end + + failing = Thread.new do + expect { instrument('/a', 'not a number') }.to raise_error(ArgumentError) + end + failing_entered.pop # {path: '/a'} is reserved + + succeeding = Thread.new { instrument('/a', 2) } + succeeding_entered.pop # joined the reservation, the client has nothing yet + + failing_resume << true + failing.join # the reservation is given back here + + succeeding_resume << true + succeeding.join # from now on the client holds {path: '/a'} + + expect { instrument('/b') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError) + expect(client_counter.values.keys).to eq([{path: '/a'}]) + end + end + end + + describe 'a metric name shared by two sections' do + # both sections instrument the same client metric, so counting per section + # would let it hold max_series_per_metric label sets twice over + let(:max_series_per_metric) { 2 } + let(:other_metric) { Fluent::Plugin::Prometheus::Counter.new(element, registry, {}, {}) } + + def instrument_other(path, value = 1) + other_metric.instrument({'foo' => value, 'path' => path}, expander) + end + + it 'wraps one and the same client metric' do + expect(other_metric.instance_variable_get(:@counter)) + .to equal(metric.instance_variable_get(:@counter)) + end + + it 'counts the label sets of both sections against one limit' do + instrument('/a') + instrument_other('/b') + + # the metric is full, whichever section the next record goes through + expect { instrument_other('/c') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError) + expect { instrument('/d') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError) + expect(client_counter.values.keys).to contain_exactly({path: '/a'}, {path: '/b'}) + end + + it 'lets both sections instrument a label set the metric already holds' do + instrument('/a') + + expect { instrument_other('/a', 2) }.not_to raise_error + expect(client_counter.values[{path: '/a'}]).to eq(3) + end + end + + describe 'initialized true with ' do + let(:initlabels) { ['/a', '/b', '/c'] } + let(:element) do + Fluent::Config::Element.new( + 'metric', '', + { + 'name' => 'limited', + 'type' => 'counter', + 'desc' => 'Something foo.', + 'key' => 'foo', + 'initialized' => 'true', + 'max_series_per_metric' => max_series_per_metric.to_s, + }, + [Fluent::Config::Element.new('labels', '', {'path' => '$.path'}, [])] + + initlabels.map { |path| Fluent::Config::Element.new('initlabels', '', {'path' => path}, []) } + ) + end + + context 'with a limit equal to the number of label sets' do + # every label set is known in advance, so the limit is reached but no + # record is dropped + let(:max_series_per_metric) { 3 } + + it 'accepts the config' do + expect { metric }.not_to raise_error + end + + it 'still counts a record on an label set' do + instrument('/a') + + expect(client_counter.values[{path: '/a'}]).to eq(1) + end + + it 'refuses a label set which is not in ' do + expect { instrument('/d') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError) + end + end + end + + describe ' overriding the plugin limit' do + # the plugin is configured with 100, which has to win over + let(:metric) do + Fluent::Plugin::Prometheus::Counter.new(element, registry, {}, {max_series_per_metric: 100}) + end + + it 'narrows down the limit given to the plugin' do + expect(metric.max_series_per_metric).to eq(1) + end + + context 'with a limit above the one given to the plugin' do + let(:max_series_per_metric) { 1000 } + + it 'widens the limit given to the plugin' do + expect(metric.max_series_per_metric).to eq(1000) + end + end + + context 'with 0' do + let(:max_series_per_metric) { 0 } + + it 'lifts the limit given to the plugin' do + expect(metric.max_series_per_metric).to eq(0) + end + end + + context 'without a limit in ' do + let(:element) do + Fluent::Config::Element.new( + 'metric', '', + { + 'name' => 'limited', + 'type' => 'counter', + 'desc' => 'Something foo.', + 'key' => 'foo', + }, + [Fluent::Config::Element.new('labels', '', {'path' => '$.path'}, [])] + ) + end + + it 'falls back to the limit given to the plugin' do + expect(metric.max_series_per_metric).to eq(100) + end + end + end +end diff --git a/spec/fluent/plugin/shared.rb b/spec/fluent/plugin/shared.rb index 7d48e21..942be79 100644 --- a/spec/fluent/plugin/shared.rb +++ b/spec/fluent/plugin/shared.rb @@ -390,6 +390,114 @@ end end +shared_examples_for 'limits label expansion' do + # the limit is enforced by the shared Metric class, but each plugin reaches + # it through its own path (instrument_single vs instrument), so both are run + # against these examples + def limited_config(options) + BASE_CONFIG + options + %[ + + name limited + type counter + desc Something foo. + key foo + + path $.path + + + ] + end + + def drop_logs + driver.logs.select { |log| log.include?('dropped a label set') } + end + + def dropped_label_sets + registry.metrics.find { |metric| metric.name == :fluentd_prometheus_dropped_label_sets_total } + end + + let(:counter) { registry.get(:limited) } + + context 'without any limit configured' do + let(:config) { limited_config('') } + + it 'is unlimited by default' do + expect(driver.instance.max_series_per_metric).to eq(0) + end + + it 'keeps every label set' 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'}, {path: '/b'}]) + expect(drop_logs).to be_empty + # nothing was dropped, so the counter is not even registered + expect(dropped_label_sets).to be_nil + end + end + + context 'with max_series_per_metric' do + let(:config) { limited_config(%[max_series_per_metric 1\n]) } + + 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'}]) + 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 reserved + 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 + + it 'counts every dropped record, while the log is throttled' 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' => '/c'}) + end + + expect(dropped_label_sets.values).to eq({{name: 'limited'} => 2.0}) + expect(drop_logs.size).to eq(1) + end + + # the same label set is refused again and again, and each record which + # brought it is lost + it 'counts a label set which is dropped more than once every time' 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' => '/b'}) + end + + expect(dropped_label_sets.values).to eq({{name: 'limited'} => 2.0}) + end + end +end + shared_examples_for 'initalized metrics' do before do driver.run(default_tag: tag) From b701d52b729b8fbe8c3d15ab4e919d870aae9722 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Fri, 28 Aug 2026 06:24:46 +0000 Subject: [PATCH 2/3] prometheus: refuse a non numeric value before it takes a slot Before: the client raised on such a value once with_label_set had reserved the slot. Summary raised only after Summary#observe had incremented its count, so releasing the slot left a half instrumented series which the metric could not take back. After: the value is validated before with_label_set, so a record which cannot be instrumented never takes a slot and never reaches the client. It is still routed to @ERROR as before. Co-Authored-By: Claude Signed-off-by: Kentaro Hayashi --- lib/fluent/plugin/prometheus.rb | 22 ++++++- spec/fluent/plugin/out_prometheus_spec.rb | 2 +- .../plugin/prometheus/series_limit_spec.rb | 64 +++++++++++++++++-- spec/fluent/plugin/shared.rb | 3 +- 4 files changed, 78 insertions(+), 13 deletions(-) diff --git a/lib/fluent/plugin/prometheus.rb b/lib/fluent/plugin/prometheus.rb index 2df90ed..87d1e5c 100644 --- a/lib/fluent/plugin/prometheus.rb +++ b/lib/fluent/plugin/prometheus.rb @@ -104,7 +104,7 @@ def confirm(label) end # Gives a reserved slot back when the instrumentation failed, so that a - # record which never reached the client does not consume the limit. One + # label set the client does not hold does not consume the limit. One # which a concurrent call confirmed in the meantime is kept: the client # holds that series. def release(label) @@ -396,8 +396,9 @@ def labels(record, expander) # Instruments a record through the given block and keeps its label set # as a series once the client holds it. The slot is taken before - # instrumenting and given back on failure, so that a record which never - # reached the client does not exhaust max_series_per_metric. + # instrumenting and given back when the client refused the record, so + # that a label set the client does not hold does not exhaust + # max_series_per_metric. def with_label_set(record, expander) label = labels(record, expander) reserved = reserve_series!(label) @@ -428,6 +429,17 @@ def self.get(registry, name, type, docstring) private + # The client refuses a value which is not a number, but it is only + # called once the label set has been reserved, and Summary refuses it + # only once it has already incremented its count: releasing the slot + # does not take that half instrumented series back from the client. + # Refuse the value before it reaches either. + def validate_value!(value) + return if value.is_a?(Numeric) + + raise ArgumentError, 'value must be a number' + end + def metric_limit(element, name, default) return default unless element.has_key?(name) @@ -512,6 +524,7 @@ def instrument(record, expander) value = @key.call(record) end if value + validate_value!(value) with_label_set(record, expander) do |label| @gauge.set(value, labels: label) end @@ -547,6 +560,7 @@ def instrument(record, expander) # ignore if record value is nil return if value.nil? + validate_value!(value) with_label_set(record, expander) do |label| @counter.increment(by: value, labels: label) end @@ -579,6 +593,7 @@ def instrument(record, expander) value = @key.call(record) end if value + validate_value!(value) with_label_set(record, expander) do |label| @summary.observe(value, labels: label) end @@ -619,6 +634,7 @@ def instrument(record, expander) value = @key.call(record) end if value + validate_value!(value) with_label_set(record, expander) do |label| @histogram.observe(value, labels: label) end diff --git a/spec/fluent/plugin/out_prometheus_spec.rb b/spec/fluent/plugin/out_prometheus_spec.rb index 339b738..907e786 100644 --- a/spec/fluent/plugin/out_prometheus_spec.rb +++ b/spec/fluent/plugin/out_prometheus_spec.rb @@ -40,7 +40,7 @@ it 'emits an error event' do driver.run(default_tag: tag) do - # a non numeric value makes Counter#increment raise + # a non numeric value is refused when the metric is instrumented driver.feed(event_time, {'foo' => 'not a number'}) end diff --git a/spec/fluent/plugin/prometheus/series_limit_spec.rb b/spec/fluent/plugin/prometheus/series_limit_spec.rb index a135c7c..7cd9d63 100644 --- a/spec/fluent/plugin/prometheus/series_limit_spec.rb +++ b/spec/fluent/plugin/prometheus/series_limit_spec.rb @@ -42,8 +42,17 @@ def instrument(path, value = 1) end it 'gives the slot back when the instrumentation failed' do - # a non numeric value makes Counter#increment raise, after the label set - # has been reserved + # a negative value makes Counter#increment raise, after the label set has + # been reserved + expect { instrument('/a', -1) }.to raise_error(ArgumentError) + + expect { instrument('/b') }.not_to raise_error + expect(client_counter.values.keys).to eq([{path: '/b'}]) + end + + it 'does not take a slot for a value which is not a number' do + # such a value is refused before the label set is reserved, so the metric + # is left as if the record had never arrived expect { instrument('/a', 'not a number') }.to raise_error(ArgumentError) expect { instrument('/b') }.not_to raise_error @@ -98,8 +107,11 @@ def stall_first_instrumentation(entered, resume) resume = Queue.new stall_first_instrumentation(entered, resume) + # a negative value is the one which fails inside the client call: a + # value which is not a number never gets there, so it could not be + # stalled failing = Thread.new do - expect { instrument('/a', 'not a number') }.to raise_error(ArgumentError) + expect { instrument('/a', -1) }.to raise_error(ArgumentError) end entered.pop # {path: '/a'} is reserved by the record which is about to fail @@ -122,8 +134,8 @@ def stall_first_instrumentation(entered, resume) pending_call = Thread.new { instrument('/a') } entered.pop # {path: '/a'} is reserved and being instrumented - # joins that reservation and fails, without owning the slot - expect { instrument('/a', 'not a number') }.to raise_error(ArgumentError) + # joins that reservation and fails in the client, without owning the slot + expect { instrument('/a', -1) }.to raise_error(ArgumentError) resume << true pending_call.join @@ -139,7 +151,7 @@ def stall_first_instrumentation(entered, resume) succeeding_resume = Queue.new allow(client_counter).to receive(:increment).and_wrap_original do |original, *args, **kwargs| case kwargs[:by] - when 'not a number' + when -1 failing_entered << true failing_resume.pop when 2 @@ -150,7 +162,7 @@ def stall_first_instrumentation(entered, resume) end failing = Thread.new do - expect { instrument('/a', 'not a number') }.to raise_error(ArgumentError) + expect { instrument('/a', -1) }.to raise_error(ArgumentError) end failing_entered.pop # {path: '/a'} is reserved @@ -169,6 +181,44 @@ def stall_first_instrumentation(entered, resume) end end + # Summary#observe increments the count and the sum one after the other, so a + # value which is not a number leaves the count incremented and raises on the + # sum. Giving the slot back would not take that half instrumented label set + # away from the client, so the value is refused before it reaches either. + describe 'a summary of a value which is not a number' do + let(:element) do + Fluent::Config::Element.new( + 'metric', '', + { + 'name' => 'limited', + 'type' => 'summary', + 'desc' => 'Something foo.', + 'key' => 'foo', + 'max_series_per_metric' => max_series_per_metric.to_s, + }, + [Fluent::Config::Element.new('labels', '', {'path' => '$.path'}, [])] + ) + end + let(:metric) { Fluent::Plugin::Prometheus::Summary.new(element, registry, {}, {}) } + let(:client_summary) do + metric + registry.get(:limited) + end + + it 'leaves the client holding nothing' do + expect { instrument('/a', 'not a number') }.to raise_error(ArgumentError) + + expect(client_summary.values).to be_empty + end + + it 'does not take a slot' do + expect { instrument('/a', 'not a number') }.to raise_error(ArgumentError) + + expect { instrument('/b', 2) }.not_to raise_error + expect(client_summary.values.keys).to eq([{path: '/b'}]) + end + end + describe 'a metric name shared by two sections' do # both sections instrument the same client metric, so counting per section # would let it hold max_series_per_metric label sets twice over diff --git a/spec/fluent/plugin/shared.rb b/spec/fluent/plugin/shared.rb index 942be79..43d209e 100644 --- a/spec/fluent/plugin/shared.rb +++ b/spec/fluent/plugin/shared.rb @@ -462,8 +462,7 @@ def dropped_label_sets 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 reserved + # a non numeric value is refused before the label set is reserved driver.feed(event_time, {'foo' => 'not a number', 'path' => '/a'}) driver.feed(event_time, {'foo' => 1, 'path' => '/b'}) end From f17543edc6b43d572d0bdc54f610db86917092a9 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Fri, 28 Aug 2026 14:48:56 +0900 Subject: [PATCH 3/3] Add note about max_series_per_metric with N workers Signed-off-by: Kentaro Hayashi --- README.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/README.md b/README.md index cd1f672..0d0b921 100644 --- a/README.md +++ b/README.md @@ -307,6 +307,11 @@ with the same `name`, in one plugin or in two, share one count. Each of them refuses a new label set once that shared count reaches its own limit, so a section which stays at `0` adds label sets without counting them. +The count is per worker process as well, since a worker has its own registry and +exposes the metrics it holds itself. With `workers N`, a metric can hold up to N +times `max_series_per_metric` label sets in total, so divide the number of label +sets the metric may reach by the number of workers. + A record which fails to be instrumented, for example when the value of `key` is not a number, does not consume the limit. A pre-initialized label set (`initialized` and ``) consumes it from the start, since the metric