Fluentd Blog

Fluentd v1.2.0 has been released

Hi users!

We have released v1.2.0. ChangeLog is here. This release includes new features and improvements.

output: Backup for broken chunks

Fluentd receives various events from various data sources. This sometimes have a problem in Output plugins.

For example, if one application generates invalid events for data destination, e.g. schema mismatch, buffer flush always failed. Other case is generated events are invalid for output configuration, e.g. required field is missing.

Fluentd has retry feature for temporal failures but there errors are never succeeded. So Fluentd should not retry unexpected "broken chunks".

Since v1.2.0, fluentd routes broken chunks to backup directory. By default, backup root directory is /tmp/fluent. If you set root_dir in <system>, root_dir is used. File path consists of several parameters for unique path and ${root_dir}/backup/worker${worker_id}/${plugin_id}/{chunk_id}.log is path template.

If you have following configuration and unrecoverable error happens inside @type sql plugin,

<system>
  root_dir /var/log/fluentd
</system>
# ...
<match app.**>
  @type sql
  @id out_sql
  <buffer>
    # ...
  </buffer>
</match>

chunk is routed to /var/log/fluentd/backup/worker0/out_sql/56644156d63a108eda2f487c58140736.log.

Currently, fluentd routes chunks to backup directory when Output plugin raises following errors during buffer flush.

• Fluent::UnrecoverableError: Output plugin can raise this error to avoid retry for plugin specific case.
• TypeError: This error sometimes happens when an event has unexpected type in target field.
• ArgumentError: This error sometimes happens when library usage is wrong in plugins.
• NoMethodError: This error sometimes happens when events and configuration are mismatched.

Fluentd continues to retry buffer flush when other error happens.

New API: Counter API

This is for plugin developers. Counter API consists of server and client. Server stores counter values and client calls APIs to store values to server.

Here is example configuration:

<system>
  <counter_server>
    scope test
    bind 127.0.0.1
    port 25000
    backup_path /tmp/counter_backup.json
  </counter_server>
  <counter_client>
    host 127.0.0.1
    port 25000    
  </counter_client>
</system>

And filter implementation example:

require 'fluent/plugin/filter'

module Fluent
  module Plugin
    class CounterFilter < Filter
      Plugin.register_filter('counter', self)

      helpers :counter

      def start
        super

        @client = counter_client_create(scope: 'test') # scope value is same with `<counter_server>`
        @client.establish('counter').get
        @client.init(:name => 'num', type: 'numeric', reset_interval: 10).get
      end

      def filter(tag, time, record)
        @client.inc(:name => 'num', :value => 5)
        p @client.get('num').data.first["current"] # Show current value
        record
      end
    end
  end
end

This API is useful for storing metrics under multiprocess environment. We will add API documents.

filter_grep: Support for <and> and <or> sections

grep filter now supports and / or condition for <regexp> / <exclude> Before v1.2.0, grep filter supports only and <regexp> and or <exclude> patterns.

<filter pattern>
  # These <regexp>s are "and"
  <regexp>
    key level
    pattern ^ERROR|WARN$
  </regexp>
  <regexp>
    key method
    pattern ^GET|POST$
  </regexp>
  # These <exclude>s are "or"
  <exclude>
    key level
    pattern ^WARN$
  </exclude>
  <exclude>
    key method
    pattern ^GET$
  </exclude>
</filter>

v1.2.0 adds <and> and <or> sections to support more patterns.

Here is configuration example:

<filter pattern>
  <or>
    <regexp>
      key level
      pattern ^ERROR|WARN$
    </regexp>
    <regexp>
      key method
      pattern ^GET|POST$
    </regexp>
  </or>
  <and>
    <exclude>
      key level
      pattern ^WARN$
    </exclude>
    <exclude>
      key method
      pattern ^GET$
    </exclude>
  </and>
</filter>

If you pass this data streams:

