1. 背景
推荐系统的推荐请求追踪日志，通过ELK收集，方便遇到问题时，可以通过唯一标识sid来复现推荐过程
最近在碰到了几个bad case，需要通过sid来查询推荐日志，但发现部分无法在kibana查询到
2. 分析
推荐日志的整个收集流程如下：
flowchart LR 线上机器日志 --> 平台Flume --> Kafka --> Logstash --> ES因为整个流程较长，且中间的各个组件也是由不同部门来负责的，其中任一环节数据都有可能丢失，所以排查起来比较困难
2.1 版本信息
Flume: 未知
Kafka: 2.4.0
ELK: 7.17
2.2 ELK排查-日志和配置
首先联系了DB的同学，主要负责ELK，他们首先查看了日志，确认无报错之后，要求先确认Kafka的数据是否有丢失
确认Logstash配置和ES Mapping没有问题
input {
kafka {
add_field => {"myid"=>"hdp_lbg_zhaopin_hrgrecservice_trace_all"}
bootstrap_servers => "xxx.xxx.xxx.xxx:9092"
client_id => "hdp_lbg_zhaopin-hdp_lbg_zhaopin_hrgrecservice_trace_all"
group_id => "hdp_lbg_zhaopin_hrgrecservice_trace_topic"
auto_offset_reset => "latest"
consumer_threads => 1
topics => ["hdp_lbg_zhaopin_hrgrecservice_trace_all"]
codec => json
}
}
filter {
if [myid] == "hdp_lbg_zhaopin_hrgrecservice_trace_all" {
json {
source => "message"
}
mutate {
convert => { "recOuts" => "string"}
join => {"recOuts" => ","}
add_field => {"context" => "%{traceContext}"}
add_field => {"trace" => "%{traces}"}
remove_field => ["message", "event", "traces", "traceContext"]
}
}
}
output {
if [myid] == "hdp_lbg_zhaopin_hrgrecservice_trace_all" {
elasticsearch {
hosts => ["ip:port"]
index => "hdp_lbg_zhaopin_hrgrecservice_trace_all-%{+YYYY_MM_dd}"
user =>
password =>
}
stdout { }
}
}
2.2 Kafka排查-Kafka数据是否丢失
然后联系的大数据的同学，他们让我们把Kafka的数据写入Hive中，然后看缺失的数据是否能在Hive中查询到，这样做之后在Hive中查到了缺失的数据，确认Kafka的数据没有问题
缺失的数据是找到在机器上的一条日志，但不在Kibana中的数据即可，因为我们缺失率也是很高的，所以也很容易找到
如果缺失的条数很少，可以选取一段时间对比Hive和Kibana的数据
flowchart LR 线上机器日志 --> 平台Flume --> Kafka --> Hive2.3 继续ELK排查-数据在哪个环节丢了
确认Kafka数据无误之后，再返回去找DB的同学查ELK问题，DB同学使用另外一个Logstash读取Kafka的数据，然后写入ES，同时开启本地输出，录制了一段时间之后，再在Logstash输出文件中寻找这段时间内缺失数据，发现都找不到，于是确认为Kafka到Logstash丢失数据
flowchart LR 线上机器日志 --> 平台Flume --> Kafka -.-> Logstash2.4 继续-丢失的日志有何特点
虽然确定了是Kafka到Logstash丢失了数据，但还是无法确定具体原因，考虑过是不是某些日志过大，Logstash无法处理，于是看了单台机器的连续的十几条日志，发现丢失的日志大小有大有小，丢的间隔条数也是毫无规律
2.5 Kafka数据格式-一个消息有多个日志
在毫无头绪的情况下，想去看看Kafka内的数据，于是探查了几条，发现了一个奇怪的现象，Kafka的消息里面有多行日志，之间看起来像是空格拼接了起来，后来知道那是个换行符，如下：
线上机器日志符格式：
A
B
C
Kakfa消息格式：
A\nB
C
所以将探查的数据在Kibana中搜索，发现了规律，对于Kafka的每一条消息，只有第一个日志能写入Logstash，对于上面的数据，A和C能写入，而B不能
对于了别的Logstash集群的配置，发现了问题
我们的Logstash Kafka input配置中是codec => json，而其他的配的是codec => line，也就是说Logstash认为Kafka的一条消息就是一个JSON，拿过来直接转，转化的过程中应该把后面换行符拼接的不合法的部分给丢弃了，而使用codec => line的方式，Logstash会把Kafka的消息按照换行符分割成多个event
不过返回来看，为什么Kafka不是一行日志一条消息呢，在我们机器配置收集日志的Flume中，有个默认选项，“是否聚合”，选择是会将不同行的日志聚合到一起，这就是Kafka数据格式看起来很奇怪的原因，后续消费者，不管是Logstash，还是通过DDS写入Hive，还是自己写代码来消费，都要处理这个换行符
3. 总结
Flume将多行日志聚合成一条消息放入Kafka，Logstash在接收的时候需要配置codec => line再将一个消息转换为多行日志