日志監控_ElasticStack-0003.Logstash輸入插件及實際生產案例應用?

新版插件:

說明: 從5.0開始,插件都獨立拆分成gem包,每個插件可獨立更新,無需等待Logstash自身整體更新,具體管理命令可參考./bin/logstash-plugin --help幫助信息../bin/logstash-plugin list其實所有的插件就位于本地./vendor/bundle/jruby/1.9/gems/目錄下

擴展: 如果GitHub上面(https://github.com/logstash-plugins/)發布了擴展插件,可通過./bin/logstash-plugin install <plugin-name>,當然升級也很方便./bin/logstash-plugin update <plugin-name>,如果要安裝更新本地已有的插件可通過./bin/logstash-plugin install/update <plugin-path>即可.

注意: 默認./bin/logstash-plugin install/update時是到https://rubygems.org/下載包,速度非常慢,所以強烈推薦手動從https://github.com/logstash-plugins/或https://rubygems.org/下載下來更新

輸入插件: https://www.elastic.co/guide/en/logstash/current/input-plugins.html

說明: 輸入插件只用于input區段中,可以通過不同的方式來采集數據,如上為目前Logstash支持的所有輸入插件.

插件名稱: udp (https://www.elastic.co/guide/en/logstash/current/plugins-inputs-udp.html)

input {
    udp {
        port => 25826
        workers => 2
        queue_size => 20000
        buffer_size => 1452
        codec => collectd { }
        type => "collectd"
    }
}
output{
    stdout{
        codec => rubydebug
    }
}

生產案例:

yum -y install collectd*
cp -rfp /etc/collectd.conf /etc/collectd.conf_org
vim /etc/collectd.conf
# Collectd基礎信息采集配置
Hostname    "xm-server-00001"
FQDNLookup  false
LoadPlugin  df
LoadPlugin  cpu
LoadPlugin  disk
LoadPlugin  memory
LoadPlugin  network
<Plugin network>
    Server "10.2.5.51" "25826"
</Plugin>
LoadPlugin  interface
<Plugin interface>
    Interface "eth0"
    IgnoreSelected false
</Plugin>
Include "/etc/collectd.d"

說明: 如上是Logstash Udp插件的生產案例,Logstash配置文件中udp區段port表示監聽的udp端口,workers表示讀取socket數據的線程數,一般設置為CPU個數即可,queue_size表示udp包可堆積在內存中個數,buffer_size表示讀取緩沖區最大大小,codec 表示編碼器,type表示手工定義的類型,可用于filter區段中以及后面前端Kibana檢索

插件名稱: file (https://www.elastic.co/guide/en/logstash/current/plugins-inputs-file.html)

input {
    file {
        path => ["/xm-workspace/xm-webs/xmcloud/logs/*.log"]
        type => "dss-pubserver"
codec => json
        start_position => "beginning"
    }
}
output{
    stdout{
        codec => rubydebug
    }
}

生產案例:

log_format json '{"@timestamp":"$time_iso8601",'
                 '"host":"$server_addr",'
                 '"clientip":"$remote_addr",'
                 '"size":$body_bytes_sent,'
                 '"responsetime":$request_time,'
                 '"upstreamtime":"$upstream_response_time",'
                 '"upstreamhost":"$upstream_addr",'
                 '"http_host":"$host",'
                 '"url":"$uri",'
                 '"xff":"$http_x_forwarded_for",'
                 '"referer":"$http_referer",'
                 '"agent":"$http_user_agent",'
                 '"status":"$status"}';

說明: 分析網站/接口訪問日志時可監聽匹配日志文件,同時會生成.sincedb數據庫文件,記錄被監聽文件的inode/major/minor/position,file區段中還支持delimiter表示行分割符,discover_interval表示檢查是否有新文件頻率,默認15秒,exclude表示排除監聽的匹配列表,close_older表示多久沒修改就關閉監聽句柄,默認3600秒,ignore_older表示只檢查多久內的文件,默認86400,sincedb_path表示sincedb存放位置,默認位于(Linux: $HOME/.,sincedb | Windows: C:\Windows\System32\config\systemprofile\.sincedb),sincedb_write_interval表示多久寫一次sincedb,默認為15秒,stat_interval為每隔多久檢查一次被監聽的文件狀態,默認為1秒,start_position表示從什么位置讀取文件數據,默認為結束位置,類似tail -f,當然也可設定beginning從頭讀取.

注意: 通常導入原有數據進Elasticsearch前需通過filter/date插件規范默認的@timestamp字段值,file插件并不支持遞歸監視,如果有需求可以數組形式匹配,支持/path/to/*/*/*/*.log寫法,如果需要反復測試可將sincedb_path定義為/dev/null,這樣每次重啟都會自動從頭開始讀,還有一點兒需要注意的是Windows由于沒有inode概念,所以監聽文件不靠譜,推薦使用nxlog作為收集端.

插件名稱: stdin (https://www.elastic.co/guide/en/logstash/current/plugins-inputs-stdin.html)

input {
    stdin {
        codec => json
        tags => ["dss-pubserver"]
        type => "stdin"
        add_field => {
            "runenv" => "docker"
        }
    }
}
output{
    stdout{
        codec => rubydebug
    }
}

生產案例:

{"status": "", "content_length": "", "body_bytes_sent": "", "http_x_forwarded_for": "", "request_time": "", "errors": {"text": "res body is invalid protocol format args", "line": 128, "func": "process_msg()", "file": "dsspub-server.lua"}, "request_body": "", "http_referer": "", "level": "", "hostname": "", "request": "", "remote_addr": "", "upstream_response_time": "", "time_local": "", "upstream_addr": ""}

說明: type和tags是Logstash事件中兩個特殊的字段,通常會在輸入區段中來標記事件類型,然后在數據處理區域根據事件類型來調用插件處理消息或添加刪除tags,最后在輸出中通過tags來判斷并選擇輸出

插件名稱: syslog(https://www.elastic.co/guide/en/logstash/current/plugins-inputs-syslog.html)

input {
    syslog {
        port => "6514"
    }
}
output{
    stdout{
        codec => rubydebug
    }
}

生產案例:

vim /etc/rsyslog.conf 
*.*                                                     @@127.0.0.1:6514

說明: syslog在收集網絡設備端日志時也許是幾乎唯一可行辦法,簡單測試只需修改/etc/rsyslog.conf配置,加入網絡傳輸支持,發送到本地的Logstash監聽端口6514,默認接收過濾操作都在input中完成,雖然啟動時可通過-w指定worker數但是同一個客戶端數據的處理以及過濾都在同一個線程中完成,導致處理性能嚴重下降,如果使用udp+grok的match的%{SYSLOGLINE}和syslog_pri實現,雖然將input和grok拆分拆分到兩個線程,但udp接收緩沖區有限制(netstat -plnu | awk 'NR==1 || $4~/:514$/{print $2}'),當大于228096時就開始丟包,所以還是推薦使用tcp方式

擴展: 如果實在沒法切換到tcp協議,可通過這個小工具來實現異步IO的UDP監聽數據輸入寫入Elasticsearch(https://gist.github.com/chenryn/7c922ac424324ee0d695)

插件名稱: tcp(https://www.elastic.co/guide/en/logstash/current/plugins-inputs-tcp.html)

input {
    tcp {
        port => "8888"
        mode => "server"
        ssl_enable => false
    }
}
output{
    stdout{
        codec => rubydebug
    }
}

生產實例:

nc 127.0.0.1 8888 < error.log

說明: Logstash也可通過Tcp組件實現簡單消息隊列,但千萬別用于生產環境,生產環境還是換用專業的Logstash broker,比較常用的就是配合nc實現舊數據導入,這種導入舊數據的方式相比file好處在于可以準確的知道何時導入完成.