Fluentd

How to Write Service Discovery Plugin

Fluentd supports pluggable service discovery which lets an output plugin resolve its destinations at run time. The plugin filenames prefixed sd_ are registered as Service Discovery Plugins.

See Plugin Base Class API for details on the common APIs for all the plugin types.

Here is an example of a custom service discovery plugin that reads a list of servers from a plain text file, one host:port per line, and re-reads it on a timer. It takes a required parameter called path, which is the pathname of that file, and an optional interval.

require 'fluent/plugin/service_discovery'
require 'fluent/plugin_helper'

module Fluent::Plugin
  class MyListServiceDiscovery < ServiceDiscovery
    include Fluent::PluginHelper::Mixin

    # Register MyListServiceDiscovery as 'my_list'.
    Fluent::Plugin.register_sd('my_list', self)

    helpers :timer

    config_param :path, :string
    config_param :interval, :time, default: 10

    def configure(conf)
      super

      unless File.readable?(@path)
        raise Fluent::ConfigError, "my_list: path=#{@path} is not readable"
      end

      @services = fetch_services
    end

    def start(queue)
      timer_execute(:sd_my_list, @interval) do
        refresh(queue)
      end

      super()
    end

    private

    # Reads one "host:port" per line. Blank lines and # comments are skipped.
    def fetch_services
      File.readlines(@path).filter_map do |line|
        line = line.sub(/#.*/, '').strip
        next if line.empty?

        host, port = line.split(':', 2)
        Service.new(:my_list, host, port.to_i, line, 60, false, nil, nil, nil)
      end
    end

    # Announces the additions before the removals, so that the owner plugin is
    # never left without a destination.
    def refresh(queue)
      services = begin
                   fetch_services
                 rescue => e
                   log.error "my_list: failed to read the list", path: @path, error: e
                   return
                 end

      (services - @services).each { |s| queue.push(ServiceDiscovery.service_in_msg(s)) }
      (@services - services).each { |s| queue.push(ServiceDiscovery.service_out_msg(s)) }
      @services = services
    end
  end
end

Save this as sd_my_list.rb in a loadable plugin path.

With out_forward output plugin:

<match test>
  @type forward
  <service_discovery>
    @type my_list
    path /path/to/servers.txt
    interval 2
  </service_discovery>
</match>

/path/to/servers.txt holds one server per line:

# host:port
127.0.0.1:24230
127.0.0.1:24231

out_forward picks up both of them at startup:

adding forwarding server '127.0.0.1:24230' host="127.0.0.1" port=24230 weight=60
adding forwarding server '127.0.0.1:24231' host="127.0.0.1" port=24231 weight=60

Replacing the second line of the file with another server swaps the destination without a restart:

Service in: name=127.0.0.1:24232 127.0.0.1:24232
Service out: name=127.0.0.1:24231 127.0.0.1:24231

Fluent::Plugin::ServiceDiscovery does not include the plugin helper mixin, so a plugin which needs a helper includes Fluent::PluginHelper::Mixin itself, as the example above does. The built-in sd_file and sd_srv do the same.

How To Use Service Discovery From Plugins

Service discovery plugins are designed to be used from output plugins. The service discovery plugin helper is there for this purpose:

# in class definition
helpers :service_discovery

# in #configure
service_discovery_configure(:my_output_sd_watcher, static_default_service_directive: 'server')

# in #write, or wherever a destination is picked
service_discovery_select_service do |node|
  send_data(node, chunk)
end

The helper owns the lifecycle of the configured plugins. It creates one instance per <service_discovery> section, starts them with the queue they push their updates onto, keeps the current set of services, and shuts them down with the owner plugin.

See Service Discovery Plugin Helper API for details.

Methods

Fluent::Plugin::ServiceDiscovery keeps the current destinations in @services, and gives the plugin two class methods to announce a change. A plugin implements #configure for the initial list, and #start for the updates.

#configure(conf)

It sets @services to the destinations the plugin knows at startup. The owner plugin reads them while it configures itself, so a destination missing here does not exist for the owner plugin until the plugin announces it later.

Raise Fluent::ConfigError here for anything the plugin cannot work with, so that a misconfiguration stops Fluentd at startup instead of showing up later as a lost destination.

#start(queue)

It starts watching wherever the plugin discovers destinations from. queue is a Thread::Queue owned by the owner plugin. A plugin which never changes its list, like sd_static, does not have to override this method.

The plugin schedules its own updates. The example above and sd_srv use the timer plugin helper, and sd_file watches the file with the event loop plugin helper.

.service_in_msg(service)

It builds the message which adds service to the destinations of the owner plugin.

.service_out_msg(service)

It builds the message which removes service from the destinations of the owner plugin. The built-in plugins push the additions before the removals, so that the owner plugin always has at least one destination left.

Services

A destination is a Fluent::Plugin::ServiceDiscovery::Service, a Struct with these members in this order:

#discovery_id identifies a service, and it is built from every member. Two services which differ in any one of them, weight included, are different services, so a changed weight is announced as one removal and one addition of the same server.

How The Services Reach The Owner Plugin

The service discovery plugin helper owns a manager which holds the plugins configured in the <service_discovery> sections. The manager builds the initial set of destinations from #services of every plugin, then polls the queue every 3 seconds by default and applies the messages it finds there. This is where the Service in: and Service out: log lines come from.

The manager skips the polling when every configured plugin is static, since a static list never changes. A plugin registered under any other name is polled.

The owner plugin may keep an object of its own per service instead of the struct. out_forward builds a node which owns the connection to the server, which is why service_discovery_select_service yields that node rather than the Service.

Writing Tests

Fluentd does not provide a test driver for service discovery plugins, so tests instantiate the plugin and drive it directly. The tests of the built-in plugins do the same.

# test/plugin/test_sd_my_list.rb

require 'test/unit'
require 'fileutils'
require 'timeout'
require 'fluent/test'
require 'fluent/test/helpers'

# Your own plugin
require 'fluent/plugin/sd_my_list'

class ServiceDiscoveryMyListTest < Test::Unit::TestCase
  include Fluent::Test::Helpers

  TMP_DIR = File.expand_path(File.dirname(__FILE__) + '/tmp/sd_my_list')
  LIST_PATH = File.join(TMP_DIR, 'list.txt')

  Service = Fluent::Plugin::ServiceDiscovery::Service

  def setup
    Fluent::Test.setup
    FileUtils.mkdir_p(TMP_DIR)
  end

  def teardown
    FileUtils.rm_rf(TMP_DIR)
  end

  def create_plugin(conf)
    sd = Fluent::Plugin::MyListServiceDiscovery.new
    sd.configure(config_element('service_discovery', '', conf))
    sd
  end

  sub_test_case 'configure' do
    test 'reads the list' do
      File.write(LIST_PATH, "# host:port\n127.0.0.1:24224\n")
      sd = create_plugin('path' => LIST_PATH)
      assert_equal([Service.new(:my_list, '127.0.0.1', 24224, '127.0.0.1:24224', 60, false, nil, nil, nil)], sd.services)
    end

    test 'raises ConfigError for an unreadable path' do
      assert_raise(Fluent::ConfigError) do
        create_plugin('path' => File.join(TMP_DIR, 'no-such-file'))
      end
    end
  end

  sub_test_case 'start' do
    test 'pushes the difference onto the queue' do
      File.write(LIST_PATH, "127.0.0.1:24224\n")
      sd = create_plugin('path' => LIST_PATH, 'interval' => '1')
      queue = []
      sd.start(queue)
      begin
        File.write(LIST_PATH, "127.0.0.1:24225\n")
        Timeout.timeout(10) { sleep 0.1 until queue.size >= 2 }
      ensure
        sd.stop; sd.shutdown; sd.close; sd.terminate
      end
      assert_equal([:service_in, :service_out], queue.map(&:type))
      assert_equal(24225, queue[0].service.port)
      assert_equal(24224, queue[1].service.port)
    end
  end
end

config_element comes from Fluent::Test::Helpers, which fluent/test does not load by itself. A plain Array works as the queue in tests, since the plugin only pushes onto it.

Overview of Tests

Testing for service discovery plugins is mainly for:

• Validation of configuration (i.e. #configure)
• Validation of the services built from the configuration
• Validation of the messages pushed while the plugin is running

The lifecycle of the plugin in a test is:

1. Instantiate the plugin
2. Configure the plugin
3. Start the plugin with a queue, and run test code
4. Assert the services and the messages on the queue
5. Shut the plugin down

For:

• configuration tests, repeat steps # 1-2
• full feature tests, repeat steps # 1-5

See Testing API for Plugins for details.