{"time" : "2013/01/13T07:02:11.124202", "level" : "INFO", "method" : "GET", "path" : "/ping"}
{"time" : "2013/01/13T07:02:13.232645", "level" : "WARN", "method" : "POST", "path" : "/auth"}
{"time" : "2013/01/13T07:02:21.542145", "level" : "WARN", "method" : "GET", "path" : "/favicon.ico"}
{"time" : "2013/01/13T07:02:43.632145", "level" : "WARN", "method" : "POST", "path" : "/login"}
{"time" : "2013/01/13T07:02:44.959307", "level" : "ERROR", "method" : "POST", "path" : "/login"}
{"time" : "2013/01/13T07:02:45.444992", "level" : "ERROR", "method" : "GET", "path" : "/ping"}
{"time" : "2013/01/13T07:02:51.247941", "level" : "WARN", "method" : "GET", "path" : "/info"}
{"time" : "2013/01/13T07:02:53.108366", "level" : "WARN", "method" : "POST", "path" : "/ban"}

filtered result is below:

{"time" : "2013/01/13T07:02:11.124202", "level" : "INFO", "method" : "GET", "path" : "/ping"}
{"time" : "2013/01/13T07:02:13.232645", "level" : "WARN", "method" : "POST", "path" : "/auth"}
{"time" : "2013/01/13T07:02:43.632145", "level" : "WARN", "method" : "POST", "path" : "/login"}
{"time" : "2013/01/13T07:02:44.959307", "level" : "ERROR", "method" : "POST", "path" : "/login"}
{"time" : "2013/01/13T07:02:45.444992", "level" : "ERROR", "method" : "GET", "path" : "/ping"}
{"time" : "2013/01/13T07:02:53.108366", "level" : "WARN", "method" : "POST", "path" : "/ban"}

Major bug fixes

• server helper: Close invalid socket when ssl error happens on reading
• log: Fix unexpected implementation bug when log rotation setting is applied

Thanks for submitting bug reports and patches :)

Enjoy logging!

Read More

Fluentd v1.1.3 has been released

Hi users!

We have released v1.1.3. ChangeLog is here. This release includes several enhancements and bug fixes.

output: Support negative array index for tag placeholder

We can use array index for tag placeholder. This is useful for accessing tag parts.

<match app.**>
  @type foo
  param value-${tag[1]} # if tag is 'app.foo.bar', ${tag[1]} is 'foo'
</source>

Since v1.1.3, you can also use negative array index for tag placeholder. The behaviour is same as ruby's negative array index.

<match app.**>
  @type foo
  param value-${tag[-1]} # if tag is 'app.foo.bar', ${tag[-1]} is 'bar'
</source>

buffer: Add queued_chunks_limit_size to control the number of queued chunks

This new queued_chunks_limit_size parameter mitigates lots of queued chunks issue with frequent enqueuing.

Sometimes users set smaller flush_interval, e.g. 1s, for log forwarding. This is no problem on healthy environment. But if the destination is slower or unstable, output's flush fails and retry is started. In such situation, lots of small queued chunks are generated in the buffer and it consumes lots of fd resources when you use file buffer. To avoid this problem, queued_chunks_limit_size is useful. If you set queued_chunks_limit_size 5, staged chunks are not enqueued until the number of waiting enqueued chunks is less than 5.

Note that this check is for interval based enqueuing for now. It means if your staged chunk reaches chunk_limit_size, its chunks is enqueued even if the number of waiting enqueued chunks is greater than queued_chunks_limit_size.

Major bug fixes

• output: Delete empty queued_num field after purging chunks. This fixes memory leak when chunk keys include time
• out_forward: The node should be disabled when TLS socket for ack returns an error

Thanks for submitting bug reports and patches :)

Enjoy logging!

Read More

Fluentd v1.1.1 has been released

Hi users!

We have released v1.1.1. ChangeLog is here. This release includes several enhancements and bug fixes..

