Fixes #20

Signed-off-by: Philipp Hossner <[email protected]>

Author: phihos

README.md

@@ -286,6 +286,8 @@ For details of each metric type, see [Prometheus documentation](http://prometheu
 - `type`: metric type (required)
 - `desc`: description of this metric (required)
 - `key`: key name of record for instrumentation (**optional**)
+- `retention`: time in seconds to remove a metric after not being updated (optional). See [Retention](#retention)
+- `retention_check_interval`: time in seconds to check for expired metrics (optional). Has no effect when `retention` not set. See [Retention](#retention)
 - `<labels>`: additional labels for this metric (optional). See [Labels](#labels)
 
 If key is empty, the metric values is treated as 1, so the counter increments by 1 on each record regardless of contents of the record.
@@ -310,6 +312,8 @@ If key is empty, the metric values is treated as 1, so the counter increments by
 - `type`: metric type (required)
 - `desc`: description of metric (required)
 - `key`: key name of record for instrumentation (required)
+- `retention`: time in seconds to remove a metric after not being updated (optional). See [Retention](#retention)
+- `retention_check_interval`: time in seconds to check for expired metrics (optional). Has no effect when `retention` not set. See [Retention](#retention)
 - `<labels>`: additional labels for this metric (optional). See [Labels](#labels)
 
 ### summary type
@@ -332,6 +336,8 @@ If key is empty, the metric values is treated as 1, so the counter increments by
 - `type`: metric type (required)
 - `desc`: description of metric (required)
 - `key`: key name of record for instrumentation (required)
+- `retention`: time in seconds to remove a metric after not being updated (optional). See [Retention](#retention)
+- `retention_check_interval`: time in seconds to check for expired metrics (optional). Has no effect when `retention` not set. See [Retention](#retention)
 - `<labels>`: additional labels for this metric (optional). See [Labels](#labels)
 
 ### histogram type
@@ -356,6 +362,8 @@ If key is empty, the metric values is treated as 1, so the counter increments by
 - `desc`: description of metric (required)
 - `key`: key name of record for instrumentation (required)
 - `buckets`: buckets of record for instrumentation (optional)
+- `retention`: time in seconds to remove a metric after not being updated (optional). See [Retention](#retention)
+- `retention_check_interval`: time in seconds to check for expired metrics (optional). Has no effect when `retention` not set. See [Retention](#retention)
 - `<labels>`: additional labels for this metric (optional). See [Labels](#labels)
 
 ## Labels
@@ -430,6 +438,33 @@ Prometheus output/filter plugin can have multiple metric section. Top-level labe
 
 In this case, `message_foo_counter` has `tag`, `hostname`, `key` and `data_type` labels.
 
+## Retention
+
+By default metrics with all encountered label combinations are preserved until the next restart of fluentd.
+Even if a label combination did not receive any update for a long time.
+That behavior is not always desired especially when having labels where the value is determined by the user like a `url` label.
+For these metrics you can set `retention`  and `retention_check_interval` like this:
+
+```
+<metric>
+  name message_foo_counter
+  type counter
+  desc The total number of foo in message.
+  key foo
+  retention 3600 # 1h
+  retention_check_interval 1800 # 30m
+  <labels>
+    bar ${bar}
+  </labels>
+</metric>
+```
+
+If `${bar}` was `baz` one time but after that no records with that value were processed, then after one hour the metric
+`foo{bar="baz"}` might be removed. 
+When this actually happens depends on `retention_check_interval` (default 60).
+It causes a background thread to check every 30 minutes for expired metrics.
+So worst case the metrics are removed 30 minutes after expiration.
+You can set this value as low as `1`, but that may put more stress on your CPU.
 
 ## Try plugin with nginx
 

lib/fluent/plugin/filter_prometheus.rb

@@ -7,6 +7,8 @@ class PrometheusFilter < Fluent::Plugin::Filter
     include Fluent::Plugin::PrometheusLabelParser
     include Fluent::Plugin::Prometheus
 
+    helpers :thread
+
     def initialize
       super
       @registry = ::Prometheus::Client.registry
@@ -22,6 +24,17 @@ def configure(conf)
       @metrics = Fluent::Plugin::Prometheus.parse_metrics_elements(conf, @registry, labels)
     end
 
+    def start
+      super
+      Fluent::Plugin::Prometheus.start_retention_threads(
+        @metrics,
+        @registry,
+        method(:thread_create),
+        method(:thread_current_running?),
+        @log
+      )
+    end
+
     def filter(tag, time, record)
       instrument_single(tag, time, record, @metrics)
       record

lib/fluent/plugin/out_prometheus.rb

@@ -7,6 +7,8 @@ class PrometheusOutput < Fluent::Plugin::Output
     include Fluent::Plugin::PrometheusLabelParser
     include Fluent::Plugin::Prometheus
 
+    helpers :thread
+
     def initialize
       super
       @registry = ::Prometheus::Client.registry
@@ -22,6 +24,17 @@ def configure(conf)
       @metrics = Fluent::Plugin::Prometheus.parse_metrics_elements(conf, @registry, labels)
     end
 
+    def start
+      super
+      Fluent::Plugin::Prometheus.start_retention_threads(
+        @metrics,
+        @registry,
+        method(:thread_create),
+        method(:thread_current_running?),
+        @log
+      )
+    end
+
     def process(tag, es)
       instrument(tag, es, @metrics)
     end

lib/fluent/plugin/prometheus.rb

@@ -81,13 +81,24 @@ def self.parse_metrics_elements(conf, registry, labels = {})
         metrics
       end
 
+      def self.start_retention_threads(metrics, registry, thread_create, thread_running, log)
+        metrics.select { |metric| metric.has_retention? }.each do |metric|
+          thread_create.call("prometheus_retention_#{metric.name}".to_sym) do
+            while thread_running.call()
+              metric.remove_expired_metrics(registry, log)
+              sleep(metric.retention_check_interval)
+            end
+          end
+        end
+      end
+
       def self.placeholder_expander(log)
         Fluent::Plugin::Prometheus::ExpandBuilder.new(log: log)
       end
 
       def stringify_keys(hash_to_stringify)
         # Adapted from: https://www.jvt.me/posts/2019/09/07/ruby-hash-keys-string-symbol/
-        hash_to_stringify.map do |k,v|
+        hash_to_stringify.map do |k, v|
           value_or_hash = if v.instance_of? Hash
                             stringify_keys(v)
                           else
@@ -151,6 +162,9 @@ class Metric
         attr_reader :name
         attr_reader :key
         attr_reader :desc
+        attr_reader :retention
+        attr_reader :retention_check_interval
+        attr_reader :last_modified_store
 
         def initialize(element, registry, labels)
           ['name', 'desc'].each do |key|
@@ -162,6 +176,11 @@ def initialize(element, registry, labels)
           @name = element['name']
           @key = element['key']
           @desc = element['desc']
+          @retention = element['retention'].to_i
+          @retention_check_interval = element.fetch('retention_check_interval', 60).to_i
+          if has_retention?
+            @last_modified_store = LastModifiedStore.new
+          end
 
           @base_labels = Fluent::Plugin::Prometheus.parse_labels_elements(element)
           @base_labels = labels.merge(@base_labels)
@@ -192,6 +211,76 @@ def self.get(registry, name, type, docstring)
 
           metric
         end
+
+        def set_value?(value)
+          if value
+            return true
+          end
+          false
+        end
+
+        def instrument(record, expander)
+          value = self.value(record)
+          if self.set_value?(value)
+            labels = labels(record, expander)
+            set_value(value, labels)
+            if has_retention?
+              @last_modified_store.set_last_updated(labels)
+            end
+          end
+        end
+
+        def has_retention?
+          @retention > 0
+        end
+
+        def remove_expired_metrics(registry, log)
+          if has_retention?
+            require 'fluent/plugin/prometheus/patch_metric_store' # require locally to patch only if necessary
+
+            metric = registry.get(@name)
+
+            expiration_time = Time.now - @retention
+            expired_label_sets = @last_modified_store.get_labels_not_modified_since(expiration_time)
+
+            expired_label_sets.each { |expired_label_set|
+              log.debug "Metric #{@name} with labels #{expired_label_set} expired. Removing..."
+              metric.remove(expired_label_set) # this method is supplied by the require at the top of this method
+              @last_modified_store.remove(expired_label_set)
+            }
+          else
+            log.warn('remove_expired_metrics should not be called when retention is not set for this metric!')
+          end
+        end
+
+        class LastModifiedStore
+          def initialize
+            @internal_store = Hash.new
+            @lock = Monitor.new
+          end
+
+          def synchronize
+            @lock.synchronize { yield }
+          end
+
+          def set_last_updated(labels)
+            synchronize do
+              @internal_store[labels] = Time.now
+            end
+          end
+
+          def remove(labels)
+            synchronize do
+              @internal_store.delete(labels)
+            end
+          end
+
+          def get_labels_not_modified_since(time)
+            synchronize do
+              @internal_store.select { |k, v| v < time }.keys
+            end
+          end
+        end
       end
 
       class Gauge < Metric
@@ -208,16 +297,17 @@ def initialize(element, registry, labels)
           end
         end
 
-        def instrument(record, expander)
+        def value(record)
           if @key.is_a?(String)
-            value = record[@key]
+            record[@key]
           else
-            value = @key.call(record)
-          end
-          if value
-            @gauge.set(value, labels: labels(record, expander))
+            @key.call(record)
           end
         end
+
+        def set_value(value, labels)
+          @gauge.set(value, labels: labels)
+        end
       end
 
       class Counter < Metric
@@ -230,20 +320,22 @@ def initialize(element, registry, labels)
           end
         end
 
-        def instrument(record, expander)
-          # use record value of the key if key is specified, otherwise just increment
+        def value(record)
           if @key.nil?
-            value = 1
+            1
           elsif @key.is_a?(String)
-            value = record[@key]
+            record[@key]
           else
-            value = @key.call(record)
+            @key.call(record)
           end
+        end
 
-          # ignore if record value is nil
-          return if value.nil?
+        def set_value?(value)
+          !value.nil?
+        end
 
-          @counter.increment(by: value, labels: labels(record, expander))
+        def set_value(value, labels)
+          @counter.increment(by: value, labels: labels)
         end
       end
 
@@ -261,16 +353,17 @@ def initialize(element, registry, labels)
           end
         end
 
-        def instrument(record, expander)
+        def value(record)
           if @key.is_a?(String)
-            value = record[@key]
+            record[@key]
           else
-            value = @key.call(record)
-          end
-          if value
-            @summary.observe(value, labels: labels(record, expander))
+            @key.call(record)
           end
         end
+
+        def set_value(value, labels)
+          @summary.observe(value, labels: labels)
+        end
       end
 
       class Histogram < Metric
@@ -294,16 +387,17 @@ def initialize(element, registry, labels)
           end
         end
 
-        def instrument(record, expander)
+        def value(record)
           if @key.is_a?(String)
-            value = record[@key]
+            record[@key]
           else
-            value = @key.call(record)
-          end
-          if value
-            @histogram.observe(value, labels: labels(record, expander))
+            @key.call(record)
           end
         end
+
+        def set_value(value, labels)
+          @histogram.observe(value, labels: labels)
+        end
       end
     end
   end

lib/fluent/plugin/prometheus/patch_metric_store.rb

@@ -0,0 +1,25 @@
+# The default Prometheus client data store has no means of removing values.
+# For the "retention" feature we need to be able to remove metrics with specific labels after some time of inactivity.
+# By patching the MetricStore and Metric classes we implement that missing feature.
+module Prometheus
+  module Client
+    module DataStores
+      class Synchronized
+        class MetricStore
+          def remove(labels:)
+            synchronize do
+              @internal_store.delete(labels)
+            end
+          end
+        end
+      end
+    end
+
+    class Metric
+      def remove(labels)
+        label_set = label_set_for(labels)
+        @store.remove(labels: label_set)
+      end
+    end
+  end
+end