Plugin Development

Search…
Plugin Development
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
filter for filter 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.)
Fluentd version and Plugin API
Fluentd now has two active versions, v0.12 and v1.0. v0.12 is old stable and many people still use this version. v1.0 is current stable version and this version has brand-new Plugin API.
The important point is plugin for v0.12 works with v0.12 and v1.0, but new v1.0 API based Plugin doesn't work with v0.12.
This document based on v0.12 Plugin API.
Send a patch or fork?
If you have a problem with existing plugins or new feature idea, sending a patch is better. If the plugin author is non-active, try to become new plugin maintainer first. Forking a plugin and release alternative plugin, e.g. fluent-plugin-xxx-alt, is final approach.
Plugin versioning policy
If you don't have a your policy, following semantic versioning is better.
​Semantic Versioning 2.0.0 - Semantic Versioning​
The important thing is if your plugin breaks the backward compatibiliy, update major version to avoid user troubles. For example, new v0.14 API based plugin doesn't work with fluentd v0.12. So if you release your existing plugin with new v0.14 API, please update major version, 0.5.0 -> 1.0.0, 1.2.0 -> 2.0.0.
Writing Input Plugins
Extend the Fluent::Input class and implement the following methods. In initialize, configure and start, super should be called to call Input plugin default behaviour.
1
require 'fluent/input'
2
​
3
module Fluent
4
  class SomeInput < Input
5
    # First, register the plugin. NAME is the name of this plugin
6
    # and identifies the plugin in the configuration file.
7
    Fluent::Plugin.register_input('NAME', self)
8
​
9
    # config_param defines a parameter. You can refer a parameter via @port instance variable
10
    # :default means this parameter is optional
11
    config_param :port, :integer, :default => 8888
12
​
13
    # This method is called before starting.
14
    # 'conf' is a Hash that includes configuration parameters.
15
    # If the configuration is invalid, raise Fluent::ConfigError.
16
    def configure(conf)
17
      super
18
​
19
      # You can also refer to raw parameter via conf[name].
20
      @port = conf['port']
21
      ...
22
    end
23
​
24
    # This method is called when starting.
25
    # Open sockets or files and create a thread here.
26
    def start
27
      super
28
      ...
29
    end
30
​
31
    # This method is called when shutting down.
32
    # Shutdown the thread and close sockets or files here.
33
    def shutdown
34
      super
35
      ...
36
    end
37
  end
38
end
Copied!
To submit events, use the router.emit(tag, time, record) method, where tag is the String, time is the UNIX time integer and record is a Hash object.
1
tag = "myapp.access"
2
time = Engine.now
3
record = {"message"=>"body"}
4
router.emit(tag, time, record)
Copied!
To submit multiple events in one call, use the router.emit_stream(tag, es) and MultiEventStream combo instead.
1
es = MultiEventStream.new
2
records.each { |record|
3
  es.add(time, record)
4
}
5
router.emit_stream(tag, es)
Copied!
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.
1
router.emit(tag, time, {'foo' => 'bar'})  # OK!
2
router.emit(tag, time, {:foo => 'bar'})   # NG!
Copied!
Writing Buffered Output Plugins
Extend the Fluent::BufferedOutput class and implement the following methods. In initialize, configure and start, super should be called to call BufferedOutput plugin default behaviour.
1
require 'fluent/output'
2
​
3
module Fluent
4
  class SomeOutput < BufferedOutput
5
    # First, register the plugin. NAME is the name of this plugin
6
    # and identifies the plugin in the configuration file.
7
    Fluent::Plugin.register_output('NAME', self)
8
​
9
    # config_param defines a parameter. You can refer a parameter via @path instance variable
10
    # Without :default, a parameter is required.
11
    config_param :path, :string
12
​
13
    # This method is called before starting.
14
    # 'conf' is a Hash that includes configuration parameters.
15
    # If the configuration is invalid, raise Fluent::ConfigError.
16
    def configure(conf)
17
      super
18
​
19
      # You can also refer raw parameter via conf[name].
20
      @path = conf['path']
21
      ...
22
    end
23
​
24
    # This method is called when starting.
25
    # Open sockets or files here.
26
    def start
27
      super
28
      ...
29
    end
30
​
31
    # This method is called when shutting down.
32
    # Shutdown the thread and close sockets or files here.
33
    def shutdown
34
      super
35
      ...
36
    end
37
​
38
    # This method is called when an event reaches to Fluentd.
39
    # Convert the event to a raw string.
