Introduce config parameter limit label expansion

README.md

@@ -266,6 +266,42 @@ You can access nested keys in records via dot or bracket notation (https://docs.
 
 See Supported Metric Type and Labels for more configuration parameters.
 
+#### Limiting label expansion
+
+Label values come from records, so a metric can grow unboundedly when a label
+is bound to a field with many distinct values. Both plugins limit it:
+
+|parameter|description|default|
+|---|---|---|
+|max_label_value_length|The maximum length of a label value. A longer value is truncated. `0` means unlimited.|256|
+|max_series_per_metric|The maximum number of label sets a metric can hold. A label set beyond the limit is dropped, while the label sets already known keep being instrumented. `0` means unlimited.|10000|
+|ignore_error_log_interval|The interval in seconds to suppress the repeated warning about the dropped label sets. `0` logs every occurrence.|3600|
+
+```
+<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>
+```
+
+Note that the number of label sets is counted per metric of each plugin
+instance. When two plugin instances instrument the same metric name, each of
+them has its own limit.
+
+A label set is counted only after the metric was instrumented successfully. A
+record which fails to be instrumented, for example when the value of `key` is
+not a number, does not consume `max_series_per_metric`.
+
 ## Supported Metric Types
 
 For details of each metric type, see [Prometheus documentation](http://prometheus.io/docs/concepts/metric_types/). Also see [metric name guide](http://prometheus.io/docs/practices/naming/).

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

@@ -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)