Logstash配置方法

時間 2020-03-12

原文原文鏈接

Logstash
#############################logstash命令經常使用參數#############################
html

-n 指定logstash實例名稱，若是沒有指定，默認是本地主機名
-f 從指定的文件或文件夾中加載logstash的配置；若是是一個文件夾，它會加載裏面全部的文件，或者能夠加上通配符只加載特定格式的文件
-w 容許filter和output的pipeline線程數量，默認是CPU核數
-b 每一個 Logstash pipeline 線程，在執行具體的 filter 和 output 函數以前，最多能累積的日誌條數，默認是 125 條。越大性能越好，一樣也會消耗越多的 JVM 內存
-u 每一個 Logstash pipeline 線程，在打包批量日誌的時候，最多等待幾毫秒。默認是 5 ms
-l 指定日誌輸出位置
-r 監控配置配置文件，若是有變化則自動重載logstash
-e 使用給定的命令行字符串做爲配置，直接運行bin/log/logstash -e 配置默認是"input { stdin { type => stdin } }" "output { stdout { codec => rubydebug } }"
-t 檢查配置文件是否有語法錯誤
-V 打印logstash版本
--log.level 指定日誌等級，包括trace、debug、info、warn、error、fatal，默認info
--http.host 指定web API綁定的主機，默認127.0.0.1
--http.port 指定web API的http端口，默認9600至9700

#############################後臺啓動logstash#############################
1.安裝supervisor
java

yum -y install supervisor --enablerepo=epel

2.在/etc/supervisord.conf配置文件最後添加:
node

[program:logstash]
directory=/usr/local/logstash
command=/usr/local/logstash/bin/logstash -f /usr/local/logstash/etc/conf.d/ -r -l /var/log/logstash

3.啓動
nginx

service supervisord start

4.單獨控制一個子進程
git

supervisorctl stop logstash

#############################編解碼配置插件#############################
#官網參考：https://www.elastic.co/guide/en/logstash/current/codec-plugins.html
#logstash不僅是一個input|filter|output的數據流，而是一個input|decode|filter|encode|output的數據流。codec就是用來decode、encode事件的。
1.JSON編解碼
#直接輸入預約義好的JSON數據，能夠省略filter/grok配置，下降logstash過濾的CPU負載消耗
github

input {
    file {
        path => "/opt/data/logs/bjape01-ngx-a1-172.16.3.2/nginx_access.log"
        codec => "json"
    }
}

2.multiline多行事件編碼
web

codec => multiline {
    pattern => "^\["
    negate => true
    what => "previous"
}

#pattern 要匹配的正則表達式，字符串類型
#negate 正則表達式是否生效，布爾類型，默認爲flase
#what 未匹配的內容是向前合併仍是向後後合併,previous,next兩個值選擇
#示例配置能夠用於Log4j方式的應用程序日誌

3.line面向行的文本數據

4.plain空的純文本編解碼
#使用已經有定義框架輸入或輸出，例如logstash輸出轉存到elasticsearch
正則表達式

output {
        if [type] == "user_audit" {
            elasticsearch {
                            hosts => ["172.16.1.25","172.16.1.26","172.16.1.27"] 
                            index => 'user_audit-%{+YYYY-MM-dd}'
                            codec=>plain{charset=>"UTF-8"}
                          }
                    }
        }

#############################輸入插件#############################
#官網參考：https://www.elastic.co/guide/en/logstash/current/input-plugins.html
1.標準輸入
示例：
數據庫

input {
    stdin {
        add_field => {"key" => "value"}
        codec => "plain"
        tags => ["add"]
        type => "std"
    }
}

#add_field 向事件添加一個字段，hash類型
#codec 設置編碼方式，默認是line
#tags 添加標記
#type 添加類型

2.文件輸入
apache

input {
    file {
        path => [
                "/opt/data/logs/idca-web1-172.16.3.2/apache_access.log",
                "/opt/data/logs/idca-web1-172.16.3.2/apache_error.log"
                ]
        stat_interval => 1
        discover_interval => 1
        type => "logstash-apache_log"
    }
}