40
    def format(tag, time, record)
41
      [tag, time, record].to_json + "\n"
42
      ## Alternatively, use msgpack to serialize the object.
43
      # [tag, time, record].to_msgpack
44
    end
45
​
46
    # This method is called every flush interval. Write the buffer chunk
47
    # to files or databases here.
48
    # 'chunk' is a buffer chunk that includes multiple formatted
49
    # events. You can use 'data = chunk.read' to get all events and
50
    # 'chunk.open {|io| ... }' to get IO objects.
51
    #
52
    # NOTE! This method is called by internal thread, not Fluentd's main thread. So IO wait doesn't affect other plugins.
53
    def write(chunk)
54
      data = chunk.read
55
      print data
56
    end
57
​
58
    ## Optionally, you can use chunk.msgpack_each to deserialize objects.
59
    #def write(chunk)
60
    #  chunk.msgpack_each {|(tag,time,record)|
61
    #  }
62
    #end
63
  end
64
end
Copied!
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.
1
require 'fluent/output'
2
​
3
module Fluent
4
  class SomeOutput < TimeSlicedOutput
5
    # configure(conf), start(), shutdown() and format(tag, time, record) are
6
    # the same as BufferedOutput.
7
    ...
8
​
9
    # You can use 'chunk.key' to get sliced time. The format of 'chunk.key'
10
    # can be configured by the 'time_format' option. The default format is %Y%m%d.
11
    def write(chunk)
12
      day = chunk.key
13
      ...
14
    end
15
  end
16
end
Copied!
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. In initialize, configure and start, super should be called to call non-buffered Output plugin default behaviour.
1
require 'fluent/output'
2
​
3
module Fluent
4
  class SomeOutput < Output
5
    # First, register the plugin. NAME is the name of this plugin
6
    # and identifies the plugin in the configuration file.
7
    Fluent::Plugin.register_output('NAME', self)
8
​
9
    # This method is called before starting.
10
    def configure(conf)
11
      super
12
      ...
13
    end
14
​
15
    # This method is called when starting.
16
    def start
17
      super
18
      ...
19
    end
20
​
21
    # This method is called when shutting down.
22
    def shutdown
23
      super
24
      ...
25
    end
26
​
27
    # This method is called when an event reaches Fluentd.
28
    # 'es' is a Fluent::EventStream object that includes multiple events.
29
    # You can use 'es.each {|time,record| ... }' to retrieve events.
30
    # 'chain' is an object that manages transactions. Call 'chain.next' at
31
    # appropriate points and rollback if it raises an exception.
32
    #
33
    # NOTE! This method is called by Fluentd's main thread so you should not write slow routine here. It causes Fluentd's performance degression.
34
    def emit(tag, es, chain)
35
      chain.next
36
      es.each {|time,record|
37
        log.info "OK!"
38
      }
39
    end
40
  end
41
end
Copied!
Filter Plugins
This section shows how to write custom filters in addition to the core filter plugins. The plugin files whose names start with "filter_" are registered as filter plugins.
Here is the implementation of the most basic filter that passes through all events as-is:
1
require 'fluent/filter'
2
​
3
module Fluent
4
  class PassThruFilter < Filter
5
    # Register this filter as "passthru"
6
    Fluent::Plugin.register_filter('passthru', self)
7
​
8
    # config_param works like other plugins
9
​
10
    def configure(conf)
11
      super
12
      # do the usual configuration here
13
    end
14
​
15
    def start
16
      super
17
      # This is the first method to be called when it starts running
18
      # Use it to allocate resources, etc.
19
    end
20
​
21
    def shutdown
22
      super
23
      # This method is called when Fluentd is shutting down.
24
      # Use it to free up resources, etc.
25
    end
26
​
27
    def filter(tag, time, record)
28
      # This method implements the filtering logic for individual filters
29
      # It is internal to this class and called by filter_stream unless
30
      # the user overrides filter_stream.
31
      #
32
      # Since our example is a pass-thru filter, it does nothing and just
33
      # returns the record as-is.
34
      # If returns nil, that records are ignored.
35
      record
36
    end
37
  end
38
end
Copied!
In initialize, configure, start and shutdown, super should be called to call Filter plugin default behaviour.
See Writing Input plugins section for the details of tag, time and record.
filter_stream method
Almost plugins could be implemented by overriding filter method. But if you want to mutate the event stream itself, you can override filter_stream method.
Here is the default implementation of filter_stream.
1
def filter_stream(tag, es)
2
  new_es = MultiEventStream.new
