【EKL】

1. logstash

1.1 启动服务

1.1.1 标准输入输出启动 (调试用)

./logstash -e 'input{stdin{}} output{stdout{}}'
Stdin：标准化输入插件
Stdout：标准化输出插件

1.1.2 指定配置文件启动

• logstash -f ./E:/ES/logstash-7.10.0/config/logstash-sample.conf
• logstash.bat -f ..\config\logstash-sample.conf (或者)

1.2 配置文件中的input

1.2 配置文件中的filter

1.2.1 基本结构

filter {插件名称 {参数 => 值...}
}

1.2.2 常用 Filter 插件

• grok - 模式匹配解析

filter {grok {match => { "message" => "%{COMBINEDAPACHELOG}" }}
}
filter {grok {match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }}
}

eg:

192.168.1.10 GET /api/v1/user 1024 12

1.2.3 mutate - 字段操作

filter {mutate {# 重命名字段rename => { "old_field" => "new_field" }# 添加字段add_field => { "new_field" => "value" }# 删除字段remove_field => [ "field1", "field2" ]# 替换字段值replace => { "field" => "new_value" }# 转换类型convert => { "numeric_field" => "integer" }}
}

1.2.4 date - 日期处理

filter {date {match => [ "timestamp", "ISO8601" ]target => "@timestamp" # 默认就是@timestamp}
}

1.2.4 json - JSON解析

filter {json {source => "message"target => "json_content"}
}

1.2.6 kv - 键值对解析

filter {kv {source => "message"field_split => "&?"value_split => "="}
}

1.2.7 geoip - IP地理位置

filter {geoip {source => "clientip"}
}

1.2.8 useragent - 用户代理解析

filter {useragent {source => "user_agent"target => "ua"}
}

1.2.9 条件处理

filter {if [type] == "apache" {grok {match => { "message" => "%{COMBINEDAPACHELOG}" }}} else if [type] == "syslog" {grok {match => { "message" => "%{SYSLOGLINE}" }}} else {drop {}}
}

1.2.10 多过滤器顺序执行

filter {grok { ... }date { ... }mutate { ... }
}

1.4 配置文件中的output