in_prometheus: do not disclosure error details for client

README.md

@@ -64,6 +64,7 @@ More configuration parameters:
 - `metrics_path`: metrics HTTP endpoint (default: /metrics)
 - `aggregated_metrics_path`: metrics HTTP endpoint (default: /aggregated_metrics)
 - `content_encoding`: encoding format for the exposed metrics (default: identity). Supported formats are {identity, gzip}
+- `ignore_error_log_interval`: Suppress repeated error logs in a certain period of time or until message was changed (default: 1h)
 
 When using multiple workers, each worker binds to port + `fluent_worker_id`.
 To scrape metrics from all workers at once, you can access http://localhost:24231/aggregated_metrics.

lib/fluent/plugin/in_prometheus.rb

@@ -36,10 +36,15 @@ class PrometheusInput < Fluent::Plugin::Input
     desc 'Content encoding of the exposed metrics, Currently supported encoding is identity, gzip. Ref: https://prometheus.io/docs/instrumenting/exposition_formats/#basic-info'
     config_param :content_encoding, :enum, list: [:identity, :gzip], default: :identity
 
+    desc 'Suppress repeated error logs in a certain period of time (1h) or until message was changed'
+    config_param :ignore_error_log_interval, :time, default: 3600
+
     def initialize
       super
       @registry = ::Prometheus::Client.registry
       @secure = nil
+      @error_log_mutex = Mutex.new
+      @last_error_logs = {} # scope => [logged_at, fingerprint, suppressed_count]
     end
 
     def configure(conf)
@@ -210,7 +215,8 @@ def start_webrick
     def all_metrics
       response(::Prometheus::Client::Formats::Text.marshal(@registry))
     rescue => e
-      [500, { 'Content-Type' => 'text/plain' }, e.to_s]
+      log_error_throttled(:metrics, "in_prometheus: failed to render metrics", error: e)
+      [500, { 'Content-Type' => 'text/plain' }, "in_prometheus server error: <#{e.class}>"]
     end
 
     def all_workers_metrics
@@ -223,7 +229,8 @@ def all_workers_metrics
       end
       response(full_result.get_metrics)
     rescue => e
-      [500, { 'Content-Type' => 'text/plain' }, e.to_s]
+      log_error_throttled(:workers_metrics, "in_prometheus: failed to render workers metrics", error: e)
+      [500, { 'Content-Type' => 'text/plain' }, "in_prometheus server error: <#{e.class}>"]
     end
 
     def send_request_to_each_worker
@@ -271,5 +278,32 @@ def response(metrics)
       end
       [200, { 'Content-Type' => ::Prometheus::Client::Formats::Text::CONTENT_TYPE, 'Content-Encoding' => @content_encoding.to_s }, body]
     end
+
+    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
+      return unless emit
+
+      if suppressed > 0
+        log.error message, error_class: error.class, error: error, suppressed_log_count: suppressed
+      else
+        log.error message, error_class: error.class, error: error
+      end
+    end
   end
 end

spec/fluent/plugin/in_prometheus_spec.rb

@@ -71,6 +71,22 @@
         expect(driver.instance.content_encoding).to eq(:gzip)
       end
     end
+
+    describe 'default ignore_error_log_interval' do
+      let(:config) { CONFIG }
+      it 'should be 3600 seconds by default' do
+        expect(driver.instance.ignore_error_log_interval).to eq(3600)
+      end
+    end
+
+    describe 'error_log_interval' do
+      let(:config) { CONFIG + %[
+    ignore_error_log_interval 60
+] }
+      it 'should be configurable' do
+        expect(driver.instance.ignore_error_log_interval).to eq(60)
+      end
+    end
   end
 
   describe '#start' do
@@ -338,4 +354,226 @@
       include_examples 'IPv6 server binding', '[::1]', '::1', 'handles pre-bracketed address correctly'
     end
   end
