《Learning ELK Stack》3 使用Logstash采集、解析和转换数据

bin/plugin list

bin/plugin list --group <group_name>
bin/plugin list --group output

path => ["value1", "value2"]

periodic_flush => false

codec => "json"

match => {
    "key1" => "value1", "key2" => "value2"
}

value => "welcome"

# 这是一个注释

if <conditional expression1> {
# 一些处理语句
}
else if <conditional expression2> {
# 一些处理语句
}
else {
# 一些其他语句
}

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

output {
    if [loglevel] == "ERROR" and [deployment] == "production" {
        email {}
    }
}

input {
    file {
        path => "/GOOG.csv"
        add_field => {"input_time" => "%{@timestamp}"}
        codec => "json"
        delimiter => "\n"
        exclude => "*.gz"
        sincedb_path => "$HOME/.sincedb*"
        sincedb_write_interval => 15
        start_position => "end"
        tags => ["login"]
        type => ["apache"]
    }
}

input {
    file {
        path => ["/var/log/syslog/*"]
        type => "syslog"
    }
    file {
        path => ["/var/log/apache/*"]
        type => "apache"
    }
}
filter {
    if [type] == "syslog" {
        grok{}
    }
    if [type] == "apache" {
        grok{}
    }
    if "login" == tags[] {}
}

csv {
    columns => ["date_of_record","open","high","low","close","volume","adj_close"]
    separator => ","
}

date {
    match => ["date_of_record", "yyyy-MM-dd"]
}

filter {
    if [fieldname == "test"] {
        drop {}
    }
}

geoip {
    source => # 必选字符串，需要使用geoip服务进行映射的ip地址或主机名
}

HOSTNAME \b(?:[0-9A-Za-z][0-9A-Za-z-]{0,62})(?:\.(?:[0-9A-Za-z][0-9A-Za-z-]{0,62}))*(\.?|\b)
DAY (?:Mon(?:day)?|Tue(?:sday)?|Wed(?:nesday)?|Thu(?:rsday)?|Fri(?:day)?|Sat(?:urday)?|Sun(?:day)?)
YEAR (?>\d\d){1,2}
HOUR (?:2[0123]|[01]?[0-9])
MINUTE (?:[0-5][0-9])

%{HOSTNAME:host_name}

54.3.245.1 GET /index.html 14562 0.056

%{IP:client_ip} %{WORD:request_method} %{URIPATHPARAM:uri_path} %{NUMBER:bytes_transfered} %{NUMBER:duration}

filter {
    grok {
        match => { "message" => "%{IP:client_ip} %{WORD:request_method} %{URIPATHPARAM:uri_path} %{NUMBER:bytes_transfered} %{NUMBER:duration}" }
    }
}

filter {
    sleep {
        time => "1"
        every => 5
    }
}

input {
    stdin { codec => "json" }
    stdin { codec => json{} }
}

input {
    stdin { codec => line{} }
    stdin { codec => "line" }
}

input {
    file {
        path => "/var/log/someapp.log"
        codec => multiline {
            pattern => "^%{TIMESTAMP_ISO8601}"
            negate => true
            what => previous
        }
    }
}