在线时间:8:00-16:00
迪恩网络APP
随时随地掌握行业动态
扫描二维码
关注迪恩网络微信公众号
开源软件名称:childe/gohangout开源软件地址:https://github.com/childe/gohangout开源编程语言:Go 97.3%开源软件介绍:之前因为 logstash 处理数据的效率比较低, 用 java 模仿 Logstash 写了一个java版本的 https://github.com/childe/hangout. 不知道现在 Logstash 效率怎么样了, 很久不用了. 后来因为Java的太吃内存了, 而且自己对java不熟, 又加上想学习一下golang, 就用golang又写了一次. 内存问题得到了很大的缓解. 目前我们使用golang版本的gohangout每天处理2000亿条以上的数据. 创建一个 QQ 群交流吧 安装从源码编译使用 go module 管理依赖. 直接 make 就可
go get
第三方 Plugin使用 Plugin 的话,自己编译一下,将 CGO_ENABLED 打开:
运行gohangout --config config.yml 一个简单的配置文件如下,从标准输入读取数据,输出到标准输出。具体的配置说明见 配置一节 inputs:
- Stdin: {}
outputs:
- Stdout: {} 日志日志模块使用 github.com/golang/glog , 几个常用参数如下:
pprof debug
prometheus metrics运行时加参数 在 Input/Output/Filter 里面配置 如下例子表示,如果数据通过 if 条件,则此 Add Filter 的计数加 1。
多线程处理默认是一个线程 --worker 4 使用四个线程(goroutine)处理数据. 每个线程拥有自己的filter, output. 比如说translate filter, 每个线程有自己的字典, 他们占用多份内存. elasticsearch output也是一样的, 如果每个 elasticsearch 设置了2并发, 那一共就是8个并发. 进一步说明一下为什么添加了这个配置: 最开始是没有这个配置的, 如果需要多线程并发处理数据, 依赖 Input 里面的配置, 比如说 Kafka 配置 但是后面出现一些矛盾, 比如说, Kafka 的 Consumer 个数多的情况下, 给 Kafka 带来更大压力, 可能导致 Rebalance 更频繁等. 所以如果 Kafka 消费数据没有瓶颈的情况下, 希望控制尽量少的 Consumer, 后面多线程的处理这些数据. 自动更新配置默认不会监听文件系统更新,只在首次初始化时加载配置 --reload 开启这个参数后,当配置文件发生改变会马上触发shutdown,然后重新加载配置文件后运行 除此之外, 开发新的插件
配置配置文件是 Yaml 格式 一个简单的配置示例filters 是一个列表,会依次执行里面的每一个 Filter。 如下例,会先执行第一个 Grok Filter,解析 message 字段,按正则表达式提取出一些其他字段。 再执行第二个 Grok Filter,在这个 Grok 中,会首先判断 if 条件是不是符合,如果不符合就跑过不执行这个 Grok 了。 然后执行第三个 Date Filter,将 logtime 字符串转成 Date 类型的字段,存到 timestamp 字段中。 如果有多个 Output,数据会串行写到每一个 Output。 如果有多个 Input,每个 Input 进来的数据会并行处理后面的 Filter 和 Output。
字段格式约定以 Add Filter 举例
格式1 JSONPATH 格式相比格式2, 更推荐使用这种格式. 更标准, 也灵活, 性能也足够 如果以 给几个下面文中的例子
具体的格式和例子参见 https://goessner.net/articles/JsonPath/ 格式2 [XX][YY]不再推荐使用, 请使用格式1
格式3 {{XXX}}如果含有 Gohangout 使用了 https://github.com/Masterminds/sprig/ 的函数库 来举个例子吧, Date Filter 得到一个 Time 类型的字段, 然后按自己的格式格式化一个字符串出来
格式4 %{XXX}{YYY}含有 2006 01 02 15 04 05 这几个数字是 golang 里面特定的数字, 代表年月日时分秒. 1月2号3点4分5秒06年. 其实就像hangout里面的YYYY MM dd HH mm SS. 如果日期月份包含英文,也可把01换成Jan,比如:02-Jan-2006. 格式5 除了1,2,3,4 之外的其它在不同Filter中, 可能意义不同. 像 Date 中的 src: logtime, 是说取 logtime 字段的值. Elasticsearch 中的 index_type: logs , 这里的 logs 不是指字段名, 就是字面值. INPUTStdin
从标准输入读取数据. codec目前有json/plain/json:not_usenumber三种.
TCP
network默认为 tcp , 可以明确指定使用 tcp4 或者 tcp6 address监听端口, 无默认值, 必须设置 codec默认 plain Kafka
特别注意 参数需要是字符串, 像 decorate_events默认为 false 配置为 true 的话, 可以把 topic/partition/offset 信息添加到 ["@metadata"]["kafka"] 字段中 topic
assignassign 配置用来只消费特定的partition, 和 consumer_settingsbootstrap.servers group.id 必须配置 auto.commit.interval.ms 是指多久commit一次offset, 太长的话有可能造成数据重复消费,太短的话可能会对kafka造成太大压力. max.partition.fetch.bytes 是指kafka client一次从kafka server读取多少数据,默认是10MB from.beginning 如果第一次消费此topic, 或者是offset已经失效, 是从头消费还是从最新消费. 默认是 false. 但是如果已经有过commit offset, 会接着之前的消费. messages_queue_length: 内部使用的消息 channel 的长度,默认为10. sasl.mechanism 认证方式, 目前还只支持 PLAIN 一种 sasl.user sasl认证的用户名 sasl.password sasl认证的密码 servername 如果 servername 不为空的话,证书中的 IP 或者 DNS 名字,需要包含servername 更多配置参见 https://github.com/childe/healer/blob/dev/config.go#L40 OUTPUTStdout
输出到标准输出 if的语法参考下面 IF语法 TCP
network默认为 tcp , 可以明确指定使用 tcp4 或者 tcp6 addressTCP 远端地址, 无默认值, 必须设置 concurrent开几个 tcp 连接一起写, 默认1 Elasticsearch
sniff功能需求 es output 支持特定节点名的 sniffer
Sniff 会调用 index_time_location渲染索引名字时, 使用什么时区. 默认是 UTC. 北京时间 2019-10-25 07:00:00 的日志, 会写到 2019.10.24 这个索引中. 内容如 两个特殊值: bulk_actions多少次提交一次Bulk请求到ES集群. 默认 5000 bulk_size单位是MB, 多少大写提交一次到ES. 默认 15MB flush_interval单位秒, 间隔多少时间提交一次到ES. 默认 30 concurrentbulk 的goroutine 最大值, 默认1 举例来说, 如果Bulk 1W条数据到ES需要5秒, 1W条数据从Input处理完所有Filters然后到Output也需要5秒. 那么把concurrent设置为1就合适, Bulk是异步的, 这5秒钟gohangout会去Filter接下来的数据. 如果Bulk 1W条数据需要10秒, Filter只要5秒, 那么concurrent设置为2可以达到更大的吞吐量. routing默认为空, 不做routing id默认为空, 不设置id (文档id由ES生成) compress默认 true, http请求时做zip压缩 es_version默认为6,可以适配es6的版本,如果设置为7,则可以适配Elasticsearch7以上版本 retry_response_code默认 [401, 502] , 当Bulk请求的返回码是401或者502时, 会重试. 两个额外的配置
没有这个配置的时候, 会把日志做 json.dump, 拿到dump后的[]byte写ES. 如果source_field或者bytes_source_field配置了, 则直接把配置的字段(上面的例子是 bytes_source_field优先级高于source_field. bytes_source_field是指字段是[]byte类型, source_field是指字段是string类型 增加这个配置的来由是这样的. 上游数据源已经是 json.dump之后的[]byte数据, 做一次json.parse, 然后再json.dump, 耗费了大量CPU做无用功. Kafka特别注意 参数需要是字符串, 像
clickhouse
Notice: 如果表中字段有 default 值, 目前只支持字符串和数字 的 DEFAULT 表达式解析和处理, 如果像 IPv4设置了default 值, 是处理不了的. 代码中写死了 IPv4 和 IPv6 的默认值都是0 table表名. 必须配置 hostsclickhouse 节点列表. 必须配置 fields初始化的时候会从 ClickHouse 里面读取所有字段。 也可以手工配置,会优先使用手工配置。 为了暂时缓解 #159 bulk_actions多少次提交一次Bulk请求到ES集群. 默认 1000 flush_interval单位秒, 间隔多少时间提交一次到ES. 默认 30 concurrentbulk 的goroutine 最大值, 默认1 conn_max_life_time到 ClickHouse 的连接的生存时间, 单位为秒. 默认不设置, 也就是生存时间无限长. FILTER通用字段 |