1. 背景

推荐系统的推荐请求追踪日志，通过ELK收集，方便遇到问题时，可以通过唯一标识sid来复现推荐过程

最近在碰到了几个bad case，需要通过sid来查询推荐日志，但发现部分无法在kibana查询到

2. 分析

推荐日志的整个收集流程如下：

因为整个流程较长，且中间的各个组件也是由不同部门来负责的，其中任一环节数据都有可能丢失，所以排查起来比较困难

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的数据

2.3 继续ELK排查-数据在哪个环节丢了

确认Kafka数据无误之后，再返回去找DB的同学查ELK问题，DB同学使用另外一个Logstash读取Kafka的数据，然后写入ES，同时开启本地输出，录制了一段时间之后，再在Logstash输出文件中寻找这段时间内缺失数据，发现都找不到，于是确认为Kafka到Logstash丢失数据

2.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再将一个消息转换为多行日志

本文来自博客园，作者：songtianer，转载请注明原文链接：https://www.cnblogs.com/songjiyang/p/17235462.html