Versions | v0.14 (td-agent3) | v0.12 (td-agent2) | v0.10 (td-agent1)

This page is for v0.10, not the latest stable version which is v0.12. For the latest stable version of this article, click here.


Writing plugins

Table of Contents

Installing custom plugins

To install a plugin, please put the ruby script in the /etc/fluent/plugin directory.

Alternatively, you can create a Ruby Gem package that includes a lib/fluent/plugin/<TYPE>_<NAME>.rb file. The TYPE is:

  • in for input plugins
  • out for output plugins
  • buf for buffer plugins
  • parser for parser plugins
  • formatter for formatter plugins

For example, an email Output plugin would have the path: lib/fluent/plugin/out_mail.rb. The packaged gem can be distributed and installed using RubyGems. For further information, please see the list of Fluentd plugins for third-party plugins.

Overview

The following slides can help the user understand how Fluentd works before they dive into writing their own plugins.

(The slides are taken from Naotoshi Seo’s RubyKaigi 2014 talk.)

Writing Input Plugins

Extend the Fluent::Input class and implement the following methods.

module Fluent
  class SomeInput < Input
    # First, register the plugin. NAME is the name of this plugin
    # and identifies the plugin in the configuration file.
    Fluent::Plugin.register_input('NAME', self)

    # config_param defines a parameter. You can refer a parameter via @port instance variable
    # :default means this parameter is optional
    config_param :port, :integer, :default => 8888

    # This method is called before starting.
    # 'conf' is a Hash that includes configuration parameters.
    # If the configuration is invalid, raise Fluent::ConfigError.
    def configure(conf)
      super

      # You can also refer to raw parameter via conf[name].
      @port = conf['port']
      ...
    end

    # This method is called when starting.
    # Open sockets or files and create a thread here.
    def start
      super
      ...
    end

    # This method is called when shutting down.
    # Shutdown the thread and close sockets or files here.
    def shutdown
      ...
    end
  end
end

In initialize, configure and start, super should be called to call Input plugin default behaviour.

To submit events, use the Fluent::Engine.emit(tag, time, record) method, where tag is the String, time is the UNIX time integer and record is a Hash object.

tag = "myapp.access"
time = Engine.now
record = {"message"=>"body"}
Fluent::Engine.emit(tag, time, record)

Record format

Fluentd plugins assume the record is a JSON so the key should be the String, not Symbol. If you emit a symbol keyed record, it may cause a problem.

Fluent::Engine.emit(tag, time, {'foo' => 'bar'})  # OK!
Fluent::Engine.emit(tag, time, {:foo => 'bar'})   # NG!

Writing Buffered Output Plugins

Extend the Fluent::BufferedOutput class and implement the following methods.

module Fluent
  class SomeOutput < BufferedOutput
    # First, register the plugin. NAME is the name of this plugin
    # and identifies the plugin in the configuration file.
    Fluent::Plugin.register_output('NAME', self)

    # config_param defines a parameter. You can refer a parameter via @path instance variable
    # Without :default, a parameter is required.
    config_param :path, :string

    # This method is called before starting.
    # 'conf' is a Hash that includes configuration parameters.
    # If the configuration is invalid, raise Fluent::ConfigError.
    def configure(conf)
      super

      # You can also refer raw parameter via conf[name].
      @path = conf['path']
      ...
    end

    # This method is called when starting.
    # Open sockets or files here.
    def start
      super
      ...
    end

    # This method is called when shutting down.
    # Shutdown the thread and close sockets or files here.
    def shutdown
      super
      ...
    end

    # This method is called when an event reaches to Fluentd.
    # Convert the event to a raw string.
    def format(tag, time, record)
      [tag, time, record].to_json + "\n"
      ## Alternatively, use msgpack to serialize the object.
      # [tag, time, record].to_msgpack
    end

    # This method is called every flush interval. Write the buffer chunk
    # to files or databases here.
    # 'chunk' is a buffer chunk that includes multiple formatted
    # events. You can use 'data = chunk.read' to get all events and
    # 'chunk.open {|io| ... }' to get IO objects.
    #
    # NOTE! This method is called by internal thread, not Fluentd's main thread. So IO wait doesn't affect other plugins.
    def write(chunk)
      data = chunk.read
      print data
    end

    ## Optionally, you can use chunk.msgpack_each to deserialize objects.
    #def write(chunk)
    #  chunk.msgpack_each {|(tag,time,record)|
    #  }
    #end
  end
