分布式框架－日志系统思路及实现

转自：https://www.jianshu.com/p/ce30c31111ca

背景

随着互联网时代数据规模的爆发式增加,传统的单机系统在性能和可用性上已经没法胜任,分布式应用和服务化应用开始走进你们的视野，可是分布式的部署也会带来另外的问题，日志分散在各个应用服务节点中，出现问题不方便及时排查，尤为是服务化的应用中，分析问题时可能须要查看多个日志文件才能定位问题，若是相关项目不是一个团队维护时沟通成本更是直线上升，怎么将日志文件归集，怎么将日志文件呈现成了不少公司须要面对的问题，所以日志系统应运而生。

dapeng日志系统的选型

日志系统一般有三部分组成，采集器、解析器、存储器

采集器一般部署在各个应用结点中，它监控本地文件的变化，对于新产生的日志变化，它实时收集并发送给对应的解析器，常见的采集器有flume、logstash、fluentd以及更轻量级的fluent-bit

解析器一般和采集器结合在一块儿，也有一部分解析器是经过接收缓冲队列，将日志解析成json格式数据后，把数据发往存储器进行存储

存储器用于存储对应的数据，提供相关的查询，常见的存储有hdfs、elasticsearch

咱们dapeng选取的是fluent-bit+fluentd+kafka+elasticsearch做为日志系统的方案，zookeeper、elasticsearch、kafka都采用集群模式，示例图中采用单结点fluent-bit收集各个docker容器中的日志文件发往fluentd，fluentd作为中转收集全部的日志发往kafak用于削峰填谷，削峰后的数据再经由fluentd发送给elasticsearch进行存储

 

image

为了支持fluent-bit<=>fluentd的高可用, 咱们改动了fluent-bit的源码. 另外, 生产环境上, 上述结构图中的每个环节都不能省, 以避免数据量太大发生不可预料的错误.
目前咱们生产环境, 小规模应用的状况下, 天天大概产生5千万条日志记录.

关于MDC的小插曲

Logback中有一项功能很好使-MDC，映射诊断环境（Mapped Diagnostic Context）MDC本质上是使用的ThreadLocal。系统调用链可能很长，为了方便日志跟踪，统一打印标识。咱们dapeng使用MDC来保存sessionTid，在一个完整的调用链中使sessionTid在各个服务中进行传递，将服务进行串联，方便问题定位，具体的logback以下

<appender name="SIMPLEFILE" class="ch.qos.logback.core.rolling.RollingFileAppender"> <prudent>false</prudent> <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"> <fileNamePattern>${soa.base}/logs/simple-dapeng-container.%d{yyyy-MM-dd}.log</fileNamePattern> <maxHistory>30</maxHistory> </rollingPolicy> <encoder> <pattern>%d{MM-dd HH:mm:ss SSS} %t %p [%X{sessionTid}] - %m%n</pattern> </encoder> </appender> 

配置采集器

