模仿logstash做的一个应用.
This product includes GeoLite2 data created by MaxMind, available from http://www.maxmind.com
我们一直用logstash从Kafka消费数据进ES, 随着数据量的越来越大, 需要的logstash实例和机器也越来越多.
于是就拿java实现了一下目前我们用到的Logstash的插件. 做为java初学者, 肯定有很多地方写的不好.
- input
- kafka
- output
- elastickseach
- kafka
- stdout
- filter
- Grok
- Date
- Json
- Gsub
- Drop
- Trim
- Translate
- Rename
- Lowercase
- Uppercase
- Remove
- Add
- KV
- URLDecode
- GeoIP2
用一个典型配置做测试, 包括Grok Date AddField If条件判断等, 吞吐量是Logstash的5倍左右 .
不压测吞吐量, 正常消费的情况下, CPU使用率大概是Logstash的50%到25%.
运行
bin/hangout -f app.yml
记录日志到文件:
bin/hangout -f app.yml -l /var/log/hangout/app.yml.log
默认日志记录是warn级别, 详细日志:
bin/hangout -f app.yml -l /var/log/hangout/app.yml.log -vvvv
-v 是info , -vv 是debug , -vvvv 是Trace
配置
配置在一定程度上也是模仿logstash, 但用的是通用的yaml格式. 因为能力有限, 不能实现logstash的那套语法, if是用了另外的格式. 可以参考一个配置的示例
Input
kafka
从Kafka读数据. 下面示例是说, 开2个线程取app这个topic的数据, 编码格式为plain纯文本. consumer_settings中的参数, 参考[kafka]官方文档](http://kafka.apache.org/documentation.html#consumerconfigs), 所有参数都可以在这里配置. 比较重要的是group.id, zookeeper.connect这两个, 一定要有. zookeeper.connect的格式为 hostname1:port1,hostname2:port2,hostname3:port3.
- Kafka:
topic:
app: 2
consumer_settings:
group.id: hangout
zookeeper.connect: 192.168.1.200:2181
auto.commit.interval.ms: "1000"
socket.receive.buffer.bytes: "1048576"
fetch.message.max.bytes: "1048576"
num.consumer.fetchers: "4"
codec: plain
Filter
Grok
Grok是为了把一个字符串切割成多个field, 用的是Joni库, 不完全支持logstash里面的patterns语法,%{INT:bytes:int}这种语法不支持, 只支持%{INT:bytes},字段类型需要在ES中定义.
会依次匹配match中的正则, 直到有一个成功的.
注意, 如果正则中的groupname和已有的字段一样, 原来的字段被覆盖
src: message #default message
match:
- '(?<logtime>\S+) (?<user>\S+) (-|(?<level>\w+))'
- '(?<logtime>\S+\s+\S+) (?<user>\S+) (-|(?<level>\w+))'
remove_fields: ['message']
tag_on_failure: grokfail #default grokfail
Date
Date是用的jona-time做解析和格式化.
会依次匹配formats里面的格式, 直到成功.
src: logtime # default logtime
formats:
- 'ISO8601'
- 'UNIX'
- 'UNIX_MS'
- 'YYYY-MM-dd HH:mm:ss.SSS'
remove_fields: ['logtime']
tag_on_failure: datefail # default datefail
Json
解析json字符串, 如果json里面的字段和原有的字段重复, 原有字段会被覆盖!
- Json:
filed: message # required
GeoIP2
geoip2用的是maxmind公司的开源数据和算法. This product includes GeoLite2 data created by MaxMind, available from http://www.maxmind.com
geoip2里面可以获取的数据也比较多, 目前我只是用到了country_code country_name city_name latitude longitude location 6个字段.
注意 geoip2里面不含有city_code值, 后续会增加GeoIP filter, 可以获取city_code值
如果不需要, 还是建议用GeoIP2, 据ELKstack-guide, 速度比GeoIP有速倍的提升.
maxmind也提供了数据的下载 http://dev.maxmind.com/geoip/geoip2/geolite2/
- GeoIP2:
source: message # required
target: geoip # default geoip
database: '/tmp/GeoLite2-City.mmdb'
Drop
没有额外参数, 配置if使用.
if:
- '<#if user?matches("huhan3")>true</#if>'
IF
应用于filter. 是一个列表, 需要满足列表中的每一个条件, 使用freemarker 模板引擎.
下面这个例子是添加一个target字段, target的值为url字段的第5个部分(下标从0开始).
只有在满足以下2个条件的时候才会触发添加字段这个行为.
- 日志含有url字段
- url中包含 "http://images4.c-ctrip.com/target/" 或者 ".c-ctrip.com/images/" 字符串.
- Add:
fields:
target: '<#assign a=url?split("/")>${a[4]}'
if:
- '<#if url??>true</#if>'
- '<#if url?contains("http://images4.c-ctrip.com/target/")>true<#elseif url?contains(".c-ctrip.com/images/")>true</#if>'
Translate
每隔一段时间, 会重新加载字典. 字典是yaml格式.
用的snakeyaml加载yaml文件的, 如果key是整数类型,会首先尝试转换成Integer, 超出Integer范围才会尝试Long.
但是Json格式的日志会被直接转成Long类型的, 导致不能匹配.
所以yaml里面需要这么写:
!!java.lang.Long 123: 信用卡
!!java.lang.Long 345: 借记卡
Tranlate 配置：
source: user
target: nick
dictionary_path: /user-nick.yml
refresh_interval: 300 # default 300 seconds
KV
将 a=1&b=2, 或者name=app id=123 type=nginx 这样的字符串拆分成{a:1,b:2} {name:app, id:123, type:nginx} 等多个字段, 放到日志里面去.
配置如下.
如果targete有定义, 会把拆分出来字段放在这个字段中, 如果没有定义,放到在顶层.
trim 是把拆分出来的字段内容做前后修整. 将不需要的字符去掉. 下面的示例就是说把双引号和tag都去掉.
trimkey和trim类似, 处理的是字段名称.
- KV:
source: msg # default message
target: kv # default null
field_split: ' ' # default " "
value_split: '=' # default "="
trim: '\t\"' #default null
trimkey: '\"' # default null
tag_on_failure: "KVfail" # default "KVfail"
remove_fields: ['msg']
其它filter
配置都比较简单, 可以参考配置文件示例
freemarker
If 条件和addfileds等插件使用freemarker模板引擎, 速度快, 功能强大.
Output
Elasticsearch
使用bulk api 批量将数据写入ES.
插入失败的时候, 只对 TOO_MANY_REQUESTS, SERVICE_UNAVAILABLE 这两种错误重试.
参数含义参考java client bulk api
concurrent_requests设置成大于0的数, 意思着多线程处理, 以我应用的经验,还有是一定OOM风险的. 因为在执行bulk的时候, bulkProcessor还在继续接收新的文档, 如果bulk失败了, 会把失败的actionrequest继续放到bulkProcessor里面. 一段时间内, 由于各种原因一直失败的话, 内存就会越用越多.
看了一下ES的代码, 现在的bulk就是对原来的bulkRequest简单封装, 又是异步又是回调, 我这种新手有些控制不了, 考虑要不要换回bulkRequest自己控制.
cluster: prod # cluster name, required
hosts: # required
- 192.168.1.100
- 192.168.1.200:9301
index: 'hangout-${@timestamp.toString("YYYY.MM.dd")}'
index_type: logs # default logs
bulk_actions: 20000 # default 20000
bulk_size: 15 #default 15
flush_interval: 10 #default 10
concurrent_requests: 0 #default 0
Kafka
#broker_list format is host1:port1,host2:port2, and the list can be a subset of brokers or a VIP pointing to a subset of brokers.
broker_list: 192.168.1.100:9092 # required
topic: test # required
Stdout
主要测试用吧.
开发:
如果需要新的插件, 可以从源码编译重新打包.
或者编译成jar包, 加入到class path(尚未开发完成).
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent