Fluentd parser detection with Kubernetes annotations

filter_detect_parser.rb

require 'fluent/plugin/filter'

# This filter detects the parser to use based on Kubernetes annotations.
#
# Annotation: fluentd.org/parser[_stream][-container]
#
# Suggest a pre-defined parser. The parser must be already registered by Fluentd. If present,
# the stream (stdout or stderr) will restrict that specific stream. If present, the container can
# override a specific container in a Pod.
#
# Annotation: fluentd.org/exclude
#
# Request to Fluentd to exclude or not the logs generated by the Pod. Must be
# a true or false (default) string value.
#
# Author: Pieter Vogelaar - pietervogelaar.nl
#
module Fluent::Plugin
  class DetectParserFilter < Filter
    Fluent::Plugin.register_filter('detect_parser', self)

    # config_param works like other plugins

    def configure(conf)
      super
      # do the usual configuration here
    end

    def filter(tag, time, record)
      if record.key?("kubernetes") and record["kubernetes"].key?("annotations") and
        annotations = record["kubernetes"]["annotations"]

        if annotations.key?("fluentd_org/exclude") and
            annotations["fluentd_org/exclude"] == 'true' then
          nil
        else
          container_name = ''
          annotations.each do |annotation, annotation_value|
            if match = annotation.match(/fluentd_org\/parser(?:_stdout-|_stderr-)?-?(.*)/i)
              if record["kubernetes"]["container_name"] == match.captures[0]
                container_name = match.captures[0]
                break
              end
            end
          end

          stream = record["stream"]

          if record["kubernetes"]["container_name"] == container_name
            if annotations.key?("fluentd_org/parser_#{stream}-#{container_name}") then
              record["detected_parser"] = annotations["fluentd_org/parser_#{stream}-#{container_name}"]
            elsif annotations.key?("fluentd_org/parser-#{container_name}") then
              record["detected_parser"] = annotations["fluentd_org/parser-#{container_name}"]
            elsif annotations.key?("fluentd_org/parser_#{stream}") then
              record["detected_parser"] = annotations["fluentd_org/parser_#{stream}"]
            elsif annotations.key?("fluentd_org/parser") then
              record["detected_parser"] = annotations["fluentd_org/parser"]
            end
          else
            if annotations.key?("fluentd_org/parser_#{stream}") then
              record["detected_parser"] = annotations["fluentd_org/parser_#{stream}"]
            elsif annotations.key?("fluentd_org/parser") then
              record["detected_parser"] = annotations["fluentd_org/parser"]
            end
          end

          record
        end
      else
        record
      end
    end
  end
end

The documentation at Fluent-bit (https://github.com/fluent/fluent-bit-docs/blob/master/filter/kubernetes.md#kubernetes-annotations) gave me the idea to built this for Fluentd.