in_prometheus: extract the error log throttling into LogThrottle

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/prometheus.rb

@@ -1,5 +1,6 @@
 require 'prometheus/client'
 require 'prometheus/client/formats/text'
+require 'fluent/plugin/prometheus/log_throttle'
 require 'fluent/plugin/prometheus/placeholder_expander'
 
 module Fluent

lib/fluent/plugin/prometheus/log_throttle.rb

@@ -0,0 +1,43 @@
+require 'fluent/clock'
+
+module Fluent
+  module Plugin
+    module Prometheus
+      # Suppresses the repeated log for the same key within the interval.
+      # in_prometheus uses it, with an instance of its own. The key decides
+      # what is throttled, an error scope for now. The fingerprint tells the
+      # logs of a key apart: one which differs from the last is not suppressed.
+      class LogThrottle
+        Entry = Struct.new(:time, :fingerprint, :suppressed)
+
+        def initialize(interval)
+          @interval = interval
+          @mutex = Mutex.new
+          # one entry per key, so this does not grow without a limit
+          @entries = {}
+        end
+
+        # Returns [emit, suppressed_count]. emit is true for the first log of a
+        # key, for a new fingerprint, and after the interval has passed.
+        # suppressed_count is how many logs were suppressed since the last one
+        # was emitted.
+        def check(key, fingerprint)
+          return [true, 0] if @interval <= 0
+
+          @mutex.synchronize do
+            now = Fluent::Clock.now
+            last = @entries[key]
+            if last.nil? || last.fingerprint != fingerprint || (now - last.time) >= @interval
+              suppressed = (last && last.fingerprint == fingerprint) ? last.suppressed : 0
+              @entries[key] = Entry.new(now, fingerprint, 0)
+              [true, suppressed]
+            else
+              last.suppressed += 1
+              [false, 0]
+            end
+          end
+        end
+      end
+    end
+  end
+end

spec/fluent/plugin/in_prometheus_spec.rb

@@ -422,6 +422,10 @@
     end
   end
 
+  # The LogThrottle spec covers the throttling itself. These examples cover how
+  # in_prometheus uses it: the scope it throttles on, the fingerprint it makes
+  # from an error, the suppressed_log_count in the log, and what the client
+  # gets meanwhile.
   describe 'error log throttling' do
     let(:config) { LOCAL_CONFIG }
     let(:secret_message) { 'dummy secret detail: password=deadbeef' }
@@ -457,22 +461,6 @@ def error_logs(message)
         end
       end
 
-      it 'logs the repeated same failure again after ignore_error_log_interval has elapsed' do
-        2.times do
-          driver.instance.send(:all_metrics)
-          clock[:now] += driver.instance.ignore_error_log_interval
-        end
-        expect(error_logs(log_message).size).to eq(2)
-      end
-
-      it 'does not log the repeated same failure just before ignore_error_log_interval has elapsed' do
-        2.times do
-          driver.instance.send(:all_metrics)
-          clock[:now] += driver.instance.ignore_error_log_interval - 0.1
-        end
-        expect(error_logs(log_message).size).to eq(1)
-      end
-
       it 'reports how many logs were suppressed in the meantime' do
         3.times { driver.instance.send(:all_metrics) }
         clock[:now] += driver.instance.ignore_error_log_interval
@@ -493,14 +481,7 @@ def log_error(scope, message, error)
         driver.instance.send(:log_error_throttled, scope, message, error: error)
       end
 
-      # the plugin raises a fresh exception object per failure, so the errors
-      # have to be compared by value, not by identity
-      it 'suppresses an equal error given as a different object' do
-        log_error(:metrics, log_message, RuntimeError.new(secret_message))
-        log_error(:metrics, log_message, RuntimeError.new(secret_message))
-        expect(error_logs(log_message).size).to eq(1)
-      end
-
+      # the class and the message both take part in the fingerprint
       it 'logs immediately when the error class differs' do
         log_error(:metrics, log_message, RuntimeError.new(secret_message))
         log_error(:metrics, log_message, ArgumentError.new(secret_message))
@@ -514,21 +495,14 @@ def log_error(scope, message, error)
       end
 
       # the scope, not the log message, picks the slot to throttle on
-      it 'keeps a separate state per scope' do
-        error = RuntimeError.new(secret_message)
-        log_error(:metrics, log_message, error)
-        log_error(:workers_metrics, workers_log_message, error)
-        expect(error_logs(log_message).size).to eq(1)
-        expect(error_logs(workers_log_message).size).to eq(1)
-      end
-
       it 'suppresses an equal error within a scope even when the log message differs' do
         error = RuntimeError.new(secret_message)
         log_error(:metrics, log_message, error)
         log_error(:metrics, workers_log_message, error)
         expect(error_logs(workers_log_message)).to be_empty
       end
 
