Merge pull request #106 from fluent-plugins-nursery/handle-CRI-O-log-on-partial_cri-mode

Handle CRI-O log on partial_cri mode

Author: cosmo0920

README.md

@@ -196,7 +196,8 @@ Handle containerd/cri in Kubernetes.
   path /var/log/containers/*.log
   <parse>
     @type regexp
-    expression /^(?<timestamp>\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z) (?<output>\w+) (?<partial_flag>[FP]) (?<message>.+)$/
+    expression /^(?<time>[^ ]+) (?<stream>stdout|stderr) (?<logtag>[^ ]*) (?<message>.*)$/
+    time_format %Y-%m-%dT%H:%M:%S.%L%z
   </parse>
   tag k8s
   @label @CONCAT
@@ -206,8 +207,9 @@ Handle containerd/cri in Kubernetes.
   <filter k8s>
     @type concat
     key message
-    partial_key partial_flag
-    partial_value P
+    use_partial_cri_logtag true
+    partial_cri_logtag_key logtag
+    partial_cri_stream_key stream
   </filter>
   <match k8s>
     @type relabel

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
@@ -52,11 +58,18 @@ def initialize
       end
     end
 
+    def required_params
+      params = [@n_lines.nil?, @multiline_start_regexp.nil?, @multiline_end_regexp.nil?, @partial_key.nil?, !@use_partial_metadata, !@use_partial_cri_logtag]
+      names = ["n_lines", "multiline_start_regexp", "multiline_end_regexp", "partial_key", "use_partial_metadata", "use_partial_cri_logtag"]
+      return params, names
+    end
+
     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"
+      params, names = required_params
+      if params.all?
+        raise Fluent::ConfigError, "Either #{[names[0..-2].join(", "), names[-1]].join(" or ")} 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 +92,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 +132,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 +202,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 +250,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 +280,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