From e8441f1d01ad0d7b973631277125e15b62934d40 Mon Sep 17 00:00:00 2001
From: Kentaro Hayashi <hayashi@clear-code.com>
Date: Wed, 14 Aug 2024 13:39:19 +0900
Subject: [PATCH] in_sample: support to renew record functionality

When in_sample plugin is used with filter parser which uses
remove_key_name_field, it raises the following error repeatedly.

  #0 dump an error event: error_class=ArgumentError error="message
  does not exist"

This kind of error occurs when key_name and remove_key_name_field
removes key from record with destructive change in filter parser.
It affects generated sample data.

To fix this issue, it is simple to just dup every record even though
it has a significant performance penalty.

Considering keeping compatibility and providing way to a workaround,
added option to enable previous behavior - reuse_record, disabled by default.

ref. https://github.com/fluent/fluentd/issues/4575

Here is the small benchmark:

<source>
    @type sample
    tag test
    size xxx
    rate 100
    reuse_record
</source>
<match test>
  @type null
</match>

size: 100000

 master:

   PID USER      PR  NI    VIRT    RES    SHR S  %CPU  %MEM     TIME+ COMMAND
   330767 kenhys    20   0  364316  81036  13620 S 100.3   0.1   0:52.10 ruby

 reuse_record: true

   PID USER      PR  NI    VIRT    RES    SHR S  %CPU  %MEM     TIME+ COMMAND
   333640 kenhys    20   0  364328  80956  13560 S 100.0   0.1   0:17.04 ruby

 reuse_record: false

   PID USER      PR  NI    VIRT    RES    SHR S  %CPU  %MEM     TIME+ COMMAND
   335843 kenhys    20   0  366188 113300  13536 S 100.3   0.2   0:17.24 ruby

Signed-off-by: Kentaro Hayashi <hayashi@clear-code.com>
---
 lib/fluent/plugin/in_sample.rb | 20 +++++++++++++-------
 1 file changed, 13 insertions(+), 7 deletions(-)

diff --git a/lib/fluent/plugin/in_sample.rb b/lib/fluent/plugin/in_sample.rb
index ccd918aa10..ec8eab841b 100644
--- a/lib/fluent/plugin/in_sample.rb
+++ b/lib/fluent/plugin/in_sample.rb
@@ -39,6 +39,8 @@ class SampleInput < Input
     config_param :auto_increment_key, :string, default: nil
     desc "The boolean to suspend-and-resume incremental value after restart"
     config_param :suspend, :bool, default: false,deprecated: 'This parameters is ignored'
+    desc "Reuse the sample data to reduce the load when sending large amounts of data. You can enable it if filter does not do destructive change"
+    config_param :reuse_record, :bool, default: false
     desc "The sample data to be generated. An array of JSON hashes or a single JSON hash."
     config_param :sample, alias: :dummy, default: [{"message" => "sample"}] do |val|
       begin
@@ -117,15 +119,19 @@ def emit(num)
       end
     end
 
-    def generate
-      d = @sample[@sample_index]
-      unless d
-        @sample_index = 0
-        d = @sample[@sample_index]
-      end
+    def next_sample
+      d = @reuse_record ? @sample[@sample_index] : @sample[@sample_index].dup
       @sample_index += 1
+      return d if d
+
+      @sample_index = 0
+      next_sample
+    end
+
+    def generate
+      d = next_sample
       if @auto_increment_key
-        d = d.dup
+        d = d.dup if @reuse_record
         d[@auto_increment_key] = @storage.update(:auto_increment_value){|v| v + 1 }
       end
       d