end

In initialize, configure and start, super should be called to call BufferedOutput plugin default behaviour.

Writing Time Sliced Output Plugins

Time Sliced Output plugins are extended versions of buffered output plugins. One example of a time sliced output is the out_file plugin.

Note that Time Sliced Output plugins use file buffer by default. Thus the buffer_path option is required.

To implement a Time Sliced Output plugin, extend the Fluent::TimeSlicedOutput class and implement the following methods.

module Fluent
  class SomeOutput < TimeSlicedOutput
    # configure(conf), start(), shutdown() and format(tag, time, record) are
    # the same as BufferedOutput.
    ...

    # You can use 'chunk.key' to get sliced time. The format of 'chunk.key'
    # can be configured by the 'time_format' option. The default format is %Y%m%d.
    def write(chunk)
      day = chunk.key
      ...
    end
  end
end

Writing Non-buffered Output Plugins

Extend the Fluent::Output class and implement the following methods. Output plugin is often used for implementing filter like plugin.

module Fluent
  class SomeOutput < Output
    # First, register the plugin. NAME is the name of this plugin
    # and identifies the plugin in the configuration file.
    Fluent::Plugin.register_output('NAME', self)

    # This method is called before starting.
    def configure(conf)
      super
      ...
    end

    # This method is called when starting.
    def start
      super
      ...
    end

    # This method is called when shutting down.
    def shutdown
      super
      ...
    end

    # This method is called when an event reaches Fluentd.
    # 'es' is a Fluent::EventStream object that includes multiple events.
    # You can use 'es.each {|time,record| ... }' to retrieve events.
    # 'chain' is an object that manages transactions. Call 'chain.next' at
    # appropriate points and rollback if it raises an exception.
    #
    # NOTE! This method is called by Fluentd's main thread so you should not write slow routine here. It causes Fluentd's performance degression.
    def emit(tag, es, chain)
      chain.next
      es.each {|time,record|
        $stderr.puts "OK!"
      }
    end
  end
end

In initialize, configure and start, super should be called to call non-buffered Output plugin default behaviour.

Parser Plugins

Starting with v0.10.46, Fluentd supports pluggable, customizable formats for input plugins. The plugin files whose names start with “parser_” are registered as Parser Plugins.

Here is an example of a custom parser that parses the following newline-delimited log format:

<timestamp><SPACE>key1=value1<DELIMITER>key2=value2<DELIMITER>key3=value...

e.g., something like this

2014-04-01T00:00:00 name=jake age=100 action=debugging

While it is not hard to write a regular expression to match this format, it is tricky to extract and save key names.

Here is the code to parse this custom format (let’s call it time_key_value). It takes one optional parameter called delimiter, which is the delimiter for key-value pairs. It also takes time_format to parse the time string. In initialize, configure and start, super should be called to call Parser plugin default behaviour.

module Fluent
  class TextParser
    class TimeKeyValueParser < Parser
      # Register this parser as "time_key_value"
      Plugin.register_parser("time_key_value", self)

      config_param :delimiter, :string, :default => " " # delimiter is configurable with " " as default
      config_param :time_format, :string, :default => nil # time_format is configurable
      # This method is called after config_params have read configuration parameters
      def configure(conf)
        super

        if @delimiter.length != 1
          raise ConfigError, "delimiter must be a single character. #{@delimiter} is not."
        end

        # TimeParser class is already given. It takes a single argument as the time format
        # to parse the time string with.
        @time_parser = TimeParser.new(@time_format)
      end

      # This is the main method. The input "text" is the unit of data to be parsed.
      # If this is the in_tail plugin, it would be a line. If this is for in_syslog,
      # it is a single syslog message.
      def call(text)
        time, key_values = text.split(" ", 2)
        time = @time_parser.parse(time)
        record = {}
        key_values.split(@delimiter).each { |kv|
          k, v = kv.split("=", 2)
          record[k] = v
        }
        yield time, record
      end
    end
  end
end

Then, save this code in parser_time_key_value.rb in a loadable plugin path. Then, if in_tail is configured as

# Other lines...
<source>
  type tail
  path /path/to/input/file
  format time_key_value
</source>

