Introduce config parameter limit label expansion

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.
+
+```
+<filter message>
+  @type prometheus
+  max_label_value_length 128
+  max_series_per_metric 1000
+  <metric>
+    name message_foo_counter
+    type counter
+    desc The total number of foo in message.
+    key foo
+    <labels>
+      path $.kubernetes.pod_name
+    </labels>
+  </metric>
+</filter>
+```
+
+Both limits can also be set in a `<metric>` 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:
+
+```
+<filter message>
+  @type prometheus
+  max_series_per_metric 1000
+  <metric>
+    name message_foo_counter
+    type counter
+    desc The total number of foo in message.
+    key foo
+    max_series_per_metric 10
+    <labels>
+      path $.kubernetes.pod_name
+    </labels>
+  </metric>
+</filter>
+```
+
+The label sets are counted per metric name, not per `<metric>` 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 `<initlabels>` 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 `<initlabels>` 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 `<initlabels>`) 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/).

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)

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

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)