Adding tests for metrics

Author: Athishpranav2003

.gitignore

@@ -1 +1,3 @@
 *.gem
+nclude-dependencies
+Gemfile.lock

Gemfile.lock

@@ -1,87 +1,107 @@
 PATH
   remote: .
   specs:
-    fluent-plugin-throttle (0.0.5)
+    fluent-plugin-throttle (0.0.6)
+      fluent-plugin-prometheus (~> 2.1)
       fluentd (~> 1.1)
+      prometheus-client (~> 4.2)
 
 GEM
   remote: https://rubygems.org/
   specs:
-    addressable (2.5.2)
-      public_suffix (>= 2.0.2, < 4.0)
-    appraisal (2.2.0)
+    addressable (2.8.6)
+      public_suffix (>= 2.0.2, < 6.0)
+    appraisal (2.5.0)
       bundler
       rake
       thor (>= 0.14.0)
-    byebug (10.0.2)
-    coderay (1.1.2)
-    concurrent-ruby (1.1.6)
-    cool.io (1.6.0)
-    crack (0.4.3)
-      safe_yaml (~> 1.0.0)
-    fluentd (1.11.1)
+    base64 (0.2.0)
+    bigdecimal (3.1.8)
+    byebug (11.1.3)
+    coderay (1.1.3)
+    concurrent-ruby (1.3.3)
+    cool.io (1.8.1)
+    crack (1.0.0)
+      bigdecimal
+      rexml
+    csv (3.3.0)
+    drb (2.2.1)
+    fluent-plugin-prometheus (2.1.0)
+      fluentd (>= 1.9.1, < 2)
+      prometheus-client (>= 2.1.0)
+    fluentd (1.17.0)
+      base64 (~> 0.2)
+      bundler
       cool.io (>= 1.4.5, < 2.0.0)
-      http_parser.rb (>= 0.5.1, < 0.7.0)
+      csv (~> 3.2)
+      drb (~> 2.2)
+      http_parser.rb (>= 0.5.1, < 0.9.0)
       msgpack (>= 1.3.1, < 2.0.0)
-      serverengine (>= 2.0.4, < 3.0.0)
-      sigdump (~> 0.2.2)
-      strptime (>= 0.2.2, < 1.0.0)
+      serverengine (>= 2.3.2, < 3.0.0)
+      sigdump (~> 0.2.5)
+      strptime (>= 0.2.4, < 1.0.0)
       tzinfo (>= 1.0, < 3.0)
       tzinfo-data (~> 1.0)
+      webrick (~> 1.4)
       yajl-ruby (~> 1.0)
-    hashdiff (0.3.7)
-    http_parser.rb (0.6.0)
-    maxitest (3.1.0)
-      minitest (>= 5.0.0, < 5.12.0)
-    metaclass (0.0.4)
-    method_source (0.9.0)
-    minitest (5.11.3)
-    mocha (1.6.0)
-      metaclass (~> 0.0.1)
-    msgpack (1.3.3)
-    power_assert (1.1.3)
-    pry (0.11.3)
-      coderay (~> 1.1.0)
-      method_source (~> 0.9.0)
-    pry-byebug (3.6.0)
-      byebug (~> 10.0)
-      pry (~> 0.10)
-    public_suffix (3.0.2)
-    rake (12.3.1)
-    safe_yaml (1.0.4)
-    serverengine (2.2.1)
+    hashdiff (1.1.0)
+    http_parser.rb (0.8.0)
+    maxitest (5.5.0)
+      minitest (>= 5.14.0, < 5.24.0)
+    method_source (1.1.0)
+    minitest (5.23.1)
+    mocha (2.3.0)
+      ruby2_keywords (>= 0.0.5)
+    msgpack (1.7.2)
+    power_assert (2.0.3)
+    prometheus-client (4.2.2)
+    pry (0.14.2)
+      coderay (~> 1.1)
+      method_source (~> 1.0)
+    pry-byebug (3.10.1)
+      byebug (~> 11.0)
+      pry (>= 0.13, < 0.15)
+    public_suffix (5.0.5)
+    rake (13.2.1)
+    rexml (3.3.0)
+      strscan
+    ruby2_keywords (0.0.5)
+    serverengine (2.3.2)
       sigdump (~> 0.2.2)
-    sigdump (0.2.4)
-    single_cov (1.1.0)
-    strptime (0.2.4)
-    test-unit (3.2.8)
+    sigdump (0.2.5)
+    single_cov (1.11.0)
+    strptime (0.2.5)
+    strscan (3.1.0)
+    test-unit (3.6.2)
       power_assert
-    thor (0.20.0)
-    tzinfo (2.0.2)
+    thor (1.3.1)
+    tzinfo (2.0.6)
       concurrent-ruby (~> 1.0)
-    tzinfo-data (1.2020.1)
+    tzinfo-data (1.2024.1)
       tzinfo (>= 1.0.0)
