日志从 Kafka 到 Loki 的 N 种方式​

最近群里有小伙伴有说到自己的日志存储路径先是从客户端到Kafka，再通过消费kafka到ElasticSearch。现在要将ES换成Loki面临需要同时支持Kafka和Loki插件的工具。小白查了下当前市面上满足需求且足够可靠的工具分别为Fluentd、Logstash以及Vector。

• Fluentd

CNCF已毕业的云原生日志采集客户端。与kubernetes结合比较紧密，插件丰富且有大厂背书。不足之处在于受ruby限制，在日志量大（建议使用FluentBit）时本身的资源消耗不小。

• Logstash

ELK栈中老牌的日志采集和聚合工具，使用广泛且插件丰富，不足之处在于资源消耗整体比较高，单机日志并发处理效率不高。

• Vector

刚开源不久的轻量级日志客户端，产品集成度比较高，资源消耗极低。不足之处就是当下产品似乎没有还没有广泛的最佳实践。

官方性能报告 https://vector.dev/#performance

以下是vector分别对上述产品做的一个性能测试对比，大家可以参考下：

那么接下来进入主题吧，当我们需要将Kafka里的日志存进Loki时，我们有哪些方法实现，先来看个简单的?

Vector

Vector内部已经集成好了kafka和loki方法，我们只需下载vector和配置就能直接用起来。

安装vector

curl --proto '=https' --tlsv1.2 -sSf https://sh.vector.dev | sh

或者你可以直接使用docker镜像

docker pull timberio/vector:0.10.0-alpine

vector配置

[sources.in]
  bootstrap_servers = ""
  group_id = "<消费组id>"
  topics = ["^(prefix1|prefix2)-.+", "topic-1", "topic-2"]  \\topic名字，支持正则
  type = "kafka"

[sinks.out]
  endpoint = "http://" 
  inputs = ["in"]      \\source.in 
  type = "loki"
  labels.key = "value"  \\自定义key
  labels.key = "{{ event_field }}" \\event的动态值

关于vector-loki更多参数可以参考:https://vector.dev/docs/reference/sinks/loki/

Fluentd

Input - fluent-plugin-kafka

fluent-plugin-kafka插件是fluent的官方处理kafka的插件，可同时用于input和output两个阶段。它的安装方式如下：

gem install fluent-plugin-kafka

当它用于input阶段时，这时fluentd就会作为一个kafka的消费者，从指定的topic中取出消息并做相关处理，它的配置如下：

  @type kafka

  brokers  borker地址>
  topics 
  format <日志处理格式，默认是json，支持text|json|ltsv|msgpack>
  message_key <日志key，仅用于text格式的日志>
  add_prefix <添加fluentd的tag前缀>
  add_suffix <添加fluentd的tag后缀>

如果你想指定从不同topic的偏移量开始消费消息的话，就需要如下配置:

  @type kafka

  brokers   borker地址>
  format   <日志处理格式，默认是json，支持text|json|ltsv|msgpack>
  
    topic     <单个topic名>
    partition 
    offset    <从offset开始>
  
  
    topic     <单个topic名>
    partition 
    offset    <从offset开始>
  

熟悉fluentd的同学可能知道，在fluentd中是以tag名来处理pipeline的，默认情况下kafka插件会用topic名来做你tag名，如果你想做一些全局的filter可以添加tag前缀/后缀来全局匹配实现。

Output - fluent-plugin-grafana-loki

fluent-plugin-grafana-loki是grafana lab贡献的一个从fluentd发送日志到loki的插件。之前小白在《Loki和Fluentd的那点事儿》里介绍过，这里不过多展开。

配置直接从以前的文章中copy过来，主要的区别在于tag的匹配，参考如下：

 $kafka.topic>  \\此处为kafka的topic
  @type loki
  @id loki.output
  url "http://loki:3100"
  
    key1          \\如果你的日志json格式，那么你可以将需要提取
    key2          \\的字段作为你的loki labels
  
   label>
    @type file
    path /var/log/fluentd-buffers/loki.buffer
    flush_mode interval
    flush_thread_count 4
    flush_interval 3s
    retry_type exponential_backoff
    retry_wait 2s
    retry_max_interval 60s
    retry_timeout 12h
    chunk_limit_size 8M
    total_limit_size 5G
    queued_chunks_limit_size 64
    overflow_action drop_oldest_chunk
  

Logstash

Input - logstash-input-kafka

logstash-input-kafka是elastic官方提供的kafka消费端插件，对于input阶段的配置也比较简单。

input {
    kafka {
        bootstrap_servers => ""
        topics => ""
        codec => "<日志类型，默认plain>"
        tags => ""
        }
    }

更多的参数参考https://www.elastic.co/guide/en/logstash/7.10/plugins-inputs-kafka.html

Output - logstash-output-loki

logstash-output-loki也是由grafana lab贡献的用于处理往loki输出的logstash插件。安装时直接执行下列命令即可：

logstash-plugin install logstash-output-loki

logstash输出到loki的参数不多，小白捡主要的几个说明如下：

output {
  loki {
    url => "http:///loki/api/v1/push"
    batch_size => 112640 #单次push的日志包大小
    retries => 5
    min_delay => 3
    max_delay => 500
    message_field => "<日志消息行的key，改key由logstas传递过来，默认为message>"
  }
}

总结

以上三个工具均没有做filter和解析，仅仅只是充当管道将日志从kafka里转存到loki，实际环境可能比较复杂，需要对日志做进一步分析。不过从小白的体验来看vector对于日志从kafka到loki的配置算是比较简单直接，fluentd和logstash整体差不多，就看大家自己的顺手程度了。