From cb9997224265a71d64d17e2a7cb1152fe906ff09 Mon Sep 17 00:00:00 2001 From: Kentaro Hayashi Date: Wed, 26 Aug 2026 01:34:08 +0000 Subject: [PATCH] Extract the error log throttling into LogThrottle Before: in_prometheus suppressed its repeated error log with a Hash and a Mutex of its own, which filter/out_prometheus could not reach. After: the same mechanism lives in LogThrottle, which in_prometheus uses. No behaviour changes. in_prometheus_spec keeps only what the unit spec cannot cover. Signed-off-by: Kentaro Hayashi Co-Authored-By: Claude --- lib/fluent/plugin/in_prometheus.rb | 22 +--- lib/fluent/plugin/prometheus.rb | 1 + lib/fluent/plugin/prometheus/log_throttle.rb | 43 +++++++ spec/fluent/plugin/in_prometheus_spec.rb | 54 +-------- .../plugin/prometheus/log_throttle_spec.rb | 108 ++++++++++++++++++ 5 files changed, 162 insertions(+), 66 deletions(-) create mode 100644 lib/fluent/plugin/prometheus/log_throttle.rb create mode 100644 spec/fluent/plugin/prometheus/log_throttle_spec.rb diff --git a/lib/fluent/plugin/in_prometheus.rb b/lib/fluent/plugin/in_prometheus.rb index 82d4907..9c4f5fc 100644 --- a/lib/fluent/plugin/in_prometheus.rb +++ b/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 diff --git a/lib/fluent/plugin/prometheus.rb b/lib/fluent/plugin/prometheus.rb index 5db615a..112fd31 100644 --- a/lib/fluent/plugin/prometheus.rb +++ b/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 diff --git a/lib/fluent/plugin/prometheus/log_throttle.rb b/lib/fluent/plugin/prometheus/log_throttle.rb new file mode 100644 index 0000000..0df9956 --- /dev/null +++ b/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 diff --git a/spec/fluent/plugin/in_prometheus_spec.rb b/spec/fluent/plugin/in_prometheus_spec.rb index 9d05506..a0c4447 100644 --- a/spec/fluent/plugin/in_prometheus_spec.rb +++ b/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,14 +495,6 @@ 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) @@ -529,6 +502,7 @@ def log_error(scope, 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 diff --git a/spec/fluent/plugin/prometheus/log_throttle_spec.rb b/spec/fluent/plugin/prometheus/log_throttle_spec.rb new file mode 100644 index 0000000..99057b4 --- /dev/null +++ b/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