How to write a Java input plugin

To develop a new Java input for Logstash, you write a new Java class that conforms to the Logstash Java Inputs API, package it, and install it with the logstash-plugin utility. We’ll go through each of those steps.

Start by copying the example input plugin. The plugin API is currently part of the Logstash codebase so you must have a local copy of that available. You can obtain a copy of the Logstash codebase with the following git command:

git clone --branch <branch_name> --single-branch https://github.com/elastic/logstash.git <target_folder>

The branch_name should correspond to the version of Logstash containing the preferred revision of the Java plugin API.

Specify the target_folder for your local copy of the Logstash codebase. If you do not specify target_folder, it defaults to a new folder called logstash under your current folder.

After you have obtained a copy of the appropriate revision of the Logstash codebase, you need to compile it to generate the .jar file containing the Java plugin API. From the root directory of your Logstash codebase ($LS_HOME), you can compile it with ./gradlew assemble (or gradlew.bat assemble if you’re running on Windows). This should produce the $LS_HOME/logstash-core/build/libs/logstash-core-x.y.z.jar where x, y, and z refer to the version of Logstash.

After you have successfully compiled Logstash, you need to tell your Java plugin where to find the logstash-core-x.y.z.jar file. Create a new file named gradle.properties in the root folder of your plugin project. That file should have a single line:

LOGSTASH_CORE_PATH=<target_folder>/logstash-core

where target_folder is the root folder of your local copy of the Logstash codebase.

The example input plugin generates a configurable number of simple events before terminating. Let’s look at the main class in the example input.

@LogstashPlugin(name="java_input_example")
public class JavaInputExample implements Input {

    public static final PluginConfigSpec<Long> EVENT_COUNT_CONFIG =
            Configuration.numSetting("count", 3);

    public static final PluginConfigSpec<String> PREFIX_CONFIG =
            Configuration.stringSetting("prefix", "message");

    private long count;
    private String prefix;
    private final CountDownLatch done = new CountDownLatch(1);
    private volatile boolean stopped;

    public JavaInputExample(Configuration config, Context context) {
        count = config.get(EVENT_COUNT_CONFIG);
        prefix = config.get(PREFIX_CONFIG);
    }

    @Override
    public void start(QueueWriter queueWriter) {
        int eventCount = 0;
        try {
            while (!stopped && eventCount < count) {
                eventCount++;
                queueWriter.push(Collections.singletonMap("message",
                        prefix + " " + StringUtils.center(eventCount + " of " + count, 20)));
            }
        } finally {
            stopped = true;
            done.countDown();
        }
    }

    @Override
    public void stop() {
        stopped = true; // set flag to request cooperative stop of input
    }

    @Override
    public void awaitStop() throws InterruptedException {
        done.await(); // blocks until input has stopped
    }

    @Override
    public Collection<PluginConfigSpec<?>> configSchema() {
        return Arrays.asList(EVENT_COUNT_CONFIG, PREFIX_CONFIG);
    }
}

Let’s step through and examine each part of that class.

