Logstash是一個具有實時管線能力的開源數據收集引擎忍法。在ELK Stack中逞力，通常選擇更輕量級的Filebeat收集日志啦吧，然后將日志輸出到Logstash進行加工處理绿店，再將處理后的日志輸出到指定的目標（ElasticSearch妙痹，Kafka等）當中铸史。
Logstash事件的處理管線是inputs → filters → outputs，三個階段都可以自定義插件细诸，本文主要介紹如何開發(fā)自定義需求最多的filter插件沛贪。
Logstash的安裝就不詳細介紹了，下載傳送門：https://www.elastic.co/downloads/logstash震贵。

生成filter插件

cd到Logstash的跟目錄，使用bin/logstash-plugin生成filter插件模板水评，如下：

bin/logstash-plugin generate --type filter --name test  --path vendor/localgems

vendor/localgems可修改為你自己的路徑猩系。
查看filter插件的目錄結構，如下：

$ tree logstash-filter-test
├── Gemfile
├── LICENSE
├── README.md
├── Rakefile
├── lib
│   └── logstash
│       └── filters
│           └── test.rb
├── logstash-filter-test.gemspec
└── spec
    └── filters
        └── test_spec.rb
    └── spec_helper.rb

filter插件初探

代碼結構

Logstash插件是用ruby寫的中燥，查看lib/logstash/filters/test.rb文件寇甸，如下：

# encoding: utf-8
require "logstash/filters/base"
require "logstash/namespace"

# This  filter will replace the contents of the default 
# message field with whatever you specify in the configuration.
#
# It is only intended to be used as an .
class LogStash::Filters::Test < LogStash::Filters::Base

  # Setting the config_name here is required. This is how you
  # configure this filter from your Logstash config.
  #
  # filter {
  #    {
  #     message => "My message..."
  #   }
  # }
  #
  config_name "test"
  
  # Replace the message with this value.
  config :message, :validate => :string, :default => "Hello World!"
  

  public
  def register
    # Add instance variables 
  end # def register

  public
  def filter(event)

    if @message
      # Replace the event message with our message as configured in the
      # config file.
      event.set("message", @message)
    end

    # filter_matched should go in the last line of our successful code
    filter_matched(event)
  end # def filter
end # class LogStash::Filters::Test

UTF-8編碼

Logstash依賴于UTF-8編碼，需要在插件代碼開始出添加：

# encoding: utf-8

require

模板代碼里面默認require了"logstash/filters/base"和"logstash/namespace"，如果需要依賴其它代碼或者gems就在這添加拿霉，可以參考后面在插件中查詢MySql的代碼吟秩。

插件名稱配置

插件名稱配置代碼如下：

config_name "test"

test就是插件名稱，在Logstash配置的filter塊中使用绽淘。

插件參數配置

插件參數配置代碼如下：

config :message, :validate => :string, :default => "Hello World!"

message是插件test的可選參數涵防，默認值是"Hello World!"。下面是參數的通用配置代碼：

config :variable_name, :validate => :variable_type, :default => "Default value", :required => boolean, :deprecated => boolean, :obsolete => string

• :variable_name：參數名稱
• :validate：驗證參數類型沪铭，如:string, :password, :boolean, :number, :array, :hash, :path等
• :required：是否必須配置
• :default：默認值
• :deprecated：是否廢棄
• :obsolete：聲明該配置不再使用壮池，通常提供升級方案

插件方法

Logstash插件必須實現兩個方法：register和filter。
register方法代碼如下：

  public
  def register
    # Add instance variables 
  end # def register

register方法相當于初始化方法杀怠，不需要手動調用椰憋，可以在這個方法里面調用配置變量，如@message赔退，也可以初始化自己的實例變量橙依。
filter方法代碼如下：

  public
  def filter(event)

    if @message
      # Replace the event message with our message as configured in the
      # config file.
      event.set("message", @message)
    end

    # filter_matched should go in the last line of our successful code
    filter_matched(event)
  end # def filter

filter方法是插件的數據處理邏輯，其中event變量封裝了數據流硕旗，可以通過接口訪問event中的內容票编，具體參見https://www.elastic.co/guide/en/logstash/5.1/event-api.html。最后一句調用了filter_matched卵渴，這個方法用于保證Logstash的配置add_field, remove_field, add_tag 和remove_tag會被正確執(zhí)行慧域。

在插件中使用其它類庫

這里以在插件中查詢MySql為例進行說明，使用jdbc操作MySql浪读，需要安裝jdbc-mysql昔榴，操作如下：
添加Logstash的環(huán)境變量：

export LOGSTASH_HOME=/opt/logstash-5.2.1
export PATH=$PATH:$LOGSTASH_HOME/vendor/jruby/bin

安裝jdbc-mysql：

