Dive into Fluentd plugin v0.12

Deprecated: Function get_magic_quotes_gpc() is deprecated in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 99

Deprecated: The each() function is deprecated. This message will be suppressed on further calls in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 619

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1169

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176

Warning: Cannot modify header information - headers already sent by (output started at /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php:99) in /hermes/walnacweb04/walnacweb04ab/b2791/pow.jasaeld/htdocs/De1337/nothing/index.php on line 1176
Dive into Fluentd plugin v0.12

Site:
repeatedly.github.com
Company:
Treasure Data Inc.
Love plugins:
input: tail 
buffer: memory 
output: mongo

developed by
The missing log collector
What's Fluentd?

Fluentd is a
buffer
router
collector
converter
aggregator
etc...

... but,
Fluentd doesn’t have such features as a built-in.

Instead,
Fluentd has flexible plugin architecture which
consists of Input, Filter, Output and Buffer.

We can customize Fluentd using plugins :)

Agenda: This slide
talk about
- an example of Fluentd plugins
- Fluentd and libraries
- how to develop a Fluentd plugin
do...

See also: 
http://docs.fluentd.org/articles/plugin-development

Example
based on bit.ly/fluentd-with-mongo

Install
Plugin name is ,
and fluent-gem is included in Fluentd gem.
fluent-plugin-xxx

Let’s type!
$ fluent-gem install fluent-plugin-mongo

<source>
type tail
format apache
path /path/to/log
tag mongo.apache
</source>
<match mongo.**>
type mongo
database apache
...

Start!
$ fluentd -c fluentd.conf
2015-10-04 00:00:14 +0900: starting fluentd-0.12.16
2015-10-04 00:00:14 +0900: reading co...

Attack!
$ ab -n 100 -c 10 http://localhost/

$ mongo --host otherhost
> use apache
> db.access.find()
{
"type": "127.0.0.1",
"method": "GET",
"path": "/",
"code": "200...

Mongo
Apache
Fluentd
write
tail
insert
I’m a log!
event
buffering

Ruby
Fluentd Stack
OS
Cool.io
MessagePack
Buffer
Input
Outpu

Fluentd and plugins are written in Ruby.

Note that
Fluentd v0.14 works on Ruby 2.1 or later.
We will remove 1.9 or 2.0 support.

Serialization:
JSON like fast and compact format.
RPC:
Async and parallelism for high performance.
IDL:
Easy to integrate ...

Binary format,
Header + Body,
and
Variable length.

Note that Ruby version can’t handle
a Time object natively.

So, we use an Integer object , second unit,
instead of a Time for now.

Since v0.14, Fluentd supports nano-second
unit with new EventTime object. 
https://github.com/fluent/fluentd/pull/653

Source:
github.com/msgpack
Wiki:
wiki.msgpack.org/display/MSGPACK
Mailing List:
groups.google.com/group/msgpack

Event driven framework built on top of libev.

Cool.io has Loop and Watchers with
Transport wrappers.
Lots of input plugins use cool.io.

Input:
$LOAD_PATH/fluent/plugin/in_<type>.rb
Buffer:
$LOAD_PATH/fluent/plugin/buf_<type>.rb
Output:
$LOAD_PATH/fluent/plug...

We use ‘register_xxx’ to register a plugin.
’xxx’ is ’input’, ’filter’, ’buffer’ and ’output’

We can load the plugin configuration using 
config_param and configure method.
config_param set config value to
@<config n...

class TailInput < Input
Plugin.register_input(’tail’, self)
config_param :path, :string
...
end
<source>
type tail
path /p...

One trick is here:
Fluentd’s configuration module does not
verify a default value. So,
we can use the nil like Tribool :)
...

Fluentd provides some useful mixins for
input and output plugins.

SetTagKeyMixin:
Provide ‘tag_key’ and ‘include_tag_key’.
SetTimeKeyMixin:
Provide ‘time_key’ and ‘include_time_key’.
Handl...

Code Flow
Mixin usage
class MongoOutput <
BufferedOutput
...
include SetTagKeyMixin
config_set_default
:include_tag_key, f...

Default 3rd party
Available plugins
exec
forward
http
syslog
tail
monitor_agent
etc...
mongo_tail
scribe
msgpack
dstat
kaf...

class NewInput < Input
...
def configure(conf)
# parse a configuration manually
end
def start
# invoke action
end
def shut...

In action method,
we use router.emit to input data.
tag = "app.tag"
time = Engine.now
record = {"key" => "value", ...}
rou...

How to read an input in an efficient way?
We use a thread and an event loop.

class ForwardInput < Fluent::Input
...
def start
...
@thread = Thread.new(&method(:run))
end
def run
...
end
end
Thread

class ForwardInput < Fluent::Input
...
def start
@loop = Coolio::Loop.new
@lsock = listen
@loop.attach(@lsock)
...
end
......

Note that
We must use Engine.now instead of Time.now

Default 3rd party
Available plugins
grep
stdout
record_transformer
parser
geoip
record_map
typecast
record_modifier
etc...

class NewFilter < Filter
# configure, start and shutdown
# are same as input plugin
def filter(tag, time, record)
# Modify...

Default 3rd party
Available plugins
memory
file

In most cases,
Memory and File are enough.

Memory type is default.
It’s fast but can’t resume data.

File type is persistent type.
It can resume data from file. 
TimeSlicedOutput’s default is file.

Default 3rd party
Available plugins
copy
exec
file
forward
null
stdout
etc...
mongo
s3
scribe
elasticsearch
webhdfs
kafka
...

class NewOutput < BufferedOutput
# configure, start and shutdown
# are same as input plugin
def format(tag, time, record)
...

Output has 3 buffering modes.
None
Buffered
ObjectBuffered
Time sliced

Buffered Time sliced
Buffering type
chunk
go out
chunk
chunk
from in
chunk
limit
queue
limit
Buffer has an internal
map to...

Input:
Fluent::Test::InputTestDriver
Buffer:
Fluent::Test::BufferedOutputTestDriver
Output:
Fluent::Test::OutputTestDriver...

class MongoOutputTest < Test::Unit::TestCase
def setup
Fluent::Test.setup
require 'fluent/plugin/out_mongo'
end
def create...

...
def test_format
# test format using emit and expect_format
end
def test_write
d = create_driver
t = emit_documents(d)
...

Gem Structure
Plugin root
|-- lib/
| |-- fluent/
| |-- plugin/
| |- out_<name>.rb
|- Gemfile
|- fluent-plugin-<name>.gemsp...

Bundle with git
$ edit lib/fluent/plugin/out_<name>.rb
$ git add / commit
$ cat VERSION
0.1.0
$ bunlde exec rake release
S...

See released plugins
for more details about each file.