小姐姐教你定製一個Logstash Java Filter~

Logstash是用來收集數據，解析處理數據，最終輸出數據到存儲組件的處理引擎。數據處理流程爲：

Logstash Java Filter 就是基於Logstash的Filter擴展API開發一個用Java語言實現的Filter,而後將Filter代碼打包構建到本身服務器上的Logstash Filter lib中。就能夠在數據流轉配置文件中(也就是logstash -f 指定的配置文件)使用這個定製的Logstash Java Filter了。

定製步驟包括如下五步：

1.準備Logstash環境

由於Logstash Java Filter須要依賴Logstash的API，咱們須要將Logstash源碼下載下來並構建

1.1.下載logstash源碼

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

其中<branch_name>需替換爲你想要使用的logstash版本，使用7.1以後的GA版本就能夠。<target_folder>需替換爲你想要下載到的logstash代碼父目錄,不指定的話就下載到當前目錄的logstash文件夾下。我這裏使用的是7.6版本：

git clone --branch 7.6  --single-branch https://github.com/elastic/logstash.git

1.2.構建logstash源碼

進入到當前目錄的logstash目錄（也就是logstash源碼目錄，後續稱爲：$LS_HOME）下，執行

./gradlew assemble

若是是Windows系統的話執行gradlew.bat assemble

這一步要等好久好久，若是下載不下來能夠試着添加gradle的國內鏡像。
vim $LS_HOME/build.gradle，而後在文件中添加

repositories {
           maven { url 'https://maven.aliyun.com/repository/google/' }
           maven { url 'https://maven.aliyun.com/repository/jcenter/'}
           mavenCentral()
   
           maven {
               url 'https://plugins.gradle.org/m2/'
           }
      }

構建成功後檢查在$LS_HOME/logstash-core/build/libs/目錄下是否生成logstash-core-x.y.z.jar。其中x,y,z是你下載的logstash版本號。個人就是

/Users/xx/corprepo/logstash/logstash-core/build/libs/logstash-core-7.6.3.jar

2.編寫Logstash Java Filter代碼

2.1.下載官方demo

官方提供了一個demo，咱們能夠下載下來基於這個demo作修改。

2.2.指定LOGSTASH_CORE_PATH

下載下來demo後，在項目根目錄建立gradle.properties文件，添加一行數據：

LOGSTASH_CORE_PATH=<target_folder>/logstash-core

2.3.開發Filter代碼

咱們須要繼承Logstash的Filter API實現咱們本身的Java Filter功能。一個實現好的Filter以下：

import co.elastic.logstash.api.Configuration;
import co.elastic.logstash.api.Context;
import co.elastic.logstash.api.Event;
import co.elastic.logstash.api.Filter;
import co.elastic.logstash.api.FilterMatchListener;
import co.elastic.logstash.api.LogstashPlugin;
import co.elastic.logstash.api.PluginConfigSpec;
import org.apache.commons.lang3.StringUtils;

import java.util.Collection;
import java.util.Collections;
//類名必須按照駝峯命名匹配這個下劃線註解名，JavaFilterExample -> java_filter_example
@LogstashPlugin(name = "java_filter_example")
public class JavaFilterExample implements Filter {
    //定義一個該Filter支持的setting配置。名字是source，默認值爲message
    //可從filter方法中看出是拿 SOURCE_CONFIG 的value值作field 的名稱使用的
    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) {
        // constructors should validate configuration options
        this.id = id;
        this.sourceField = config.get(SOURCE_CONFIG);
    }

    /**
     * 該Filter的過濾邏輯，能夠對輸入的event數據作各類CRUD操做
     * @param events
     * @param matchListener
     * @return 最終流轉到下一個pipeline的數據，若是有符合條件的event必須返回
     */
    @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;
    }
    /**
     *
     * @return 返回該Filter支持的全部setting配置
     */
    @Override
    public Collection<PluginConfigSpec<?>> configSchema() {
        // should return a list of all configuration options for this plugin
        return Collections.singletonList(SOURCE_CONFIG);
    }

    /**
     *
     * @return 該Filter的ID，Logstash會幫咱們賦值
     */
    @Override
    public String getId() {
        return this.id;
    }
}

