fix(metrics): prevent negative gauge values for buffer metrics

lib/fluent/plugin/buffer.rb

@@ -917,7 +917,11 @@ def write_step_by_step(metadata, data, format, splits_count, &block)
       ]
 
       def statistics
-        stage_size, queue_size = @stage_size_metrics.get, @queue_size_metrics.get
+        # Clamp to non-negative: historical races can leave counters slightly
+        # negative (issues #5303, #2712). Exported Prometheus gauges must not
+        # report negative buffer sizes.
+        stage_size = [@stage_size_metrics.get, 0].max
+        queue_size = [@queue_size_metrics.get, 0].max
         buffer_space = 1.0 - ((stage_size + queue_size * 1.0) / @total_limit_size)
         @stage_length_metrics.set(@stage.size)
         @queue_length_metrics.set(@queue.size)

lib/fluent/plugin/metrics_local.rb

@@ -62,7 +62,10 @@ def inc
 
       def dec_gauge
         @monitor.synchronize do
+          # Buffer size / length gauges must never go negative even if a race
+          # causes sub/dec to run more times than add/inc (see #5303).
           @store -= 1
+          @store = 0 if @store < 0
         end
       end
 
@@ -74,7 +77,10 @@ def add(value)
 
       def sub_gauge(value)
         @monitor.synchronize do
+          # Prevent negative values that leak into Prometheus buffer metrics
+          # (fluentd_output_status_buffer_total_bytes, etc.). See #5303 / #2712.
           @store -= value
+          @store = 0 if @store < 0
         end
       end
 

test/plugin/test_buffer.rb

@@ -1536,5 +1536,19 @@ def create_chunk_es(metadata, es)
     test 'returns available_buffer_space_ratios' do
       assert_equal 10.0, @p.statistics['buffer']['available_buffer_space_ratios']
     end
+
+    # Export path clamps independently of LocalMetrics sub/dec (#5303).
+    # set_gauge can still hold a negative value (e.g. external metrics backends).
+    test 'exports non-negative stage/queue byte sizes when gauges are negative' do
+      @p.stage_size_metrics.set(-50)
+      @p.queue_size_metrics.set(-100)
+
+      stats = @p.statistics['buffer']
+      assert_equal 0, stats['stage_byte_size']
+      assert_equal 0, stats['queue_byte_size']
+      assert_equal 0, stats['total_queued_size']
+      assert stats['total_queued_size'] >= 0
+      assert stats['available_buffer_space_ratios'] >= 0
+    end
   end
 end

test/plugin/test_metrics_local.rb

@@ -91,6 +91,28 @@ class LocalMetricsTest < ::Test::Unit::TestCase
         @m.set(10)
         assert_equal 10, @m.get # On gauge, value always should be overwritten.
       end
+
+      # Prevents negative buffer size metrics exported to Prometheus (#5303)
+      test 'gauge sub does not go below zero' do
+        @m.set(5)
+        @m.sub(10)
+        assert_equal 0, @m.get
+
+        @m.sub(1)
+        assert_equal 0, @m.get
+      end
+
+      test 'gauge dec does not go below zero' do
+        assert_equal 0, @m.get
+        @m.dec
+        assert_equal 0, @m.get
+
+        @m.set(1)
+        @m.dec
+        assert_equal 0, @m.get
+        @m.dec
+        assert_equal 0, @m.get
+      end
     end
   end
 end