如何编写 Java 过滤插件

要为 Logstash 开发新的 Java 过滤器，你需要编写一个符合 Logstash Java 过滤器 API 的新 Java 类，将其打包，并使用 logstash-plugin 实用程序进行安装。我们将逐步介绍这些步骤。

首先，复制 示例过滤器插件。插件 API 目前是 Logstash 代码库的一部分，因此你必须拥有一个本地副本。你可以使用以下 git 命令获取 Logstash 代码库的副本

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

branch_name 应对应包含所需 Java 插件 API 版本的 Logstash 版本。

为你的 Logstash 代码库本地副本指定 target_folder。如果你不指定 target_folder，它默认为当前文件夹下的一个名为 logstash 的新文件夹。

获取适当版本的 Logstash 代码库副本后，你需要对其进行编译以生成包含 Java 插件 API 的 .jar 文件。从 Logstash 代码库的根目录 ($LS_HOME) 中，你可以使用 ./gradlew assemble（或者如果你在 Windows 上运行，则使用 gradlew.bat assemble）对其进行编译。这应该会生成 $LS_HOME/logstash-core/build/libs/logstash-core-x.y.z.jar，其中 x、y 和 z 指的是 Logstash 的版本。

成功编译 Logstash 后，你需要告诉你的 Java 插件到哪里查找 logstash-core-x.y.z.jar 文件。在插件项目的根文件夹中创建一个名为 gradle.properties 的新文件。该文件应包含一行

LOGSTASH_CORE_PATH=<target_folder>/logstash-core
		

其中 target_folder 是你的 Logstash 代码库本地副本的根文件夹。

该示例过滤器插件允许用户配置每个事件中将被反转的字段。例如，如果过滤器被配置为反转 day_of_week 字段，则带有 day_of_week: "Monday" 的事件将被转换为 day_of_week: "yadnoM"。让我们看一下该示例过滤器中的主类

@LogstashPlugin(name = "java_filter_example")
public class JavaFilterExample implements Filter {

    public static final PluginConfigSpec<String> SOURCE_CONFIG =
            PluginConfigSpec.stringSetting("source", "message");

    private String id;
    private String sourceField;

    public JavaFilterExample(String id, Configuration config, Context context) {
        this.id = id;
        this.sourceField = config.get(SOURCE_CONFIG);
    }

    @Override
    public Collection<Event> filter(Collection<Event> events, FilterMatchListener matchListener) {
        for (Event e : events) {
            Object f = e.getField(sourceField);
            if (f instanceof String) {
                e.setField(sourceField, StringUtils.reverse((String)f));
                matchListener.filterMatched(e);
            }
        }
        return events;
    }

    @Override
    public Collection<PluginConfigSpec<?>> configSchema() {
        return Collections.singletonList(SOURCE_CONFIG);
    }

    @Override
    public String getId() {
        return this.id;
    }

    @Override
    public void close() {
        this.sourceField = null;
        return;
    }
}
		

让我们逐步检查该类的每个部分。

@LogstashPlugin(name = "java_filter_example")
public class JavaFilterExample implements Filter {
		

关于类声明的注意事项

• 所有 Java 插件都必须使用 @LogstashPlugin 注解进行标注。此外

