如何编写 Logstash 输入插件

每个 Logstash 插件都位于其自己的 GitHub 存储库中。要为您的插件创建一个新的存储库，请执行以下操作

您可以在几秒钟内创建自己的 Logstash 插件！ bin/logstash-plugin 的 generate 子命令使用模板化的文件为新的 Logstash 插件创建基础。它创建正确的目录结构、gemspec 文件和依赖项，以便您可以开始添加自定义代码以使用 Logstash 处理数据。

有关更多信息，请参阅 生成插件

或者，您可以使用我们在 github.com 上托管的示例存储库

您的文件结构应如下所示

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

有关 Ruby gem 文件结构以及 Ruby gem 创建过程的优秀演练的更多信息，请参阅 http://timelessrepo.com/making-ruby-gems

在我们深入了解细节之前，请在您喜欢的文本编辑器中打开插件文件并查看一下。

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 输入插件需要 logstash/inputs/base 和 logstash/namespace 中定义的父类

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

当然，您构建的插件可能依赖于其他代码，甚至 gem。只需将它们与这些 Logstash 依赖项放在一起即可。

输入插件类应该是 LogStash::Inputs::Base 的子类

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

类名应与插件名称紧密对应，例如

LogStash::Inputs::Example

  config_name "example"

这是您的插件在输入配置块内部调用的名称。

如果您在插件代码中设置 config_name "example"，则相应的 Logstash 配置块需要如下所示

input {
  example {...}
}

  public
  def register
  end # def register

Logstash register 方法类似于 initialize 方法。它最初是为了强制调用 super 而创建的，以避免新手的头痛。（注意：为了支持 initialize，可能会删除它，同时强制执行一些测试以确保调用了 super。）

public 表示该方法可以在任何地方调用，而不仅仅是在类中。这是 Ruby 中方法的默认行为，但这里仍然明确指定了它。

您也可以在此处分配实例变量（以 @ 开头的变量）。配置变量现在作为实例变量在作用域中，例如 @message

示例输入插件具有以下 run 方法

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

run 方法是从输入获取的数据流变为事件的地方。

该流可以是纯文本或生成的，如 heartbeat 输入插件所示。在这些情况下，尽管未使用编解码器，但必须在代码中设置 默认编解码器，以避免错误。

这是另一个 run 方法示例

  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

在此示例中，data 被发送到配置块中定义的编解码器，以 decode 数据流并返回事件。

在两个示例中，生成的 event 都被传递给 decorate 方法

      decorate(event)

这将应用您可能在输入配置块中设置的任何标签。例如，tags => ["tag1", "tag2"]。

同样在两个示例中，在“修饰”之后，event 将追加到队列中

      queue << event

这会将事件插入到管道中。

Ruby 中的 require 语句用于包含必要的代码。在某些情况下，您的插件可能需要其他文件。例如，collectd 插件 使用 collectd 提供的 types.db 文件。在插件的主目录中，一个名为 vendor.json 的文件描述了这些文件。

vendor.json 文件包含一个 JSON 对象数组，每个对象描述一个文件依赖项。此示例来自 collectd 编解码器插件

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

另一个 vendor.json 文件示例是 geoip 过滤器

用于下载这些依赖项的过程是调用 rake vendor。这将在本文档的测试部分中进一步讨论。

另一种外部依赖项是 jar 文件。这将在“添加 gemspec 文件”部分中描述。

随着插件的发展，某个选项或功能可能不再满足预期目的，并且开发人员可能希望弃用其用法。弃用会警告用户有关该选项的状态，以便他们在以后的版本中删除该选项时不会措手不及。

Logstash 7.6 引入了弃用记录器，使处理这些情况更容易。您可以使用 适配器 来确保您的插件可以使用弃用记录器，同时仍支持较旧版本的 Logstash。有关更多信息以及有关使用适配器的说明，请参阅 readme。

弃用信息会记录在 log 目录下的 logstash-deprecation.log 文件中。

Gemfiles 允许 Ruby 的 Bundler 管理插件的依赖项。目前，我们只需要 Logstash gem 用于测试，但如果需要其他 gem，你应该在此处添加它们。

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

gemspec 文件的底部有一个注释部分：Gem dependencies。这是任何其他需要的 gem 必须提到的地方。如果一个 gem 对于你的插件运行是必需的，那么它是一个运行时依赖项。如果一个 gem 仅用于测试，那么它将是一个开发依赖项。

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

在某些情况下，例如 Elasticsearch 输出插件，你的代码可能依赖于 jar 文件。在这种情况下，依赖项以这种方式添加到 gemspec 文件中

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

定义这两个之后，安装过程将在 http://mvnrepository.com 上搜索所需的 jar 文件并下载指定的版本。

你已经拥有所有必要的要素，所以让我们继续运行构建命令

gem build logstash-input-example.gemspec

就这样！你的 gem 应该已经构建完成，并且位于同一路径下，名称为

logstash-input-mypluginname-0.1.0.gem

gemspec 文件中的 s.version 数字将提供 gem 版本，在本例中为 0.1.0。

你应该将你的插件测试安装到 Logstash 的全新安装中。从 Logstash 下载页面下载最新版本。

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

示例输入插件将每秒发送 message 的内容（默认消息为“Hello World!”）。

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

随意尝试并通过更改 message 和 interval 参数来测试它

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

恭喜你！你已经构建、部署并成功运行了 Logstash 输入。

Logstash 及其所有插件均根据 Apache License, version 2 ("ALv2") 许可。如果你通过 RubyGems.org 公开你的插件，请确保你的 gemspec 中包含以下行

首先，你需要在 RubyGems.org 上拥有一个帐户

创建帐户后，从 RubyGems.org 获取一个 API 密钥。默认情况下，RubyGems 使用文件 ~/.gem/credentials 来存储你的 API 密钥。这些凭据将用于发布 gem。将 username 和 password 替换为你在 RubyGems.org 上创建的凭据

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

在继续操作之前，请确保你的 gemspec 文件中具有正确的版本，并提交你的更改。

要发布新 logstash gem 的 0.1.0 版本

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

就这样！你的插件已发布！Logstash 用户现在可以通过运行以下命令来安装你的插件

bin/logstash-plugin install logstash-input-mypluginname

在 logstash-plugins 存储库中拥有你的插件的许多好处包括

要开始将你的插件迁移到 logstash-plugins，只需在 Logstash 存储库中创建一个新的 issue 即可。当验收指南完成后，我们将使用推荐的 github 流程来促进迁移到 logstash-plugins 组织。