add docker 20.10 partial metadata improvements

Author: lobeck

NEWS.md

@@ -1,5 +1,11 @@
 # ChangeLog
 
+## next
+
+### Improvements
+
+* adds `partial_metadata_format` to support Docker 20.10 journald log driver improvements [moby/moby#41407](https://github.com/moby/moby/pull/41407) 
+
 ## v2.3.0
 
 ### Improvements

README.md

@@ -97,6 +97,17 @@ Use partial metadata to concatenate multiple records
 
 If true, keep partial metadata
 
+**partial\_metadata\_format** (string) (optional)
+
+Configure based on the input plugin, that is used.
+The docker fluentd and journald log drivers are behaving differently, so the plugin needs to know, what to look for.
+Use `docker-journald-lowercase`, if you have `fields_lowercase true` in the `journald` source config
+
+Valid options: `docker-fluentd`, `docker-journald`, `docker-journald-lowercase`
+
+Default value is `docker-fluentd`
+
+
 ## Usage
 
 Every 10 events will be concatenated into one event.
@@ -186,6 +197,20 @@ Handle Docker logs splitted in several parts (using `partial_message`), and do n
 </filter>
 ```
 
+(Docker v20.10+) Handle Docker logs splitted in several parts (using `use_partial_metadata`), and do not add new line between parts.
+
+Docker v20.10 improved partial message handling by adding better metadata in the journald log driver, this works now similarily to the fluentd log driver, but requires one additional setting
+
+```aconf
+<filter>
+  @type concat
+  key log
+  use_partial_metadata true
+  partial_metadata_format docker-journald
+  separator ""
+</filter>
+```
+
 Handle Docker logs splitted in several parts (using newline detection), and do not add new line between parts (prior to Docker 18.06).
 
 ```aconf

lib/fluent/plugin/filter_concat.rb

@@ -34,6 +34,8 @@ class ConcatFilter < Filter
     config_param :keep_partial_key, :bool, default: false
     desc "Use partial metadata to concatenate multiple records"
     config_param :use_partial_metadata, :bool, default: false
+    desc "Input format of the partial metadata (fluentd or journald docker log driver)"
+    config_param :partial_metadata_format, :string, default: "docker-fluentd"
     desc "If true, keep partial metadata"
     config_param :keep_partial_metadata, :bool, default: false
 
@@ -77,6 +79,9 @@ def configure(conf)
       if @use_partial_metadata && @partial_key
         raise Fluent::ConfigError, "use_partial_metadata and partial_key are exclusive"
       end
+      unless ["docker-fluentd", "docker-journald", "docker-journald-lowercase"].include?(@partial_metadata_format)
+        raise Fluent::ConfigError, "partial_metadata_format only supports docker-fluentd, docker-journald or docker-journald-lowercase"
+      end
 
       @mode = nil
       case
@@ -86,6 +91,28 @@ def configure(conf)
         @mode = :partial
       when @use_partial_metadata
         @mode = :partial_metadata
+        case @partial_metadata_format
+        when "docker-fluentd"
+          @partial_message_field     = "partial_message".freeze
+          @partial_id_field          = "partial_id".freeze
+          @partial_ordinal_field     = "partial_ordinal".freeze
+          @partial_last_field        = "partial_last".freeze
+          @partial_message_indicator = @partial_message_field
+        when "docker-journald"
+          @partial_message_field     = "CONTAINER_PARTIAL_MESSAGE".freeze
+          @partial_id_field          = "CONTAINER_PARTIAL_ID".freeze
+          @partial_ordinal_field     = "CONTAINER_PARTIAL_ORDINAL".freeze
+          @partial_last_field        = "CONTAINER_PARTIAL_LAST".freeze
+          # the journald log driver does not add CONTAINER_PARTIAL_MESSAGE to the last message
+          # so we help ourself by using another indicator
+          @partial_message_indicator = @partial_id_field
+        when "docker-journald-lowercase"
+          @partial_message_field     = "container_partial_message".freeze
+          @partial_id_field          = "container_partial_id".freeze
+          @partial_ordinal_field     = "container_partial_ordinal".freeze
+          @partial_last_field        = "container_partial_last".freeze
+          @partial_message_indicator = @partial_id_field
+        end
       when @multiline_start_regexp || @multiline_end_regexp
         @mode = :regexp
         if @multiline_start_regexp
@@ -130,7 +157,7 @@ def filter_stream(tag, es)
           end
         end
         if @mode == :partial_metadata
-          unless record.key?("partial_message".freeze)
+          unless record.key?(@partial_message_indicator)
             new_es.add(time, record)
             next
           end
@@ -146,10 +173,10 @@ def filter_stream(tag, es)
                 merged_record.delete(@partial_key) unless @keep_partial_key
               when :partial_metadata
                 unless @keep_partial_metadata
-                  merged_record.delete("partial_message".freeze)
-                  merged_record.delete("partial_id".freeze)
-                  merged_record.delete("partial_ordinal".freeze)
-                  merged_record.delete("partial_last".freeze)
+                  merged_record.delete(@partial_message_field)
+                  merged_record.delete(@partial_id_field)
+                  merged_record.delete(@partial_ordinal_field)
+                  merged_record.delete(@partial_last_field)
                 end
               end
               new_es.add(time, merged_record)
@@ -175,9 +202,9 @@ def on_timer
     def process(tag, time, record)
       if @mode == :partial_metadata
         if @stream_identity_key
-          stream_identity = %Q(#{tag}:#{record[@stream_identity_key]}#{record["partial_id"]})
+          stream_identity = %Q(#{tag}:#{record[@stream_identity_key]}#{record[@partial_id_field]})
         else
-          stream_identity = %Q(#{tag}:#{record["partial_id"]})
+          stream_identity = %Q(#{tag}:#{record[@partial_id_field]})
         end
       else
         if @stream_identity_key
@@ -227,7 +254,7 @@ def process_partial(stream_identity, tag, time, record)
     def process_partial_metadata(stream_identity, tag, time, record)
       new_es = Fluent::MultiEventStream.new
       @buffer[stream_identity] << [tag, time, record]
-      if record["partial_last"] == "true"
+      if record[@partial_last_field] == "true"
         new_time, new_record = flush_buffer(stream_identity)
         time = new_time if @use_first_timestamp
         new_record.delete(@partial_key)
@@ -306,7 +333,7 @@ def continuous_line?(text)
     def flush_buffer(stream_identity, new_element = nil)
       lines = if @mode == :partial_metadata
                 @buffer[stream_identity]
-                  .sort_by {|_tag, _time, record| record["partial_ordinal"].to_i }
+                  .sort_by {|_tag, _time, record| record[@partial_ordinal_field].to_i }
                   .map {|_tag, _time, record| record[@key] }
               else
                 @buffer[stream_identity].map {|_tag, _time, record| record[@key] }