diff --git a/README.md b/README.md
index f353f09..639ab06 100644
--- a/README.md
+++ b/README.md
@@ -266,6 +266,128 @@ 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 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
+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
+
+
+
+```
+
+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.
+
+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,
+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
+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.
+
+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.
+
+##### 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 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/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..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
@@ -19,7 +24,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..078c906 100644
--- a/lib/fluent/plugin/prometheus.rb
+++ b/lib/fluent/plugin/prometheus.rb
@@ -1,5 +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
@@ -31,6 +33,154 @@ 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.'
+
+ # 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.'
+
+ # 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.'
+ 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 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 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)
+
+ def initialize(interval)
+ @interval = interval
+ @mutex = Mutex.new
+ # bounded by the number of keys a caller can build, 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
+
+ # 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' }
@@ -119,7 +269,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 +280,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 +315,75 @@ 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)
+ @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 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 ||=
+ 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 })
+
+ 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
+
+ details = details.merge(suppressed_log_count: suppressed) if suppressed > 0
+ log.warn(message, details)
end
def instrument_single(tag, time, record, metrics)
@@ -180,6 +399,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 +423,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 +439,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 +457,22 @@ def initialize(element, registry, labels)
@base_labels = Fluent::Plugin::Prometheus.parse_labels_elements(element)
@base_labels = labels.merge(@base_labels)
+ # 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))
+ check_label_value_length!
+ @on_label_value_truncated = opts[:on_label_value_truncated]
+
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)
+ .map { |initlabels| truncate_label_set(initlabels) }
end
end
@@ -252,14 +493,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), k)
else
- label[k] = v.call(record)
+ label[k] = truncate_label_value(v.call(record), k)
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 +542,122 @@ 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
+
+ # 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
+
+ # 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
+ return if @series_set.size <= @max_series_per_metric
+
+ 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
+
+ # 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
+
+ # 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 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|
+ 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.
+ # 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_set.reserve(label, @max_series_per_metric, @name)
+ end
+
+ def confirm_series(label)
+ return if @max_series_per_metric <= 0
+
+ @series_set.confirm(label)
+ end
+
+ def release_series(label)
+ @series_set.release(label)
+ end
end
class Gauge < Metric
- def initialize(element, registry, labels)
+ def initialize(element, registry, labels, opts = {})
super
if @key.nil?
raise ConfigError, "gauge metric requires 'key' option"
@@ -287,6 +668,7 @@ def initialize(element, registry, labels)
rescue ::Prometheus::Client::Registry::AlreadyRegisteredError
@gauge = Fluent::Plugin::Prometheus::Metric.get(registry, element['name'].to_sym, :gauge, element['desc'])
end
+ bind_series_set(@gauge)
if @initialized
Fluent::Plugin::Prometheus::Metric.init_label_set(@gauge, @base_initlabels, @base_labels)
@@ -300,19 +682,22 @@ def instrument(record, expander)
value = @key.call(record)
end
if value
- @gauge.set(value, labels: labels(record, expander))
+ with_label_set(record, expander) do |label|
+ @gauge.set(value, labels: label)
+ end
end
end
end
class Counter < Metric
- def initialize(element, registry, labels)
+ def initialize(element, registry, labels, opts = {})
super
begin
@counter = registry.counter(element['name'].to_sym, docstring: element['desc'], labels: @base_labels.keys)
rescue ::Prometheus::Client::Registry::AlreadyRegisteredError
@counter = Fluent::Plugin::Prometheus::Metric.get(registry, element['name'].to_sym, :counter, element['desc'])
end
+ bind_series_set(@counter)
if @initialized
Fluent::Plugin::Prometheus::Metric.init_label_set(@counter, @base_initlabels, @base_labels)
@@ -332,12 +717,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"
@@ -348,6 +735,7 @@ def initialize(element, registry, labels)
rescue ::Prometheus::Client::Registry::AlreadyRegisteredError
@summary = Fluent::Plugin::Prometheus::Metric.get(registry, element['name'].to_sym, :summary, element['desc'])
end
+ bind_series_set(@summary)
if @initialized
Fluent::Plugin::Prometheus::Metric.init_label_set(@summary, @base_initlabels, @base_labels)
@@ -361,13 +749,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"
@@ -385,6 +775,7 @@ def initialize(element, registry, labels)
rescue ::Prometheus::Client::Registry::AlreadyRegisteredError
@histogram = Fluent::Plugin::Prometheus::Metric.get(registry, element['name'].to_sym, :histogram, element['desc'])
end
+ bind_series_set(@histogram)
if @initialized
Fluent::Plugin::Prometheus::Metric.init_label_set(@histogram, @base_initlabels, @base_labels)
@@ -398,7 +789,9 @@ def instrument(record, expander)
value = @key.call(record)
end
if value
- @histogram.observe(value, labels: labels(record, expander))
+ with_label_set(record, expander) do |label|
+ @histogram.observe(value, labels: label)
+ end
end
end
end
diff --git a/spec/fluent/plugin/filter_prometheus_spec.rb b/spec/fluent/plugin/filter_prometheus_spec.rb
index c22c884..b23fee5 100644
--- a/spec/fluent/plugin/filter_prometheus_spec.rb
+++ b/spec/fluent/plugin/filter_prometheus_spec.rb
@@ -114,4 +114,99 @@
)
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 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
+
+ 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, max_label_value_length: 4) }
+
+ before do
+ allow(Fluent::Clock).to receive(:now) { clock[:now] }
+ end
+
+ def logs_about(text)
+ driver.logs.select { |log| log.include?(text) }
+ end
+
+ def drop
+ driver.instance.send(:warn_label_set_limit, metric)
+ end
+
+ 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/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/label_truncation_spec.rb b/spec/fluent/plugin/prometheus/label_truncation_spec.rb
new file mode 100644
index 0000000..4b8e20d
--- /dev/null
+++ b/spec/fluent/plugin/prometheus/label_truncation_spec.rb
@@ -0,0 +1,159 @@
+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
+
+ 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.
+ 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/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..f6a2ef6
--- /dev/null
+++ b/spec/fluent/plugin/prometheus/series_limit_spec.rb
@@ -0,0 +1,318 @@
+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 'a metric name shared by two sections' do
+ # both sections instrument the same client metric, so counting per section
+ # would let it hold max_series_per_metric label sets twice over
+ let(:max_series_per_metric) { 2 }
+ let(:other_metric) { Fluent::Plugin::Prometheus::Counter.new(element, registry, {}, {}) }
+
+ def instrument_other(path, value = 1)
+ other_metric.instrument({'foo' => value, 'path' => path}, expander)
+ end
+
+ it 'wraps one and the same client metric' do
+ expect(other_metric.instance_variable_get(:@counter))
+ .to equal(metric.instance_variable_get(:@counter))
+ end
+
+ it 'counts the label sets of both sections against one limit' do
+ instrument('/a')
+ instrument_other('/b')
+
+ # the metric is full, whichever section the next record goes through
+ expect { instrument_other('/c') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError)
+ expect { instrument('/d') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError)
+ expect(client_counter.values.keys).to contain_exactly({path: '/a'}, {path: '/b'})
+ end
+
+ it 'lets both sections instrument a label set the metric already holds' do
+ instrument('/a')
+
+ expect { instrument_other('/a', 2) }.not_to raise_error
+ expect(client_counter.values[{path: '/a'}]).to eq(3)
+ end
+ end
+
+ describe 'initialized true with ' do
+ let(:initlabels) { ['/a', '/b', '/c'] }
+ let(:element) do
+ Fluent::Config::Element.new(
+ 'metric', '',
+ {
+ 'name' => 'limited',
+ 'type' => 'counter',
+ 'desc' => 'Something foo.',
+ 'key' => 'foo',
+ 'initialized' => 'true',
+ 'max_series_per_metric' => max_series_per_metric.to_s,
+ },
+ [Fluent::Config::Element.new('labels', '', {'path' => '$.path'}, [])] +
+ initlabels.map { |path| Fluent::Config::Element.new('initlabels', '', {'path' => path}, []) }
+ )
+ end
+
+ context 'with a limit 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,
+ /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, so the limit is reached but no
+ # record is dropped
+ let(:max_series_per_metric) { 3 }
+
+ it 'accepts the config' do
+ expect { metric }.not_to raise_error
+ end
+
+ it 'still counts a record on an label set' do
+ instrument('/a')
+
+ expect(client_counter.values[{path: '/a'}]).to eq(1)
+ end
+
+ it 'refuses a label set which is not in ' do
+ expect { instrument('/d') }.to raise_error(Fluent::Plugin::Prometheus::LabelSetLimitError)
+ end
+ end
+
+ 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
+ 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..17a837d 100644
--- a/spec/fluent/plugin/shared.rb
+++ b/spec/fluent/plugin/shared.rb
@@ -390,6 +390,167 @@
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 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
+ 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
+ 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
+
+ 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
+
+ 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
+ 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)
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'