output: add metrics for number of stored chunks in secondary

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

Author: Watson1978

lib/fluent/plugin/output.rb

@@ -213,6 +213,7 @@ def initialize
         @rollback_count_metrics = nil
         @flush_time_count_metrics = nil
         @slow_flush_count_metrics = nil
+        @secondary_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
@@ -281,6 +282,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)")
+        @secondary_chunk_count_metrics = metrics_create(namespace: "fluentd", subsystem: "output", name: "secondary_chunk_count", help_text: "Number of stored chunks in secondary")
 
         if has_buffer_section
           unless implement?(:buffered) || implement?(:delayed_commit)
@@ -1117,6 +1119,7 @@ def commit_write(chunk_id, delayed: @delayed_commit, secondary: false)
           if @retry # success to flush chunks in retries
             if secondary
               log.warn "retry succeeded by secondary.", chunk_id: dump_unique_id_hex(chunk_id)
+              @secondary_chunk_count_metrics.inc
             else
               log.warn "retry succeeded.", chunk_id: dump_unique_id_hex(chunk_id)
             end
@@ -1594,6 +1597,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,
+          'secondary_chunk_count' => @secondary_chunk_count_metrics.get,
         }
 
         if @buffer && @buffer.respond_to?(:statistics)

test/plugin/test_in_monitor_agent.rb

@@ -184,6 +184,7 @@ def test_enable_input_metrics(with_config)
         "rollback_count"  => Integer,
         "slow_flush_count" => Integer,
         "flush_time_count" => Integer,
+        "secondary_chunk_count" => Integer,
       }
       output_info.merge!("config" => {"@id" => "test_out", "@type" => "test_out"}) if with_config
       error_label_info = {
@@ -206,6 +207,7 @@ def test_enable_input_metrics(with_config)
         "rollback_count"  => Integer,
         "slow_flush_count" => Integer,
         "flush_time_count" => Integer,
+        "secondary_chunk_count" => Integer,
       }
       error_label_info.merge!("config" => {"@id"=>"null", "@type" => "null"}) if with_config
       opts = {with_config: with_config}
@@ -323,6 +325,7 @@ def test_enable_input_metrics(with_config)
         "rollback_count"  => Integer,
         "slow_flush_count" => Integer,
         "flush_time_count" => Integer,
+        "secondary_chunk_count" => Integer,
       }
       output_info.merge!("config" => {"@id" => "test_out", "@type" => "test_out"}) if with_config
       error_label_info = {
@@ -345,6 +348,7 @@ def test_enable_input_metrics(with_config)
         "rollback_count"  => Integer,
         "slow_flush_count" => Integer,
         "flush_time_count" => Integer,
+        "secondary_chunk_count" => Integer,
       }
       error_label_info.merge!("config" => {"@id"=>"null", "@type" => "null"}) if with_config
       opts = {with_config: with_config}
@@ -419,6 +423,7 @@ def test_enable_input_metrics(with_config)
         "rollback_count"  => Integer,
         "slow_flush_count" => Integer,
         "flush_time_count" => Integer,
+        "secondary_chunk_count" => Integer,
       }
       expect_test_out_record = {
         "plugin_id"       => "test_out",
@@ -433,6 +438,7 @@ def test_enable_input_metrics(with_config)
         "rollback_count"  => Integer,
         "slow_flush_count" => Integer,
         "flush_time_count" => Integer,
+        "secondary_chunk_count" => Integer,
       }
       assert_fuzzy_equal(expect_relabel_record, d.events[1][2])
       assert_fuzzy_equal(expect_test_out_record, d.events[3][2])
@@ -571,6 +577,7 @@ def get(uri, header = {})
         "rollback_count"  => Integer,
         "slow_flush_count" => Integer,
         "flush_time_count" => Integer,
+        "secondary_chunk_count" => Integer,
       }
       expected_null_response.merge!("config" => {"@id" => "null", "@type" => "null"}) if with_config
       expected_null_response.merge!("retry" => {}) if with_retry
@@ -685,6 +692,7 @@ def get(uri, header = {})
         "rollback_count"  => Integer,
         "slow_flush_count" => Integer,
         "flush_time_count" => Integer,
+        "secondary_chunk_count" => Integer,
       }
       response = JSON.parse(get("http://127.0.0.1:#{@port}/api/plugins.json?with_config=no&with_retry=no&with_ivars=id,num_errors").body)
       test_in_response = response["plugins"][0]
@@ -816,6 +824,7 @@ def write(chunk)
           "rollback_count" => Integer,
           'slow_flush_count' => Integer,
           'flush_time_count' => Integer,
+          "secondary_chunk_count" => Integer,
       }
       output.emit_events('test.tag', Fluent::ArrayEventStream.new([[event_time, {"message" => "test failed flush 1"}]]))
       # flush few times to check steps

test/test_plugin_classes.rb

@@ -154,6 +154,7 @@ def initialize
       @rollback_count_metrics = FluentTest::FluentTestCounterMetrics.new
       @flush_time_count_metrics = FluentTest::FluentTestCounterMetrics.new
       @slow_flush_count_metrics = FluentTest::FluentTestCounterMetrics.new
+      @secondary_chunk_count_metrics = FluentTest::FluentTestCounterMetrics.new
     end
 
     attr_reader :events
@@ -284,6 +285,7 @@ def initialize
       @rollback_count_metrics = FluentTest::FluentTestCounterMetrics.new
       @flush_time_count_metrics = FluentTest::FluentTestCounterMetrics.new
       @slow_flush_count_metrics = FluentTest::FluentTestCounterMetrics.new
+      @secondary_chunk_count_metrics = FluentTest::FluentTestCounterMetrics.new
     end
 
     def format(tag, time, record)