ELK生态下，构建日志分析系统的选择

说起开源的日志分析系统，ELK几乎无人不晓，这个生态并非是Elastic特意而为，毕竟Elasticsearch的初心是分布式的搜索引擎，被广泛用作日志系统纯粹一个“美丽的意外”，这是社区使用者推动而成。而现在各大云厂商推广自己的日志服务时，也往往将各种指标对标于ELK，可见其影响之广。

但其实，流行的架构中并非只有ELKB，当我们使用ELKB搭建一套日志系统时，除了Elasticsearch, Logstash, Kibana, beats之外，其实被广泛使用的还有另一个工具 —— Kafka。在这当中，Kafka的作用是明显的，作为一个中间件，一个缓冲，它起到了提高吞吐，隔离峰值影响，缓存日志数据，快速落盘，同时通过producer/consumer模式，让Logstash能够横向拓展的作用，还能够用作数据的多路分发。因此，大多数时候，我们看到的实际架构，按数据流转顺序排列，应该是BKLEK架构。

但我们在使用Kafka的时候，也并非是没有成本的，额外的一套分布式系统，更长的数据链路，都会是我们在做最后架构选型时的一些痒点，特别是随着我们产生的数据越来越多，BKLEK架构会变得越来越大，越来越重，成本、性能、运维的简易性都会成为我们评估日志系统的重要指标。因此，我们的问题是：在Elastic Stack都已经进化到了8.1的当下，我们是否还需要延续一直以来的惯性思维，认为在我们仍然在任何情况下都是需要BKLEK的架构呢？

在我们开始正式探讨之前，我们可以从现在普遍看到的新的架构图可以一猜端倪：

在这个架构中，所有的Integration的输出都是Elasticsearch，所有的数据处理都由ingest pipeline完成，数据的完整性和可靠性，由端点和elasticsearch之间的应答确认来保证。因此，我们本文探讨的是：

1. 我们数据采集端直接到ES的架构，是否可取
2. 我们什么时候可以使用这种架构

数据采集端直接到ES的架构分析

虽然Elastic原厂的Fleet与Elastic Agent已经处于GA（普遍可用）的阶段，但因为其本身的一些限制，比如：

1. Integration Repository需要连接外网
2. 需要单独的Fleet Server (在最新版本已经与APM server合并为Integration Server)
3. Elastic Agent还没有能完全覆盖Beats所支持的数据源

因此，现阶段，我们讨论数据采集端直接到ES的架构时，会主要集中在Beats->Elasticsearch这一架构。这一架构，相对于BKLEK架构来说，少了中间的Kafka，甚至我们可以忽略Logstash，因此，架构会相对精简，带来的好处包括：

• 相对更低的成本
• 更高的传输和数据处理效率
• 更一致的安全特性
• 更容易进行监控

当然，在带来以上好处的同时，我们也会失去Kafka所带来的各种好处。不过不用担心，Kafka的特性只是构建一个稳健的日志分析系统的充分条件，而非必要条件，在不少场景下，我们不一定是非Kafka不可。接下来，我们将讨论几个我们相对会比较关心的问题，以让大家了解，我们是否可以选择这种架构，什么时候选择这种架构，以及相应的最佳实践。

Beats -> Elasticsearch链路的健康保障

对于不使用Kafka的场景，我们可能始终会有点觉得不踏实。因为对Beats -> Elasticsearch这个简单架构不够了解，以至于我们信心不足。接下来，我们讨论一下，在这种简单架构下，我们是如何面对各种可能出现的问题的。

大多数架构师会担心的问题是流量波动的问题，如果突然出现日志流量的洪峰，是不是会影响到后端的日志系统。这个问题的答案是，有影响，但影响有限。

我们先明确一下日志系统的主要作用：即日志的集中管理，在统一的日志平台上提供所有日志的准实时的关联查询与分析能力。这里的核心是准实时和查询能力。

在流量洪峰的情况下，受影响的是“准实时”的能力，因为受限于日志系统的处理能力，如果日志产生的速度，大于日志系统处理的速度，则我们无法读取到最新的数据。这个问题，即便是我们有了Kafka也是无法解决的。

而查询的能力，几乎不受影响。原因如下：

• ES的读写是由不同的线程处理的，有各自独立的线程池。过载的写流量，会导致后到的写请求，因为写线程池已满，而被拒绝。但不会导致查询情况的拒接服务
• ES从7.0开始，使用真实内存断路器，能够避免由于过多的请求，导致节点内存OOM。但在日志场景，特定时间内日志请求的数量是有明确上限的，该上限为Beats的数量，相比于高并发的读场景，Beats几乎不可能造成请求数量的过载
• Beats与Elasticsearch之间有背压检测机制，当Beats检测到ES有拒绝写服务的情况出现时，会主动限流降低，避免对ES产生持续的压力与影响。具体参见，附录1

