How to write a Logstash input plugin

Each Logstash plugin lives in its own GitHub repository. To create a new repository for your plugin:

You can create your own Logstash plugin in seconds! The generate subcommand of bin/logstash-plugin creates the foundation for a new Logstash plugin with templatized files. It creates the correct directory structure, gemspec files, and dependencies so you can start adding custom code to process data with Logstash.

For more information, see Generating plugins

Alternatively, you can use the examples repo we host on github.com

Your file structure should look like this:

$ tree logstash-input-mypluginname
├── Gemfile
├── LICENSE
├── README.md
├── Rakefile
├── lib
│   └── logstash
│       └── inputs
│           └── mypluginname.rb
├── logstash-input-mypluginname.gemspec
└── spec
    └── inputs
        └── mypluginname_spec.rb

For more information about the Ruby gem file structure and an excellent walkthrough of the Ruby gem creation process, see http://timelessrepo.com/making-ruby-gems

Before we dive into the details, open up the plugin file in your favorite text editor and take a look.

require "logstash/inputs/base"
require "logstash/namespace"
require "stud/interval"
require "socket" # for Socket.gethostname

# Add any asciidoc formatted documentation here
# Generate a repeating message.
#
# This plugin is intended only as an example.

class LogStash::Inputs::Example < LogStash::Inputs::Base
  config_name "example"

  # If undefined, Logstash will complain, even if codec is unused.
  default :codec, "plain"

  # The message string to use in the event.
  config :message, :validate => :string, :default => "Hello World!"

  # Set how frequently messages should be sent.
  #
  # The default, `1`, means send a message every second.
  config :interval, :validate => :number, :default => 1

  public
  def register
    @host = Socket.gethostname
  end # def register

  def run(queue)
    Stud.interval(@interval) do
      event = LogStash::Event.new("message" => @message, "host" => @host)
      decorate(event)
      queue << event
    end # loop
  end # def run

end # class LogStash::Inputs::Example

Logstash input plugins require parent classes defined in logstash/inputs/base and logstash/namespace:

require "logstash/inputs/base"
require "logstash/namespace"

Of course, the plugin you build may depend on other code, or even gems. Just put them here along with these Logstash dependencies.

The input plugin class should be a subclass of LogStash::Inputs::Base:

class LogStash::Inputs::Example < LogStash::Inputs::Base

The class name should closely mirror the plugin name, for example:

LogStash::Inputs::Example

  config_name "example"

This is the name your plugin will call inside the input configuration block.

If you set config_name "example" in your plugin code, the corresponding Logstash configuration block would need to look like this:

input {
  example {...}
}

  public
  def register
  end # def register

The Logstash register method is like an initialize method. It was originally created to enforce having super called, preventing headaches for newbies. (Note: It may go away in favor of initialize, in conjunction with some enforced testing to ensure super is called.)

public means the method can be called anywhere, not just within the class. This is the default behavior for methods in Ruby, but it is specified explicitly here anyway.

You can also assign instance variables here (variables prepended by @). Configuration variables are now in scope as instance variables, like @message

The example input plugin has the following run Method:

  def run(queue)
    Stud.interval(@interval) do
      event = LogStash::Event.new("message" => @message, "host" => @host)
      decorate(event)
      queue << event
    end # loop
  end # def run

The run method is where a stream of data from an input becomes an event.

The stream can be plain or generated as with the heartbeat input plugin. In these cases, though no codec is used, a default codec must be set in the code to avoid errors.

Here’s another example run method:

  def run(queue)
    while true
      begin
        # Based on some testing, there is no way to interrupt an IO.sysread nor
        # IO.select call in JRuby.
        data = $stdin.sysread(16384)
        @codec.decode(data) do |event|
          decorate(event)
          event.set("host", @host) if !event.include?("host")
          queue << event
        end
      rescue IOError, EOFError, LogStash::ShutdownSignal
        # stdin closed or a requested shutdown
        break
      end
    end # while true
    finished
  end # def run

In this example, the data is being sent to the codec defined in the configuration block to decode the data stream and return an event.

In both examples, the resulting event is passed to the decorate method:

      decorate(event)

This applies any tags you might have set in the input configuration block. For example, tags => ["tag1", "tag2"].

Also in both examples, the event, after being "decorated," is appended to the queue:

      queue << event

This inserts the event into the pipeline.