3
  es.each { |time, record|
4
    begin
5
      filtered_record = filter(tag, time, record)
6
      new_es.add(time, filtered_record) if filtered_record
7
    rescue => e
8
      router.emit_error_event(tag, time, record, e)
9
    end
10
  }
11
  new_es
12
end
Copied!
filter_stream should return EventStream object.
Parser Plugins
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:
1
<timestamp><SPACE>key1=value1<DELIMITER>key2=value2<DELIMITER>key3=value...
Copied!
e.g., something like this
1
2014-04-01T00:00:00 name=jake age=100 action=debugging
Copied!
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.
1
require 'fluent/parser'
2
​
3
module Fluent
4
  class TextParser
5
    class TimeKeyValueParser < Parser
6
      # Register this parser as "time_key_value"
7
      Plugin.register_parser("time_key_value", self)
8
​
9
      config_param :delimiter, :string, :default => " " # delimiter is configurable with " " as default
10
      config_param :time_format, :string, :default => nil # time_format is configurable
11
​
12
      # This method is called after config_params have read configuration parameters
13
      def configure(conf)
14
        super
15
​
16
        if @delimiter.length != 1
17
          raise ConfigError, "delimiter must be a single character. #{@delimiter} is not."
18
        end
19
​
20
        # TimeParser class is already given. It takes a single argument as the time format
21
        # to parse the time string with.
22
        @time_parser = TimeParser.new(@time_format)
23
      end
24
​
25
      # This is the main method. The input "text" is the unit of data to be parsed.
26
      # If this is the in_tail plugin, it would be a line. If this is for in_syslog,
27
      # it is a single syslog message.
28
      def parse(text)
29
        time, key_values = text.split(" ", 2)
30
        time = @time_parser.parse(time)
31
        record = {}
32
        key_values.split(@delimiter).each { |kv|
33
          k, v = kv.split("=", 2)
34
          record[k] = v
35
        }
36
        yield time, record
37
      end
38
    end
39
  end
40
end
Copied!
Then, save this code in parser_time_key_value.rb in a loadable plugin path. Then, if in_tail is configured as
1
# Other lines...
2
<source>
3
  @type tail
4
  path /path/to/input/file
5
  format time_key_value
6
</source>
Copied!
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.
1
# OK
2
parser.parse(text) { |time, record| ... }
3
# NG. This API will be removed
4
time, record = parser.parse(text) # or parser.call(text)
Copied!
Text Formatter Plugins
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.
1
require 'fluent/formatter'
2
​
3
module Fluent
4
  module TextFormatter
5
    class MyCSVFormatter < Formatter
6
      # Register MyCSVFormatter as "my_csv".
7
      Plugin.register_formatter("my_csv", self)
8
​
9
      include HandleTagAndTimeMixin # If you wish to use tag_key, time_key, etc.          
10
      config_param :csv_fields, :array, value_type: :string
11
​
12
      # This method does further processing. Configuration parameters can be
13
      # accessed either via "conf" hash or member variables.
14
      def configure(conf)
15
        super
16
      end
17
​
18
      # This is the method that formats the data output.
19
      def format(tag, time, record)
20
        values = []
21
​
22
        # Look up each required field and collect them from the record
23
        @csv_fields.each do |field|
24
          v = record[field]
25
          if not v
26
            $log.error "#{field} is missing."
27
          end
28
          values << v
29
        end
30
​
31
        # Output by joining the fields with a comma
32
        values.join(",")
33
      end
34
    end        
35
  end
36
end
Copied!
Then, save this code in formatter_my_csv.rb in a loadable plugin path. Then, if out_file is configured as
1
# Other lines...
2
<match test>
3
  @type file
4
  path /path/to/output/file
5
  format my_csv
6
  csv_fields k1,k2
7
</match>
Copied!
and if the record {"k1": 100, "k2": 200} is matched, the output file should look like 100,200
Error stream
router has emit_error_event API to rescue invalid events. Emitted events via emit_error_event are routed to @ERROR label.
There are several use cases:
Rescue invalid event which hard to apply filter routine, e.g. don't
have geoip target field.
Rescue invalid event which hard to serialize in the output, e.g.
can't convert a record into BSON.
API
1
 :::text
2
 router.emit_error_event(tag, time, record, error)
