如何定制一個(gè)Logstash Java Filter

Logstash是用來(lái)收集數(shù)據(jù)，解析處理數(shù)據(jù)，最終輸出數(shù)據(jù)到存儲(chǔ)組件的處理引擎。數(shù)據(jù)處理流程為：

Logstash Java Filter 就是基于Logstash的Filter擴(kuò)展API開(kāi)發(fā)一個(gè)用Java語(yǔ)言實(shí)現(xiàn)的Filter,然后將Filter代碼打包構(gòu)建到自己服務(wù)器上的Logstash Filter lib中。就可以在數(shù)據(jù)流轉(zhuǎn)配置文件中(也就是logstash -f 指定的配置文件)使用這個(gè)定制的Logstash Java Filter了。

定制步驟包括以下五步：

1.準(zhǔn)備Logstash環(huán)境

因?yàn)長(zhǎng)ogstash Java Filter需要依賴Logstash的API，我們需要將Logstash源碼下載下來(lái)并構(gòu)建
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代碼父目錄,不指定的話就下載到當(dāng)前目錄的logstash文件夾下。我這里使用的是7.6版本：

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

1.2.構(gòu)建logstash源碼
進(jìn)入到當(dāng)前目錄的logstash目錄（也就是logstash源碼目錄，后續(xù)稱為：$LS_HOME）下，執(zhí)行

./gradlew assemble

如果是Windows系統(tǒng)的話執(zhí)行gradlew.bat assemble

這一步要等很久很久，如果下載不下來(lái)可以試著添加gradle的國(guó)內(nèi)鏡像。
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/'
           }
      }

構(gòu)建成功后檢查在$LS_HOME/logstash-core/build/libs/目錄下是否生成logstash-core-x.y.z.jar。其中x,y,z是你下載的logstash版本號(hào)。我的就是

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

2.編寫(xiě)Logstash Java Filter代碼

2.1.下載官方demo
官方提供了一個(gè)demo，我們可以下載下來(lái)基于這個(gè)demo做修改。

2.2.指定LOGSTASH_CORE_PATH
下載下來(lái)demo后，在項(xiàng)目根目錄創(chuàng)建gradle.properties文件，

添加一行數(shù)據(jù)：

LOGSTASH_CORE_PATH=<target_folder>/logstash-core

2.3.開(kāi)發(fā)Filter代碼
我們需要繼承Logstash的Filter API實(shí)現(xiàn)我們自己的Java Filter功能。一個(gè)實(shí)現(xiàn)好的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;
//類名必須按照駝峰命名匹配這個(gè)下劃線注解名，JavaFilterExample -> java_filter_example
@LogstashPlugin(name = "java_filter_example")
public class JavaFilterExample implements Filter {
    //定義一個(gè)該Filter支持的setting配置。名字是source，默認(rèn)值為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的過(guò)濾邏輯，可以對(duì)輸入的event數(shù)據(jù)做各種CRUD操作
     * @param events
     * @param matchListener
     * @return 最終流轉(zhuǎn)到下一個(gè)pipeline的數(shù)據(jù)，如果有符合條件的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會(huì)幫我們賦值
     */
    @Override
    public String getId() {
        return this.id;
    }
}

其中需要注意兩點(diǎn)：

  • @LogstashPlugin注解的name必須和類名高度保持一致。如java_filter_example-> JavaFilterExample（我特么反正是被坑了。。）
  • 需要實(shí)現(xiàn)co.elastic.logstash.api.Filter類，如果你import不成功，那就是gradle.properties配置不成功 或者 構(gòu)建logstash源碼不成功。重寫(xiě)其三個(gè)方法：

getId方法
返回該Filter的ID，Logstash會(huì)幫我們賦值。我們只需要定義一個(gè)成員變量，構(gòu)造方法中賦值進(jìn)去就好了。

configSchema方法
返回該Filter支持的所有setting配置集合。PluginConfigSpec定義的setting配置就是我們?cè)趌ogstash的配置文件中使用該Filter時(shí)，可以傳的參數(shù)，如在使用grok Filter時(shí)傳進(jìn)去的patterns_dir和match。

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

這個(gè)setting配置PluginConfigSpec支持的配置參數(shù)有name, type, deprecation status, required status, 和 default value。

在我們的Filter類中我們定義了PluginConfigSpec<String> SOURCE_CONFIG = PluginConfigSpec.stringSetting("source", "message");其中name=source, default value= message

filter方法
過(guò)濾器當(dāng)然要干過(guò)濾邏輯的事了。其中入?yún)?code>Collection<Event> events是我們要處理的輸入過(guò)來(lái)的數(shù)據(jù)，我們可以針對(duì)邏輯做一些CURD操作。入?yún)?code>FilterMatchListener matchListener是該 Filter將滿足自己邏輯的event數(shù)據(jù)通知給matchListener. 如Logstash中matchListener的實(shí)現(xiàn)為DecoratingFilterMatchListener。它能做的操作比如有ADD_FIELD
同樣需要我們先定義PluginConfigSpec，然后在使用該Filter時(shí)配置add_field參數(shù)。如grok Filter就支持該參數(shù)和該DecoratingFilterMatchListener

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

沒(méi)有通知matchListener的需求時(shí)就不用調(diào)用matchListener.filterMatched(e)了。

3.單元測(cè)試

demo里面也有測(cè)試類，run一下就完了。。

4.打包部署Filter

我們需要使用gradle將我們的Filter項(xiàng)目達(dá)成ruby gem包，所以最好一定要基于demo項(xiàng)目中的gradle配置文件修改。

4.1.配置gradle打包任務(wù)
編輯項(xiàng)目根路徑下的build.gradle文件


plugin info部分是我們Filter的信息，其中需要修改的特別注意點(diǎn)我已經(jīng)用TODO標(biāo)示出來(lái)了。
4.2.運(yùn)行g(shù)radle打包任務(wù)
在項(xiàng)目根目錄下執(zhí)行

./gradlew gem

Windows系統(tǒng)執(zhí)行gradlew.bat gem

執(zhí)行成功之后會(huì)看到在根目錄下生成logstash-{plugintype}-<pluginName>-<version>.gem文件


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

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

其中 /path/to/javaPlugin.gem就是我們4.2步驟中生成的gem絕對(duì)路徑。成功可以看到

5.使用我們的Java Filter運(yùn)行Logstash

5.1.在$LS_HOME/config目錄下創(chuàng)建logstash運(yùn)行配置文件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.啟動(dòng)Logstash
在$LS_HOME運(yùn)行

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
}

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

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容