added first default label support for metrics

Author: Athishpranav2003

fluent-plugin-throttle.gemspec

@@ -4,7 +4,7 @@ $LOAD_PATH.unshift(lib) unless $LOAD_PATH.include?(lib)
 
 Gem::Specification.new do |spec|
   spec.name          = "fluent-plugin-throttle"
-  spec.version       = "0.0.5"
+  spec.version       = "0.0.6"
   spec.authors       = ["François-Xavier Bourlet"]
   spec.email         = ["[email protected]"]
   spec.summary       = %q{Fluentd filter for throttling logs based on a configurable key.}

lib/fluent/plugin/filter_throttle.rb

@@ -1,9 +1,11 @@
 # frozen_string_literal: true
 require 'fluent/plugin/filter'
+require 'fluent/plugin/prometheus'
 
 module Fluent::Plugin
   class ThrottleFilter < Filter
     Fluent::Plugin.register_filter('throttle', self)
+    include Fluent::Plugin::Prometheus
 
     desc "Used to group logs. Groups are rate limited independently"
     config_param :group_key, :array, :default => ['kubernetes.container_name']
@@ -41,7 +43,13 @@ class ThrottleFilter < Filter
       :aprox_rate,
       :bucket_count,
       :bucket_last_reset,
-      :last_warning)
+      :last_warning,
+      :rate_count_last_exceeded)
+
+    def initialize
+      super
+      @registry = ::Prometheus::Client.registry
+    end
 
     def configure(conf)
       super
@@ -68,12 +76,15 @@ def configure(conf)
 
       raise "group_warning_delay_s must be >= 1" \
         unless @group_warning_delay_s >= 1
+
+      @base_labels = {}
     end
 
     def start
       super
 
       @counters = {}
+      @metrics = {throttle_rate_limit_exceeded: get_counter(:fluentd_throttle_rate_limit_exceeded, "The exceeded rate of pods in the group")}
     end
 
     def shutdown
@@ -85,17 +96,18 @@ def filter(tag, time, record)
       now = Time.now
       rate_limit_exceeded = @group_drop_logs ? nil : record # return nil on rate_limit_exceeded to drop the record
       group = extract_group(record)
-      
-      # Ruby hashes are ordered by insertion. 
+
+      # Ruby hashes are ordered by insertion.
       # Deleting and inserting moves the item to the end of the hash (most recently used)
-      counter = @counters[group] = @counters.delete(group) || Group.new(0, now, 0, 0, now, nil)
+      counter = @counters[group] = @counters.delete(group) || Group.new(0, now, 0, 0, now, nil, 0)
 
       counter.rate_count += 1
       since_last_rate_reset = now - counter.rate_last_reset
       if since_last_rate_reset >= 1
         # 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_last_reset = now
       end
 
@@ -151,6 +163,10 @@ 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
       emit = counter.last_warning == nil ? true \
         : (now - counter.last_warning) >= @group_warning_delay_s
       if emit
@@ -176,5 +192,13 @@ def log_items(now, group, counter)
        'rate_limit_s': @group_rate_limit,
        'reset_rate_s': @group_reset_rate_s}
     end
+
+    def get_counter(name, docstring)
+      if @registry.exist?(name)
+        @registry.get(name)
+      else
+        @registry.counter(name, docstring: docstring, labels: @base_labels.keys + [:podname])
+      end
+    end
   end
 end