Logstash 簡介：
Logstash 是一個實時數據收集引擎，可收集各類型數據並對其進行分析，過濾和歸納。按照自己條件分析過濾出符合數據導入到可視化界面。Logstash 建議使用java1.8 有些版本是不支持的，比如java1.9。
一. 下載安裝jdk1.8
下載地址：http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html
下載好的安裝包上傳到cpy04.dev.xjh.com的/usr/local/ 目錄下並三執行如下操作：
#解壓文件 tar xf /usr/local/jdk1.8.0_111.tar.gz -C /usr/local mv /usr/local/jdk1.8.0_111 /usr/local/jdk-1.8.0 #添加環境變量 alternatives --install /usr/bin/java java /usr/local/jdk1.8.0/jre/bin/java 3000 alternatives --install /usr/bin/jar jar /usr/local/jdk1.8.0/bin/jar 3000 alternatives --install /usr/bin/javac javac /usr/local/jdk1.8.0/bin/javac 3000 alternatives --install /usr/bin/javaws javaws /usr/local/jdk1.8.0/jre/bin/javaws 3000 alternatives --set java /usr/local/jdk1.8.0/jre/bin/java alternatives --set jar /usr/local/jdk1.8.0/bin/jar alternatives --set javac /usr/local/jdk1.8.0/bin/javac alternatives --set javaws /usr/local/jdk1.8.0/jre/bin/javaws #切換java 版本 alternatives --config java
二. 安裝logstash
1. 登陸cpy04.dev.xjh.com(需下載其他版本請點擊：https://www.elastic.co/downloads/logstash )
wget https://artifacts.elastic.co/downloads/logstash/logstash-5.6.1.tar.gz -o /opt/logstash-5.6.1.tar.gz tar xf /opt/logstash-5.6.1.tar.gz -C /usr/local mv /usr/local/logstash-5.6.1 /usr/local/logstash
三、配置logstash
1. 編輯 /usr/local/logstash/config/logstash.yml配置文件修改如下內容：
node.name: cpy04.dev.xjh.com #設置節點名稱，一般寫主機名 path.data: /usr/local/logstash/plugin-data #創建logstash 和插件使用的持久化目錄 config.reload.automatic: true #開啟配置文件自動加載 config.reload.interval: 10 #定義配置文件重載時間周期 http.host: "cpy04.dev.xjh.com" #定義訪問主機名，一般為域名或IP
2. 新建持久化目錄：
mkdir -p /usr/local/logstash/plugin-data
3. 配置logstash 從Filebeat 輸入、過濾、輸出至elasticsearch（logstash 有非常多插件，詳見官網,此處不列舉）
3.1 安裝logstash-input-jdbc 和logstash-input-beats-master 插件
/usr/local/logstash/bin/logstash-plugin install logstash-input-jdbc wget https://github.com/logstash-plugins/logstash-input-beats/archive/master.zip -O /opt/master.zip unzip -d /usr/local/logstash /opt/master.zip
3.2 配置logstash input 段
vim /usr/local/logstash/from_beat.conf
input {
beats {
port => 5044
}
}
output {
stdout { codec => rubydebug }
}
啟動logstash 看是否能接收到filebeat 傳過來的日志內容，要確保filebeat 在日志節點上啟動正常。此時只測試傳入是否正常，並未對原始日志進行過濾和篩選
/usr/local/logstash/bin/logstash -f /usr/local/logstash/config/from_beat.conf
啟動后如果沒有報錯需要等待logstash 完成，此時間可能比較長
3.3 配置 logstash filter 段，修改/usr/local/logstash/from_beat.conf 為以下內容，配置完成后再次啟動logstash，此時如果成功，輸出內容應該是自己正則表達式捕獲后的字段切分內容。
input {
beats {
port => 5044
}
}
filter {
#過濾access 日志
if ( [source] =~ "localhost\_access\_log" ) {
grok {
match => {
message => [ "%{COMMONAPACHELOG}" ]
}
}
date {
match => [ "request_time", "ISO8601" ]
locale => "cn"
target => "request_time"
}
#過濾tomcat日志
} else if ( [source] =~ "catalina" ) {
#使用正則匹配內容到字段
grok {
match => {
message => [ "(?<webapp_name>\[\w+\])\s+(?<request_time>\d{4}\-\d{2}\-\d{2}\s+\w{2}\:\w{2}\:\w{2}\,\w{3})\s+(?<log_level>\w+)\s+(?<class_package>[^.^\s]+(?:\.[^.\s]+)+)\.(?<class_name>[^\s]+)\s+(?<message_content>.+)" ]
}
}
#解析請求時間
date {
match => [ "request_time", "ISO8601" ]
locale => "cn"
target => "request_time"
}
} else {
drop {}
}
}
output {
stdout { codec => rubydebug }
}
3.4 配置 過濾后內容輸出至elasticsearch，修改from_beat.conf 文件為以下內容：
input {
beats {
port => 5044
}
}
filter {
#過濾access 日志
if ( [source] =~ "localhost\_access\_log" ) {
grok {
match => {
message => [ "%{COMMONAPACHELOG}" ]
}
}
date {
match => [ "request_time", "ISO8601" ]
locale => "cn"
target => "request_time"
}
#過濾tomcat日志
} else if ( [source] =~ "catalina" ) {
#匹配內容到字段
grok {
match => {
message => [ "(?<webapp_name>\[\w+\])\s+(?<request_time>\d{4}\-\d{2}\-\d{2}\s+\w{2}\:\w{2}\:\w{2}\,\w{3})\s+(?<log_level>\w+)\s+(?<class_package>[^.^\s]+(?:\.[^.\s]+)+)\.(?<class_name>[^\s]+)\s+(?<message_content>.+)" ]
}
}
#解析請求時間
date {
match => [ "request_time", "ISO8601" ]
locale => "cn"
target => "request_time"
}
} else {
drop {}
}
}
output {
if ( [source] =~ "localhost_access_log" ) {
elasticsearch {
hosts => ["cpy04.dev.xjh.com:9200"]
index => "access_log"
}
} else {
elasticsearch {
hosts => ["cpy04.dev.xjh.com:9200"]
index => "tomcat_log"
}
}
stdout { codec => rubydebug }
}
至此，logstash 配置完成。如果需要做其他過濾或者輸出至除elasticsearch 以外插件，如kafka 詳見：https://www.elastic.co/guide/en/logstash/current/index.html
input {
beats {
port =>
5044
}
}
filter {
#過濾access 日志
if
( [source] =~
"localhost\_access\_log"
) {
grok {
match => {
message => [
"%{COMMONAPACHELOG}"
]
}
}
date {
match => [
"request_time"
,
"ISO8601"
]
locale =>
"cn"
target =>
"request_time"
}
#過濾tomcat日志
}
else
if
( [source] =~
"catalina"
) {
#匹配內容到字段
grok {
match => {
message => [
"(?<webapp_name>\[\w+\])\s+(?<request_time>\d{4}\-\d{2}\-\d{2}\s+\w{2}\:\w{2}\:\w{2}\,\w{3})\s+(?<log_level>\w+)\s+(?<class_package>[^.^\s]+(?:\.[^.\s]+)+)\.(?<class_name>[^\s]+)\s+(?<message_content>.+)"
]
}
}
#解析請求時間
date {
match => [
"request_time"
,
"ISO8601"
]
locale =>
"cn"
target =>
"request_time"
}
}
else
{
drop {}
}
}
output {
if
( [source] =~
"localhost_access_log"
) {
elasticsearch {
hosts => [
"cpy04.dev.xjh.com:9200"
]
index =>
"access_log"
}
}
else
{
elasticsearch {
hosts => [
"cpy04.dev.xjh.com:9200"
]
index =>
"tomcat_log"
}
}
stdout { codec => rubydebug }
}