Then, the log line like 2014-01-01T00:00:00 k=v a=b is parsed as 2013-01-01 00:00:00 +0000 test: {"k":"v","a":"b"}.

Parser API

Current Parser#parse API is called with block. We will remove Parser#parse with return value API since v0.14 or later.

# OK
parser.parse(text) { |time, record| ... }
# NG. This API will be removed
time, record = parser.parse(text) # or parser.call(text)

Text Formatter Plugins

Starting with v0.10.49, Fluentd supports pluggable, customizable formats for output plugins. The plugin files whose names start with “formatter_” are registered as Formatter Plugins.

Here is an example of a custom formatter that outputs events as CSVs. It takes a required parameter called “csv_fields” and outputs the fields. It assumes that the values of the fields are already valid CSV fields. In initialize, configure and start, super should be called to call Formatter plugin default behaviour.

module Fluent
  module TextFormatter
    class MyCSVFormatter < Formatter
      # Register MyCSVFormatter as "my_csv".
      Plugin.register_formatter("my_csv", self)

      include HandleTagAndTimeMixin # If you wish to use tag_key, time_key, etc.          
      config_param :csv_fields, :string

      # This method does further processing. Configuration parameters can be
      # accessed either via "conf" hash or member variables.
      def configure(conf)
        super
        @csv_fields = conf['csv_fields'].split(",")
      end

      # This is the method that formats the data output.
      def format(tag, time, record)
        values = []

        # Look up each required field and collect them from the record
        @csv_fields.each do |field|
          v = record[field]
          if not v
            $log.error "#{field} is missing."
          end
          values << v
        end

        # Output by joining the fields with a comma
        values.join(",")
      end
    end        
  end
end

Then, save this code in formatter_my_csv.rb in a loadable plugin path. Then, if out_file is configured as

# Other lines...
<match test>
  type file
  path /path/to/output/file
  format my_csv
  csv_fields k1,k2
</match>

and if the record {"k1": 100, "k2": 200} is matched, the output file should look like 100,200

Logging

In the past, Fluentd used $log objects to output logs. Fluentd (v0.10.43 and above) now provides the log method to support the plugin-specific log_level parameter. Newer plugins should use the log method instead of $log objects:

log.info "setup foo plugin"

Supporting Older Fluentd Versions

The log method only works with Fluentd v0.10.43 or above. Please use the following code in your plugin to support older Fluentd versions:

module Fluent
  module FooPluginOutput < Output
    # Define `log` method for v0.10.42 or earlier
    unless method_defined?(:log)
      define_method(:log) { $log }
    end
    ...
  end
end

This code defines the log method using $log when the log method is not defined, so log.error becomes $log.error for older Fluentd versions.

Debugging plugins

Run fluentd with the -vv option to show debug messages:

$ fluentd -vv

The stdout and copy output plugins are useful for debugging. The stdout output plugin dumps matched events to the console. It can be used as follows:

# You want to debug this plugin.
<source>
  type your_custom_input_plugin
</source>

# Dump all events to stdout.
<match **>
  type stdout
</match>

The copy output plugin copies matched events to multiple output plugins. You can use it in conjunction with the stdout plugin:

<source>
  type tcp
</source>

# Use the tcp Input plugin and the fluent-cat command to feed events:
#  $ echo '{"event":"message"}' | fluent-cat test.tag
<match test.tag>
  type copy

  # Dump the matched events.
  <store>
    type stdout
  </store>

  # Feed the dumped events to your plugin.
  <store>
    type your_custom_output_plugin
  </store>
</match>

Writing test cases

Fluentd provides unit test frameworks for plugins:

Fluent::Test::InputTestDriver
  Test driver for input plugins.

Fluent::Test::BufferedOutputTestDriver
  Test driver for buffered output plugins.

Fluent::Test::OutputTestDriver
  Test driver for non-buffered output plugins.

Please see Fluentd’s source code for details.

Further Reading

Last updated: 2016-06-06 04:47:13 UTC

Versions | v0.14 (td-agent3) | v0.12 (td-agent2) | v0.10 (td-agent1)

If this article is incorrect or outdated, or omits critical information, please let us know. Fluentd is a open source project under Cloud Native Computing Foundation (CNCF), originally invented by Treasure Data, Inc. All components are available under the Apache 2 License.

Interested in the Fluentd Newsletters?