in_forward/server helper: Support mutual TLS authentication

We added ca_path and client_cert_path parameters to server plugin helper. You can now send data between fluent-bit and fluentd with mutual TLS authentication. In in_forward, put these parameters into <transport tls>.

<source>
  @type forward
  # ... other parameters
  <transport tls>
    # ... other parameters
    ca_path /path/to/ca_file
    client_cert_auth true
  </transport>
</source>

buf_file: Skip and delete broken chunk files during resume

fluentd can't restart if unexpected broken file exist in buffer directory, because such files cause errors in resume routine. Since v1.1.1, if fluentd found broken chunks during resume, these files are skipped and deleted from buffer directory.

We have a plan to add backup feature for broken files in v1.2.0. It moves broken files into backup directory instead of deletion.

Major bug fixes

• json parser: Fix error handling for oj 3.x
• in_tail: Fix watcher race condition during shutdown
• in_http: Emit event time instead of raw time value in batch mode

Thanks for submitting bug reports and patches :)

Enjoy logging!

Read More

Fluentd v1.1.0 has been released

Hi users!

We have released v1.1.0. ChangeLog is here. This release includes several new features.

config: Support hostname and worker_id short-cut in embed Ruby code

Fluentd configuration supports embedded ruby code in "#{}" string. Many users use this feature to embed runtime value in plugin parameters. This is conf example:

@id "out_foo#{ENV['SERVERENGINE_WORKER_ID']}}" # add worker id to plugin id under multi-process environment
tag "log.#{Socket.gethostname}"                # Use hostname in tag parts
<record>
  metadata "#{ENV['SERVER_ROLE']}-#{ENV['FOO']}"
</record>

We noticed setting hostname and worker_id is popular and current configuration is bit messy because it depends on fluentd internal, e.g. SERVERENGINE_WORKER_ID comes from serverengine used in Fluentd. So we added hostname and worker_id short-cut to cover popular cases. Here is new conf:

@id "out_foo#{worker_id}" # add worker id to plugin id under multi-process environment
tag "log.#{hostname}"     # Use hostname in tag parts

If other popular case found, we will add new short-cut.

record_accessor helper: Support nested field deletion

We have record_accessor helper for accessing nested field. Since v1.1.0, this helper supports nested field deletion. This feature is useful in record_transformer like plugins.

Syntax is same and you can delete nested field via accessor object in your plugin code.

deleter = record_accessor_create("$.key1.key2")
deleter.delete(record) # delete record["key1"]["key2"] field

record_transfomer filter plugin supports this feature with remove_keys parameter.

fluent-ca-generate: command for certificates generation

This is port from fluent-plugin-secure-forward. Use-case is same with Using private CA file and key. Change command name to fluent-ca-generate.

You can change several values like CN/country/etc via command option. Check --help for all options.

Major bug fixes

• server helper: Fix TCP + TLS degradation

We changes buffer management in in_tcp plugin and it breaks TLS socket handling. We changed TLS socket structure and fixed the regression.

Enjoy logging!

Read More

Fluentd v1.0.2 has been released

Hi users!

We have released v1.0.2. ChangeLog is here. This announcement includes v1.0.1 changes.

in_udp: Add receive_buffer_size parameter

This parameter is corresponding to SO_RCVBUF of socket. If you want to improve the performance of in_udp, set larger size to this parameter.

Major bug fixes

• in_tail: Handle truncated/moved back case

In rename(file -> file.bak) -> truncate -> rename(file.bak -> file) case, fluentd reads logs from the bottom. It causes log lost when application appends logs after truncate. Latest version reads logs from the head.

• out_forward: Rebuild weight array to apply server setting on active/standby environment

After active node is backed, standby node was still listed in active nodes. v1.0.2 fixes this regression. standby node is excluded from active nodes properly.

We have released v0.12.42 together. v0.12.42 includes same bug fixes.

Enjoy logging!

Read More