ELK logstash 配置語法

數據類型

logstash支持的數據類型有：

• array
  數組能夠是單個或者多個字符串值。
  path => [ "/var/log/messages", "/var/log/*.log" ]
  path => "/data/mysql/mysql.log"
  若是指定了屢次，追加數組。此實例path數組包含三個字符串元素。
• boolean
  布爾值必須是TRUE或者false。true和false不能有引號。
  ssl_enable => true
• bytes
  指定字節單位。支持的單位有SI (k M G T P E Z Y) 和 Binary (Ki Mi Gi Ti Pi Ei Zi Yi)。Binary單位基於1024，SI單位基於1000。不區分大小寫和忽略值與單位之間的空格。若是沒有指定單位，默認是byte。
  my_bytes => "1113" # 1113 bytes
  my_bytes => "10MiB" # 10485760 bytes
  my_bytes => "100kib" # 102400 bytes
  my_bytes => "180 mb" # 180000000 bytes
• Codec
  logstash編碼名稱用來表示數據編碼。用於input和output段。便於數據的處理。若是input和output使用合適的編碼，就無需單獨的filter對數據進行處理。
  codec => "json"
• hash
  鍵值對，注意多個鍵值對用空格分隔，而不是逗號。
  match => {
  "field1" => "value1"
  "field2" => "value2"
  ... }
• number
  必須是有效的數值，浮點數或者整數。
  port => 33
• password
  一個單獨的字符串。
  my_password => "password"
• path
  一個表明有效的操做系統路徑。
  my_path => "/tmp/logstash"
• string
  name => "Hello world"
  name => 'It\'s a beautiful day'

字段引用

logstash字段引用語法。要在 Logstash 配置中使用字段的值，只須要把字段的名字寫在中括號 [] 裏就好了，這就叫字段引用。還需注意字段層次。若是引用的是一個頂級字段，能夠省略[]，直接指定字段名。要引用嵌套的字段，須要指定完整的路徑，如[top-level field][nested field]。

下面有五個頂級字段(agent, ip, request, response, ua) 和三個嵌套字段 (status, bytes, os)。