或许这种“轻描淡写”的描述，让大家觉得有点难以相信，甚至与平时的经验有点不符。但实际上真正的问题来自于不合理的架构：

• 大多数企业会选择维护一个“怪物”级别的ES集群，即，将业务搜索和日志查询两个完全不同的场景混用与同一个ES集群。因此，当某些日志流量洪峰来临，打满了“write”线程池之后，会发现业务数据也无法写入了。但仍然的，kafka也解决不了这个问题，因为，即便有kafka在前面挡住真正的日志数据洪流，你也很难判断kafka后面的logtash集群，会不会导致ES集群拒绝写服务，因为消费日志永远比索引日志要快，logstash会持续消费kafka中的数据，直到感受到背压，才会减缓输出的频率，这点和Filebeat是一样的。
• 大多数在早期就已经在使用ES的企业，由于缺乏合理的架构设计，或者缺乏足够的升级集群的信心，集群一直停留在5,6等老版本的，因此，整个架构缺乏足够的健壮性和可恢复性，一些关键能力，比如背压检测，内存断路器，Heap memory offload等功能是没有的。导致集群较为脆弱，在流量洪峰时，出现OOM等问题，也是有的。但仍然，kafka也解决不了这个问题。

因此，我们在做架构选型之前，确定要不要使用Beats -> Elasticsearch这一架构之前，不妨先审视一下，我们当前是不是在业务和日志混用ES集群，我们的ES集群版本是不是过低，运维缺乏升级的能力？这两个反而是更致命的问题

回到流量洪峰问题的解决。即便加上kafka，也只是治标不治本，看起来架构更加健壮了一点，但实际上也并不能帮助我们提高后端的消费能力，我们整体面临的情况和使用Beats -> Elasticsearch这一架构是没有多大的区别的（Kafka最大的好处，就是把这些数据落盘到自己的磁盘上了，但这真的是我们在这个场景下需要解决的问题吗？）。

流量洪峰的治本方案，应该是着眼于快速的故障排查，快速的找到出现故障的机器并解决故障。（出现业务暴增也不是不可能的情况，但正常的业务日志撑爆日志系统的可能性非常小，因为首先会撑爆的是产生日志的业务系统，因此，一旦业务暴增，我们需要考虑业务系统和日志系统的同步扩容）

我们可以从Elastic Stack上快速的找到这样问题的答案，而且方法很多，这里举例一二：。

1，监控日志集群的bulk拒绝率

数据产生于那一个瞬间，并且在产生数据的地方，我们并没有什么机制去存储这些数据，因此，在数据真正进入后端的存储分析之前，我们往往需要一个Queue能够帮我们把这些数据落盘。这时，Kafka几乎就是我们在技术选型时的不二选择。

数据需要高速转移

再说“快速”，Kafka的高吞吐也是非常重要的一个能力，也是其得以让人追捧的主要原因之一。可以设想这样的一种场景，在云原生环境下，我们非常动态的去创建各种计算资源，以应对业务流量的变化，虽然每个计算资源产生的数据落盘了，但由于它可能会被销毁，因此，我们需要在它被销毁之前，把其产生的日志数据搬移到Pods之外，对于这种转瞬即逝的资源所产生的数据，也需要Kafka的能力。

应对数据高峰

日志数据或基于事件的数据很少具有一致的、可预测的大小或流速。考虑一个在周五晚上升级应用程序的场景，您部署的应用程序有一个严重的错误，即信息被过度记录，淹没了您的日志基础设施。在其他多租户用例（例如游戏和电子商务行业）中，这种峰值或数据爆发也相当普遍。在这种情况下使用像 Kafka 这样的消息队列也能来缓冲数据，减缓影响。

数据的多路分发

还有一种情况，是我们可能需要Kafka将数据分发到别的地方，我们可以定义多个消费端，分别去消费Kafka里的数据，将数据分发到不同的数据处理系统。注意，不要图省事，只建立Logstash作为消费端，然后尝试用Logstash的多个Output去分发。因为Logstash并不能保证过个Output之间的数据同步与一致，我相信不仅是Logstash，其他的消费端也无法做此保证，因此需要多个消费端分别消费Kafka里的数据。当然，这里不是只有kafka一个选择，也可以数据入湖，再从湖中消费。

因此，总结一下，如果你的数据因为没有快速落盘，而存在丢失的风险；如果你的数据吞吐很大，需要及时转移，以及需要应对可能出现的数据洪峰，则你需要Kafka。

我们在什么情况下，可以不需要Kafka

