logstash 配置文件語法

需要一個配置文件 管理輸入、過濾器和輸出相關的配置。配置文件內容格式如下：

#　輸入
input {
  ...
}

# 過濾器
filter {
  ...
}

# 輸出
output {
  ...
}

 

先來看一個標准輸入輸出

root@c201b7b32a32# ./logstash -e 'input { stdin{} } output { stdout{} }'
Sending Logstash's logs to /opt/logstash/logs which is now configured via log4j2.properties
[2018-04-26T06:47:20,724][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/opt/logstash/modules/fb_apache/configuration"}
……

[2018-04-26T06:47:24,124][INFO ][logstash.pipeline ] Pipeline started succesfully {:pipeline_id=>"main", :thread=>"#<Thread:0x5fec99f4 run>"}
The stdin plugin is now waiting for input:
[2018-04-26T06:47:24,253][INFO ][logstash.agent ] Pipelines running {:count=>1, :pipelines=>["main"]}

hello ==>輸入 2018-04-26T06:47:31.957Z c201b7b32a32 hello         ==>輸出 this is test  ==>輸入
2018-04-26T06:50:29.743Z c201b7b32a32 this is test   ==>輸出

 

使用rubudebug顯示詳細輸出，codec為一種編解碼器

# ./logstash -e 'input { stdin{} } output { stdout{ codec => rubydebug} }'

test2 ==>輸入 { "message" => "test2", "@timestamp" => 2018-04-26T07:00:00.652Z, "@version" => "1", "host" => "c201b7b32a32" } ==>使用rubydebug輸出

 

 input輸入設置

input { # file為常用文件插件，插件內選項很多，可根據需求自行判斷 file { path => "/var/log/httpd/access_log" # 要導入的文件的位置，可以使用*，例如/var/log/nginx/*.log Excude =>”*.gz”  # 要排除的文件 start_position => "beginning" # 從文件開始的位置開始讀,默認是end ignore_older => 0 # 多久之內沒修改過的文件不讀取，0為無限制，單位為秒 sincedb_path => "/dev/null" # 記錄文件上次讀取位置；輸出到null表示每次都從文件首行開始解析
 add_field=>{"test"="test"} # 增加一個字段  type => "apache-log" # type字段，可表明導入的日志類型 } }

 也可以使用多個file

input { file { path => "/var/log/messages" type => "syslog" } 

 file { path => "/var/log/apache/access.log" type => "apache" } }

也可以使用數組方式   或者用*匹配

path => ["/var/log/messages","/var/log/*.log"] path => ["/data/mysql/mysql.log"]

 

 

  filter過濾設置

 Logstash三個組件的第二個組件，也是真個Logstash工具中最復雜的一個組件，
當然，也是最有作用的一個組件。

1、grok插件 grok插件有非常強大的功能，他能匹配一切數據，但是他的性能和對資源的損耗同樣讓人詬病。

filter{ grok{ #首先要說明的是，所有文本數據都是在Logstash的message字段中的，我們要在過濾器里操作的數據就是message。
        #只說一個match屬性，他的作用是從message 字段中把時間給摳出來，並且賦值給另個一個字段logdate。

 #第二點需要明白的是grok插件是一個十分耗費資源的插件。 #第三點需要明白的是，grok有超級多的預裝正則表達式，這里是沒辦法完全搞定的，也許你可以從這個大神的文章中找到你需要的表達式 #http://blog.csdn.net/liukuan73/article/details/52318243
        #但是，我還是不建議使用它，因為他完全可以用別的插件代替，當然，對於時間這個屬性來說，grok是非常便利的。
        
         match => ['message','%{TIMESTAMP_ISO8601:logdate}']
 } 
}

 

再看下match 另一種用法，將message中   ip、訪問方法、url、數據量、持續時間   提取出來 
並賦值給 clientip、method、request、bytes、duration 字段

filter { grok { match => {"message"=>"%{IPORHOST:clientip}\s+%{WORD:method}\s+%{URIPATHPARAM:request}\s+%{NUMBER:bytes}\s+%{NUMBER:duration}"} } }  

顯示數據

{  "message" => "9.9.8.6 GET /xx.hmtl 343 44",  
      "@version" => "1",  
    "@timestamp" => "2017-01-18T00:12:37.490Z",  
          "path" => "/home/elk/0204/nginx.log",  
          "host" => "db01",  
          "type" => "nginx",  
      "clientip" => "9.9.8.6",  
       "method" => "GET", "request" => "/xx.hmtl", "bytes" => "343", "duration" => "44"   } 

 

繼續修改，提取后刪除message

filter { grok { match => {"message"=>"%{IPORHOST:clientip}\s+%{WORD:method}\s+%{URIPATHPARAM:request}\s+%{NUMBER:bytes}\s+%{NUMBER:duration}"}  remove_field =>["message"] } }  

顯示結果

{ "@version" => "1", "@timestamp" => "2017-01-18T00:15:03.879Z", "path" => "/home/elk/0204/nginx.log", "host" => "db01", "type" => "nginx", "clientip" => "55.9.3.6", "method" => "GET", "request" => "/zz.xml", "bytes" => "3", "duration" => "44" }  

 

比較常用的是 %{COMBINEDAPACHELOG}   是logstash自帶的匹配模式，內置的正則，用來匹配apache access日志

filter { grok { match => { "message" => "%{COMBINEDAPACHELOG}" } remove_field => "message" } }

顯示結果

{
  "_index": "logstash-2018.05.03",
  "_type": "apache_logs",
  "_id": "VFHkI2MBPZdRHaSpwnN-",
  "_version": 1,
  "_score": null,
  "_source": {
    "agent": "\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.75 Safari/537.36 Maxthon/5.1.5.2000\"",
    "path": "/var/log/httpd/access_log",
    "referrer": "\"http://10.10.12.81/cacti/data_sources.php\"",
    "host": "cacti",
    "verb": "GET",
    "clientip": "10.0.7.99",
    "request": "/cacti/graphs.php",
    "auth": "-",
    "@version": "1",
    "ident": "-",
    "httpversion": "1.1",
    "response": "200",
    "bytes": "37138",
    "@timestamp": "2018-05-03T02:46:26.477Z",
    "timestamp": "03/May/2018:10:46:25 +0800"
  },
  "fields": {
    "@timestamp": [
      "2018-05-03T02:46:26.477Z"
    ]
  },
  "sort": [
    1525315586477
  ]
}

 

 

其它插件暫時不講……

 

  
output輸入設置

輸出到elasticserarch

 elasticsearch{ hosts=>["10.10.10.11:9200"] # elasticsearch 地址 端口 action=>"index" # 索引 index=>"indextemplate-logstash" # 索引名稱 #document_type=>"%{@type}" document_id=>"ignore" template=>"/opt/logstash-conf/es-template.json" # 模板文件的路徑 template_name=>"es-template.json" # 在es內部模板的名字 template_overwrite=>true # 

    protocol => "http"         #目前支持三種協議    node、http 和tranaport  

 }

 

 

 

寫幾個實例

1.配置文件

input {
file {
    path => ['/var/log/httpd/access_log']
    start_position => "beginning"
}
}
filter {
grok {
    match => {
        "message" => "%{COMBINEDAPACHELOG}"
    }

    remove_field => "message"   
}
}
output {
elasticsearch {
    hosts => ["10.10.15.95:9200"]
    index => "12.81-cacti-%{+YYYY.MM.dd}"
    action => "index"
    document_type => "apache_logs"
              }
}

數據

{
"_index": "logstash-2018.05.03",
"_type": "apache_logs",
"_id": "U1HkI2MBPZdRHaSpMXPM",
"_version": 1,
"_score": 1,
"_source": {
"agent": ""Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.75 Safari/537.36 Maxthon/5.1.5.2000"",
"path": "/var/log/httpd/access_log",
"referrer": ""http://10.10.12.81/cacti/include/themes/modern/jquery-ui.css"",
"host": "cacti",
"verb": "GET",
"clientip": "10.0.7.99",
"request": "/cacti/include/themes/modern/images/ui-icons_454545_256x240.png",
"auth": "-",
"@version": "1",
"ident": "-",
"httpversion": "1.1",
"response": "200",
"bytes": "6992",
"@timestamp": "2018-05-03T02:45:49.442Z",
"timestamp": "03/May/2018:10:45:49 +0800"
           }
}

 

 

2.一台機器上傳輸兩種日志

input {
    file {
        path => "/var/log/messages"
        type => "system"
        start_position => "beginning"
    }
    file {
        path => "/var/log/elasticsearch/chuck-cluster.log"
        type => "es-error"
        start_position => "beginning"
    }
}
output {
    if [type] == "system" {
        elasticsearch {
            hosts => ["192.168.56.11:9200"]
            index => "system-%{+YYYY.MM.dd}"
        }
    }
    if [type] == "es-error" {
        elasticsearch {
            hosts => ["192.168.56.11:9200"]
            index => "es-error-%{+YYYY.MM.dd}"
        }
    }
}

 

 

 

 

123