#path 處理的文件的路徑, 能夠定義多個路徑
#stat_interval 每隔多久檢查一次被監聽文件狀態（是否更新），默認是1秒
#discover_interval 每隔多久去檢查一次被監聽的path下是否有新文件，默認是15秒
#start_position 從什麼位置開始讀取文件數據，"beginning"從頭開始讀取，讀到最後一行不會終止，繼續tailf；"end"從結束位置tailf。默認值是"end"
#坑：start_position僅在該文件從未被監聽過的時候起做用，若是sincedb文件中已經有這個文件的inode記錄了，那麼logstash依然會衝記錄過的pos開始讀取數據。因此重複測試的時候每回須要刪除sincedb文件，該文件通常在安裝目錄下的data/plugins/inputs/file/中
#優化：在file中添加sincedb_path => "/dev/null"，能夠直接將sincedb寫到黑洞中
#exclude 不想被監聽的文件能夠排除出去
#sincedb_path 配置sincedb文件位置
#sincedb_write_interval 每隔多久寫一次sincedb文件，默認15秒

3.syslog輸入

input {
    syslog {
        port => "514"
    }
}

4.kafka輸入

input {
kafka {
    type => "logstash-apache_log"
    codec => "plain"
    topics => "apache_log"
    bootstrap_servers => "kafka1:9092,kafka2:9092,kafka3:9092"
  }
}

#topics 主題名，多個主題時使用列表形式

#bootstrap_servers kafka地址，多個地址時使用「host1:port1,host2:port2」格式

#############################過濾器配置#############################
#官網參考：https://www.elastic.co/guide/en/logstash/current/filter-plugins.html
1.grok正則捕獲
#用於匹配對應格式的message，而後過濾出來

filter {
  grok {
    break_on_match => false
    patterns_dir => ["/usr/local/logstash/etc/conf.d/patterns"]
    match => { "message" => "\[%{TIMESTAMP_ISO8601:time}\] \[%{GLASSFISHVERSION:gfversion}\] %{OTHER}"}
  }
   grok {
    break_on_match => false
    patterns_dir => ["/usr/local/logstash/etc/conf.d/patterns"]
    match => {"path" => "/%{USER}/%{USER}/%{USER}/%{USER:host}/%{OTHER}" }
    overwrite => [ "host" ]
  }
}

#break_on_match 表示匹配即中止，默認爲true，若是但願grok嘗試全部模式，設置爲false
#patterns_dir 將grok表達式統一寫到文件中，該選項指定文件目錄。grok表達式能夠寫到配置文件中，可是日誌格式可能有多種，每種都寫一行本身的表達式就可能會出問題，因此建議將全部的grok表達式統一寫到一個地方。

#grok表達式格式：

USERNAME [a-zA-Z0-9._-]+
USER %{USERNAME}
#第一行用普通的正則表達式來定義一個grok表達式；第二行用以定義好的grok表達式來定義另外一個grok表達式。logstash內置了多種grok表達式，查看https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns，一樣能夠自定義grok表達式。

%{NUMBER:num}
%{NUMBER:num:int}
#grok表達式的打印複製格式，第一行中NUMBER表示匹配的模式，num是匹配內容的標識，表明匹配的內容；第二行中int表示轉換匹配內容的格式，默認是字符串格式。

#overwrite用來重寫message字段，上面例子第二個grok中match的是一個路徑，可是咱們只想保留路徑中的一個host字段，因此overwrite => [ "host" ]就能夠了

2.date時間處理
#日誌產生到logstash處理通常都會有一段時間差，致使logstash實際處理時間都比日誌產生時間晚。logstash-filter-date插件能夠解析日誌中的時間，變成LogStash:Timestamp對象，而後轉存到@timestamp字段裏，做爲事件的時間戳；若是未使用date插件或沒有解析到時間，logstash將使用日誌輸入的時間做爲事件的時間戳。

date {
      match => [ "time", "MMM dd yyyy HH:mm:ss",
            "MMM  d yyyy, HH:mm:ss", "ISO8601" ]
    }

#配置選項包括：locale、match、tag_on_failure、target、timezone。經常使用的是match，值的類型一樣是列表，默認是[]
#實例中的time就是grok中匹配字段的標識，由咱們自定義。
#date插件支持五種時間格式，經常使用的是ISO8601和Joda-Time庫