[SERVICE]
    Flush        5
    Daemon       On
    Log_Level    error
    Log_File     /fluent-bit/log/fluent-bit.log Parsers_File parse_dapeng.conf [INPUT] Name tail Path /dapeng-container/logs/*.log Exclude_Path /dapeng-container/logs/fluent*.log,/dapeng-container/logs/gc*.log Tag dapeng Multiline on Buffer_Chunk_Size 2m buffer_max_size 30m Mem_Buf_Limit 32m DB.Sync Normal db_count 400 Parser_Firstline dapeng_multiline db /fluent-bit/db/logs.db [FILTER] Name record_modifier Match * Record hostname ${HOSTNAME} Record tag ${serviceName} [OUTPUT] Name Forward Match * Host fluentd Port 24224 HostStandby fluentdStandby PortStandby 24224 

record_modifer用于在解析出的json中增长hostname标识和tag标识方便日志检索
chunk及buffer块的设置根据各系统日志的大小来进行设置
HostStandby和PortStandby是咱们dapeng基于原生fluent-bit进行改造添，当主fluentd挂掉后，日志事件会相应的发送给fluentdstandBy进行处理

解析器的配置

[PARSER]
    Name        dapeng_multiline
    Format      regex
    Regex       (?<logtime>\d{1,2}-\d{1,2} \d{1,2}:\d{1,2}:\d{1,2} \d{1,3}) (?<threadPool>.*) (?<level>.*) \[(?<sessionTid>.*)\] - (?<message>.*) 

解析器这块对应上面的logback配置，将日志消息处理成比较直观的JSON数据进行存储

转发器fluentd的配置(用于接收消息发送kafka)

<system>
        log_level error
        flush_thread_count 8
        workers 8
</system>
<source> @type forward port 24224 </source> <source> @type monitor_agent port 24225 </source> <match dapeng tomcat> @type kafka_buffered brokers ${kafkabrokers} topic_key messages #zookeeper 192.168.20.200:2181 buffer_type file buffer_path /tmp/buffer flush_interval 60s default_topic messages output_data_type json compression_codec gzip max_send_retries 3 required_acks -1 discard_kafka_delivery_failed true </match> 

monitor_agent是fluentd的一个插件，能够及时获取fluentd响应用于fluentd的健康度检查

[root@monitor-elk etc]# curl 192.168.20.200:24225/api/plugins.json {"plugins":[{"plugin_id":"object:3ff681f97a88","plugin_category":"input","type":"forward","config":{"@type":"forward","port":"24224"},"output_plugin":false,"retry_count":null},{"plugin_id":"object:3ff681c37078","plugin_category":"input","type":"monitor_agent","config":{"@type":"monitor_agent","port":"24225"},"output_plugin":false,"retry_count":null},{"plugin_id":"object:3ff681c19ca8","plugin_category":"output","type":"kafka_buffered","config":{"@type":"kafka_buffered","brokers":"192.168.20.200:9092","topic_key":"messages","buffer_type":"file","buffer_path":"/tmp/buffer","flush_interval":"60s","default_topic":"messages","output_data_type":"json","compression_codec":"gzip","max_send_retries":"3","required_acks":"-1","discard_kafka_delivery_failed":"true"},"output_plugin":true,"buffer_queue_length":0,"buffer_total_queued_size":1174144,"retry_count":6,"retry":{}}]} 

转发器fluentd的配置(用于接收kafka中的消息发送elasticsearch)

<system>
        log_level info
        flush_thread_count 8
        workers 8
</system>
<source> @type kafka_group brokers 192.168.20.200:9092 consumer_group dapeng_consume2 topics messages format json start_from_beginning false </source> <source> @type monitor_agent port 24225 </source> <match> @type elasticsearch host 192.168.20.200 port 9200 index_name dapeng_log_index type_name dapeng_log content_type application/x-ndjson buffer_type file buffer_path /tmp/buffer_file buffer_chunk_limit 30m buffer_queue_limit 512 flush_mode interval flush_interval 60s request_timeout 15s flush_thread_count 8 reload_on_failure true resurrect_after 30s reconnect_on_error true with_transporter_log true logstash_format true logstash_prefix dapeng_log_index template_name dapeng_log_index template_file /fluentd/etc/template.json num_threads 8 utc_index false </match> 

start_from_beginning默认为true，表明从消息队列中起始读取数据，当fluentd重启会形成日志消息冗余，所以这里配置false,若是须要恢复日志索引，能够配置成true让日志消息再消息一次（咱们日志kafka消息保留的策略是保留1天，所以当出现故障时咱们能够快速恢复1天内的日志）
logstash_format 用于配置将日志索引按天数来存放

template.json模板配置

{
 "mappings": { "dapeng_log": { "properties": { "logtime": { "type": "date", "format": "MM-dd HH:mm:ss SSS" }, "threadPool": { "type": "string", "index": "not_analyzed" }, "level": { "type": "string", "index": "not_analyzed" }, "tag": { "type": "string", "index": "not_analyzed" }, "message": { "type": "string", "index": "not_analyzed", "ignore_above":256 }, "hostname":{ "type": "string", "index": "not_analyzed" }, "sessionTid":{ "type": "string", "index": "not_analyzed" }, "log":{ "type": "string", "index": "not_analyzed" } } } }, "settings": { "index": { "max_result_window": "100000000", "number_of_shards": "1", "number_of_replicas": "1", "refresh_interval": "60s" } }, "warmers": {}, "template": "dapeng_log_index-*" } 

配置使用的是es2的配置，线上咱们使用的是5.6.9的版本，es这块能够向下兼容将string类型的转换为keyword

日志查询

查询服务调用关系

经过sessionTid来查询服务间的调用关系，这里sessionTid正是上面MDC中设置的，在服务的调用中经过InvocationContext(dapeng上下文)进行传递

 

服务调用关系.png

查询堆栈异常

 

堆栈异常.png

 

按天进行错误分组

GET dapeng_log_index-2018.07.25/_search
{
  "size": 0, "query": { "bool": { "must": [ { "term": { "level": "ERROR" } } ], "filter": { "script": { "script": { "source": "doc['message'].values.length==0" } } } } }, "aggs": { "group_by_tag": { "terms": { "field": "tag", "size": 100 } } } } 

坑及优化

fluent-bit报Invalid indentation level
fluent-bit对配置文件的要求比较高，请保持配置用空格对齐，不要使用tab键

fluent-bit高内存占用
根据官方文档描述，在某些环境中，一般会发现被摄取的日志或数据比将其刷新到某些目的地的速度要快。 常见的状况是从大日志文件读取并经过网络将日志分派到后端，这须要一些时间来响应，这样会产生背压，致使服务中的高内存消耗。为了不背压，Fluent Bit在引擎中实现了一种限制数据量的机制，经过配置参数Mem_Buf_Limit完成的。

咱们这里经过配置Mem_Buf_Limit来优化，另外fluent-bit默认使用Glibc来管理分配内存，这里咱们使用jmalloc,这是一种替代内存分配器，它具备更好的策略来减小其余碎片以得到更好的性能

 

image

fluentd隔天写入索引
写入es中的日志会比当前时间提早8个小时，例如0-8点的日志会写入到昨天的索引中，这里咱们配置utc-index为false便可

elasticsearch长期报GC
因为业务高峰日志量致使瞬时写入较大，es会长时间报gc，影响数据的写入，这里咱们引入kafka做消息缓冲，另外咱们弃用elasticsearch默认的垃圾回收器，使用G1回收器

jdb2高io使用
最开始，咱们在网站上检索关于jdb2高iowait的解决方案，给出的方案都是ext4的bug，差一点我就信了，linux的bug也能遇到，可是转过来一想这bug也好多年了，内核早就修复了，应该不是这方面的问题，咱们使用top查看cpu的使用状况,比较空闲，可是wait比较高

 

image

使用iotop来查看磁盘的io使用状况,基本都是fluent-bit产生的

 

image

接下来咱们使用 blktrace来收集更进一步的详细信息

 

image

最后咱们使用wc来统计43这一秒内fluent-bit产生的IO请求数(Q表示即将生成IO请求)

 

image

问题元凶找到了，fluent-bit读取的日志文件后会在写出的时候更新文件位置索引，将索引保存在sqllite中，根据上面的统计，每秒钟产生的IO操做在101次（因为有4个fluent-bit）正是因为fluent-bit频繁的更新sqlite中的文件索引，形成文件合并引发的高iowait，所以须要对sqlite的写入次数加限制，这里咱们基于fluent-bit改造了两种 方案，第一种，每次都只从尾部读取文件，这样就省掉了文件索引的保存达到减小磁盘IO，第二种，增长db_count参数用于对chunk块计数，当发送chunk块计数达到配置的参数时保存文件的位置索引，咱们dapeng对这两块都进行了个性化改造实现，改造后的效果对比图以下

 

image

elasticsearch内存使用优化

es这块按天来存放对应的日志索引，长期不用的索引会占用大量内存。通常日志索引只须要开放近三天的索引便可，日志索引保留近一月便可

#!/bin/bash date=`date -d "3 days ago" +%Y.%m.%d` date1=`date -d "30 days ago" +%Y.%m.%d` echo $date echo $date1 curl -XPOST http://192.168.20.200:9200/dapeng_log_index-$date/_close curl -XDELETE "http://192.168.20.200:9200/dapeng_log_index-${date1}" 

基于日志系统的衍生扩展

目前咱们基于现有的日志系统，作了生产故障实时告警系统，直接钉钉推送给相关的业务系统负责人，具体方案有两种，一种是根据索引去过滤近30分钟的日志异常推送，另一种是从kafak中提取消息后过滤推送，第一种是假实时，错误有所延迟，第二种是彻底实时，咱们如今采起的是第一种方案，第二种方案有待实现

 

image

总结

到这一步，咱们的日志系统已经搭建成功了，当服务器扩容时，因为fluent-bit是集成在dapeng容器中，只须要在环境变量中简单配置serviceName和hostname以及fluentdhost便可，日志消息就会写入到es存储中。

日志系统是一个很是重要的功能组成部分，咱们可使用日志系统来进行错误编排，系统优化，根据这些信息调整系统的行为，提升系统的可用性。(想了解更多？请关注dapeng开源)