+
+  describe 'error handling (information disclosure)' do
+    let(:config) { LOCAL_CONFIG }
+    let(:secret_message) { 'dummy secret detail: password=deadbeef' }
+
+    shared_examples 'suppressed exception response' do
+      it 'returns 500 with text/plain' do
+        status, header, _body = subject
+        expect(status).to eq(500)
+        expect(header['Content-Type']).to eq('text/plain')
+      end
+
+      it 'exposes the exception class only' do
+        _status, _header, body = subject
+        expect(body).to eq('in_prometheus server error: <RuntimeError>')
+        expect(body).not_to include(secret_message)
+      end
+
+      it 'logs the detail on the server side' do
+        subject
+        expect(driver.logs.any? { |log| log.include?(log_message) }).to be true
+        expect(driver.logs.any? { |log| log.include?(secret_message) }).to be true
+      end
+    end
+
+    context '#all_metrics' do
+      subject { driver.instance.send(:all_metrics) }
+
+      let(:log_message) { 'in_prometheus: failed to render metrics' }
+
+      before do
+        allow(::Prometheus::Client::Formats::Text).to receive(:marshal).and_raise(RuntimeError, secret_message)
+      end
+
+      include_examples 'suppressed exception response'
+    end
+
+    context '#all_workers_metrics' do
+      subject { driver.instance.send(:all_workers_metrics) }
+
+      let(:log_message) { 'in_prometheus: failed to render workers metrics' }
+
+      before do
+        allow(driver.instance).to receive(:send_request_to_each_worker).and_raise(RuntimeError, secret_message)
+      end
+
+      include_examples 'suppressed exception response'
+    end
+
+    context 'over HTTP' do
+      before do
+        allow(::Prometheus::Client::Formats::Text).to receive(:marshal).and_raise(RuntimeError, secret_message)
+      end
+
+      it 'does not leak the exception message to the client' do
+        driver.run(timeout: 1) do
+          Net::HTTP.start('127.0.0.1', port) do |http|
+            req = Net::HTTP::Get.new('/metrics')
+            res = http.request(req)
+            expect(res.code).to eq('500')
+            expect(res.body).to eq('in_prometheus server error: <RuntimeError>')
+            expect(res.body).not_to include(secret_message)
+          end
+        end
+      end
+    end
+  end
+
+  describe 'error log throttling' do
+    let(:config) { LOCAL_CONFIG }
+    let(:secret_message) { 'dummy secret detail: password=deadbeef' }
+    let(:log_message) { 'in_prometheus: failed to render metrics' }
+    let(:workers_log_message) { 'in_prometheus: failed to render workers metrics' }
+
+    # Fluent::Clock.now is monotonic, so a plain Hash is enough to drive it
+    let(:clock) { { now: 1000.0 } }
+
+    def error_logs(message)
+      driver.logs.select { |log| log.include?(message) }
+    end
+
+    context 'when rendering metrics keeps failing' do
+      before do
+        allow(Fluent::Clock).to receive(:now) { clock[:now] }
+        allow(::Prometheus::Client::Formats::Text).to receive(:marshal).and_raise(RuntimeError, secret_message)
+      end
+
+      # every iteration raises from the same line, so the exceptions are equal
+      # to each other and only ignore_error_log_interval can let a log through
+      it 'logs the repeated same failure only once within ignore_error_log_interval' do
+        5.times { driver.instance.send(:all_metrics) }
+        expect(error_logs(log_message).size).to eq(1)
+      end
+
+      it 'keeps returning 500 to the client even while the log is suppressed' do
+        responses = 5.times.map { driver.instance.send(:all_metrics) }
+        expect(error_logs(log_message).size).to eq(1)
+        responses.each do |status, _header, body|
+          expect(status).to eq(500)
+          expect(body).to eq('in_prometheus server error: <RuntimeError>')
+        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
+        driver.instance.send(:all_metrics)
+        logs = error_logs(log_message)
+        expect(logs.size).to eq(2)
+        expect(logs.first).not_to include('suppressed_log_count')
+        expect(logs.last).to include('suppressed_log_count=2')
+      end
+    end
+
+    describe 'telling the errors apart' do
+      before do
+        allow(Fluent::Clock).to receive(:now) { clock[:now] }
+      end
+
+      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
+
+      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))
+        expect(error_logs(log_message).size).to eq(2)
+      end
+
+      it 'logs immediately when the error differs' do
+        log_error(:metrics, log_message, RuntimeError.new(secret_message))
+        log_error(:metrics, log_message, RuntimeError.new('another failure'))
+        expect(error_logs(log_message).size).to eq(2)
+      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
+
+      context 'with ignore_error_log_interval 0' do
+        let(:config) { LOCAL_CONFIG + %[
+  ignore_error_log_interval 0
+] }
+
+        it 'logs every occurrence of the same error' do
+          3.times { log_error(:metrics, log_message, RuntimeError.new(secret_message)) }
+          expect(error_logs(log_message).size).to eq(3)
+        end
+      end
+    end
+
+    # /metrics and /aggregated_metrics are usually scraped in turn, so both of
+    # them must be throttled on their own slot
+    context 'when both endpoints keep failing alternately' do
+      before do
+        allow(Fluent::Clock).to receive(:now) { clock[:now] }
+        allow(::Prometheus::Client::Formats::Text).to receive(:marshal).and_raise(RuntimeError, secret_message)
+        allow(driver.instance).to receive(:send_request_to_each_worker).and_raise(ArgumentError, 'another failure')
+      end
+
+      it 'logs each failure only once within ignore_error_log_interval' do
+        5.times do
+          driver.instance.send(:all_metrics)
+          driver.instance.send(:all_workers_metrics)
+        end
+        expect(error_logs(log_message).size).to eq(1)
+        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