output: rename metrics

Signed-off-by: Shizuo Fujita <fujita@clear-code.com>

Author: Watson1978

lib/fluent/plugin/output.rb

@@ -210,10 +210,10 @@ def initialize
         @emit_records_metrics = nil
         @emit_size_metrics = nil
         @write_count_metrics = nil
+        @write_secondary_count_metrics = nil
         @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
@@ -279,10 +279,10 @@ def configure(conf)
         @emit_records_metrics = metrics_create(namespace: "fluentd", subsystem: "output", name: "emit_records", help_text: "Number of emit records")
         @emit_size_metrics =  metrics_create(namespace: "fluentd", subsystem: "output", name: "emit_size", help_text: "Total size of emit events")
         @write_count_metrics = metrics_create(namespace: "fluentd", subsystem: "output", name: "write_count", help_text: "Number of writing events")
+        @write_secondary_count_metrics = metrics_create(namespace: "fluentd", subsystem: "output", name: "secondary_chunk_count", help_text: "Number of writing events in secondary")
         @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)
@@ -1119,7 +1119,6 @@ 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
@@ -1227,6 +1226,7 @@ def try_flush
             dump_chunk_id = dump_unique_id_hex(chunk_id)
             log.trace "adding write count", instance: self.object_id
             @write_count_metrics.inc
+            @write_secondary_count_metrics.inc if using_secondary
             log.trace "executing sync write", chunk: dump_chunk_id
 
             output.write(chunk)
@@ -1594,10 +1594,10 @@ def statistics
           'retry_count' => @num_errors_metrics.get,
           'emit_count' => @emit_count_metrics.get,
           'write_count' => @write_count_metrics.get,
+          'write_secondary_count' => @write_secondary_count_metrics.get,
           '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

@@ -181,10 +181,10 @@ def test_enable_input_metrics(with_config)
         "emit_records"    => Integer,
         "emit_size"       => Integer,
         "write_count"     => Integer,
+        "write_secondary_count" => Integer,
         "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 = {
@@ -204,10 +204,10 @@ def test_enable_input_metrics(with_config)
         "emit_records"    => Integer,
         "emit_size"       => Integer,
         "write_count"     => Integer,
+        "write_secondary_count" => Integer,
         "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}
@@ -322,10 +322,10 @@ def test_enable_input_metrics(with_config)
         "emit_records"    => Integer,
         "emit_size"       => Integer,
         "write_count"     => Integer,
+        "write_secondary_count" => Integer,
         "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,10 +345,10 @@ def test_enable_input_metrics(with_config)
         "emit_records"    => Integer,
         "emit_size"       => Integer,
         "write_count"     => Integer,
+        "write_secondary_count" => Integer,
         "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}
@@ -420,10 +420,10 @@ def test_enable_input_metrics(with_config)
         "emit_records"    => Integer,
         "emit_size"       => Integer,
         "write_count"     => Integer,
+        "write_secondary_count" => Integer,
         "rollback_count"  => Integer,
         "slow_flush_count" => Integer,
         "flush_time_count" => Integer,
-        "secondary_chunk_count" => Integer,
       }
       expect_test_out_record = {
         "plugin_id"       => "test_out",
@@ -435,10 +435,10 @@ def test_enable_input_metrics(with_config)
         "emit_records"    => Integer,
         "emit_size"       => Integer,
         "write_count"     => Integer,
+        "write_secondary_count" => Integer,
         "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])
@@ -574,10 +574,10 @@ def get(uri, header = {})
         "emit_records"    => Integer,
         "emit_size"       => Integer,
         "write_count"     => Integer,
+        "write_secondary_count" => Integer,
         "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
@@ -689,10 +689,10 @@ def get(uri, header = {})
         "emit_records"    => Integer,
         "emit_size"       => Integer,
         "write_count"     => Integer,
+        "write_secondary_count" => Integer,
         "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]
@@ -821,10 +821,10 @@ def write(chunk)
           "emit_records" => Integer,
           "emit_size" => Integer,
           "write_count" => Integer,
+          "write_secondary_count" => Integer,
           "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/plugin/test_output_as_buffered_secondary.rb

@@ -308,7 +308,7 @@ def dummy_event_stream
 
       assert_equal 0, @i.write_count
       assert_equal 0, @i.num_errors
-      assert_equal 0, @i.instance_variable_get(:@secondary_chunk_count_metrics).get
+      assert_equal 0, @i.instance_variable_get(:@write_secondary_count_metrics).get
 
       @i.enqueue_thread_wait
       @i.flush_thread_wakeup
@@ -349,7 +349,7 @@ def dummy_event_stream
       assert{ @i.write_count > prev_write_count }
       assert{ @i.num_errors == prev_num_errors }
 
-      assert{ @i.instance_variable_get(:@secondary_chunk_count_metrics).get > 0 }
+      assert{ @i.instance_variable_get(:@write_secondary_count_metrics).get > 0 }
 
       assert_nil @i.retry
 
@@ -391,7 +391,6 @@ def dummy_event_stream
 
       assert_equal 0, @i.write_count
       assert_equal 0, @i.num_errors
-      assert_equal 0, @i.instance_variable_get(:@secondary_chunk_count_metrics).get
 
       @i.enqueue_thread_wait
       @i.flush_thread_wakeup
@@ -447,8 +446,6 @@ def dummy_event_stream
       assert{ @i.buffer.dequeued[chunks[0].unique_id].nil? }
       assert{ chunks.first.empty? }
 
-      assert{ @i.instance_variable_get(:@secondary_chunk_count_metrics).get > 0 }
-
       assert_nil @i.retry
 
       logs = @i.log.out.logs
@@ -743,7 +740,7 @@ def dummy_event_stream
 
       assert_equal 0, @i.write_count
       assert_equal 0, @i.num_errors
-      assert_equal 0, @i.instance_variable_get(:@secondary_chunk_count_metrics).get
+      assert_equal 0, @i.instance_variable_get(:@write_secondary_count_metrics).get
 
       @i.enqueue_thread_wait
       @i.flush_thread_wakeup
@@ -772,7 +769,7 @@ def dummy_event_stream
         prev_write_count = @i.write_count
       end
 
-      assert{ @i.instance_variable_get(:@secondary_chunk_count_metrics).get > 0 }
+      assert{ @i.instance_variable_get(:@write_secondary_count_metrics).get > 0 }
 
       # retry_timeout == 60(sec), retry_secondary_threshold == 0.8
 

test/test_plugin_classes.rb

@@ -151,10 +151,10 @@ def initialize
       @emit_records_metrics = FluentTest::FluentTestCounterMetrics.new
       @emit_size_metrics = FluentTest::FluentTestCounterMetrics.new
       @write_count_metrics = FluentTest::FluentTestCounterMetrics.new
+      @write_secondary_count_metrics = FluentTest::FluentTestCounterMetrics.new
       @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
@@ -282,10 +282,10 @@ def initialize
       @emit_records_metrics = FluentTest::FluentTestCounterMetrics.new
       @emit_size_metrics = FluentTest::FluentTestCounterMetrics.new
       @write_count_metrics = FluentTest::FluentTestCounterMetrics.new
+      @write_secondary_count_metrics = FluentTest::FluentTestCounterMetrics.new
       @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)