  • 必须提供注解的 name 属性，它定义了插件在 Logstash 管道定义中使用的名称。例如，此过滤器将在 Logstash 管道定义的 filter 部分中被引用为 filter { java_filter_example => { .... } }
  • name 属性的值必须与类名匹配（不区分大小写和下划线）。

• 该类必须实现 co.elastic.logstash.api.Filter 接口。

• Java 插件不得在 org.logstash 或 co.elastic.logstash 包中创建，以防止与 Logstash 本身中的类发生潜在冲突。

下面的代码片段包含了设置定义以及引用该定义的方法

public static final PluginConfigSpec<String> SOURCE_CONFIG =
        PluginConfigSpec.stringSetting("source", "message");

@Override
public Collection<PluginConfigSpec<?>> configSchema() {
    return Collections.singletonList(SOURCE_CONFIG);
}
		

PluginConfigSpec 类允许开发者指定插件支持的设置，包括设置名称、数据类型、弃用状态、是否必需以及默认值。在此示例中，source 设置定义了每个事件中将被反转的字段名称。它不是必需设置，如果未显式设置，其默认值为 message。

configSchema 方法必须返回插件支持的所有设置列表。在 Java 插件项目的未来阶段，Logstash 执行引擎将验证是否存在所有必需的设置，并确保不存在不受支持的设置。

private String id;
private String sourceField;

public JavaFilterExample(String id, Configuration config, Context context) {
    this.id = id;
    this.sourceField = config.get(SOURCE_CONFIG);
}
		

所有 Java 过滤器插件都必须有一个构造函数，该构造函数接受 String 类型的 id 以及 Configuration 和 Context 参数。这是运行时用于实例化插件的构造函数。所有插件设置的检索和验证都应在此构造函数中进行。在此示例中，每个事件中要反转的字段名称从其设置中检索，并存储在局部变量中，以便稍后在 filter 方法中使用。

任何额外的初始化也可以在构造函数中进行。如果在过滤器插件的配置或初始化过程中遇到任何无法恢复的错误，则应抛出描述性的异常。该异常将被记录，并阻止 Logstash 启动。

@Override
public Collection<Event> filter(Collection<Event> events, FilterMatchListener matchListener) {
    for (Event e : events) {
        Object f = e.getField(sourceField);
        if (f instanceof String) {
            e.setField(sourceField, StringUtils.reverse((String)f));
            matchListener.filterMatched(e);
        }
    }
    return events;
		

最后，我们来到 filter 方法，该方法由 Logstash 执行引擎在事件流经事件处理管道时对批量事件进行调用。要过滤的事件由 events 参数提供，该方法应返回过滤后的事件集合。过滤器在事件流经管道时可以对事件执行多种操作，包括

• 变更（Mutation）——过滤器可以添加、删除或更改事件中的字段。这是对事件执行各种增强功能的过滤器的最常见场景。在这种情况下，可以返回未修改的传入 events 集合，因为集合中的事件是在原地（in place）变更的。
• 删除（Deletion）——过滤器可以从事件管道中移除事件，以便后续的过滤器和输出接收不到它们。在这种情况下，必须在返回之前从过滤后的事件集合中删除要删除的事件。
• 创建（Creation）——过滤器可以在事件管道中插入新的事件，这些事件将仅被后续的过滤器和输出看到。在这种情况下，必须在返回之前将新事件添加到过滤后的事件集合中。
• 观察（Observation）——事件可以在过滤器不进行修改的情况下通过事件管道。这在过滤器根据事件管道中观察到的事件执行外部操作（例如，更新外部缓存）的场景中很有用。在这种情况下，可以返回未修改的传入 events 集合，因为没有进行任何更改。

在上面的示例中，从每个事件中检索 source 字段的值，如果是字符串值，则将其反转。由于每个事件都是在原地进行变更的，因此可以直接返回传入的 events 集合。

matchListener 是过滤器指示哪些事件“匹配”的机制。过滤器常见的操作（如 add_field 和 add_tag）仅应用于被指定为“匹配”的事件。一些过滤器（如 grok 过滤器）对什么是匹配事件有明确的定义，并且只会通知监听器匹配的事件。其他过滤器（如 UUID 过滤器）没有特定的匹配标准，应该为过滤的每个事件通知监听器。在此示例中，过滤器会针对任何在 source 字段中具有 String 值且因此能够被反转的事件通知匹配监听器。

@Override
public String getId() {
    return id;
}
		

对于过滤器插件，getId 方法应始终返回在实例化时通过构造函数提供给插件的 id。

@Override
public void close() {
    // shutdown a resource that was instantiated during the filter initialization phase.
    this.sourceField = null;
    return;
}
		

过滤器插件可以使用额外的资源来执行操作，例如创建新的数据库连接。实现 close 方法将允许插件在关闭管道时释放这些资源。

最后，但同样重要的是，强烈建议编写单元测试。示例过滤器插件包含一个 示例单元测试，你可以将其作为自己测试的模板。

Java 插件被打包为 Ruby gem，用于依赖管理和与 Ruby 插件的互操作性。一旦它们被打包为 gem，就可以像 Ruby 插件一样使用 logstash-plugin 实用程序进行安装。由于 Java 插件开发不需要具备 Ruby 或其工具链的知识，因此将 Java 插件打包为 Ruby gem 的过程已通过示例 Java 插件中提供的 Gradle 构建文件中的自定义任务实现了自动化。以下各节介绍如何配置和执行该打包任务，以及如何将打包好的 Java 插件安装到 Logstash 中。

以下部分出现在示例 Java 插件随附的 build.gradle 文件顶部附近

// ===========================================================================
// plugin info
// ===========================================================================
group                      'org.logstashplugins'
version                    "${file("VERSION").text.trim()}"
description                = "Example Java filter implementation"
pluginInfo.licenses        = ['Apache-2.0']
pluginInfo.longDescription = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using \$LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
pluginInfo.authors         = ['Elasticsearch']
pluginInfo.email           = ['info@elastic.co']
pluginInfo.homepage        = "https://esdocs.cn/guide/en/logstash/current/index.html"
pluginInfo.pluginType      = "filter"
pluginInfo.pluginClass     = "JavaFilterExample"
pluginInfo.pluginName      = "java_filter_example"
// ===========================================================================
		

1. 必须与主插件类的包匹配
2. 从必需的 VERSION 文件中读取
3. SPDX 许可证 ID 列表

你应该为你的插件配置上述值。

• version 值将自动从插件代码库根目录中的 VERSION 文件中读取。
• pluginInfo.pluginType 应设置为 input、filter、codec 或 output 之一。
• pluginInfo.pluginName 必须与主插件类上 @LogstashPlugin 注解中指定的名称匹配。Gradle 打包任务将验证这一点，如果它们不匹配，将返回错误。

将插件打包为 Ruby gem 需要几个 Ruby 源文件以及一个 gemspec 文件和一个 Gemfile。这些 Ruby 文件仅用于定义 Ruby gem 结构或在 Logstash 启动时注册 Java 插件。它们在运行时事件处理期间不会被使用。Gradle 打包任务会根据上述部分中配置的值自动生成所有这些文件。

你可以使用以下命令运行 Gradle 打包任务

./gradlew gem
		

对于 Windows 平台：根据需要在命令中将 ./gradlew 替换为 gradlew.bat。

该任务将在你的插件代码库的根目录中生成一个 gem 文件，文件名为 logstash-{{plugintype}}-<pluginName>-<version>.gem

将 Java 插件打包为 Ruby gem 后，你可以使用以下命令将其安装在 Logstash 中

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

对于 Windows 平台：根据需要在命令中用反斜杠替换正斜杠。

以下是一个最小的 Logstash 配置，可用于测试 Java 过滤器插件是否已正确安装并正常工作。

input {
  generator { message => "Hello world!" count => 1 }
}
filter {
  java_filter_example {}
}
output {
  stdout { codec => rubydebug }
}
		

将上述 Logstash 配置复制到文件中，例如 java_filter.conf。使用以下命令启动 Logstash

bin/logstash -f /path/to/java_filter.conf
		

使用上述配置时，预期的 Logstash 输出（不包括初始化信息）为

{
      "sequence" => 0,
      "@version" => "1",
       "message" => "!dlrow olleH",
    "@timestamp" => yyyy-MM-ddThh:mm:ss.SSSZ,
          "host" => "<yourHostName>"
}
		

如果你对 Logstash 中的 Java 插件支持有任何反馈，请在我们的 GitHub 主问题页面上发表评论或在 Logstash 论坛上发帖。