先总结一下我们在什么时候必须要用Kafka:

我们稍微回想一下，在通常的日志场景下我们是如何采集日志的：

    filebeat.autodiscover:
      providers:
        - type: kubernetes
          node: ${NODE_NAME}
          hints.enabled: true
          templates:
            - condition:
                contains:
                  kubernetes.container.name: "opbeans-"
              config:
                - type: container
                  paths:
                    - "/var/log/containers/*-${data.kubernetes.container.id}.log"
                  include_lines: ['^{']
                  tail_files: true

我们需要指定特定的日志目录，也就意味着，我们是从磁盘上采集日志数据的。而在云环境上，如果我们采用了云盘作为数据盘，这些日志数据在产生端，即已经实现了冗余的落盘存储。因此，相对于物联网数据，指标数据，apm数据这种ephemeral数据，已经落盘的日志数据大多数时候，并不需要额外用一个Queue再去做一次数据的落盘与冗余存储。

但在之前的分析中，我们也说过，在某些ephemeral环境下，如果存储数据的本地文件缓存，在我们将数据搬移出去之前就被销毁的话，就会有数据丢失的风险。但幸运的是，大多数情况下，我们是能够轻松判断出这种风险的，大多数非动态扩缩容的应用，比如那些部署在虚拟机上的应用，这种风险的可能性极小，并且可以控制。相对于快速将数据搬出到kafka，增加本地文件存储的大小，反而是一个更简单，更低成本的选择。

并且，当我们发现数据没有正确的写入日志分析系统时，重新让filebeat采集一次日志，会比让Logstash去调整offset，在Kafka上重新consume来得简单得多，直观得多，你不需要了解被采集的文件和kafka topic、partition之间的映射关系，你不需要知道文件与offset的映射关系，以及如何操作re-consume。而且通过Beats与Elasticsearch之间的端到端的应答确认机制和背压探测机制，我们更容易保证数据的正确写入。

因此，总结一下，如果你的数据是日志，这些日志被写入到磁盘，并且不会在短期内被删除。我们可以选择不用Kafka

总结

为了防止大家漏掉关键的点，在最后再强调三遍：

• 日志系统切勿与业务搜索系统混用一个ES集群
• 日志系统切勿与业务搜索系统混用一个ES集群
• 日志系统切勿与业务搜索系统混用一个ES集群

这是我看过的最悲剧的日志分析系统的架构，没有之一。

回到我们本文内容，一个常见的问题是，我们常常忽略架构设计的重要性，懒于去思考其中变量的影响，希望通过一刀切的简单方案去做事情（一上来就是BKLEK架构，不去思考这样做是否值得，我们是否还有更多的选择）。但即便是我们全部使用了Beats, Kafka, Logstash, Elasticsearch, Kibana五个组件，也不是所有的数据链路都走一样的路径，我们可以针对接入系统的特性，优先级，重要程度，要求的指标，有所取舍。

因此，虽然选择Elastic Stack是一个一定不会错的选择，但如果你觉得这套系统复杂了，成本高了，那希望本文可能给你帮助，或许在只有日志数据的这个阶段，你的系统还是可以简化的，有更多选择的。

附录

1, filebeat与Elasticsearch之间的背压检测协议

Filebeat 在将数据发送到 Logstash 或 Elasticsearch 时使用背压敏感协议来处理大量数据。如果 Logstash 忙于处理数据，它会让 Filebeat 知道减慢其读取速度。一旦拥塞得到解决，Filebeat 将恢复到原来的速度并继续发送数据。

关于细节，可以查看以下配置：

  # The number of seconds to wait before trying to reconnect to Elasticsearch
  # after a network error. After waiting backoff.init seconds, the Beat
  # tries to reconnect. If the attempt fails, the backoff timer is increased
  # exponentially up to backoff.max. After a successful connection, the backoff
  # timer is reset. The default is 1s.
  backoff.init: 1s

  # The maximum number of seconds to wait before attempting to connect to
  # Elasticsearch after a network error. The default is 60s.
  backoff.max: 60s

与Libbeat的实现（https://github.com/elastic/beats/下）：

/libbeat/outputs/elasticsearch/elasticsearch.go (elasticsearch output会使用backoff初始化client)

/libbeat/outputs/elasticsearch/client.go#L204 (client会判断接口调用是否成功，如果失败，会返回err)

/libbeat/outputs/backoff.go#L38(判断接口的返回，如果有err，等待）

  func (b *backoffClient) Publish(ctx context.Context, batch publisher.Batch) error {
	err := b.client.Publish(ctx, batch)
	if err != nil {
		b.client.Close()
	}
	backoff.WaitOnError(b.backoff, err)
	return err
}