logstash
logstash是一款轻量级的日志搜集处理框架,可以方便的把分散的、多样化的日志搜集起来,并进行自定义的处理,然后传输到指定的位置(比如落地到本地磁盘文件或直接入到elasticsearch)。
配置文件
配置logstash的kafka数据源,并指定消费的topics,如果日志格式为json,设定输入数据格式为json。由于message字段里面的数据应该是json,但存储的时候是以字符串形式存储,logstash如果不作处理直接入es,在es中无法以json格式查看,因此这里利用 logstash json filter 将message字段转为json格式
input {
kafka {
bootstrap_servers => "192.168.0.111:9091"
topics => ["logtest"]
codec => json {
charset => "UTF-8"
}
}
# 如果有其他数据源,直接在下面追加
}
filter {
# 将message转为json格式
if [type] == "log" {
json {
source => "message"
target => "message"
}
}
}
output {
# 处理后的日志落到本地文件
file {
path => "/config-dir/test.log"
flush_interval => 0
}
# 处理后的日志入es
elasticsearch {
hosts => "192.168.0.112:9201"
index => "logstash-%{[type]}-%{+YYYY.MM.dd}"
}
}
启动logstash
为测试方便直接使用logstash官方docker镜像,快速启动测试程序
docker pull logstash
docker run -it --rm --net=host -v "$PWD":/config-dir logstash -f /config-dir/logstash.conf
查看es中的数据
kafka的logtest队列有新数据进入时,logstash会对其进行消费,处理后输入es,存储的数据格式如下: