From b75cfb1017bee135e7c4847349a8191e77bb9c3b Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Mon, 10 Aug 2026 10:32:19 +0000 Subject: [PATCH 1/7] 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 for filter_prometheus and out_prometheus: * 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. 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. Additionally the following counter is introduced too. * fluentd_prometheus_dropped_label_sets_total About internal design: Both limits are unlimited by default and have to be opted in. Enabling them changes what an existing metric exposes without telling anyone: a truncated label value merges label sets which were distinct so far and sums their values, and a dropped label set loses the record. Label values above a few hundred characters are not exotic, and a deployment which legitimately runs above 10000 label sets would start losing records right after an upgrade. That is a decision for the operator who knows the records, so each limit can be overridden per as well. A metric is instrumented through with_label_set, which takes the slot before instrumenting instead of counting the label set afterwards: two threads which build a new label set at the same time would otherwise both pass the check and expand the metric beyond max_series_per_metric. The slot is taken as :reserved and marked :confirmed once the client actually holds the label set, so that a record which fails to be instrumented, for example when the value of `key` is not a number, gives its own reservation back without dropping a series a concurrent record established in the meantime. Records which fail that way must not consume the limit, otherwise they could exhaust it and make the following valid label sets dropped. A drop is hard to notice, since it is not routed to @ERROR and the warning is throttled by ignore_error_log_interval, which both plugins now take as well. Count the drops in a new fluentd_prometheus_dropped_label_sets_total counter, labelled with the metric name and registered on the first drop, so that an operator can alert on a metric which is losing records. The throttling itself is extracted into LogThrottle, which in_prometheus shares. 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 Co-Authored-By: Claude --- README.md | 89 ++++++++ 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 | 275 +++++++++++++++++++++++-- 5 files changed, 354 insertions(+), 36 deletions(-) diff --git a/README.md b/README.md index f353f09..6bace7f 100644 --- a/README.md +++ b/README.md @@ -266,6 +266,95 @@ 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 can limit it: + +|parameter|description|default| +|---|---|---| +|max_label_value_length|The maximum length of a label value. A longer value is truncated. `0` means unlimited.|0| +|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.|0| +|ignore_error_log_interval|The interval in seconds to suppress the repeated warning about the dropped label sets. `0` logs every occurrence.|3600| + +**Both limits are disabled by default and must be enabled explicitly.** They +change what a metric exposes, so turning them on is a decision for the operator +who knows the records: + +* `max_label_value_length` truncates a label value, which merges label sets + that differ only after the limit into a single series. Their values are + summed from then on, and the series which existed before disappear. Values + above a few hundred characters are not exotic: URLs, Kubernetes annotations + and SQL statements routinely exceed them. +* `max_series_per_metric` drops a record whose label set is new once the limit + is reached. The record is lost and cannot be recovered. + +``` + + @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 + + + +``` + +Both limits can also be set in a `` section, which overrides the value +given to the plugin. A metric whose labels are known to be bounded can stay +unlimited with `0` while the plugin limits the others, and a metric which +expands faster than the others can be limited on its own: + +``` + + @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 + + + +``` + +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 consumes `max_series_per_metric` from the moment the metric is +about to be instrumented, so that two records which expand a metric at the same +time cannot both pass the limit. A record which fails to be instrumented, for +example when the value of `key` is not a number, gives its slot back, unless +another record gave the very same label set to the metric in the meantime. + +##### Observing the dropped label sets + +A dropped label set is not routed to `@ERROR`, because dropping it is what the +configuration asks for. It is reported in two ways instead: + +* a warning in the Fluentd log, suppressed for `ignore_error_log_interval` + seconds per metric and reporting how many warnings were suppressed in the + meantime. +* a counter named `fluentd_prometheus_dropped_label_sets_total`, labelled with + the metric `name`. It is registered on the first drop, so it does not show up + as long as nothing is dropped. Alert on it 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/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..b61e12b 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,72 @@ 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. Both limits are unlimited by default, because + # enabling them changes the existing metrics silently: truncating a label + # value merges label sets which were distinct so far, and 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_LABEL_VALUE_LENGTH = 0 + DEFAULT_MAX_SERIES_PER_METRIC = 0 + DEFAULT_IGNORE_ERROR_LOG_INTERVAL = 3600 + + # Counts the label sets dropped by max_series_per_metric, so that the + # drops are visible in Prometheus itself and 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 label sets dropped because the metric reached max_series_per_metric.' + + def self.included(klass) + klass.class_eval do + desc 'The maximum length of a label value. Longer values are truncated. 0 (default) 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 (default) 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 +186,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 +197,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 +232,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_label_value_length: @max_label_value_length, + max_series_per_metric: @max_series_per_metric, + } + end + + # Registered on the first drop only, so that a plugin which never drops a + # label set does not expose a metric which stays 0 forever. Its only label + # is the metric name, which comes from the configuration and not from a + # record, so this metric cannot blow up the cardinality by itself. + def dropped_label_sets_counter + @dropped_label_sets_counter ||= + begin + @registry.counter(DROPPED_LABEL_SETS_METRIC_NAME, + docstring: DROPPED_LABEL_SETS_METRIC_DESC, + labels: [:name]) + rescue ::Prometheus::Client::Registry::AlreadyRegisteredError + # another plugin instance shares the registry and registered it first + Fluent::Plugin::Prometheus::Metric.get(@registry, DROPPED_LABEL_SETS_METRIC_NAME, + :counter, DROPPED_LABEL_SETS_METRIC_DESC) + end + 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 }) + + 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 +290,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 +314,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 +330,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 +348,21 @@ 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, and the + # client already holds them, so they are established right away + @base_initlabels.each do |initlabels| + @series[normalize_label_set(initlabels)] = :confirmed + end end end @@ -252,14 +383,42 @@ 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 label end + # Instruments a record through the given block and keeps its label set + # as a series once the client holds it. A record which fails to be + # instrumented (e.g. its value is not a number) gives its slot back, + # since such records must not exhaust max_series_per_metric and make + # the following valid label sets dropped. A label set which a + # concurrent call gave to the client in the meantime keeps its slot. + def with_label_set(record, expander) + label = labels(record, expander) + # The slot is taken before instrumenting and given back on failure, + # instead of being taken afterwards: two threads which build a new + # label set at the same time would otherwise both pass the check and + # let the client create more series than max_series_per_metric. + reserved = reserve_series!(label) + begin + result = yield label + rescue + # only the call which took the slot may give it back: a concurrent + # call which joined an already reserved label set has nothing of + # its own to release + release_series(label) if reserved + raise + end + # the client holds the label set now, whether this call reserved it + # or joined a reservation made by a concurrent one + confirm_series(label) + result + end + def self.get(registry, name, type, docstring) metric = registry.get(name) @@ -273,10 +432,86 @@ 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. Checking the limit and taking the slot happen under the same + # lock, so that concurrent calls cannot both take the last one. + # A slot is taken as :reserved until the instrumentation confirms it, + # so that a failing call can tell an in-flight reservation from a + # series the client already holds. + # Returns true when this call took the slot, which is what tells + # #with_label_set whether it has something to give back on failure. + def reserve_series!(label) + return false if @max_series_per_metric <= 0 + + @series_mutex.synchronize do + next false 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] = :reserved + next true + end + end + + # Marks a label set as established, once the client actually holds it. + # The slot is (re)taken without checking the limit 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_series(label) + return if @max_series_per_metric <= 0 + + @series_mutex.synchronize do + @series[label] = :confirmed + end + end + + # Gives a reserved slot back when the instrumentation failed, so that a + # record which never reached the client does not consume the limit. A + # label set which a concurrent call confirmed in the meantime is kept: + # the client holds that series, and dropping it here would let the + # metric grow past max_series_per_metric. + def release_series(label) + @series_mutex.synchronize do + @series.delete(label) if @series[label] == :reserved + 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 +535,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 +569,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 +600,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 +639,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 From e68526877edd50c3166546b10246fff7202299d1 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Mon, 24 Aug 2026 10:17:45 +0900 Subject: [PATCH 2/7] spec: test the label expansion limits and the log throttling max_series_per_metric refuses a label set once a metric is full, max_label_value_length truncates a label value, and the warning about a dropped label set is throttled by LogThrottle, which in_prometheus shares. in_prometheus_spec used to cover the throttling logic itself, from before LogThrottle was extracted. Some test cases were removed from there. Keep there only what a unit spec cannot reach: which slot a failure is throttled on, how a fingerprint is built out of an exception, the suppressed_log_count the log carries, and the 500 the client keeps receiving meanwhile. Signed-off-by: Kentaro Hayashi Co-Authored-By: Claude --- lib/fluent/plugin/out_prometheus.rb | 5 + spec/fluent/plugin/filter_prometheus_spec.rb | 46 ++++ spec/fluent/plugin/in_prometheus_spec.rb | 65 +----- spec/fluent/plugin/out_prometheus_spec.rb | 4 + .../plugin/prometheus/log_throttle_spec.rb | 107 +++++++++ .../plugin/prometheus/series_limit_spec.rb | 217 ++++++++++++++++++ spec/fluent/plugin/shared.rb | 131 +++++++++++ 7 files changed, 516 insertions(+), 59 deletions(-) create mode 100644 spec/fluent/plugin/prometheus/log_throttle_spec.rb create mode 100644 spec/fluent/plugin/prometheus/series_limit_spec.rb diff --git a/lib/fluent/plugin/out_prometheus.rb b/lib/fluent/plugin/out_prometheus.rb index 9c611e3..979eb9b 100644 --- a/lib/fluent/plugin/out_prometheus.rb +++ b/lib/fluent/plugin/out_prometheus.rb @@ -7,6 +7,11 @@ class PrometheusOutput < Fluent::Plugin::Output include Fluent::Plugin::PrometheusLabelParser include Fluent::Plugin::Prometheus + # a record which cannot be instrumented is emitted as an error event, the + # same way as filter_prometheus does. Filter gets its router from the + # plugin base, while Output has to ask for the helper. + helpers :event_emitter + def initialize super @registry = ::Prometheus::Client.registry diff --git a/spec/fluent/plugin/filter_prometheus_spec.rb b/spec/fluent/plugin/filter_prometheus_spec.rb index c22c884..3fd1c95 100644 --- a/spec/fluent/plugin/filter_prometheus_spec.rb +++ b/spec/fluent/plugin/filter_prometheus_spec.rb @@ -114,4 +114,50 @@ ) 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 + 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/in_prometheus_spec.rb b/spec/fluent/plugin/in_prometheus_spec.rb index 9d05506..734694f 100644 --- a/spec/fluent/plugin/in_prometheus_spec.rb +++ b/spec/fluent/plugin/in_prometheus_spec.rb @@ -422,6 +422,11 @@ end end + # The throttling itself (the interval and the fingerprint comparison) is + # covered by the LogThrottle spec. What is left here is how in_prometheus + # wires it: which slot a failure is throttled on, which fingerprint tells two + # failures apart, the suppressed_log_count the log carries, and what the + # client sees meanwhile. describe 'error log throttling' do let(:config) { LOCAL_CONFIG } let(:secret_message) { 'dummy secret detail: password=deadbeef' } @@ -457,22 +462,6 @@ def error_logs(message) end end - it 'logs the repeated same failure again after ignore_error_log_interval has elapsed' do - 2.times do - driver.instance.send(:all_metrics) - clock[:now] += driver.instance.ignore_error_log_interval - end - expect(error_logs(log_message).size).to eq(2) - end - - it 'does not log the repeated same failure just before ignore_error_log_interval has elapsed' do - 2.times do - driver.instance.send(:all_metrics) - clock[:now] += driver.instance.ignore_error_log_interval - 0.1 - end - expect(error_logs(log_message).size).to eq(1) - end - it 'reports how many logs were suppressed in the meantime' do 3.times { driver.instance.send(:all_metrics) } clock[:now] += driver.instance.ignore_error_log_interval @@ -493,14 +482,7 @@ def log_error(scope, message, error) driver.instance.send(:log_error_throttled, scope, message, error: error) end - # the plugin raises a fresh exception object per failure, so the errors - # have to be compared by value, not by identity - it 'suppresses an equal error given as a different object' do - log_error(:metrics, log_message, RuntimeError.new(secret_message)) - log_error(:metrics, log_message, RuntimeError.new(secret_message)) - expect(error_logs(log_message).size).to eq(1) - end - + # the class and the message both take part in the fingerprint it 'logs immediately when the error class differs' do log_error(:metrics, log_message, RuntimeError.new(secret_message)) log_error(:metrics, log_message, ArgumentError.new(secret_message)) @@ -514,31 +496,12 @@ def log_error(scope, message, error) end # the scope, not the log message, picks the slot to throttle on - it 'keeps a separate state per scope' do - error = RuntimeError.new(secret_message) - log_error(:metrics, log_message, error) - log_error(:workers_metrics, workers_log_message, error) - expect(error_logs(log_message).size).to eq(1) - expect(error_logs(workers_log_message).size).to eq(1) - end - it 'suppresses an equal error within a scope even when the log message differs' do error = RuntimeError.new(secret_message) log_error(:metrics, log_message, error) log_error(:metrics, workers_log_message, error) expect(error_logs(workers_log_message)).to be_empty end - - context 'with ignore_error_log_interval 0' do - let(:config) { LOCAL_CONFIG + %[ - ignore_error_log_interval 0 -] } - - it 'logs every occurrence of the same error' do - 3.times { log_error(:metrics, log_message, RuntimeError.new(secret_message)) } - expect(error_logs(log_message).size).to eq(3) - end - end end # /metrics and /aggregated_metrics are usually scraped in turn, so both of @@ -559,21 +522,5 @@ def log_error(scope, message, error) expect(error_logs(workers_log_message).size).to eq(1) end end - - context 'when errors occur concurrently' do - # long enough to keep every call within the same interval - let(:config) { LOCAL_CONFIG + %[ - ignore_error_log_interval 3600 -] } - - it 'logs the error only once' do - instance = driver.instance - error = RuntimeError.new(secret_message) - 10.times.map { - Thread.new { instance.send(:log_error_throttled, :metrics, log_message, error: error) } - }.each(&:join) - expect(error_logs(log_message).size).to eq(1) - end - end end end diff --git a/spec/fluent/plugin/out_prometheus_spec.rb b/spec/fluent/plugin/out_prometheus_spec.rb index e58f2e2..94adc03 100644 --- a/spec/fluent/plugin/out_prometheus_spec.rb +++ b/spec/fluent/plugin/out_prometheus_spec.rb @@ -19,6 +19,10 @@ describe '#testinitlabels' do it_behaves_like 'initalized metrics' end + + describe 'limiting label expansion' do + it_behaves_like 'limits label expansion' + end describe '#run' do let(:message) { {"foo" => 100, "bar" => 100, "baz" => 100, "qux" => 10} } 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 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..71c64e4 --- /dev/null +++ b/spec/fluent/plugin/prometheus/series_limit_spec.rb @@ -0,0 +1,217 @@ +require 'spec_helper' + +# The limits are 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 ' 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..86fa1f4 100644 --- a/spec/fluent/plugin/shared.rb +++ b/spec/fluent/plugin/shared.rb @@ -390,6 +390,137 @@ end end +shared_examples_for 'limits label expansion' do + # the limits are 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_label_value_length).to eq(0) + expect(driver.instance.max_series_per_metric).to eq(0) + end + + it 'keeps every label set and the whole label value' do + long_value = 'a' * 300 + + 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' => long_value}) + end + + expect(counter.values.keys).to eq([{path: '/a'}, {path: '/b'}, {path: long_value}]) + 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_label_value_length' do + let(:config) { limited_config(%[max_label_value_length 4\n]) } + + it 'truncates a longer label value' do + driver.run(default_tag: tag) do + driver.feed(event_time, {'foo' => 1, 'path' => '/abcdefg'}) + end + + expect(counter.values.keys).to eq([{path: '/abc'}]) + end + + it 'merges the label sets which differ only after the limit' do + driver.run(default_tag: tag) do + driver.feed(event_time, {'foo' => 1, 'path' => '/abcdefg'}) + driver.feed(event_time, {'foo' => 1, 'path' => '/abcxyz'}) + end + + # this is why the limit is opt-in: the two series became one and their + # values are summed + expect(counter.values).to eq({{path: '/abc'} => 2.0}) + end + + it 'keeps a shorter label value as is' do + driver.run(default_tag: tag) do + driver.feed(event_time, {'foo' => 1, 'path' => '/ab'}) + end + + expect(counter.values.keys).to eq([{path: '/ab'}]) + 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 label set, 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 + end +end + shared_examples_for 'initalized metrics' do before do driver.run(default_tag: tag) From 4dadc11cba0663dd2f14028ffecad8bae2d7ca1f Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Mon, 24 Aug 2026 10:47:38 +0900 Subject: [PATCH 3/7] Implement fluentd_prometheus_truncated_label_values_total metric That metric is visible when label was truncated and merged into series. Signed-off-by: Kentaro Hayashi Co-Authored-By: Claude --- README.md | 39 +++++++--- lib/fluent/plugin/prometheus.rb | 128 +++++++++++++++++++++++--------- 2 files changed, 122 insertions(+), 45 deletions(-) diff --git a/README.md b/README.md index 6bace7f..201611a 100644 --- a/README.md +++ b/README.md @@ -275,7 +275,7 @@ is bound to a field with many distinct values. Both plugins can limit it: |---|---|---| |max_label_value_length|The maximum length of a label value. A longer value is truncated. `0` means unlimited.|0| |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.|0| -|ignore_error_log_interval|The interval in seconds to suppress the repeated warning about the dropped label sets. `0` logs every occurrence.|3600| +|ignore_error_log_interval|The interval in seconds to suppress the repeated warning about the drops and the truncations. `0` logs every occurrence.|3600| **Both limits are disabled by default and must be enabled explicitly.** They change what a metric exposes, so turning them on is a decision for the operator @@ -338,23 +338,42 @@ time cannot both pass the limit. A record which fails to be instrumented, for example when the value of `key` is not a number, gives its slot back, unless another record gave the very same label set to the metric in the meantime. -##### Observing the dropped label sets +A pre-initialized label set (`initialized` and ``) is truncated by +`max_label_value_length` as well, and consumes `max_series_per_metric` from the +start. A record which expands to it then lands on that very series, instead of +creating a second one under the truncated value. -A dropped label set is not routed to `@ERROR`, because dropping it is what the -configuration asks for. It is reported in two ways instead: +##### Observing what the limits leave out + +A dropped label set and a truncated label value are not routed to `@ERROR`, +because both are what the configuration asks for. They are reported in two ways +instead: * a warning in the Fluentd log, suppressed for `ignore_error_log_interval` - seconds per metric and reporting how many warnings were suppressed in the - meantime. -* a counter named `fluentd_prometheus_dropped_label_sets_total`, labelled with - the metric `name`. It is registered on the first drop, so it does not show up - as long as nothing is dropped. Alert on it to notice that a metric is losing - records: + seconds and reporting how many warnings were suppressed in the meantime. A + drop is throttled per metric, a truncation per label of a metric. +* a counter, which is what makes the loss visible in Prometheus itself. Both + are registered on their first occurrence, so they do not show up as long as + nothing is dropped or truncated: + +|metric|labels|meaning| +|---|---|---| +|fluentd_prometheus_dropped_label_sets_total|`name`|A record was not instrumented, because the metric `name` reached `max_series_per_metric`.| +|fluentd_prometheus_truncated_label_values_total|`name`, `label`|A record was instrumented under a shortened value of `label`, because it exceeded `max_label_value_length`. This is how many records went into a merged series.| + +Their labels come from the configuration and not from a record, so these +counters cannot expand on their own. Alert on them to notice that a metric is +losing records or merging series: ``` rate(fluentd_prometheus_dropped_label_sets_total[5m]) > 0 +rate(fluentd_prometheus_truncated_label_values_total[5m]) > 0 ``` +A truncation counter which keeps growing means that `max_label_value_length` is +shorter than what the records carry: the metric is still exported, but its +series no longer tell those records apart. + ## 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 b61e12b..6bc9d40 100644 --- a/lib/fluent/plugin/prometheus.rb +++ b/lib/fluent/plugin/prometheus.rb @@ -49,22 +49,29 @@ class LabelSetLimitError < StandardError; end DROPPED_LABEL_SETS_METRIC_NAME = :fluentd_prometheus_dropped_label_sets_total DROPPED_LABEL_SETS_METRIC_DESC = 'The total number of label sets dropped because the metric reached max_series_per_metric.' + # Counts the label values truncated by max_label_value_length, for the + # same reason: truncating merges label sets which were distinct, and the + # merged series looks like any other one. + TRUNCATED_LABEL_VALUES_METRIC_NAME = :fluentd_prometheus_truncated_label_values_total + TRUNCATED_LABEL_VALUES_METRIC_DESC = 'The total number of label values truncated because they exceeded max_label_value_length.' + def self.included(klass) klass.class_eval do desc 'The maximum length of a label value. Longer values are truncated. 0 (default) 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 (default) 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.' + desc 'The interval to suppress the repeated warning about the drops and the truncations.' 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 + # Shared by filter/out_prometheus and in_prometheus. Each plugin owns its + # own instance, since the lifetime differs; only the implementation is + # shared. The granularity is absorbed by the key, which a caller builds + # out of what it throttles on: a metric, a metric and one of its labels, + # or an error scope. 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) @@ -72,7 +79,7 @@ class LogThrottle def initialize(interval) @interval = interval @mutex = Mutex.new - # bounded by the number of keys (metrics / scopes), so it never grows + # bounded by the number of keys a caller can build, so it never grows # unexpectedly @entries = {} end @@ -233,48 +240,74 @@ def configure(conf) @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) + @label_value_truncated_log_throttle = Fluent::Plugin::Prometheus::LogThrottle.new(@ignore_error_log_interval) @dropped_label_sets_counter = nil + @truncated_label_values_counter = nil end def metric_options { max_label_value_length: @max_label_value_length, max_series_per_metric: @max_series_per_metric, + # a metric does not know its plugin, so it reports a truncation here + on_label_value_truncated: method(:warn_label_value_truncated), } end - # Registered on the first drop only, so that a plugin which never drops a - # label set does not expose a metric which stays 0 forever. Its only label - # is the metric name, which comes from the configuration and not from a - # record, so this metric cannot blow up the cardinality by itself. + # Registered on the first occurrence only, so that a plugin which never + # drops or truncates anything does not expose a counter which stays 0 + # forever. Their labels come from the configuration and not from a + # record, so they cannot blow up the cardinality themselves. + 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 ||= - begin - @registry.counter(DROPPED_LABEL_SETS_METRIC_NAME, - docstring: DROPPED_LABEL_SETS_METRIC_DESC, - labels: [:name]) - rescue ::Prometheus::Client::Registry::AlreadyRegisteredError - # another plugin instance shares the registry and registered it first - Fluent::Plugin::Prometheus::Metric.get(@registry, DROPPED_LABEL_SETS_METRIC_NAME, - :counter, DROPPED_LABEL_SETS_METRIC_DESC) - end + limit_counter(DROPPED_LABEL_SETS_METRIC_NAME, DROPPED_LABEL_SETS_METRIC_DESC, [:name]) + end + + def truncated_label_values_counter + @truncated_label_values_counter ||= + limit_counter(TRUNCATED_LABEL_VALUES_METRIC_NAME, TRUNCATED_LABEL_VALUES_METRIC_DESC, [:name, :label]) 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 }) - emit, suppressed = @label_set_limit_log_throttle.check(metric.name) + 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 + + # Called by a metric which truncated a label value. This is not an error + # either: the record is still instrumented. It is reported because the + # label sets which differ only after the limit become a single series, + # and nothing else shows that. + def warn_label_value_truncated(metric, label_key) + truncated_label_values_counter.increment(labels: { name: metric.name.to_s, label: label_key.to_s }) + + # the label is part of the key, so that a label truncated once in a + # while is not hidden by one truncated constantly + warn_throttled(@label_value_truncated_log_throttle, [metric.name, label_key], + "prometheus: truncated a label value because it exceeded max_label_value_length.", + name: metric.name, label: label_key, + max_label_value_length: metric.max_label_value_length) + end + + # The counters above are never throttled, only the log which comes with + # them: 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 - 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 + details = details.merge(suppressed_log_count: suppressed) if suppressed > 0 + log.warn(message, details) end def instrument_single(tag, time, record, metrics) @@ -348,18 +381,25 @@ def initialize(element, registry, labels, opts = {}) @base_labels = Fluent::Plugin::Prometheus.parse_labels_elements(element) @base_labels = labels.merge(@base_labels) - # can narrow down the limits given by the plugin + # overrides 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)) + @on_label_value_truncated = opts[:on_label_value_truncated] @series = {} @series_mutex = Mutex.new if @initialized + # A pre-initialized label set is given to the client as is, so it + # has to be truncated like the label sets built from records. + # Otherwise the client would hold the long value, a record + # expanding to the same label set would land on a second series, + # and the metric would grow past max_series_per_metric. @base_initlabels = Fluent::Plugin::Prometheus.parse_initlabels_elements(element, @base_labels) - # the pre-initialized label sets consume the limit as well, and the - # client already holds them, so they are established right away + .map { |initlabels| truncate_label_set(initlabels) } + # the client holds them from now on, so they consume the limit as + # well and are established right away @base_initlabels.each do |initlabels| @series[normalize_label_set(initlabels)] = :confirmed end @@ -383,9 +423,9 @@ def labels(record, expander) label = {} @base_labels.each do |k, v| if v.is_a?(String) - label[k] = truncate_label_value(expander.expand(v)) + label[k] = truncate_label_value(expander.expand(v), k) else - label[k] = truncate_label_value(v.call(record)) + label[k] = truncate_label_value(v.call(record), k) end end label @@ -446,14 +486,32 @@ def metric_limit(element, name, default) end end - def truncate_label_value(value) + # A truncation is reported with the label it happened on, so that an + # operator knows which label merges its label sets. A caller omits the + # key to truncate quietly: the pre-initialized label sets below come + # from the configuration and not from a record, so they merge nothing + # an operator can act on. + def truncate_label_value(value, key = nil) # 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 + return value if value.length <= @max_label_value_length + + @on_label_value_truncated.call(self, key) if key && @on_label_value_truncated + value[0, @max_label_value_length] + end - value.length > @max_label_value_length ? value[0, @max_label_value_length] : value + # Truncates a label set which is given to the client as is, keeping the + # type of its values: a pre-initialized label set may hold a non-String + # value (${worker_id}), which the client has always been given as such. + def truncate_label_set(label) + label.each_with_object({}) do |(k, v), truncated| + truncated[k] = v.is_a?(String) ? truncate_label_value(v) : v + end end + # Same, for a label set kept in @series, whose values are always + # Strings: the ones built from records go through to_s. def normalize_label_set(label) label.each_with_object({}) do |(k, v), normalized| normalized[k] = truncate_label_value(v) From 3dff32df18df52aeb5ce4d1942cb92dff33b0979 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Mon, 24 Aug 2026 10:49:03 +0900 Subject: [PATCH 4/7] spec: test the label truncation Signed-off-by: Kentaro Hayashi Co-Authored-By: Claude --- spec/fluent/plugin/filter_prometheus_spec.rb | 81 +++++++++--- .../prometheus/label_truncation_spec.rb | 121 ++++++++++++++++++ spec/fluent/plugin/shared.rb | 32 ++++- 3 files changed, 217 insertions(+), 17 deletions(-) create mode 100644 spec/fluent/plugin/prometheus/label_truncation_spec.rb diff --git a/spec/fluent/plugin/filter_prometheus_spec.rb b/spec/fluent/plugin/filter_prometheus_spec.rb index 3fd1c95..b23fee5 100644 --- a/spec/fluent/plugin/filter_prometheus_spec.rb +++ b/spec/fluent/plugin/filter_prometheus_spec.rb @@ -120,8 +120,9 @@ 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 - describe 'label set limit log throttling' do + # is the warnings warn_label_set_limit and warn_label_value_truncated build + # out of it + describe 'limit log throttling' do let(:config) { BASE_CONFIG + %[ ignore_error_log_interval 3600 @@ -135,29 +136,77 @@ } # 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(:metric) { double('metric', name: :throttled, max_series_per_metric: 5, max_label_value_length: 4) } before do allow(Fluent::Clock).to receive(:now) { clock[:now] } end - def drop_logs - driver.logs.select { |log| log.include?('dropped a label set') } + def logs_about(text) + driver.logs.select { |log| log.include?(text) } 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) + def drop + driver.instance.send(:warn_label_set_limit, metric) 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') + def truncate(label_key = :path) + driver.instance.send(:warn_label_value_truncated, metric, label_key) + end + + shared_examples_for 'a throttled warning' do + it 'warns only once within ignore_error_log_interval' do + 5.times { warn_once } + expect(logs_about(text).size).to eq(1) + end + + it 'reports how many warnings were suppressed in the meantime' do + 3.times { warn_once } + clock[:now] += driver.instance.ignore_error_log_interval + warn_once + 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 + end + + describe 'about a dropped label set' do + let(:text) { 'dropped a label set' } + def warn_once + drop + end + + it_behaves_like 'a throttled warning' + end + + describe 'about a truncated label value' do + let(:text) { 'truncated a label value' } + def warn_once + truncate + end + + it_behaves_like 'a throttled warning' + + it 'names the label it truncated' do + truncate + expect(logs_about(text).first).to include('label=:path') + end + + # a label truncated once in a while must not be hidden by one truncated + # constantly + it 'keeps a separate slot per label' do + truncate(:path) + truncate(:host) + expect(logs_about(text).size).to eq(2) + end + end + + it 'throttles the two warnings independently' do + drop + truncate + expect(logs_about('dropped a label set').size).to eq(1) + expect(logs_about('truncated a label value').size).to eq(1) end end end diff --git a/spec/fluent/plugin/prometheus/label_truncation_spec.rb b/spec/fluent/plugin/prometheus/label_truncation_spec.rb new file mode 100644 index 0000000..0f1fb4e --- /dev/null +++ b/spec/fluent/plugin/prometheus/label_truncation_spec.rb @@ -0,0 +1,121 @@ +require 'spec_helper' + +# The truncation is exercised through the plugins as well, by the 'limits label +# expansion' shared examples. These examples stay at the Metric level, where the +# callback given to a metric can be observed directly. +describe Fluent::Plugin::Prometheus::Metric do + let(:registry) { ::Prometheus::Client::Registry.new } + let(:max_label_value_length) { 4 } + let(:truncations) { [] } + let(:opts) do + { on_label_value_truncated: ->(metric, key) { truncations << [metric.name, key] } } + end + let(:element) do + Fluent::Config::Element.new( + 'metric', '', + { + 'name' => 'truncated', + 'type' => 'counter', + 'desc' => 'Something foo.', + 'key' => 'foo', + 'max_label_value_length' => max_label_value_length.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, {}, opts) } + # 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(:truncated) + end + + def instrument(path) + metric.instrument({'foo' => 1, 'path' => path}, expander) + end + + describe 'max_label_value_length' do + it 'reports which label it truncated' do + instrument('/abcdefg') + + expect(truncations).to eq([['truncated', :path]]) + end + + # two records, one series: this is what the report makes countable + it 'reports every truncation, not only the first one' do + instrument('/abcdefg') + instrument('/abcxyz') + + expect(truncations.size).to eq(2) + expect(client_counter.values.keys).to eq([{path: '/abc'}]) + end + + it 'reports nothing when the label value fits in the limit' do + instrument('/ab') + + expect(truncations).to be_empty + end + + it 'reports nothing when the label value is exactly as long as the limit' do + instrument('/abc') + + expect(truncations).to be_empty + expect(client_counter.values.keys).to eq([{path: '/abc'}]) + end + + context 'with 0' do + let(:max_label_value_length) { 0 } + + it 'reports nothing, since nothing is truncated' do + instrument('/abcdefg') + + expect(truncations).to be_empty + expect(client_counter.values.keys).to eq([{path: '/abcdefg'}]) + end + end + end + + # They are given to the client as is, so they have to be truncated like the + # label sets built from records. They come from the configuration though, so + # the truncation is not reported: nothing an operator can act on is merged. + describe 'the pre-initialized label sets' do + let(:element) do + Fluent::Config::Element.new( + 'metric', '', + { + 'name' => 'truncated', + 'type' => 'counter', + 'desc' => 'Something foo.', + 'key' => 'foo', + 'initialized' => 'true', + 'max_label_value_length' => max_label_value_length.to_s, + }, + [ + Fluent::Config::Element.new('labels', '', {'path' => '$.path'}, []), + Fluent::Config::Element.new('initlabels', '', {'path' => '/abcdefg'}, []), + ] + ) + end + + it 'are given to the client truncated' do + expect(client_counter.values.keys).to eq([{path: '/abc'}]) + end + + # otherwise the client would hold both '/abcdefg' and '/abc', and the + # metric would grow past max_series_per_metric + it 'take the same series as a record which expands to them' do + instrument('/abcdefg') + + expect(client_counter.values).to eq({{path: '/abc'} => 1.0}) + end + + it 'are not reported as a truncation' do + metric + + expect(truncations).to be_empty + end + end +end diff --git a/spec/fluent/plugin/shared.rb b/spec/fluent/plugin/shared.rb index 86fa1f4..17a837d 100644 --- a/spec/fluent/plugin/shared.rb +++ b/spec/fluent/plugin/shared.rb @@ -412,10 +412,18 @@ def drop_logs driver.logs.select { |log| log.include?('dropped a label set') } end + def truncation_logs + driver.logs.select { |log| log.include?('truncated a label value') } + end + def dropped_label_sets registry.metrics.find { |metric| metric.name == :fluentd_prometheus_dropped_label_sets_total } end + def truncated_label_values + registry.metrics.find { |metric| metric.name == :fluentd_prometheus_truncated_label_values_total } + end + let(:counter) { registry.get(:limited) } context 'without any limit configured' do @@ -437,8 +445,10 @@ def dropped_label_sets expect(counter.values.keys).to eq([{path: '/a'}, {path: '/b'}, {path: long_value}]) expect(drop_logs).to be_empty - # nothing was dropped, so the counter is not even registered + expect(truncation_logs).to be_empty + # nothing was dropped or truncated, so the counters are not even registered expect(dropped_label_sets).to be_nil + expect(truncated_label_values).to be_nil end end @@ -471,6 +481,26 @@ def dropped_label_sets expect(counter.values.keys).to eq([{path: '/ab'}]) end + + it 'counts every truncated label value, while the log is throttled' do + driver.run(default_tag: tag) do + driver.feed(event_time, {'foo' => 1, 'path' => '/abcdefg'}) + driver.feed(event_time, {'foo' => 1, 'path' => '/abcxyz'}) + end + + # two records, one series: this is what the counter makes visible + expect(truncated_label_values.values).to eq({{name: 'limited', label: 'path'} => 2.0}) + expect(truncation_logs.size).to eq(1) + end + + it 'reports nothing when the label value fits in the limit' do + driver.run(default_tag: tag) do + driver.feed(event_time, {'foo' => 1, 'path' => '/ab'}) + end + + expect(truncated_label_values).to be_nil + expect(truncation_logs).to be_empty + end end context 'with max_series_per_metric' do From 1e8ea50ef9dfac1d0df1db0bb23c3ea0b594d36e Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Tue, 25 Aug 2026 10:45:48 +0900 Subject: [PATCH 5/7] Guard mis-configuration with initlabels and limits It should block inconsistent setting in early stage. Signed-off-by: Kentaro Hayashi --- README.md | 7 ++ lib/fluent/plugin/prometheus.rb | 12 ++++ .../plugin/prometheus/series_limit_spec.rb | 68 +++++++++++++++++++ 3 files changed, 87 insertions(+) diff --git a/README.md b/README.md index 201611a..91b4749 100644 --- a/README.md +++ b/README.md @@ -332,6 +332,13 @@ 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 metric with `initialized true` creates its `` label sets at +startup, so they count towards `max_series_per_metric` before any record +arrives. When the limit is smaller than the number of `` label sets, +no record can ever be counted, so the plugin stops at startup with a +configuration error instead of dropping every record. A limit equal to that +number is fine: it means every label set of the metric is known in advance. + A label set consumes `max_series_per_metric` from the moment the metric is about to be instrumented, so that two records which expand a metric at the same time cannot both pass the limit. A record which fails to be instrumented, for diff --git a/lib/fluent/plugin/prometheus.rb b/lib/fluent/plugin/prometheus.rb index 6bc9d40..fe8bb10 100644 --- a/lib/fluent/plugin/prometheus.rb +++ b/lib/fluent/plugin/prometheus.rb @@ -403,6 +403,7 @@ def initialize(element, registry, labels, opts = {}) @base_initlabels.each do |initlabels| @series[normalize_label_set(initlabels)] = :confirmed end + check_initlabels_fit_series_limit! end end @@ -486,6 +487,17 @@ def metric_limit(element, name, default) end end + def check_initlabels_fit_series_limit! + return if @max_series_per_metric <= 0 + # two blocks with the same values make one label set, so + # count the label sets and not the blocks + return if @series.size <= @max_series_per_metric + + raise ConfigError, "metric #{@name} has #{@series.size} label sets, " \ + "but max_series_per_metric is #{@max_series_per_metric}: " \ + "no record could ever be counted" + end + # A truncation is reported with the label it happened on, so that an # operator knows which label merges its label sets. A caller omits the # key to truncate quietly: the pre-initialized label sets below come diff --git a/spec/fluent/plugin/prometheus/series_limit_spec.rb b/spec/fluent/plugin/prometheus/series_limit_spec.rb index 71c64e4..c722bae 100644 --- a/spec/fluent/plugin/prometheus/series_limit_spec.rb +++ b/spec/fluent/plugin/prometheus/series_limit_spec.rb @@ -169,6 +169,74 @@ def stall_first_instrumentation(entered, resume) 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 below the number of label sets' do + # 3 label sets exist at startup, so a limit of 1 would drop every record + let(:max_series_per_metric) { 1 } + + it 'stops at startup instead of dropping every record' do + expect { metric }.to raise_error(Fluent::ConfigError, + /has 3 label sets.*max_series_per_metric is 1/) + end + end + + context 'with a limit equal to the number of label sets' do + # every label set is known in advance, which is what is for: + # 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 + + context 'with two holding the same values' do + # both make the same label set, so they take one slot + let(:initlabels) { ['/a', '/a'] } + let(:max_series_per_metric) { 1 } + + it 'counts the label sets and not the blocks' do + expect { metric }.not_to raise_error + end + end + + context 'without a limit' do + let(:max_series_per_metric) { 0 } + + it 'accepts any number of label sets' do + expect { metric }.not_to raise_error + expect { instrument('/d') }.not_to raise_error + end + end + end + describe ' overriding the plugin limit' do # the plugin is configured with 100, which has to win over let(:metric) do From bc629f7e5d509c4ff3b9643f297626bd46e10279 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Tue, 25 Aug 2026 11:37:27 +0900 Subject: [PATCH 6/7] Share cardinality limit among same metrics Signed-off-by: Kentaro Hayashi --- README.md | 8 +- lib/fluent/plugin/prometheus.rb | 136 ++++++++++++------ .../plugin/prometheus/series_limit_spec.rb | 39 ++++- 3 files changed, 132 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index 91b4749..5c87823 100644 --- a/README.md +++ b/README.md @@ -328,9 +328,11 @@ expands faster than the others can be limited on its own: ``` -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. +The label sets are counted per metric name, not per `` section. Sections +with the same `name`, in one plugin or in two, instrument the same metric and +share one count. Each of them refuses a new label set once that shared count +reaches its own limit, so a section which leaves `max_series_per_metric` at `0` +adds label sets without counting them. A metric with `initialized true` creates its `` label sets at startup, so they count towards `max_series_per_metric` before any record diff --git a/lib/fluent/plugin/prometheus.rb b/lib/fluent/plugin/prometheus.rb index fe8bb10..601ba8e 100644 --- a/lib/fluent/plugin/prometheus.rb +++ b/lib/fluent/plugin/prometheus.rb @@ -106,6 +106,69 @@ def check(key, fingerprint = nil) 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. Counting per + # section would let the metric hold max_series_per_metric label sets per + # section instead of max_series_per_metric in total. + 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. + # A slot is taken as :reserved until the instrumentation confirms it, + # so that a failing call can tell an in-flight reservation from a + # series the client already holds. + # Returns true when this call took the slot. + 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 slot is (re)taken without checking the limit 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. A + # label set which a concurrent call confirmed in the meantime is kept: + # the client holds that series, and dropping it here would let the + # metric grow past max_series_per_metric. + 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' } if labels.size > 1 @@ -387,8 +450,6 @@ def initialize(element, registry, labels, opts = {}) @max_series_per_metric = metric_limit(element, 'max_series_per_metric', opts.fetch(:max_series_per_metric, DEFAULT_MAX_SERIES_PER_METRIC)) @on_label_value_truncated = opts[:on_label_value_truncated] - @series = {} - @series_mutex = Mutex.new if @initialized # A pre-initialized label set is given to the client as is, so it @@ -398,12 +459,6 @@ def initialize(element, registry, labels, opts = {}) # and the metric would grow past max_series_per_metric. @base_initlabels = Fluent::Plugin::Prometheus.parse_initlabels_elements(element, @base_labels) .map { |initlabels| truncate_label_set(initlabels) } - # the client holds them from now on, so they consume the limit as - # well and are established right away - @base_initlabels.each do |initlabels| - @series[normalize_label_set(initlabels)] = :confirmed - end - check_initlabels_fit_series_limit! end end @@ -487,13 +542,26 @@ def metric_limit(element, name, default) end end + # Ties this metric to the label sets of its client metric. A subclass + # calls it once it has that 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| + @series_set.confirm(normalize_label_set(initlabels)) + end + check_initlabels_fit_series_limit! + end + end + def check_initlabels_fit_series_limit! return if @max_series_per_metric <= 0 - # two blocks with the same values make one label set, so - # count the label sets and not the blocks - return if @series.size <= @max_series_per_metric + # two blocks with the same values make one label set + return if @series_set.size <= @max_series_per_metric - raise ConfigError, "metric #{@name} has #{@series.size} label sets, " \ + raise ConfigError, "metric #{@name} holds #{@series_set.size} label sets from , " \ "but max_series_per_metric is #{@max_series_per_metric}: " \ "no record could ever be counted" end @@ -522,7 +590,7 @@ def truncate_label_set(label) end end - # Same, for a label set kept in @series, whose values are always + # Same, for a label set kept in the SeriesSet, whose values are always # Strings: the ones built from records go through to_s. def normalize_label_set(label) label.each_with_object({}) do |(k, v), normalized| @@ -532,51 +600,25 @@ 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. Checking the limit and taking the slot happen under the same - # lock, so that concurrent calls cannot both take the last one. - # A slot is taken as :reserved until the instrumentation confirms it, - # so that a failing call can tell an in-flight reservation from a - # series the client already holds. + # refused. # Returns true when this call took the slot, which is what tells # #with_label_set whether it has something to give back on failure. def reserve_series!(label) + # with the limit off nothing is counted, 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_mutex.synchronize do - next false 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] = :reserved - next true - end + @series_set.reserve(label, @max_series_per_metric, @name) end - # Marks a label set as established, once the client actually holds it. - # The slot is (re)taken without checking the limit 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_series(label) return if @max_series_per_metric <= 0 - @series_mutex.synchronize do - @series[label] = :confirmed - end + @series_set.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. A - # label set which a concurrent call confirmed in the meantime is kept: - # the client holds that series, and dropping it here would let the - # metric grow past max_series_per_metric. def release_series(label) - @series_mutex.synchronize do - @series.delete(label) if @series[label] == :reserved - end + @series_set.release(label) end end @@ -592,6 +634,7 @@ def initialize(element, registry, labels, opts = {}) 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) @@ -620,6 +663,7 @@ def initialize(element, registry, labels, opts = {}) 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) @@ -657,6 +701,7 @@ def initialize(element, registry, labels, opts = {}) 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) @@ -696,6 +741,7 @@ def initialize(element, registry, labels, opts = {}) 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) diff --git a/spec/fluent/plugin/prometheus/series_limit_spec.rb b/spec/fluent/plugin/prometheus/series_limit_spec.rb index c722bae..f6a2ef6 100644 --- a/spec/fluent/plugin/prometheus/series_limit_spec.rb +++ b/spec/fluent/plugin/prometheus/series_limit_spec.rb @@ -169,6 +169,39 @@ def stall_first_instrumentation(entered, resume) 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 @@ -193,13 +226,13 @@ def stall_first_instrumentation(entered, resume) it 'stops at startup instead of dropping every record' do expect { metric }.to raise_error(Fluent::ConfigError, - /has 3 label sets.*max_series_per_metric is 1/) + /holds 3 label sets from .*max_series_per_metric is 1/) end end context 'with a limit equal to the number of label sets' do - # every label set is known in advance, which is what is for: - # the limit is reached but no record is dropped + # 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 From 200f2dc9536dbe8464775cf3844ec87df6f7ad7e Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Tue, 25 Aug 2026 02:58:21 +0000 Subject: [PATCH 7/7] Require the same max_label_value_length per metric Sections which instrument the same metric share one client metric, so they have to truncate the same way. With a shorter and a longer limit, one label value ends up in two label sets, once cut at each limit: the metric holds "/abc" and "/abcdefg" for the same "/abcdefghij", the merge the truncation is meant to do happens or not depending on which section took the record, and fluentd_prometheus_truncated_label_values_total sums both under one series, so which limit did the cut cannot be told. The length every metric was configured with is kept in the VariableStore, which fluentd empties for the duration of a graceful reload and puts back when the reload fails. Keeping it next to the metric would work until a reload: the client registry is global to the process, so the metric outlives the reload and a length stored on it would refuse the new one. max_series_per_metric stays per section, since it only decides whether a label set is accepted and not what it looks like. Signed-off-by: Kentaro Hayashi Co-Authored-By: Claude --- README.md | 5 +++ lib/fluent/plugin/prometheus.rb | 34 +++++++++++++++++ .../prometheus/label_truncation_spec.rb | 38 +++++++++++++++++++ spec/spec_helper.rb | 7 ++++ 4 files changed, 84 insertions(+) diff --git a/README.md b/README.md index 5c87823..639ab06 100644 --- a/README.md +++ b/README.md @@ -334,6 +334,11 @@ share one count. Each of them refuses a new label set once that shared count reaches its own limit, so a section which leaves `max_series_per_metric` at `0` adds label sets without counting them. +Those sections have to set the same `max_label_value_length` though, because it +decides which label set the metric is given: a shorter and a longer limit would +put one label value into two label sets, once cut at each limit. A section which +sets a different one is refused at startup. + A metric with `initialized true` creates its `` label sets at startup, so they count towards `max_series_per_metric` before any record arrives. When the limit is smaller than the number of `` label sets, diff --git a/lib/fluent/plugin/prometheus.rb b/lib/fluent/plugin/prometheus.rb index 601ba8e..078c906 100644 --- a/lib/fluent/plugin/prometheus.rb +++ b/lib/fluent/plugin/prometheus.rb @@ -1,6 +1,7 @@ require 'prometheus/client' require 'prometheus/client/formats/text' require 'fluent/clock' +require 'fluent/variable_store' require 'fluent/plugin/prometheus/placeholder_expander' module Fluent @@ -55,6 +56,18 @@ class LabelSetLimitError < StandardError; end TRUNCATED_LABEL_VALUES_METRIC_NAME = :fluentd_prometheus_truncated_label_values_total TRUNCATED_LABEL_VALUES_METRIC_DESC = 'The total number of label values truncated because they exceeded max_label_value_length.' + # The truncation length every metric was configured with, so that two + # sections which instrument the same metric cannot truncate + # differently. It is kept in the VariableStore and not next to the metric, + # because the store is emptied for the duration of a graceful reload and + # put back when the reload fails. A length which outlived the reload would + # refuse the new one. + LABEL_VALUE_LENGTHS = :fluent_plugin_prometheus_label_value_lengths + + def self.label_value_lengths + Fluent::VariableStore.fetch_or_build(LABEL_VALUE_LENGTHS) + end + def self.included(klass) klass.class_eval do desc 'The maximum length of a label value. Longer values are truncated. 0 (default) means unlimited.' @@ -449,6 +462,7 @@ def initialize(element, registry, labels, opts = {}) 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)) + check_label_value_length! @on_label_value_truncated = opts[:on_label_value_truncated] if @initialized @@ -556,6 +570,26 @@ def bind_series_set(client_metric) end end + # Sections which instrument the same metric have to truncate the same + # way. A shorter and a longer limit turn one label value into two label + # sets, so the metric would hold the same value twice, once cut at each + # limit. max_series_per_metric stays per section instead, since it only + # decides whether a label set is accepted, not what it looks like. + def check_label_value_length! + lengths = Fluent::Plugin::Prometheus.label_value_lengths + length = lengths[@name] + if length.nil? + lengths[@name] = @max_label_value_length + return + end + return if length == @max_label_value_length + + raise ConfigError, "metric #{@name} is instrumented with " \ + "max_label_value_length #{length} already, but this " \ + "gives #{@max_label_value_length}: one label value would end up " \ + "in two label sets" + end + def check_initlabels_fit_series_limit! return if @max_series_per_metric <= 0 # two blocks with the same values make one label set diff --git a/spec/fluent/plugin/prometheus/label_truncation_spec.rb b/spec/fluent/plugin/prometheus/label_truncation_spec.rb index 0f1fb4e..4b8e20d 100644 --- a/spec/fluent/plugin/prometheus/label_truncation_spec.rb +++ b/spec/fluent/plugin/prometheus/label_truncation_spec.rb @@ -78,6 +78,44 @@ def instrument(path) end end + describe 'a metric name shared by two sections' do + def build_metric(length) + element = Fluent::Config::Element.new( + 'metric', '', + { + 'name' => 'truncated', + 'type' => 'counter', + 'desc' => 'Something foo.', + 'key' => 'foo', + 'max_label_value_length' => length.to_s, + }, + [Fluent::Config::Element.new('labels', '', {'path' => '$.path'}, [])] + ) + Fluent::Plugin::Prometheus::Counter.new(element, registry, {}, opts) + end + + it 'accepts the same max_label_value_length twice' do + build_metric(4) + + expect { build_metric(4) }.not_to raise_error + end + + # cutting '/abcdefg' at 4 and at 8 would give the metric both '/abc' and + # '/abcdefg', two label sets for one label value + it 'refuses a second max_label_value_length' do + build_metric(4) + + expect { build_metric(8) }.to raise_error(Fluent::ConfigError, + /max_label_value_length 4 already.*gives 8/) + end + + it 'refuses a section which turns the truncation off' do + build_metric(4) + + expect { build_metric(0) }.to raise_error(Fluent::ConfigError) + end + end + # They are given to the client as is, so they have to be truncated like the # label sets built from records. They come from the configuration though, so # the truncation is not reported: nothing an operator can act on is merged. diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index d2a12f7..10f4747 100644 --- a/spec/spec_helper.rb +++ b/spec/spec_helper.rb @@ -9,6 +9,13 @@ Fluent::Test.setup include Fluent::Test::Helpers +RSpec.configure do |config| + # The plugins keep the truncation length of every metric in the VariableStore, + # which lives as long as the process. Each example builds its own registry, so + # the store has to start empty as well. + config.before { Fluent::VariableStore.try_to_reset {} } +end + def ipv6_enabled? require 'socket'