ISO8601：相似"2018-10-19T14:25:25.905+0800"，這是咱們的北京時間，若是是UTC時間後面跟的是一個大寫的Z（時區偏移），nginx、apache、java應用均可以設置爲這種格式。
UNIX：UNIX時間戳格式，記錄從1970年起至今的總秒數。
UNIX_MS：從1970年至今的總毫秒數。
TAI64N：tai64n格式，不多用。
Joda-Time庫：y年，yyyy例如2018，yy例如18；M月，MMMM例如January，MMM例如Jan，MM例如01，M例如1；d日，dd例如01，d例如1；H時，HH例如01，H例如1；m分，mm例如01，m例如1；s秒，ss例如01，s例如1

3.GeoIP地址查詢
#GeoIP過濾器根據來自Maxmind GeoLite2數據庫的數據，添加關於IP地址地理位置的信息。
#插件名稱logstash-filter-geoip，能夠先使用logstash-plugin list命令查看插件是否安裝，沒有安裝可使用logstash-plugin install logstash-filter-geoip命令來安裝
#基本配置

geoip {
                source => "clientip"
        }

#debug標準輸出爲

{
           "path" => "/var/log/logstash_test/apache_access.log",
           "type" => "apache_log",
      "timestamp" => "26/Oct/2018:14:44:25 +0800",
       "@version" => "1",
       "response" => "302",
        "usetime" => "0",
       "referrer" => "/hsbp2pUser/view/html/errorSkip.html;jsessionid=f1deac55291cdf60c1e66108a89c.bjape01-wgw-b2",
     "@timestamp" => 2018-10-26T06:44:25.000Z,
          "geoip" => {
         "country_code2" => "CN",
              "timezone" => "Asia/Shanghai",
              "latitude" => 22.8167,
        "continent_code" => "AS",
           "region_code" => "45",
           "region_name" => "Guangxi",
          "country_name" => "China",
              "location" => {
            "lon" => 108.3167,
            "lat" => 22.8167
        },
         "country_code3" => "CN",
                    "ip" => "113.15.241.43",
             "longitude" => 108.3167,
             "city_name" => "Nanning"
    },
          "agent" => "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; WOW64; Trident/5.0)",
           "host" => "opsmanage",
        "message" => "113.15.241.43 - - [26/Oct/2018:14:44:25 +0800] \"/hsbp2pUser/view/html/errorSkip.html;jsessionid=f1deac55291cdf60c1e66108a89c.bjape01-wgw-b2\"  302 460 0 \"Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; WOW64; Trident/5.0)\"",
       "clientip" => "113.15.241.43",
          "bytes" => "460",
    "remote_user" => "-"
}

#經常使用配置
#source 指定來源IP字段，即便用GeoIP庫來分析的IP地址，若是是一個列表，只有第一個會被使用
#database 指定GeoIP庫的路徑，若是不指定該配置項默認隨Logstash的GeoLite2城市數據庫
#fields 指定輸出那些字段，格式是一個列表。若是不指定默認爲[city_name,continent_code,country_code2,country_code3,country_name,dma_code,ip,latitude,longitude,postal_code,region_name,timezone]
#target 指定應該將geoip數據存儲在哪一個Logstash的字段，通常設置爲geoip

#############################輸出插件#############################
#官網參考：https://www.elastic.co/guide/en/logstash/current/output-plugins.html
1.輸出到Elasticsearch

output {
    elasticsearch {
        hosts => ["172.16.1.25","172.16.1.26","172.16.1.27"]
        index => "%{type}-%{+YYYY-MM-dd}"
        codec=>plain{charset=>"UTF-8"}
    }
}

#hosts Elasticsearch地址，列表形式，能夠寫多個。#index 索引名，寫入Elasticsearch索引的名稱，可使用變量。注意：%{+YYYY-MM-dd}}在語法解析時，看到已+開頭的，就會自動認爲後面是時間格式，因此以前處理過程當中不要給自定義字段取加號開頭的名字；索引名中不要有大寫字母，不然Elasticsearch會報InvalidindexNameException

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。