其中須要注意兩點：

• @LogstashPlugin註解的name必須和類名高度保持一致。如java_filter_example-> JavaFilterExample（我特麼反正是被坑了。。）
• 須要實現co.elastic.logstash.api.Filter類，若是你import不成功，那就是gradle.properties配置不成功 或者 構建logstash源碼不成功。重寫其三個方法：

getId方法

返回該Filter的ID，Logstash會幫咱們賦值。咱們只須要定義一個成員變量，構造方法中賦值進去就行了。

configSchema方法

返回該Filter支持的全部setting配置集合。PluginConfigSpec定義的setting配置就是咱們在logstash的配置文件中使用該Filter時，能夠傳的參數，如在使用grok Filter時傳進去的patterns_dir和match。

filter {
      grok {
        patterns_dir => ["./patterns"]
        match => { "message" => "%{SYSLOGBASE} %{POSTFIX_QUEUEID:queue_id}: %{GREEDYDATA:syslog_message}" }
      }
    }

這個setting配置PluginConfigSpec支持的配置參數有name, type, deprecation status, required status, 和 default value。

在咱們的Filter類中咱們定義了`PluginConfigSpec<String> SOURCE_CONFIG =

PluginConfigSpec.stringSetting("source", "message");`其中`name=source, default value= message`

filter方法

過濾器固然要幹過濾邏輯的事了。其中入參Collection<Event> events是咱們要處理的輸入過來的數據，咱們能夠針對邏輯作一些CURD操做。入參FilterMatchListener matchListener是該 Filter將知足本身邏輯的event數據通知給matchListener. 如Logstash中matchListener的實現爲DecoratingFilterMatchListener。它能作的操做好比有ADD_FIELD
一樣須要咱們先定義PluginConfigSpec，而後在使用該Filter時配置add_field參數。如grok Filter就支持該參數和該DecoratingFilterMatchListener

filter {
      grok {
        add_field => { "foo_%{somefield}" => "Hello world, from %{host}" }
      }
    }

沒有通知matchListener的需求時就不用調用matchListener.filterMatched(e)了。

3.單元測試

demo裏面也有測試類，run一下就完了。。

4.打包部署Filter

咱們須要使用gradle將咱們的Filter項目達成ruby gem包，因此最好必定要基於demo項目中的gradle配置文件修改。

4.1.配置gradle打包任務
編輯項目根路徑下的build.gradle文件

plugin info部分是咱們Filter的信息，其中須要修改的特別注意點我已經用TODO標示出來了。
4.2.運行gradle打包任務
在項目根目錄下執行

./gradlew gem

Windows系統執行gradlew.bat gem

執行成功以後會看到在項目根目錄下生成logstash-{plugintype}-<pluginName>-<version>.gem文件

4.3.到Logstash中安裝filter gem包
到logstash目錄（$LS_HOME）下執行

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

其中 /path/to/javaPlugin.gem就是咱們4.2步驟中生成的gem絕對路徑。

5.使用咱們的Java Filter運行Logstash

5.1.在$LS_HOME/config目錄下建立logstash運行配置文件java_filter.conf

input {
  generator { message => "Hello world!" count => 1 }
}
filter {
# java_filter_example：咱們的filter中@LogstashPlugin註解的name
  java_filter_example {}
}
output {
  stdout { codec => rubydebug }
}

5.2.啓動Logstash
在$LS_HOME運行

bin/logstash -f  config/java_filter.conf

至此就成功啦～

{
       "message" => "!dlrow olleH",
      "sequence" => 0,
      "@version" => "1",
          "host" => "xxdeMacBook-Pro.local",
    "@timestamp" => 2020-04-12T13:15:30.376Z
}

感謝您的閱讀，我是Monica23334 || Monica2333 。立下每週寫一篇原創文章flag的小姐姐，關注我並期待打臉吧～

參考官方文檔：https://www.elastic.co/guide/en/logstash/7.6/java-filter-plugin.html