plugin: metrics: Simplify metrics code

Signed-off-by: Hiroshi Hatake <hatake@calyptia.com>
fluent · Aug 1, 2021 · aa79a4c · aa79a4c
1 parent daf62f8
commit aa79a4c
Show file tree

Hide file tree

Showing 6 changed files with 22 additions and 88 deletions.
diff --git a/lib/fluent/compat/output.rb b/lib/fluent/compat/output.rb
@@ -352,14 +352,8 @@ def handle_stream_simple(tag, es, enqueue: false)
           write_guard do
             @buffer.write({meta => data}, format: ->(_data){ _data }, size: ->(){ size }, enqueue: enqueue)
           end
-          if @enable_size_metrics
-            @counter_mutex.synchronize do
-              @emit_records_metrics.add(es.size)
-              @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-            end
-          else
-            @emit_records_metrics.add(es.size)
-          end
+          @emit_records_metrics.add(es.size)
+          @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
           return [meta]
         end
 
@@ -370,14 +364,8 @@ def handle_stream_simple(tag, es, enqueue: false)
           write_guard do
             @buffer.write({meta => bulk}, format: ->(_data){ _data }, size: ->(){ size }, enqueue: enqueue)
           end
-          if @enable_size_metrics
-            @counter_mutex.synchronize do
-              @emit_records_metrics.add(es.size)
-              @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-            end
-          else
-            @emit_records_metrics.add(es.size)
-          end
+          @emit_records_metrics.add(es.size)
+          @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
           return [meta]
         end
 
@@ -387,14 +375,8 @@ def handle_stream_simple(tag, es, enqueue: false)
         write_guard do
           @buffer.write({meta => data}, enqueue: enqueue)
         end
-        if @enable_size_metrics
-          @counter_mutex.synchronize do
-            @emit_records_metrics.add(es.size)
-            @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-          end
-        else
-          @emit_records_metrics.add(es.size)
-        end
+        @emit_records_metrics.add(es.size)
+        @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
         [meta]
       end
 

diff --git a/lib/fluent/plugin/bare_output.rb b/lib/fluent/plugin/bare_output.rb
@@ -91,14 +91,8 @@ def emit_sync(tag, es)
         @emit_count_metrics.inc
         begin
           process(tag, es)
-          if @enable_size_metrics
-            @counter_mutex.synchronize do
-              @emit_records_metrics.add(es.size)
-              @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-            end
-          else
-            @emit_records_metrics.add(es.size)
-          end
+          @emit_records_metrics.add(es.size)
+          @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
         rescue
           @num_errors_metrics.inc
           raise

diff --git a/lib/fluent/plugin/filter.rb b/lib/fluent/plugin/filter.rb
@@ -67,14 +67,8 @@ def statistics
       end
 
       def measure_metrics(es)
-        if @enable_size_metrics
-          @counter_mutex.synchronize do
-            @emit_records_metrics.add(es.size)
-            @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-          end
-        else
-          @emit_records_metrics.add(es.size)
-        end
+        @emit_records_metrics.add(es.size)
+        @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
       end
 
       def filter(tag, time, record)

diff --git a/lib/fluent/plugin/input.rb b/lib/fluent/plugin/input.rb
@@ -63,14 +63,8 @@ def statistics
       end
 
       def metric_callback(es)
-        if @enable_size_metrics
-          @counter_mutex.synchronize do
-            @emit_records_metrics.add(es.size)
-            @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-          end
-        else
-          @emit_records_metrics.add(es.size)
-        end
+        @emit_records_metrics.add(es.size)
+        @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
       end
 
       def multi_workers_ready?

diff --git a/lib/fluent/plugin/multi_output.rb b/lib/fluent/plugin/multi_output.rb
@@ -182,14 +182,8 @@ def emit_sync(tag, es)
         @emit_count_metrics.inc
         begin
           process(tag, es)
-          if @enable_size_metrics
-            @counter_mutex.synchronize do
-              @emit_records_metrics.add(es.size)
-              @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-            end
-          else
-            @emit_records_metrics.add(es.size)
-          end
+          @emit_records_metrics.add(es.size)
+          @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
         rescue
           @num_errors_metrics.inc
           raise

diff --git a/lib/fluent/plugin/output.rb b/lib/fluent/plugin/output.rb
@@ -836,14 +836,8 @@ def emit_sync(tag, es)
         @emit_count_metrics.inc
         begin
           process(tag, es)
-          if @enable_size_metrics
-            @counter_mutex.synchronize do
-              @emit_records_metrics.add(es.size)
-              @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-            end
-          else
-            @emit_records_metrics.add(es.size)
-          end
+          @emit_records_metrics.add(es.size)
+          @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
         rescue
           @num_errors_metrics.inc
           raise
@@ -1009,14 +1003,8 @@ def handle_stream_with_custom_format(tag, es, enqueue: false)
         write_guard do
           @buffer.write(meta_and_data, enqueue: enqueue)
         end
-        if @enable_size_metrics
-          @counter_mutex.synchronize do
-            @emit_records_metrics.add(es.size)
-            @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-          end
-        else
-          @emit_records_metrics.add(es.size)
-        end
+        @emit_records_metrics.add(es.size)
+        @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
         true
       end
 
@@ -1033,14 +1021,8 @@ def handle_stream_with_standard_format(tag, es, enqueue: false)
         write_guard do
           @buffer.write(meta_and_data, format: format_proc, enqueue: enqueue)
         end
-        if @enable_size_metrics
-          @counter_mutex.synchronize do
-            @emit_records_metrics.add(es.size)
-            @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-          end
-        else
-          @emit_records_metrics.add(es.size)
-        end
+        @emit_records_metrics.add(es.size)
+        @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
         true
       end
 
@@ -1065,14 +1047,8 @@ def handle_stream_simple(tag, es, enqueue: false)
         write_guard do
           @buffer.write({meta => data}, format: format_proc, enqueue: enqueue)
         end
-        if @enable_size_metrics
-          @counter_mutex.synchronize do
-            @emit_records_metrics.add(es.size)
-            @emit_size_metrics.add(es.to_msgpack_stream.bytesize)
-          end
-        else
-          @emit_records_metrics.add(es.size)
-        end
+        @emit_records_metrics.add(es.size)
+        @emit_size_metrics.add(es.to_msgpack_stream.bytesize) if @enable_size_metrics
         true
       end