+      # the only example which takes the interval from the configuration
       context 'with ignore_error_log_interval 0' do
         let(:config) { LOCAL_CONFIG + %[
   ignore_error_log_interval 0
@@ -559,21 +533,5 @@ def log_error(scope, message, error)
         expect(error_logs(workers_log_message).size).to eq(1)
       end
     end
-
-    context 'when errors occur concurrently' do
-      # long enough to keep every call within the same interval
-      let(:config) { LOCAL_CONFIG + %[
-  ignore_error_log_interval 3600
-] }
-
-      it 'logs the error only once' do
-        instance = driver.instance
-        error = RuntimeError.new(secret_message)
-        10.times.map {
-          Thread.new { instance.send(:log_error_throttled, :metrics, log_message, error: error) }
-        }.each(&:join)
-        expect(error_logs(log_message).size).to eq(1)
-      end
-    end
   end
 end

spec/fluent/plugin/prometheus/log_throttle_spec.rb

@@ -0,0 +1,108 @@
+require 'spec_helper'
+require 'fluent/plugin/prometheus/log_throttle'
+
+describe Fluent::Plugin::Prometheus::LogThrottle do
+  # Fluent::Clock.now is monotonic, so a Hash is enough to fake it
+  let(:clock) { { now: 1000.0 } }
+  let(:interval) { 3600 }
+  # in_prometheus builds it out of an error
+  let(:fingerprint) { [RuntimeError, 'a'] }
+  subject(:throttle) { described_class.new(interval) }
+
+  before do
+    allow(Fluent::Clock).to receive(:now) { clock[:now] }
+  end
+
+  describe '#check' do
+    it 'emits on the first occurrence of a key' do
+      emit, suppressed = throttle.check(:foo, fingerprint)
+      expect(emit).to be true
+      expect(suppressed).to eq(0)
+    end
+
+    it 'suppresses the same key within the interval' do
+      throttle.check(:foo, fingerprint)
+      clock[:now] += interval - 1
+      emit, _ = throttle.check(:foo, fingerprint)
+      expect(emit).to be false
+    end
+
+    it 'emits again once the interval has elapsed' do
+      throttle.check(:foo, fingerprint)
+      clock[:now] += interval
+      emit, _ = throttle.check(:foo, fingerprint)
+      expect(emit).to be true
+    end
+
+    it 'reports how many occurrences were suppressed in the meantime' do
+      throttle.check(:foo, fingerprint)             # emits, suppressed=0
+      2.times { throttle.check(:foo, fingerprint) } # suppressed 1, then 2
+      clock[:now] += interval
+      emit, suppressed = throttle.check(:foo, fingerprint)
+      expect(emit).to be true
+      expect(suppressed).to eq(2)
+    end
+
+    it 'resets the suppressed count after emitting' do
+      throttle.check(:foo, fingerprint)
+      2.times { throttle.check(:foo, fingerprint) }
+      clock[:now] += interval
+      throttle.check(:foo, fingerprint)             # emits with suppressed=2
+      clock[:now] += interval
+      _, suppressed = throttle.check(:foo, fingerprint)
+      expect(suppressed).to eq(0)
+    end
+
+    it 'keeps a separate slot per key' do
+      expect(throttle.check(:foo, fingerprint).first).to be true
+      expect(throttle.check(:bar, fingerprint).first).to be true
+    end
+
+    it 'emits immediately when the fingerprint changes within the interval' do
+      expect(throttle.check(:foo, [RuntimeError, 'a']).first).to be true
+      expect(throttle.check(:foo, [RuntimeError, 'b']).first).to be true
+    end
+
+    # the caller makes a new fingerprint for each event, so it has to be
+    # compared by value, not by object identity
+    it 'suppresses an equal fingerprint given as a different object' do
+      expect(throttle.check(:foo, [RuntimeError, 'a']).first).to be true
+      expect(throttle.check(:foo, [RuntimeError, 'a']).first).to be false
+    end
+
+    it 'does not carry the suppressed count across a fingerprint change' do
+      throttle.check(:foo, [RuntimeError, 'a'])
+      2.times { throttle.check(:foo, [RuntimeError, 'a']) }
+      emit, suppressed = throttle.check(:foo, [RuntimeError, 'b'])
+      expect(emit).to be true
+      expect(suppressed).to eq(0)
+    end
+
+    context 'when interval is zero' do
+      let(:interval) { 0 }
+
+      it 'always emits without consulting the clock' do
+        expect(Fluent::Clock).not_to receive(:now)
+        3.times do
+          emit, suppressed = throttle.check(:foo, fingerprint)
+          expect(emit).to be true
+          expect(suppressed).to eq(0)
+        end
+      end
+    end
+
+    context 'when interval is negative' do
+      let(:interval) { -1 }
+
+      it 'always emits' do
+        expect(throttle.check(:foo, fingerprint).first).to be true
+        expect(throttle.check(:foo, fingerprint).first).to be true
+      end
+    end
+
+    it 'serializes concurrent checks for the same key into a single emission' do
+      results = 10.times.map { Thread.new { throttle.check(:foo, fingerprint).first } }.map(&:value)
+      expect(results.count(true)).to eq(1)
+    end
+  end
+end