{ "agent": "Mozilla/5.0 (compatible; MSIE 9.0)", "ip": "192.168.24.44", "request": "/index.html" "response": { "status": 200, "bytes": 52353 }, "ua": { "os": "<a href="http://www.ttlsa.com/windows/" title="windows"target="_blank">Windows</a> 7" } }

1 2 3 4 5 6 7 8 9 10 11 12

{   "agent": "Mozilla/5.0 (compatible; MSIE 9.0)",   "ip": "192.168.24.44",   "request": "/index.html"   "response": {     "status": 200,     "bytes": 52353   },   "ua": {     "os": "Windows 7"   } }

爲了引用os字段，需指定[ua][os]。引用頂級字段如request，能夠簡單指定request便可。

sprintf格式

字段引用格式也能夠用於logstash調用sprintf格式。這種格式能夠從其餘字符串中引用字段值。如：

output { statsd { increment => "apache.%{[response][status]}" } }

1 2 3 4 5

output {   statsd {     increment => "apache.%{[response][status]}"   } }

也能夠格式化時間。如：

output { file { path => "/var/log/%{type}.%{+yyyy.MM.dd.HH}" } }

1 2 3 4 5

output {   file {     path => "/var/log/%{type}.%{+yyyy.MM.dd.HH}"   } }

 

條件判斷

使用條件來決定filter和output處理特定的事件。

logstash條件相似於編程語言。條件支持if、else if、else語句，能夠嵌套。

條件語法以下：

if EXPRESSION { ... } else if EXPRESSION { ... } else { ... }

1 2 3 4 5 6 7

if EXPRESSION {   ... } else if EXPRESSION {   ... } else {   ... }

比較操做有：

• 相等: ==, !=, <, >, <=, >=
• 正則: =~(匹配正則), !~(不匹配正則)
• 包含: in(包含), not in(不包含)

布爾操做：

• and(與), or(或), nand(非與), xor(非或)

一元運算符：

• !(取反)
• ()(複合表達式), !()(對複合表達式結果取反)

如mutate filter刪除secret字段對於action是login的：

filter { if [action] == "login" { mutate { remove => "secret" } } }

1 2 3 4 5

filter {   if [action] == "login" {     mutate { remove => "secret" }   } }

在一個條件裏指定多個表達式：

output { # Send production errors to pagerduty if [loglevel] == "ERROR" and [deployment] == "production" { pagerduty { ... } } }

1 2 3 4 5 6 7 8

output {   # Send production errors to pagerduty   if [loglevel] == "ERROR" and [deployment] == "production" {     pagerduty {     ...     }   } }

在in條件，能夠比較字段值：

filter { if [foo] in [foobar] { mutate { add_tag => "field in field" } } if [foo] in "foo" { mutate { add_tag => "field in string" } } if "hello" in [greeting] { mutate { add_tag => "string in field" } } if [foo] in ["hello", "world", "foo"] { mutate { add_tag => "field in list" } } if [missing] in [alsomissing] { mutate { add_tag => "shouldnotexist" } } if !("foo" in ["hello", "world"]) { mutate { add_tag => "shouldexist" } } }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

filter {   if [foo] in [foobar] {     mutate { add_tag => "field in field" }   }   if [foo] in "foo" {     mutate { add_tag => "field in string" }   }   if "hello" in [greeting] {     mutate { add_tag => "string in field" }   }   if [foo] in ["hello", "world", "foo"] {     mutate { add_tag => "field in list" }   }   if [missing] in [alsomissing] {     mutate { add_tag => "shouldnotexist" }   }   if !("foo" in ["hello", "world"]) {     mutate { add_tag => "shouldexist" }   } }

 

output { if "_grokparsefailure" not in [tags] { elasticsearch { ... } } }

1 2 3 4 5

output {   if "_grokparsefailure" not in [tags] {     elasticsearch { ... }   } }

字段引用、sprintf格式、條件判斷只能用於filter和output，不能用於input。

@metadata字段

在logstash1.5版本開始，有一個特殊的字段，叫作@metadata。@metadata包含的內容不會做爲事件的一部分輸出。

input { stdin { } } filter { mutate { add_field => { "show" => "This data will be in the output" } } mutate { add_field => { "[@metadata][test]" => "Hello" } } mutate { add_field => { "[@metadata][no_show]" => "This data will not be in the output" } } } output { if [@metadata][test] == "Hello" { stdout { codec => rubydebug } } }

1 2 3 4 5 6 7 8 9 10 11 12 13

input { stdin { } }   filter {   mutate { add_field => { "show" => "This data will be in the output" } }   mutate { add_field => { "[@metadata][test]" => "Hello" } }   mutate { add_field => { "[@metadata][no_show]" => "This data will not be in the output" } } }   output {   if [@metadata][test] == "Hello" {     stdout { codec => rubydebug }   } }

查看輸出：

$ bin/logstash -f ../test.conf Logstash startup completed asdf { "message" => "asdf", "@version" => "1", "@timestamp" => "2015-03-18T23:09:29.595Z", "host" => "www.ttlsa.com", "show" => "This data will be in the output" }

1 2 3 4 5 6 7 8 9 10

$ bin/logstash -f ../test.conf Logstash startup completed asdf {        "message" => "asdf",       "@version" => "1",     "@timestamp" => "2015-03-18T23:09:29.595Z",           "host" => "www.ttlsa.com",           "show" => "This data will be in the output" }

"asdf"變成message字段內容。條件與@metadata內嵌的test字段內容判斷成功，可是輸出並無展現@metadata字段和其內容。

不過，若是指定了metadata => true，rubydebug codec容許顯示@metadata字段的內容。

stdout { codec => rubydebug { metadata => true } }

1	stdout { codec => rubydebug { metadata => true } }

下面是輸出的內容：

$ bin/logstash -f ../test.conf Logstash startup completed asdf { "message" => "asdf", "@version" => "1", "@timestamp" => "2015-03-18T23:10:19.859Z", "host" => "www.ttlsa.com", "show" => "This data will be in the output", "@metadata" => { "test" => "Hello", "no_show" => "This data will not be in the output" } }

1 2 3 4 5 6 7 8 9 10 11 12 13 14

$ bin/logstash -f ../test.conf Logstash startup completed asdf {        "message" => "asdf",       "@version" => "1",     "@timestamp" => "2015-03-18T23:10:19.859Z",           "host" => "www.ttlsa.com",           "show" => "This data will be in the output",      "@metadata" => {            "test" => "Hello",         "no_show" => "This data will not be in the output"     } }

能夠看到@metadata字段及其子字段內容。

注意：只有rubydebug codec能夠顯示@metadata字段內容。

確保@metadata字段臨時須要，不但願最終輸出。最多見的情景是filter的時間字段，須要一臨時的時間戳。如：

input { stdin { } } filter { grok { match => [ "message", "%{HTTPDATE:[@metadata][timestamp]}" ] } date { match => [ "[@metadata][timestamp]", "dd/MMM/yyyy:HH:mm:ss Z" ] } } output { stdout { codec => rubydebug } }

1 2 3 4 5 6 7 8 9 10

input { stdin { } }   filter {   grok { match => [ "message", "%{HTTPDATE:[@metadata][timestamp]}" ] }   date { match => [ "[@metadata][timestamp]", "dd/MMM/yyyy:HH:mm:ss Z" ] } }   output {   stdout { codec => rubydebug } }

輸出結果：

$ bin/logstash -f ../test.conf Logstash startup completed 02/Mar/2014:15:36:43 +0100 { "message" => "02/Mar/2014:15:36:43 +0100", "@version" => "1", "@timestamp" => "2014-03-02T14:36:43.000Z", "host" => "example.com" }

1 2 3 4 5 6 7 8 9

$ bin/logstash -f ../test.conf Logstash startup completed 02/Mar/2014:15:36:43 +0100 {        "message" => "02/Mar/2014:15:36:43 +0100",       "@version" => "1",     "@timestamp" => "2014-03-02T14:36:43.000Z",           "host" => "example.com" }