Merge pull request #1079 from fluent/plugin-helper-compat-parameters-…

…parser-filter

Add parser&filter support for compat_parameters plugin helper

lib/fluent/compat/filter.rb

@@ -17,23 +17,43 @@
 require 'fluent/plugin'
 require 'fluent/plugin/filter'
 require 'fluent/compat/call_super_mixin'
+require 'fluent/compat/formatter_utils'
+require 'fluent/compat/parser_utils'
 
 module Fluent
   module Compat
     class Filter < Fluent::Plugin::Filter
       # TODO: warn when deprecated
 
+      helpers :inject
+
       def initialize
         super
         unless self.class.ancestors.include?(Fluent::Compat::CallSuperMixin)
           self.class.prepend Fluent::Compat::CallSuperMixin
         end
       end
 
+      def configure(conf)
+        ParserUtils.convert_parser_conf(conf)
+        FormatterUtils.convert_formatter_conf(conf)
+
+        super
+      end
+
       # These definitions are to get instance methods of superclass of 3rd party plugins
       # to make it sure to call super
       def start
         super
+
+        if instance_variable_defined?(:@formatter) && @inject_config
+          unless @formatter.class.ancestors.include?(Fluent::Compat::HandleTagAndTimeMixin)
+            if @formatter.respond_to?(:owner) && !@formatter.owner
+              @formatter.owner = self
+              @formatter.singleton_class.prepend FormatterUtils::InjectMixin
+            end
+          end
+        end
       end
 
       def before_shutdown

lib/fluent/compat/formatter.rb

@@ -16,6 +16,8 @@
 
 require 'fluent/plugin'
 require 'fluent/plugin/formatter'
+require 'fluent/compat/handle_tag_and_time_mixin'
+require 'fluent/compat/structured_format_mixin'
 
 require 'fluent/plugin/formatter_out_file'
 require 'fluent/plugin/formatter_stdout'
@@ -66,8 +68,8 @@ def self.create(conf)
         formatter
       end
 
-      HandleTagAndTimeMixin = Fluent::Plugin::Formatter::HandleTagAndTimeMixin
-      StructuredFormatMixin = Fluent::Plugin::Formatter::StructuredFormatMixin
+      HandleTagAndTimeMixin = Fluent::Compat::HandleTagAndTimeMixin
+      StructuredFormatMixin = Fluent::Compat::StructuredFormatMixin
 
       class ProcWrappedFormatter < Fluent::Plugin::ProcWrappedFormatter
         # TODO: warn when deprecated

lib/fluent/compat/formatter_utils.rb

@@ -0,0 +1,85 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+require 'fluent/plugin_helper/compat_parameters'
+
+module Fluent
+  module Compat
+    module FormatterUtils
+      INJECT_PARAMS = Fluent::PluginHelper::CompatParameters::INJECT_PARAMS
+      FORMATTER_PARAMS = Fluent::PluginHelper::CompatParameters::FORMATTER_PARAMS
+
+      module InjectMixin
+        def format(tag, time, record)
+          r = owner.inject_values_to_record(tag, time, record)
+          super(tag, time, r)
+        end
+      end
+
+      def self.convert_formatter_conf(conf)
+        return if conf.elements(name: 'inject').first || conf.elements(name: 'format').first
+
+        inject_params = {}
+        INJECT_PARAMS.each do |older, newer|
+          next unless newer
+          if conf.has_key?(older)
+            inject_params[newer] = conf[older]
+          end
+        end
+
+        if conf.has_key?('include_time_key') && Fluent::Config.bool_value(conf['include_time_key'])
+          inject_params['time_key'] ||= 'time'
+          inject_params['time_type'] ||= 'string'
+        end
+        if conf.has_key?('time_as_epoch') && Fluent::Config.bool_value(conf['time_as_epoch'])
+          inject_params['time_type'] = 'unixtime'
+        end
+        if conf.has_key?('localtime') || conf.has_key?('utc')
+          if conf.has_key?('localtime') && conf.has_key?('utc')
+            raise Fluent::ConfigError, "both of utc and localtime are specified, use only one of them"
+          elsif conf.has_key?('localtime')
+            inject_params['localtime'] = Fluent::Config.bool_value(conf['localtime'])
+          elsif conf.has_key?('utc')
+            inject_params['localtime'] = !(Fluent::Config.bool_value(conf['utc']))
+            # Specifying "localtime false" means using UTC in TimeFormatter
+            # And specifying "utc" is different from specifying "timezone +0000"(it's not always UTC).
+            # There are difference between "Z" and "+0000" in timezone formatting.
+            # TODO: add kwargs to TimeFormatter to specify "using localtime", "using UTC" or "using specified timezone" in more explicit way
+          end
+        end
+
+        if conf.has_key?('include_tag_key') && Fluent::Config.bool_value(conf['include_tag_key'])
+          inject_params['tag_key'] ||= 'tag'
+        end
+
+        unless inject_params.empty?
+          conf.elements << Fluent::Config::Element.new('inject', '', inject_params, [])
+        end
+
+        formatter_params = {}
+        FORMATTER_PARAMS.each do |older, newer|
+          next unless newer
+          if conf.has_key?(older)
+            formatter_params[newer] = conf[older]
+          end
+        end
+        unless formatter_params.empty?
+          conf.elements << Fluent::Config::Element.new('format', '', formatter_params, [])
+        end
+      end
+    end
+  end
+end

lib/fluent/compat/handle_tag_and_time_mixin.rb

@@ -0,0 +1,60 @@
+#
+# Fluentd
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#        http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+#
+
+module Fluent
+  module Compat
+    module HandleTagAndTimeMixin
+      def self.included(klass)
+        klass.instance_eval {
+          config_param :include_time_key, :bool, default: false
+          config_param :time_key, :string, default: 'time'
+          config_param :time_format, :string, default: nil
+          config_param :time_as_epoch, :bool, default: false
+          config_param :include_tag_key, :bool, default: false
+          config_param :tag_key, :string, default: 'tag'
+          config_param :localtime, :bool, default: true
+          config_param :timezone, :string, default: nil
+        }
+      end
+
+      def configure(conf)
+        super
+
+        if conf['utc']
+          @localtime = false
+        end
+        @timef = Fluent::TimeFormatter.new(@time_format, @localtime, @timezone)
+        if @time_as_epoch && !@include_time_key
+          log.warn "time_as_epoch will be ignored because include_time_key is false"
+        end
+      end
+
+      def filter_record(tag, time, record)
+        if @include_tag_key
+          record[@tag_key] = tag
+        end
+        if @include_time_key
+          if @time_as_epoch
+            record[@time_key] = time.to_i
+          else
+            record[@time_key] = @timef.format(time)
+          end
+        end
+      end
+    end
+  end
+end
+