-    webmock (3.4.2)
-      addressable (>= 2.3.6)
+    webmock (3.23.1)
+      addressable (>= 2.8.0)
       crack (>= 0.3.2)
-      hashdiff
-    yajl-ruby (1.4.1)
+      hashdiff (>= 0.4.0, < 2.0.0)
+    webrick (1.8.1)
+    yajl-ruby (1.4.3)
 
 PLATFORMS
   ruby
+  x86_64-darwin-23
 
 DEPENDENCIES
   appraisal (~> 2.2)
-  bundler (~> 1.16)
+  bundler (~> 2.5)
   fluent-plugin-throttle!
   maxitest
   mocha
   pry
   pry-byebug
-  rake (~> 12.3)
+  rake (~> 13.2)
   single_cov
   test-unit (~> 3.2)
   webmock (~> 3.3)
 
 BUNDLED WITH
-   1.17.2
+   2.5.12

fluent-plugin-throttle.gemspec

@@ -16,15 +16,16 @@ Gem::Specification.new do |spec|
   spec.test_files    = spec.files.grep(%r{^(test|spec|features)/})
   spec.require_paths = ["lib"]
 
-  spec.add_development_dependency "bundler", "~> 1.16"
-  spec.add_development_dependency "rake", "~> 12.3"
+  spec.add_development_dependency "bundler", "~> 2.5"
+  spec.add_development_dependency "rake", "~> 13.2"
   spec.add_development_dependency "webmock", "~> 3.3"
   spec.add_development_dependency "test-unit", "~> 3.2"
   spec.add_development_dependency "appraisal", "~> 2.2"
   spec.add_development_dependency "mocha"
   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"
+  spec.add_dependency "prometheus-client", '~> 4.2'
+  spec.add_dependency "fluentd", "~> 1.1"
+  spec.add_dependency "fluent-plugin-prometheus", "~> 2.1"
 end

lib/fluent/plugin/filter_throttle.rb

@@ -7,6 +7,7 @@ class ThrottleFilter < Filter
     Fluent::Plugin.register_filter('throttle', self)
     include Fluent::Plugin::PrometheusLabelParser
     include Fluent::Plugin::Prometheus
+    attr_reader :registry
 
     desc "Used to group logs. Groups are rate limited independently"
     config_param :group_key, :array, :default => ['kubernetes.container_name']
@@ -116,15 +117,15 @@ def filter(tag, time, record)
 
       # 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, 0)
+      counter = @counters[group] = @counters.delete(group) || Group.new(0, now, 0, 0, now, nil, @group_bucket_limit)
 
       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 = 0
+        counter.rate_count_last = @group_bucket_limit
         counter.rate_last_reset = now
       end
 

test/fluent/plugin/filter_throttle_test.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 require_relative '../../helper'
 require 'fluent/plugin/filter_throttle'
+require 'fluent/plugin/prometheus'
 
 SingleCov.covered!
 
@@ -178,5 +179,33 @@ def create_driver(conf='')
 
       assert driver.logs.any? { |log| log.include?('rate back down') }
     end
+
+    it 'emit metrics when enabled and rate exceeds - multiple keys' do
+      driver = create_driver <<~CONF
+        group_key "group1,group2"
+        group_bucket_period_s 1
+        group_bucket_limit 5
+        group_emit_metrics true
+      CONF
+
+      driver.run(default_tag: "test") do
+        driver.feed([[event_time, {"msg": "test", "group1": "a", "group2": "b"}]] * 100)
+        driver.feed([[event_time, {"msg": "test", "group1": "b", "group2": "b"}]] * 50)
+        driver.feed([[event_time, {"msg": "test", "group1": "c"}]] * 25)
+        driver.feed([[event_time, {"msg": "test", "group2": "c"}]] * 10)
+      end
+
+      groups = driver.filtered_records.group_by { |r| r[:group1] }
+      groups.each { |k, g| groups[k] = g.group_by { |r| r[:group2] } }
+
+      assert_equal(5, groups["a"]["b"].size)
+      assert_equal(5, groups["b"]["b"].size)
+      assert_equal(5, groups["c"][nil].size)
+      assert_equal(5, groups[nil]["c"].size)
+      assert_equal(95, driver.instance.registry.get(:fluentd_throttle_rate_limit_exceeded).get(labels: {group1: "a", group2: "b"}))
+      assert_equal(45, driver.instance.registry.get(:fluentd_throttle_rate_limit_exceeded).get(labels: {group1: "b", group2: "b"}))
+      assert_equal(20, driver.instance.registry.get(:fluentd_throttle_rate_limit_exceeded).get(labels: {group1: "c", group2: nil}))
+      assert_equal(5, driver.instance.registry.get(:fluentd_throttle_rate_limit_exceeded).get(labels: {group1: nil, group2: "c"}))
+    end
   end
 end