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
}
參考官方文檔:https://www.elastic.co/guide/en/logstash/7.6/java-filter-plugin.html
感謝您的閱讀,我是Monica23334 || Monica2333 。立下每周寫一篇原創文章flag的小姐姐,關注我並期待打臉吧~