logstash日志處理管道通過(guò)插件完成日志數(shù)據(jù)的采集處理和輸出數(shù)據(jù)源 ↓ Input輸入插件 ↓ Filter過(guò)濾/處理插件 ↓ Output輸出插件 ↓ 目標(biāo)系統(tǒng)Elasticsearch、文件、數(shù)據(jù)庫(kù)等部署logstashyum install -y jdk-11.0.15_linux-x64_bin.rpmyum install -y logstash-7.6.1.rpmelasticsearch輸出插件屬于Output階段logstash采集并處理完的日志數(shù)據(jù)發(fā)送到Elasticsearch1、編輯Logstash Pipeline 配置文件用來(lái)描述數(shù)據(jù)處理流程vim /etc/logstash/conf.d/test.conf寫(xiě)入input { stdin { } } output { stdout {} elasticsearch { hosts 192.168.56.11:9200 index logstash-%{YYYY.MM.dd} } }2、啟動(dòng)/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/test.conf輸入內(nèi)容 westos linux返回內(nèi)容file輸入插件Input階段從文件讀取日志Logstash Pipeline 配置文件格式如下input { file { path /var/log/httpd/*_log start_position beginning } } output { stdout {} elasticsearch { hosts 192.168.56.11:9200 index filelog-%{YYYY.MM.dd} } }啟動(dòng)/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/filelog.conf終端輸出可視化界面輸出sincedb文件讀取進(jìn)度記錄本配合file input使用用來(lái)記錄文件讀取到哪里了下次logstash重啟時(shí)不需要從頭再讀而是從上次位置繼續(xù)避免了大量重復(fù)日志cd /usr/share/logstash/data/plugins/inputs/file/l.sincedb每列解析68801905 0 64768 5596 1786502152.007266 /var/log/httpd/access_log ① ② ③ ④ ⑤ ⑥ ① inode ② 文件系統(tǒng) major device number ③ 文件系統(tǒng) minor device number ④ 已經(jīng)讀取到的字節(jié)位置 offset ⑤ 最后活躍時(shí)間 ⑥ 最后一次匹配到的文件路徑syslog插件input階段接收別的主機(jī)系統(tǒng)日志Linux的許多服務(wù)會(huì)產(chǎn)生日志通過(guò)syslog協(xié)議發(fā)送給Logstash進(jìn)而發(fā)送給Elasticsearch別的主機(jī)是日志產(chǎn)生端logstash所在主機(jī)是日志接收轉(zhuǎn)發(fā)端充當(dāng)日志服務(wù)器編輯Logstash Pipeline 配置文件input { syslog {} } output { stdout {} elasticsearch { hosts 172.20.10.2:9200 index syslog-%{YYYY.MM.dd} } }啟動(dòng)開(kāi)始監(jiān)聽(tīng)本機(jī)514端口/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/syslog.conf配置客戶(hù)端日志輸出vim /etc/rsyslog.conflogstash端接收到多行過(guò)濾插件Filter階段有些日志一條記錄占很多行例如Java異常一個(gè)錯(cuò)誤但普通采集會(huì)拆成三個(gè)日志來(lái)記錄通過(guò)多行過(guò)濾插件可以合并成一條完整的日志Exception in thread main java.lang.NullPointerException at com.test.User.login(User.java:20) at com.test.Main.main(Main.java:10)模擬不使用多行過(guò)濾的日志采集1、新建一個(gè)測(cè)試日志vim /var/log/my-es.log寫(xiě)入[2026-08-12 11:50:01] INFO Elasticsearch started successfully [2026-08-12 11:50:05] ERROR Failed to connect to cluster java.net.ConnectException: Connection refused at java.net.PlainSocketImpl.socketConnect(PlainSocketImpl.java:107) at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350) at org.elasticsearch.client.RestClient.performRequest(RestClient.java:300) Caused by: java.net.SocketException: Network is unreachable at java.net.Socket.connect(Socket.java:633) [2026-08-12 11:50:10] INFO Retry connection2、配置編輯Logstash Pipeline 配置文件input { file { path /var/log/my-es.log start_position beginning } } output { stdout { codec rubydebug } }啟動(dòng)/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d /test.conf輸出使用多行過(guò)濾配置文件input { file { path /var/log/my-es.log start_position beginning codec multiline { pattern ^\[ negate true what previous } } } output { stdout { codec rubydebug } }^表示行首 ^\[表示每一行是不是以[開(kāi)頭 negate篩選不以[ 開(kāi)頭的行 what previous如果當(dāng)前行滿(mǎn)足上面的條件就把當(dāng)前行合并到 previous也就是上一條事件變成一條日志了grok過(guò)濾Filter階段把普通文本日志解析成結(jié)構(gòu)化數(shù)據(jù)例如192.168.1.10 GET /index.html 200 1234轉(zhuǎn)化為{ ip:192.168.1.10, method:GET, url:/index.html, status:200, size:1234 }編輯配置文件input { file { path /var/log/httpd/access_log start_position beginning } } filter { grok { match { message %{HTTPD_COMBINEDLOG} } } } output { stdout {} elasticsearch { hosts 172.20.10.2:9200 index apachelog-filebeat-%{YYYY.MM.dd} } }訪問(wèn)ab -c1 -n 300 http://172.20.10.6/index.html啟動(dòng)/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conffilebeat 日志采集輕量級(jí)實(shí)際生產(chǎn)環(huán)境中不推薦讓logstash直接跑在每臺(tái)業(yè)務(wù)服務(wù)器上占用內(nèi)存大CPU消耗高浪費(fèi)資源所以采用業(yè)務(wù)服務(wù)器使用filebeat采集日志發(fā)送給logstash再給Elasticsearch最后由kibana展示采集web服務(wù)日志直接輸出到ES安裝filebeatrpm -ivh filebeat-7.6.1-x86_64.rpm激活apache模塊cd /etc/filebeat/modules.dfilebeat modules enable apache輸出到ES修改配置文件vim /etc/filebeat/filebeat.yml檢查語(yǔ)法正確性filebeat test config -c /etc/filebeat/filebeat.yml測(cè)試連通性能不能連接到ESfilebeat test output啟動(dòng)systemctl enable --now filebeat采集日志輸出到logstash由各個(gè)主機(jī)上的filebeat將采集到的日志發(fā)送給logstash主機(jī)上過(guò)濾分析后輸出給ES1、編輯配置文件vim /etc/logstash/conf.d/beats.conf寫(xiě)入input { beats { port 5044 } } filter { grok { match { message %{HTTPD_COMBINEDLOG} } } } output { elasticsearch { hosts [http://172.20.10.2:9200] index apachelog-%{YYYY.MM.dd} } }啟動(dòng)logstash/usr/share/logstash/bin/logstash -rf /etc/logstash/conf.d/beats.conf修改filebeat配置文件vim /etc/filebeat/filebeat.yml重啟systemctl restart filebeat.serviceab -c1 -n 300 http://172.20.10.5/index.html