A require statement in Ruby is used to include necessary code. In some cases your plugin may require additional files. For example, the collectd plugin uses the types.db file provided by collectd. In the main directory of your plugin, a file called vendor.json is where these files are described.

The vendor.json file contains an array of JSON objects, each describing a file dependency. This example comes from the collectd codec plugin:

[{
        "sha1": "a90fe6cc53b76b7bdd56dc57950d90787cb9c96e",
        "url": "http://collectd.org/files/collectd-5.4.0.tar.gz",
        "files": [ "/src/types.db" ]
}]

Another example of the vendor.json file is the geoip filter

The process used to download these dependencies is to call rake vendor. This will be discussed further in the testing section of this document.

Another kind of external dependency is on jar files. This will be described in the "Add a gemspec file" section.

As a plugin evolves, an option or feature may no longer serve the intended purpose, and the developer may want to deprecate its usage. Deprecation warns users about the option’s status, so they aren’t caught by surprise when it is removed in a later release.

Logstash 7.6 introduced a deprecation logger to make handling those situations easier. You can use the adapter to ensure that your plugin can use the deprecation logger while still supporting older versions of Logstash. See the readme for more information and for instructions on using the adapter.

Deprecations are noted in the logstash-deprecation.log file in the log directory.

Gemfiles allow Ruby’s Bundler to maintain the dependencies for your plugin. Currently, all we’ll need is the Logstash gem, for testing, but if you require other gems, you should add them in here.

source 'https://rubygems.org'
gemspec
gem "logstash", :github => "elastic/logstash", :branch => "8.4"

At the bottom of the gemspec file is a section with a comment: Gem dependencies. This is where any other needed gems must be mentioned. If a gem is necessary for your plugin to function, it is a runtime dependency. If a gem are only used for testing, then it would be a development dependency.

  # Gem dependencies
  s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
  s.add_development_dependency 'logstash-devutils'

In some cases, such as the Elasticsearch output plugin, your code may depend on a jar file. In cases such as this, the dependency is added in the gemspec file in this manner:

  # Jar dependencies
  s.requirements << "jar 'org.elasticsearch:elasticsearch', '5.0.0'"
  s.add_runtime_dependency 'jar-dependencies'

With these both defined, the install process will search for the required jar file at http://mvnrepository.com and download the specified version.

You already have all the necessary ingredients, so let’s go ahead and run the build command:

gem build logstash-input-example.gemspec

That’s it! Your gem should be built and be in the same path with the name

logstash-input-mypluginname-0.1.0.gem

The s.version number from your gemspec file will provide the gem version, in this case, 0.1.0.

You should test install your plugin into a clean installation of Logstash. Download the latest version from the Logstash downloads page.

bin/logstash -e 'input { example{} } output {stdout { codec => rubydebug }}'

The example input plugin will send the contents of message (with a default message of "Hello World!") every second.

{
       "message" => "Hello World!",
      "@version" => "1",
    "@timestamp" => "2015-01-27T19:17:18.932Z",
          "host" => "cadenza"
}

Feel free to experiment and test this by changing the message and interval parameters:

bin/logstash -e 'input { example{ message => "A different message" interval => 5 } } output {stdout { codec => rubydebug }}'

Congratulations! You’ve built, deployed and successfully run a Logstash input.

Logstash and all its plugins are licensed under Apache License, version 2 ("ALv2"). If you make your plugin publicly available via RubyGems.org, please make sure to have this line in your gemspec:

To begin, you’ll need an account on RubyGems.org

After creating an account, obtain an API key from RubyGems.org. By default, RubyGems uses the file ~/.gem/credentials to store your API key. These credentials will be used to publish the gem. Replace username and password with the credentials you created at RubyGems.org:

curl -u username:password https://rubygems.org/api/v1/api_key.yaml > ~/.gem/credentials
chmod 0600 ~/.gem/credentials

Before proceeding, make sure you have the right version in your gemspec file and commit your changes.

To publish version 0.1.0 of your new logstash gem:

bundle install
bundle exec rake vendor
bundle exec rspec
bundle exec rake publish_gem

That’s it! Your plugin is published! Logstash users can now install your plugin by running:

bin/logstash-plugin install logstash-input-mypluginname

Some of the many benefits of having your plugin in the logstash-plugins repository are:

To begin migrating your plugin to logstash-plugins, simply create a new issue in the Logstash repository. When the acceptance guidelines are completed, we will facilitate the move to the logstash-plugins organization using the recommended github process.