added multiple group labeling for metrics and support to stop emitting metrics

Author: Athishpranav2003

README.md

@@ -129,6 +129,14 @@ When a group reaches its limit and as long as it is not reset, a warning
 message with the current log rate of the group is emitted repeatedly. This is
 the delay between every repetition.
 
+#### group\_emit\_metrics
+
+Default: `false`.
+
+When a group reaches its limit, metrics will be emitted for the logs being dropped if this value is true . This metrics can be scraped like any other metrics emitted in prometheus format. `podname` is a additional label available to identify the throttled groups. \\
+Metrics for the filter is
+- `fluentd_throttle_rate_limit_exceeded`
+
 ## License
 
 Apache License, Version 2.0

fluent-plugin-throttle.gemspec

@@ -25,5 +25,6 @@ Gem::Specification.new do |spec|
   spec.add_development_dependency "maxitest"
   spec.add_development_dependency "single_cov"
 
+  spec.add_runtime_dependency "prometheus-client", '>= 4.2.2'
   spec.add_runtime_dependency "fluentd", "~> 1.1"
 end

lib/fluent/plugin/filter_throttle.rb

@@ -37,14 +37,19 @@ class ThrottleFilter < Filter
     DESC
     config_param :group_warning_delay_s, :integer, :default => 10
 
+    desc <<~DESC
+      Whether to emit a metric when the rate limit is exceeded. The metric is fluentd_throttle_rate_limit_exceeded
+    DESC
+    config_param :group_emit_metrics, :bool, :default => false
+
     Group = Struct.new(
       :rate_count,
       :rate_last_reset,
       :aprox_rate,
       :bucket_count,
       :bucket_last_reset,
       :last_warning,
-      :rate_count_last_exceeded)
+      :rate_count_last)
 
     def initialize
       super
@@ -55,7 +60,8 @@ def configure(conf)
       super
 
       @group_key_paths = group_key.map { |key| key.split(".") }
-
+      @group_key_symbols = (group_key.map {|str| str.gsub(/[^a-zA-Z0-9_]/,'_')}).map(&:to_sym)
+      
       raise "group_bucket_period_s must be > 0" \
         unless @group_bucket_period_s > 0
 
@@ -107,7 +113,7 @@ def filter(tag, time, record)
         # compute and store rate/s at most every second
         counter.aprox_rate = (counter.rate_count / since_last_rate_reset).round()
         counter.rate_count = 0
-        counter.rate_count_last_exceeded = 0
+        counter.rate_count_last = 0
         counter.rate_last_reset = now
       end
 
@@ -163,10 +169,15 @@ def extract_group(record)
     end
 
     def log_rate_limit_exceeded(now, group, counter)
-      metric = @metrics[:throttle_rate_limit_exceeded]
-      log.debug("current rate",counter.rate_count,"current metric",metric.get(labels: @base_labels.merge(podname: group)))
-      metric.increment(by: counter.rate_count - counter.rate_count_last_exceeded, labels: @base_labels.merge(podname: group))
-      counter.rate_count_last_exceeded = counter.rate_count
+      # Check if metrics are enabled
+      if @group_emit_metrics
+        groupped_label = @group_key_symbols.zip(group).to_h
+        metric = @metrics[:throttle_rate_limit_exceeded]
+        log.debug("current rate",counter.rate_count,"current metric",metric.get(labels: @base_labels.merge(groupped_label)))
+        metric.increment(by: counter.rate_count - counter.rate_count_last, labels: @base_labels.merge(groupped_label))
+        counter.rate_count_last = counter.rate_count
+      end
+
       emit = counter.last_warning == nil ? true \
         : (now - counter.last_warning) >= @group_warning_delay_s
       if emit
@@ -194,10 +205,12 @@ def log_items(now, group, counter)
     end
 
     def get_counter(name, docstring)
-      if @registry.exist?(name)
-        @registry.get(name)
-      else
-        @registry.counter(name, docstring: docstring, labels: @base_labels.keys + [:podname])
+      if @group_emit_metrics
+        if @registry.exist?(name)
+          @registry.get(name)
+        else
+          @registry.counter(name, docstring: docstring, labels: @base_labels.keys + @group_key_symbols)
+        end
       end
     end
   end