Handle partial CRI-O log format on CRI-O specific mode

In the newer kubernetes, it start to use CRI-O format for logging.
We should handle this format on concat plugin with specific mode.

ref: https://github.com/kubernetes/community/blob/master/contributors/design-proposals/node/kubelet-cri-logging.md

On kubelet implementation for parsing, stream type is just for checking
stream names validity and then they are thrown away.
Users want to obtain concatenated log contents not for attached metadata.

ref: https://github.com/kubernetes/kubernetes/blob/629d5ab21349021cf7d38236620785071ee541b4/pkg/kubelet/kuberuntime/logs/logs.go#L124-L168

When we use regexp to concat CRI-O log, it is hard to remove attached
metadata records such as stream names(stdout|stderr) and partial_flag(P
or F).

Signed-off-by: Hiroshi Hatake <[email protected]>

Author: cosmo0920

lib/fluent/plugin/filter_concat.rb

@@ -38,6 +38,12 @@ class ConcatFilter < Filter
     config_param :partial_metadata_format, :enum, list: [:"docker-fluentd", :"docker-journald", :"docker-journald-lowercase"], default: :"docker-fluentd"
     desc "If true, keep partial metadata"
     config_param :keep_partial_metadata, :bool, default: false
+    desc "Use cri log tag to concatenate multiple records"
+    config_param :use_partial_cri_logtag, :bool, default: false
+    desc "The key name that is referred to concatenate records on cri log"
+    config_param :partial_cri_logtag_key, :string, default: nil
+    desc "The key name that is referred to detect stream name on cri log"
+    config_param :partial_cri_stream_key, :string, default: "stream"
 
     class TimeoutError < StandardError
     end
@@ -55,8 +61,8 @@ def initialize
     def configure(conf)
       super
 
-      if @n_lines.nil? && @multiline_start_regexp.nil? && @multiline_end_regexp.nil? && @partial_key.nil? && !@use_partial_metadata
-        raise Fluent::ConfigError, "Either n_lines, multiline_start_regexp, multiline_end_regexp, partial_key or use_partial_metadata is required"
+      if @n_lines.nil? && @multiline_start_regexp.nil? && @multiline_end_regexp.nil? && @partial_key.nil? && !@use_partial_metadata && !@use_partial_cri_logtag
+        raise Fluent::ConfigError, "Either n_lines, multiline_start_regexp, multiline_end_regexp, partial_key, use_partial_metadata or use_partial_cri_logtag is required"
       end
       if @n_lines && (@multiline_start_regexp || @multiline_end_regexp)
         raise Fluent::ConfigError, "n_lines and multiline_start_regexp/multiline_end_regexp are exclusive"
@@ -79,6 +85,15 @@ def configure(conf)
       if @use_partial_metadata && @partial_key
         raise Fluent::ConfigError, "use_partial_metadata and partial_key are exclusive"
       end
+      if @use_partial_cri_logtag && @n_lines
+        raise Fluent::ConfigError, "use_partial_cri_logtag and n_lines are exclusive"
+      end
+      if @use_partial_cri_logtag && (@multiline_start_regexp || @multiline_end_regexp)
+        raise Fluent::ConfigError, "use_partial_cri_logtag and multiline_start_regexp/multiline_end_regexp are exclusive"
+      end
+      if @use_partial_cri_logtag && @partial_key
+        raise Fluent::ConfigError, "use_partial_cri_logtag and partial_key are exclusive"
+      end
 
       @mode = nil
       case
@@ -110,6 +125,11 @@ def configure(conf)
           @partial_last_field        = "container_partial_last".freeze
           @partial_message_indicator = @partial_id_field
         end
+      when @use_partial_cri_logtag
+        @mode = :partial_cri
+        @partial_logtag_delimiter = ":".freeze
+        @partial_logtag_continue = "P".freeze
+        @partial_logtag_full = "F".freeze
       when @multiline_start_regexp || @multiline_end_regexp
         @mode = :regexp
         if @multiline_start_regexp
