ELK日志分析平臺(三）——Logstash簡介+安裝+常用插件示例

什么是Logstash

Logstash是資料處理管道，它的主要作用是做資料采集，

logstash擁有200多個插件，能夠同時從多個來源采集資料，轉換資料，然后將資料發送到的 “存盤庫” 中，（大多都是 Elasticsearch，）

Logstash管道有兩個必需的元素，輸入和輸出，以及一個可選元素過濾器，

輸入： 采集各種樣式、大小和來源的資料

Logstash 支持各種輸入選擇 ，同時從眾多常用來源捕捉事件，

能夠以連續的流式傳輸方式，輕松地從日志、指標、Web 應用、資料存盤以及各種 AWS 服務采集資料，

過濾器：實時決議和轉換資料

資料從源傳輸到存盤庫的程序中，Logstash 過濾器能夠決議各個事件，識別已命名的欄位以構建結構，并將它們轉換成通用格式，以便更輕松、更快速地分析和實作商業價值，

利用 Grok 從非結構化資料中派生出結構
從 IP 地址破譯出地理坐標
將 PII 資料匿名化，完全排除敏感欄位
簡化整體處理，不受資料源、格式或架構的影響

輸出：選擇您的存盤庫，匯出您的資料

盡管 Elasticsearch 是我們的首選輸出方向，能夠為我們的搜索和分析帶來無限可能，但它并非唯一選擇，

Logstash 提供眾多輸出選擇，您可以將資料發送到您要指定的地方，并且能夠靈活地解鎖眾多下游用例，

Logstash安裝與配置

軟體下載
https://elasticsearch.cn/download/
logstash安裝
# rpm -ivh jdk-8u171-linux-x64.rpm
# rpm -ivh logstash-7.6.1.rpm

file 輸出插件

1.標準輸入到標準輸出

/usr/share/logstash/bin/logstash -e 'input { stdin { } } output { stdout {} }'

2.檔案指定標準輸入輸出

-f選擇指定檔案就可

3.將輸入資料指定檔案輸出

# vim /etc/logstash/conf.d/file.conf
input {
        stdin { }
}
output {
        file {
                path => "/tmp/logstash.txt"        	#輸出的檔案路徑
                codec => line { format => "custom format: %{message}"}							#定制資料格式
        }
}

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/file.conf

在標準輸入中鍵入后可以查看：
# cat /tmp/logstash.txt

elasticsearch輸出插件

1.標準輸入到es主機

# vim /etc/logstash/conf.d/es.conf
input {
        stdin { }
}
output {
        elasticsearch {
                hosts => "192.168.122.101:9200"    #輸出到的ES主機與埠
                index => "logstash-%{+YYYY.MM.dd}"   #定制索引名稱
        }
}

直接輸出到es主機

洗掉創建的資訊索引

index數值為unknown

查看輸出到主機的資料

也可以將資料輸出到終端，然后將資料輸出到主機上去

# vim /etc/logstash/conf.d/es.conf
input {
        stdin { }
}
output {
       stdout { }
        elasticsearch {
                hosts => "192.168.122.101:9200"    #輸出到的ES主機與埠
                index => "logstash-%{+YYYY.MM.dd}"   #定制索引名稱
        }
}

file輸入插件

1.把檔案內容輸出到es主機

# vim /etc/logstash/conf.d/es.conf

input {
        file {
		path => "/var/log/messages"
		start_position => "beginning"
	}
}
output {
        stdout {}
        elasticsearch {
                hosts => "192.168.122.101:9200"
                index => "syslog-%{+YYYY.MM.dd}"
        }
}

logstash程式是以普通用戶身份來運行的要確保可以操作該檔案

瀏覽器中查看收集到的日志資料

logstash如何區分設備、檔案名、檔案的不同版本

logstash會把進度保存到sincedb檔案中

sincedb檔案內容解釋

# cat .sincedb_452905a167cf4509fd08acb964fdb20c
34507 0 64768 247396 1603976120.589161 /var/log/messages

sincedb檔案一共6個欄位

再一次執行傳輸日志資料的命令時，不會重復執行，因為檔案中沒有變動的資料，指標走到結尾，重啟服務也不會改變，
當洗掉es主機的索引，要創建新的索引也不能重復執行，
當洗掉.sincedb_452905a167cf4509fd08acb964fdb20c檔案時，可以重復執行，

syslog輸入插件

logstash可以偽裝成日志服務器，直接接受遠程日志，

# vim /etc/logstash/conf.d/syslog.conf
input {
        syslog {
                type => "rsyslog"
                port => 514		#日志收集埠
        }
}
output {
        stdout { }
}

指定組態檔運行

# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/syslog.conf
# netstat -antulp |grep :514
tcp6       0      0 :::514                  :::*                    LISTEN      3171/java           
udp        0      0 0.0.0.0:514             0.0.0.0:*                           3171/java     

重新打開一個終端

配置server2客戶端傳入日志

# vim /etc/rsyslog.conf 
*.*         @@192.168.122.201:514    
# systemctl restart rsyslog  

此時查看logstash終端上，已經開始采集server2的日志

多行過濾插件

多行過濾可以把多行日志記錄合并為一行事件

# vim /etc/logstash/conf.d/multiline.conf
input {
        stdin {
        codec => multiline {
                pattern => "^EOF"
                negate => true
                what => previous
                }
        }
}
output {
 stdout {}
}

遇見EOF關鍵字會向上合并

多行過濾插件過濾遠程主機的日志

[root@git1 conf.d]# vim es.conf 


input {
        file {
                path => "/var/log/my-es.log"
                start_position => "beginning"
                codec => multiline {
                pattern => "^\[" 
                negate => true
                what => previous
        }
      }  
}
output {
      
        elasticsearch {
                hosts => ["192.168.122.101:9200"]
                index => "server2log-%{+YYYY.MM.dd}" ##保存在es的名字
        }
}

==檔案每行是以[開頭，所以有pattern => “^[” ==

[root@git1 conf.d]#  /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/es.conf 

grok過濾插件

# vim /etc/logstash/conf.d/grok.conf

input {
	stdin {}
}
filter {
	grok {
	match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
	}
}
output {
	stdout {}
	}

指定組態檔運行

# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf
查看輸出
55.3.244.1 GET /index.html 15824 0.043

將非結構化資料轉變為結構化資料

apache服務日志過濾實戰

# vim /etc/logstash/conf.d/apache.conf

 

input {
        file {
		path => "/var/log/httpd/access_log"
		start_position => "beginning"
	}
}
filter{
	grok {
		match => { "message" => "%{HTTPD_COMBINEDLOG}" }
	}
}
output {
        elasticsearch {
                hosts => "172.25.0.13:9200"
                index => "apachelog-%{+YYYY.MM.dd}"
        }
}

[root@git1 conf.d]# systemctl start httpd
[root@git1 conf.d]# cd /var/www/html/
[root@git1 html]# ls
[root@git1 html]# echo hello > index.html

修改權限

root@git1 html]# cd /var/log/httpd/
[root@git1 httpd]# ll
總用量 16
-rw-r--r-- 1 root root 10400 10月 29 22:52 access_log
-rw-r--r-- 1 root root   814 10月 29 22:49 error_log
[root@git1 httpd]# chmod 755 access_log
[root@git1 httpd]# chmod 755 error_log 

日志格式的模塊

[root@git1 httpd]# cd /usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns/
[root@git1 patterns]# ls
aws     exim           httpd         maven                 nagios      ruby
bacula  firewalls      java          mcollective           postgresql  squid
bind    grok-patterns  junos         mcollective-patterns  rails
bro     haproxy        linux-syslog  mongodb               redis

[root@git1 conf.d]# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/apache.conf