Logstash配置安裝

時間 2020-09-14

原文原文鏈接

logstash配置

官方文檔：https://www.elastic.co/guide/en/logstash/current/index.htmlhtml

http://www.javashuo.com/article/p-rchibiya-nh.htmlnginx

在config/logstash.yml中redis

http.host: "192.168.31.102"json

xpack.monitoring.enabled: truebootstrap

xpack.monitoring.elasticsearch.username: "logstash_system"vim

xpack.monitoring.elasticsearch.password: "mPpO8ceIo14Z7k0ZQIHJ"ruby

xpack.monitoring.elasticsearch.hosts: ["http://localhost:9200"]bash

在bin文件下面新建logstash.confelasticsearch

logstash pipeline 包含兩個必須的元素：input和output，和一個可選元素：filter。從input讀取事件源，（通過filter解析和處理以後），從output輸出到目標存儲庫（elasticsearch或其餘）。ide

input{

　　輸入插件
}

filter{

　　過濾器插件
}

outer{

　　輸出插件
}

一、input

能夠接收來自任何地方的源數據:

https://www.elastic.co/guide/en/logstash/current/input-plugins.html

事件源能夠是從stdin屏幕輸入讀取，能夠從file指定的文件，也能夠從es，filebeat，kafka，redis等讀取

file{
    path => ['/var/log/nginx/access.log']  #要輸入的文件路徑
    type => 'nginx_access_log'
    start_position => "beginning"
}
# path  能夠用/var/log/*.log,/var/log/**/*.log，若是是/var/log則是/var/log/*.log
# type 通用選項. 用於激活過濾器
# start_position 選擇logstash開始讀取文件的位置，begining或者end。
還有一些經常使用的例如：discover_interval，exclude，sincedb_path,sincedb_write_interval等能夠參考官網

syslog{
    port =>"514" 
    type => "syslog"
}
# port 指定監聽端口(同時創建TCP/UDP的514端口的監聽)

#從syslogs讀取須要實現配置rsyslog：
# cat /etc/rsyslog.conf   加入一行
*.* @172.17.128.200:514　  #指定日誌輸入到這個端口，而後logstash監聽這個端口，若是有新日誌輸入則讀取
# service rsyslog restart   #重啓日誌服務

beats {
    port => 5044   #要監聽的端口
}
# 還有host等選項

# 從beat讀取須要先配置beat端，從beat輸出到logstash。
# vim /etc/filebeat/filebeat.yml 
..........
output.logstash:
hosts: ["localhost:5044"]

kafka{
    bootstrap_servers=> "kafka01:9092,kafka02:9092,kafka03:9092"
    topics => ["access_log"]
    group_id => "logstash-file"
    codec => "json"
}
kafka{
    bootstrap_servers=> "kafka01:9092,kafka02:9092,kafka03:9092"
    topics => ["weixin_log","user_log"]  
    codec => "json"
}
# bootstrap_servers 用於創建羣集初始鏈接的Kafka實例的URL列表。
# topics  要訂閱的主題列表，kafka topics
# group_id 消費者所屬組的標識符，默認爲logstash。kafka中一個主題的消息將經過相同的方式分發到Logstash的group_id
# codec 通用選項，用於輸入數據的編解碼器。

二、filter

Filter數據中轉層,主要進行格式處理,數據類型轉換,數據過濾,字段添加,修改等,經常使用的過濾器:

https://www.elastic.co/guide/en/logstash/current/filter-plugins.html

三、output

Output是logstash工做的最後一個階段,負責將數據輸出到指定位置,兼容大多數應用:

https://www.elastic.co/guide/en/logstash/current/output-plugins.html

rabbitmq同步

https://www.elastic.co/guide/en/logstash/current/plugins-inputs-rabbitmq.html

input { 
    rabbitmq {
    host => "127.0.0.1"
    port=>5672
    vhost=>"CRM"
    subscription_retry_interval_seconds => "5"
    codec => "plain"
    exchange => "log.msg.direct"
    queue => "CRM_LOG"
    durable => true
    key => "rk.log.direct"
    user => "admin"
    password => "admin"
    }
}
output {
  elasticsearch { 
        hosts => ["localhost:9200"] 
        #flush_size=> 10  
        #workers => 5  
        #document_id => "%{jobid}" 
        #document_type => "BaseSysLog"
        index => "position-%{+YYYYMMdd}" 
        user => "elastic"
        password => "caKBawShu0Pm7BsjSFbd"
  }
  stdout { codec => rubydebug }
}