gem install jdbc-mysql

使用sequel（代碼和文檔請查看vendor/bundle/jruby/1.9/gems/sequel-4.43.0）操作MySql，首先需要在logstash-filter-test.gemspec配置文件中添加對sequel的依賴碘橘，如下：

# Gem dependencies
s.add_runtime_dependency "logstash-core-plugin-api", "~> 2.0"
s.add_runtime_dependency 'sequel'
s.add_development_dependency 'logstash-devutils'

然后在test.rb中require相關代碼：

require "sequel"
require "sequel/adapters/jdbc"

在test.rb中添加:jdbc_driver_library配置參數互订，用于配置jdbc驅動庫的path，我這的路徑是"/usr/local/lib/ruby/gems/2.3.0/gems/jdbc-mysql-5.1.40/lib/mysql-connector-java-5.1.40-bin.jar"痘拆。

config :jdbc_driver_library, :validate => :string, :required => true

register方法中做了兩件事仰禽，一是初始化了幾個實例變量，二是require依賴的jdbc庫纺蛆。簡單說明下實例變量的用途吐葵，@logger用于輸出日志，@connection_retry_attempts和@connection_retry_attempts_wait_time用于數據庫連接重試桥氏，@connection_wait_timeout用于設置MySql的session超時時間温峭，避免與MySql連接過多，這是一個雙保險策略字支，正常情況下MySql會設置全局的超時時間凤藏，并且查詢完成之后我們會主動斷開連接（見fetch_info方法）奸忽，在斷開失敗且MySql的超時時間過長時@connection_wait_timeout才會起作用。

public
def register
  # Add instance variables 
  @logger = self.logger
  @connection_retry_attempts = 5
  @connection_retry_attempts_wait_time = 1
  @connection_wait_timeout = 10
  begin
    require @jdbc_driver_library
  rescue => e
    @logger.error("Failed to load #{@jdbc_driver_library}", :exception => e)
  end
end # def register

創(chuàng)建db實例：

private 
def create_db(conn_str)
  db = nil
  retry_attempts = @connection_retry_attempts
  while retry_attempts > 0 do
    retry_attempts -= 1
    begin
      tmp_db = Sequel.connect(conn_str)
    rescue Sequel::PoolTimeout => e
      if retry_attempts <= 0
        @logger.error("Failed to connect to database. 5 second timeout exceeded. Tried #{@connection_retry_attempts} times.")
        raise e
      else
        @logger.error("Failed to connect to database. 5 second timeout exceeded. Trying again.")  
      end
    rescue Sequel::Error => e
      if retry_attempts <= 0
        @logger.error("Unable to connect to database. Tried #{@connection_retry_attempts} times", :error_message => e.message)
        raise e
      else
        @logger.error("Unable to connect to database. Trying again", :error_message => e.message)
      end
    else
      db = tmp_db
      break
    end
    sleep(@connection_retry_attempts_wait_time)
  end
  db
end

查詢數據：

private
def fetch_info(db, sql, key)
  all_info = {}
  retry_attempts = @connection_retry_attempts
  while retry_attempts > 0 do
    retry_attempts -= 1
    begin
      db.fetch(sql) do |row|
        all_info[row[key]] = row
      end
      db.run "set wait_timeout = " + @connection_wait_timeout.to_s
    rescue Sequel::DatabaseConnectionError, Sequel::DatabaseError => e
      if retry_attempts <= 0
        @logger.warn("Exception when executing JDBC query", :exception => e)
        raise e
      else
        @logger.error("Failed to execute query. Trying again.", :error_message => e.message)
      end
    else
      break
    end
    sleep(@connection_retry_attempts_wait_time)
  end
  db.disconnect()
  all_info
end

接下來就可以根據需要在register和filter中使用create_db和fetch_info方法了揖庄。
注意：這里只是以查詢MySql為例進行說明栗菜，處理Logstash事件時需要考慮對性能和吞吐量的影響。

在Logstash中配置定制的插件

cd到Logstash根目錄下蹄梢，在Gemfile添加以下配置：

gem "logstash-filter-test", :path => "vendor/localgems/logstash-filter-test"

啟動Logstash

啟動Logstash疙筹，配置我們定制的test插件，如下：

bin/logstash -e 'input { beats { port => "5043" } } filter { test { jdbc_driver_library => "/usr/local/lib/ruby/gems/2.3.0/gems/jdbc-mysql-5.1.40/lib/mysql-connector-java-5.1.40-bin.jar" } } output { stdout { codec => rubydebug }}'

也可以寫配置文件检号，與上面的-e參數內容一致腌歉，然后使用配置文件啟動Logstash。
啟動Logstash的傳送門：https://www.elastic.co/guide/en/logstash/5.1/running-logstash-command-line.html齐苛。