output: add metrics for dropped oldest chunk count

Signed-off-by: Shizuo Fujita <[email protected]>

Author: Watson1978

lib/fluent/plugin/output.rb

@@ -190,6 +190,7 @@ def initialize
         @rollback_count_metrics = nil
         @flush_time_count_metrics = nil
         @slow_flush_count_metrics = nil
+        @drop_oldest_chunk_count_metrics = nil
         @enable_size_metrics = false
 
         # How to process events is decided here at once, but it will be decided in delayed way on #configure & #start
@@ -259,6 +260,7 @@ def configure(conf)
         @rollback_count_metrics = metrics_create(namespace: "fluentd", subsystem: "output", name: "rollback_count", help_text: "Number of rollbacking operations")
         @flush_time_count_metrics = metrics_create(namespace: "fluentd", subsystem: "output", name: "flush_time_count", help_text: "Count of flush time")
         @slow_flush_count_metrics = metrics_create(namespace: "fluentd", subsystem: "output", name: "slow_flush_count", help_text: "Count of slow flush occurred time(s)")
+        @drop_oldest_chunk_count_metrics = metrics_create(namespace: "fluentd", subsystem: "output", name: "drop_oldest_chunk_count", help_text: "Number of count that old chunk were discarded with drop_oldest_chunk")
 
         if has_buffer_section
           unless implement?(:buffered) || implement?(:delayed_commit)
@@ -977,6 +979,7 @@ def write_guard(&block)
               if oldest
                 log.warn "dropping oldest chunk to make space after buffer overflow", chunk_id: dump_unique_id_hex(oldest.unique_id)
                 @buffer.purge_chunk(oldest.unique_id)
+                @drop_oldest_chunk_metrics.inc
               else
                 log.error "no queued chunks to be dropped for drop_oldest_chunk"
               end
@@ -1575,6 +1578,7 @@ def statistics
           'rollback_count' => @rollback_count_metrics.get,
           'slow_flush_count' => @slow_flush_count_metrics.get,
           'flush_time_count' => @flush_time_count_metrics.get,
+          'drop_oldest_chunk_count' => @drop_oldest_chunk_count_metrics.get,
         }
 
         if @buffer && @buffer.respond_to?(:statistics)

test/plugin/test_output.rb

@@ -227,7 +227,8 @@ def waiting(seconds)
       @i.configure(config_element())
 
       %w[num_errors_metrics emit_count_metrics emit_size_metrics emit_records_metrics write_count_metrics
-         write_secondary_count_metrics rollback_count_metrics flush_time_count_metrics slow_flush_count_metrics].each do |metric_name|
+         write_secondary_count_metrics rollback_count_metrics flush_time_count_metrics slow_flush_count_metrics
+         drop_oldest_chunk_count_metrics].each do |metric_name|
         assert_true @i.instance_variable_get(:"@#{metric_name}").is_a?(Fluent::Plugin::Metrics)
       end
 
@@ -241,6 +242,7 @@ def waiting(seconds)
       assert_equal 0, @i.rollback_count
       assert_equal 0, @i.flush_time_count
       assert_equal 0, @i.slow_flush_count
+      assert_equal 0, @i.drop_oldest_chunk_count
     end
 
     data(:new_api => :chunk,