Copied!
tag: String: recommend to use incoming event tag
time: Integer: recommend to use incoming event time
record: Hash: recommend to use incoming event record
error: Exception: use a raised exception
config_param
config_param helper defines plugin parameter. You don't need to parse parameters manually. config_param syntax is config_param :name, :type, options. Here is simple example:
1
config_param :param1, :string
2
config_param :param2, :integer, default: 10
Copied!
In this example, param1 is required string parameter. If a user doesn't specify param1, fluentd raises an ConfigError. On the other hand, param2 is optional integer parameter. If a user doesn't specify param2, fluentd set 10 to param2 automatically. If a user sets "5" to param2 parameter, fluentd converts it into integer type automatically.
Access parameter value
config_param sets parsed result to :name instance variable after configure call. See example below:
1
config_param :param, :string
2
​
3
def configure(conf)
4
  super # This super is needed to parse conf by config_param
5
​
6
  p @param # You can access parsed result via instance variable. :param is used for variable name
7
end
Copied!
Supported types
Fluentd supports following built-in types for plugin parameter:
1
# hello, /path/to/file, etc
2
config_param :str_param, :string
3
# -1, 100, 100000, etc
4
config_param :int_param, :integer
5
# 0.1, 999.9, etc
6
config_param :float_param, :float
7
# true: yes / true, false: no / false
8
config_param :bool_param, :bool
9
# json object: {"k":"v"}, {"k", 10}
10
config_param :hash_param, :hash
11
# json array: [1, 10], ["foo", "bar"]
12
config_param :array_param, :array, value_type: :integer
13
# integer with size unit: 1k, 2m, 3g, 4t
14
config_param :size_param, :size
15
# integer with time unit: 1s, 2m, 3h, 4d
16
config_param :time_param, :time
17
# fixed list with `list` option: [:text, :gzip]
18
config_param :enum_param, :enum, list: [:tcp, :udp]
Copied!
Supported options
default: Specified parameter becomes optional. type value or
nil are available: default: 10, default: nil.
secret: Specified parameter is masked when dump a configuration,
e.g. start logs, in_monitor_agent result: secret: true
deprecated: Specified parameter is showed in warning log. Need
deprecated message for value: deprecated: "Use xxx instead"
obsoleted: Specified parameter is showed in error log with
configuration error. Need obsoleted message for value:
obsoleted: "Use xxx instead"
These options can be combined.
1
config_param :param, :array, default: [1, 2], secret: true, deprecated: "Use new_param instead"
Copied!
Debugging plugins
Run fluentd with the -vv option to show debug messages:
1
$ fluentd -vv
Copied!
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:
1
# You want to debug this plugin.
2
<source>
3
  @type your_custom_input_plugin
4
</source>
5
​
6
# Dump all events to stdout.
7
<match **>
8
  @type stdout
9
</match>
Copied!
The copy output plugin copies matched events to multiple output plugins. You can use it in conjunction with the stdout plugin:
1
<source>
2
  @type forward
3
</source>
4
​
5
# Use the forward Input plugin and the fluent-cat command to feed events:
6
#  $ echo '{"event":"message"}' | fluent-cat test.tag
7
<match test.tag>
8
  @type copy
9
​
10
  # Dump the matched events.
11
  <store>
12
    @type stdout
13
  </store>
14
​
15
  # Feed the dumped events to your plugin.
16
  <store>
17
    @type your_custom_output_plugin
18
  </store>
19
</match>
Copied!
You can use stdout filter instead of copy and stdout combination. The result is same as above but more simpler.
1
<source>
2
  @type forward
3
</source>
4
​
5
<filter>
6
  @type stdout
7
</filter>
8
​
9
<match test.tag>
10
  @type your_custom_output_plugin
11
</match>
Copied!
Writing test cases
Fluentd provides unit test frameworks for plugins:
1
Fluent::Test::InputTestDriver
2
  Test driver for input plugins.
3
​
4
Fluent::Test::BufferedOutputTestDriver
5
  Test driver for buffered output plugins.
6
​
7
Fluent::Test::OutputTestDriver
8
  Test driver for non-buffered output plugins.
Copied!
Please see Fluentd's source code for details.
Run test
Fluentd test follows standard gem way and uses test-unit library. Use rake command.
1
$ bundle install --path vendor/bundle # Install related libraries.
2
$ bundle exec rake test
Copied!
If you want to run only one file, use TEST environment variable:
1
$ bundle exec rake test TEST=test/plugin/test_out_foo.rb
Copied!
Further Reading
​Slides: Dive into Fluentd Plugin​
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). All components are available under the Apache 2 License.
Developer
Community
Last modified 2yr ago