@LogstashPlugin(name="java_input_example")
public class JavaInputExample implements Input {

Notes about the class declaration:

The snippet below contains both the setting definition and the method referencing it.

public static final PluginConfigSpec<Long> EVENT_COUNT_CONFIG =
        Configuration.numSetting("count", 3);

public static final PluginConfigSpec<String> PREFIX_CONFIG =
        Configuration.stringSetting("prefix", "message");

@Override
public Collection<PluginConfigSpec<?>> configSchema() {
    return Arrays.asList(EVENT_COUNT_CONFIG, PREFIX_CONFIG);
}

The PluginConfigSpec class allows developers to specify the settings that a plugin supports complete with setting name, data type, deprecation status, required status, and default value. In this example, the count setting defines the number of events that will be generated and the prefix setting defines an optional prefix to include in the event field. Neither setting is required and if it is not explicitly set, the settings default to 3 and message, respectively.

The configSchema method must return a list of all settings that the plugin supports. In a future phase of the Java plugin project, the Logstash execution engine will validate that all required settings are present and that no unsupported settings are present.

private long count;
private String prefix;

public JavaInputExample(Configuration config, Context context) {
    count = config.get(EVENT_COUNT_CONFIG);
    prefix = config.get(PREFIX_CONFIG);
}

All Java input plugins must have a constructor taking both a Configuration and Context argument. This is the constructor that will be used to instantiate them at runtime. The retrieval and validation of all plugin settings should occur in this constructor. In this example, the values of the two plugin settings are retrieved and stored in local variables for later use in the start method.

Any additional initialization may occur in the constructor as well. If there are any unrecoverable errors encountered in the configuration or initialization of the input plugin, a descriptive exception should be thrown. The exception will be logged and will prevent Logstash from starting.

@Override
public void start(QueueWriter queueWriter) {
    int eventCount = 0;
    try {
        while (!stopped && eventCount < count) {
            eventCount++;
            queueWriter.push(Collections.singletonMap("message",
                    prefix + " " + StringUtils.center(eventCount + " of " + count, 20)));
        }
    } finally {
        stopped = true;
        done.countDown();
    }
}

The start method begins the event-producing loop in an input. Inputs are flexible and may produce events through many different mechanisms including:

Events should be constructed as instances of Map<String, Object> and pushed into the event pipeline via the QueueWriter.push() method.

private final CountDownLatch done = new CountDownLatch(1);
private volatile boolean stopped;

@Override
public void stop() {
    stopped = true; // set flag to request cooperative stop of input
}

@Override
public void awaitStop() throws InterruptedException {
    done.await(); // blocks until input has stopped
}

The stop method notifies the input to stop producing events. The stop mechanism may be implemented in any way that honors the API contract though a volatile boolean flag works well for many use cases.

Inputs stop both asynchronously and cooperatively. Use the awaitStop method to block until the input has completed the stop process. Note that this method should not signal the input to stop as the stop method does. The awaitStop mechanism may be implemented in any way that honors the API contract though a CountDownLatch works well for many use cases.

Lastly, but certainly not least importantly, unit tests are strongly encouraged. The example input plugin includes an example unit test that you can use as a template for your own.

Java plugins are packaged as Ruby gems for dependency management and interoperability with Ruby plugins.

The Java plugin should be compiled and assembled into a fat jar with the vendor task in the Gradle build file. This will package all Java dependencies into a single jar and write it to the correct folder for later packaging into a Ruby gem.

Several Ruby source files are required to package the jar file as a Ruby gem. These Ruby files are used only at Logstash startup time to identify the Java plugin and are not used during runtime event processing.

logstash-input-<input-name>.gemspec

Gem::Specification.new do |s|
  s.name            = 'logstash-input-java_input_example'
  s.version         = PLUGIN_VERSION
  s.licenses        = ['Apache-2.0']
  s.summary         = "Example input using Java plugin API"
  s.description     = ""
  s.authors         = ['Elasticsearch']
  s.email           = '[email protected]'
  s.homepage        = "http://www.elastic.co/guide/en/logstash/current/index.html"
  s.require_paths = ['lib', 'vendor/jar-dependencies']

  # Files
  s.files = Dir["lib/**/*","spec/**/*","*.gemspec","*.md","CONTRIBUTORS","Gemfile","LICENSE","NOTICE.TXT", "vendor/jar-dependencies/**/*.jar", "vendor/jar-dependencies/**/*.rb", "VERSION", "docs/**/*"]

  # Special flag to let us know this is actually a logstash plugin
  s.metadata = { 'logstash_plugin' => 'true', 'logstash_group' => 'input'}

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

  s.add_development_dependency 'logstash-devutils'
end

You can use this file with the following modifications:

lib/logstash/inputs/<input-name>.rb

# encoding: utf-8
require "logstash/inputs/base"
require "logstash/namespace"
require "logstash-input-java_input_example_jars"
require "java"

class LogStash::Inputs::JavaInputExample < LogStash::Inputs::Base
  config_name "java_input_example"

  def self.javaClass() org.logstash.javaapi.JavaInputExample.java_class; end
end

Modify these items in the file above:

lib/logstash-input-<input-name>_jars.rb

require 'jar_dependencies'
require_jar('org.logstash.javaapi', 'logstash-input-java_input_example', '0.0.1')

In the file above:

After you have created the previous files and the plugin JAR file, build the gem using the following command:

gem build logstash-input-<input-name>.gemspec

After you have packaged your Java plugin as a Ruby gem, you can install it in Logstash with this command:

bin/logstash-plugin install --no-verify --local /path/to/javaPlugin.gem

For Windows platforms: Substitute backslashes for forward slashes as appropriate in the command.

The following is a minimal Logstash configuration that can be used to test that the Java input plugin is correctly installed and functioning.

input {
  java_input_example {}
}
output {
  stdout { codec => rubydebug }
}

Copy the above Logstash configuration to a file such as java_input.conf. Start Logstash with:

bin/logstash --java-execution -f /path/to/java_input.conf

Note that the --java-execution flag to enable the Java execution engine is required as Java plugins are not supported in the Ruby execution engine.

The expected Logstash output (excluding initialization) with the configuration above is:

{
      "@version" => "1",
       "message" => "message        1 of 3       ",
    "@timestamp" => yyyy-MM-ddThh:mm:ss.SSSZ
}
{
      "@version" => "1",
       "message" => "message        2 of 3       ",
    "@timestamp" => yyyy-MM-ddThh:mm:ss.SSSZ
}
{
      "@version" => "1",
       "message" => "message        3 of 3       ",
    "@timestamp" => yyyy-MM-ddThh:mm:ss.SSSZ
}

If you have any feedback on Java plugin support in Logstash, please comment on our main Github issue or post in the Logstash forum.