@@ -175,6 +195,9 @@ def filter_stream(tag, es)
                   merged_record.delete(@partial_ordinal_field)
                   merged_record.delete(@partial_last_field)
                 end
+              when :partial_cri
+                merged_record.delete(@partial_cri_logtag_key) unless @keep_partial_key
+                merged_record.delete(@partial_cri_stream_key)
               end
               new_es.add(time, merged_record)
             end
@@ -220,6 +243,8 @@ def process(tag, time, record)
         process_partial(stream_identity, tag, time, record)
       when :partial_metadata
         process_partial_metadata(stream_identity, tag, time, record)
+      when :partial_cri
+        process_partial_cri(stream_identity, tag, time, record)
       when :regexp
         process_regexp(stream_identity, tag, time, record)
       end
@@ -248,6 +273,18 @@ def process_partial(stream_identity, tag, time, record)
       new_es
     end
 
+    def process_partial_cri(stream_identity, tag, time, record)
+      new_es = Fluent::MultiEventStream.new
+      @buffer[stream_identity] << [tag, time, record]
+      if record[@partial_cri_logtag_key].split(@partial_logtag_delimiter)[0] == @partial_logtag_full
+        new_time, new_record = flush_buffer(stream_identity)
+        time = new_time if @use_first_timestamp
+        new_record.delete(@partial_cri_logtag_key)
+        new_es.add(time, new_record)
+      end
+      new_es
+    end
+
     def process_partial_metadata(stream_identity, tag, time, record)
       new_es = Fluent::MultiEventStream.new
       @buffer[stream_identity] << [tag, time, record]

test/plugin/test_filter_concat.rb

@@ -59,7 +59,7 @@ def filter_with_time(conf, messages, wait: nil)
     end
 
     test "either" do
-      assert_raise(Fluent::ConfigError.new("Either n_lines, multiline_start_regexp, multiline_end_regexp, partial_key or use_partial_metadata is required")) do
+      assert_raise(Fluent::ConfigError.new("Either n_lines, multiline_start_regexp, multiline_end_regexp, partial_key, use_partial_metadata or use_partial_cri_logtag is required")) do
         create_driver(<<-CONFIG)
           key message
         CONFIG
@@ -125,6 +125,15 @@ def filter_with_time(conf, messages, wait: nil)
       CONFIG
       assert_equal(:regexp, d.instance.instance_variable_get(:@mode))
     end
+
+    test "use_partial_cri_logtag" do
+      d = create_driver(<<-CONFIG)
+        key message
+        use_partial_cri_logtag true
+        partial_cri_logtag_key logtag
+      CONFIG
+      assert_equal(:partial_cri, d.instance.instance_variable_get(:@mode))
+    end
   end
 
   sub_test_case "lines" do
@@ -1000,6 +1009,27 @@ def filter_with_time(conf, messages, wait: nil)
     end
   end
 
+  sub_test_case "CRI-O format log" do
+    test "filter with CRI-O style events" do
+      config = <<-CONFIG
+        key message
+        use_partial_cri_logtag true
+        partial_cri_logtag_key logtag
+      CONFIG
+      messages = [
+        {"stream" => "stdout", "logtag" => "F", "message" => "The content of the log entry 1"},
+        {"stream" => "stdout", "logtag" => "P", "message" => "First line of log entry 2"},
+        {"stream" => "stdout", "logtag" => "P", "message" => "Second line of the log entry 2"},
+        {"stream" => "stderr", "logtag" => "F", "message" => "Last line of the log entry 2"},
+      ]
+      filtered = filter(config, messages, wait: 3)
+      expected = [
+        {"message" => "The content of the log entry 1"},
+        {"message" => "First line of log entry 2\nSecond line of the log entry 2\nLast line of the log entry 2"}]
+      assert_equal(expected, filtered)
+    end
+  end
+
   sub_test_case "raise exception in on_timer" do
     # See also https://github.com/fluent/fluentd/issues/1946
     test "failed to flush timeout buffer" do