日志收集是一个很普遍的需求，各个服务的log日志，打点日志都需要收集起来做离线etl或实时分析。日志收集工具也有很多开源的可供选择，flume,  logstash, filebeat等等。  目前360商业化最主要的日志收集场景是投放引擎端的打点日志收集，中间件服务日志收集，这两种场景都是日志先持久化在本地，然后通过日志收集工具收集。

技术选型

投放引擎对资源占用比较敏感， 所以在收集工具的选择上我们对比下来filebeat相比于基于jvm的工具在资源占用上优势还是很明显的，而且支持的一些特性也是能很好满足我们要求的。如果需要日志汇聚层我们选用的是logstash, 因为与filebeat都是elastic家的产品，集成起来方便。

fliebeat版本选用的是目前stable版本7.14，支持kafka sasl scram授权认证方式， 老的版本不一定支持。logstash版本选用的是目前stable版本7.14。

当前我们主要使用的是filebeat的log input,  kafka output,  得益于filebeat的可扩展性，其他input, output也能很好支持。

filebeat简介

filebeat底层主要依赖libbeat库来完成日志收集，libbeat提供publisher来对接inputs, 进入到libbeat后会经过processors的一系列加工处理。 libbeat封装了retry与ack反馈持久化机制，可以保证日志收集的完整性。

要用好filebeat涉及到很多配置，以下列出比较重要的:

  • processors配置: 可以对event做很多加工， 比如add_fields,add_labels, add_tags, rename, drop_event等等

  • Inputs配置: 定义各种input配置。

  • Moduels: 对各种开源组件的日志收集支持，这些modules针对各个组件的日志做了parse与结构化处理，同时集成了ES, Kibana, 为这些开源组件提供一整套日志收集展示方案。

  • HTTP endpoint: 配置http rest api获取metrics。

接下来介绍filebeat两个比较重要的机制，收集meta持久化机制和反压机制， 正是通过这两个机制filebeat才能保证日志收集的at least once语义。


01

filebeat收集meta持久化机制

这个机制很重要，因为这个涉及到日志收集的完整性保障，日志重放等问题。在filebeat run之后，在其根目录下会有data/registry/filebeat/log.json 文件,  这其中记录了filebeat收集meta持久化信息：

{"op":"set","id":1}
{"k":"filebeat::logs::native::3162611-64768","v":{"id":"native::3162611-64768","prev_id":"","source":"/root/ysptest1.log","offset":0,"ttl":-1,"type":"log","timestamp":[2061828726851,1628151202],"FileStateOS":{"inode":3162611,"device":64768},"identifier_name":"native"}}
{"op":"set","id":2}
{"k":"filebeat::logs::native::3162611-64768","v":{"id":"native::3162611-64768","prev_id":"","timestamp":[2061975276931,1628151203],"ttl":-1,"FileStateOS":{"inode":3162611,"device":64768},"identifier_name":"native","source":"/root/ysptest1.log","offset":165,"type":"log"}}其中记录了input type, source端文件路径以及当前收集到的offset，时间戳， 还有用于唯一标识文件的inode, device信息。ack确认反馈会从output端一直反馈到input端，input端收到ack确认后会负责将meta信息持久化到这个文件中。

通过收集meta持久化机制我们能清楚得知道哪些文件收集完成了，哪些文件正在收集以及正在收集文件的offset， 在必要的情况下，通过修改这个持久化文件可以做到日志重新收集的目的， 或者干脆将data文件夹删除，那么所有本地日志都会重新收集。

02

filebeat反压机制

如果outputs一直发送失败，会有retry机制尝试，然后日志在internel queue堆积，internel queue满了inputs端就会停止收集，queue中日志的ack反馈也会迟迟得不到确认， 文件的offset信息也不会更新.  这样就达到了一个反压的效果。

举个例子如果下游kafka集群不可用， 那么outputs就会失败，慢慢internel queue就会full，inputs收集也会停止。这正是我们想要的效果， 在下游出问题的时候，文件保持在本地磁盘是安全的。

通过filebeat的收集meta持久化机制以及反压机制就可以保证在极端情况下的at least once语义。

日志收集方案


设计目标：

  • 提供统一运维监控管理，降低运维成本。

  • 收集的日志可以同时满足实时，离线需求。

  • 日志收集pipeline支持反压，支持at least once语义，支持日志重放。

  • 跨IDC容灾，支持动态修改agent配置，将日志收集定向到其他IDC。


设计方案：

  • 提供一键agent部署脚本，提供agentManager管理metrcis上报与agent配置修改动态感知

  • 将kafka作为统一日志收集目的地。

  • logstash 作为日志汇聚层， 避免agent过多对kafka带来压力。

  • 监控告警: 获取filebeat，logstash metrics并解析成prom格式上报给prometheus, 通过grafana展示，提供实例级别的监控，对收集延迟，失败及时告警， 对收集的日志count进行统计，方便对数。

  • 产品化集成到ultron平台，基于项目粒度进行统一管理，运维，监控。

以下是日志收集的总体架构示意图:


对于日志收集处理可以分为如下四个层次:

#Q1 日志收集agent层：

我们选用filebeat作为日志收集agent，同时会有一个伴生的agentManager来对agent做管理，抽取metrics并发送到pushgateway,  对配置修改动态感知

#Q2 日志汇聚层:

这是个可选层，分以下两种情况:

  • 对于agent不是很多的case， 直接采用filebeat + kafka的方案是很高效的

  • 对于agent很多的case，我们提供logstash汇聚层来对收集数据汇总然后发送给kafka, 避免对kafka造成连接过多的问题

#Q3 DataBus层:

我们将日志采集统一到kafka, 这样离线实时需求都可以得到满足, 我们在每个IDC都有kafka集群，这样当某个IDC不可用时，动态修改filebeat配置即可完成重新收集，相当于具备了跨IDC容灾的能力。

#Q4 异构传输层:

这一层主要是对收集日志的处理使用， 我们当前通过自研的hamal2落地hdfs/hive为离线etl提供支持， 通过flink/spark/storm/druid/clickhouse等实时消费处理。hamal2是我们基于flink实现的一个异构数据同步传输框架，用于Kafka数据实时入仓入湖。

下面主要介绍两种方案:

  • filebeat —> kafka 直连， 这是目前主要在用的方案。

  • filebeat —> logstash —> kafka  这个方案加入了logstash汇聚层。


filebeat —> kafka 

对于agent不是很多的场景， 直接使用filebeat kafka output写入kafka是高效简洁的方式， 根据我们上面阐述的filebeat收集meta持久化机制和反压机制，在kafka有问题写入不成功的情况下，会触发filebeat反压， 日志收集文件的offset也将停止持久化，这样是符合我们预期的。目前360商业化没有有很多agent的场景， 所以主要使用这种模式。

下面是filebeat-->kafka的简单架构示意图:


filebeat.yml 配置示例:

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /root/test*.log
  fields:
    topic_name: test
- type: log
  enabled: true
  paths:
    - /root/test2*.log
  fields:
    topic_name: test2


output.kafka:
  version: 2.0.0
  hosts: ["xxx:9092", "xxx:9092", "xxx:9092"]
  # message topic selection + partitioning
  topic: '%{[fields.topic_name]}'
  partition.round_robin:
    reachable_only: false


  required_acks: 1
  compression: lz4
  max_message_bytes: 10000000
  sasl.mechanism: SCRAM-SHA-256
  username: xxx
  password: xxx
  worker: 1   # producer实例数
  refresh_frequency: 5m
  codec.format:
    string: '%{[message]}'   # 定义输出的日志，默认是带有很多meta信息的json string


filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false


# 定义http endpoint, 用于获取metrics
http.enabled: true
http.host: your host
http.port: 5066

简单说明下上面的配置， 这个配置是filebeat直接使用kafka output写入kafka,  filebeat.inputs部分定义了两个log type收集目录，并且通过fields添加了对应的topic名， 在output.kafka中通过topic: '%{[fields.topic_name]}’ 动态定义了topic.

还有比较重要的配置是codec.format，可以重新定义输出日志的格式。

filebeat-->logstash-->kafka 

对于agent有很多的场景， 我们需要加入logstash汇聚层,  因为这可以避免大量filebeat直连kafka，对kafka连接负载造成压力。至于为什么选择logstash, 因为filebeat与logstash都是elastic家的产品，集成起来很方便， filebeat直接有logstash的output.  需要注意的是加入汇聚层对日志的完整性，稳定性保障都会带来挑战。

为了保障日志的完整性，我们依赖logstash的persistent queues(PQ)机制与反压机制：

  • persistent queues(PQ): 默认是不开启的，日志会先写内存queue再output出去，但这种方式在异常情况下会丢数据， 为了保证日志完整性，我们必须开启PQ，开启后所有日志将先持久化到disk然后再output出去， 这样可以做到at least once语义，可以通过配置queue.type来开启PQ：

queue.type: persisted
queue.max_bytes: 8gb
queue.max_events: 3000000
path.queue: /path/to/queuefile

  • 反压机制:  在PQ写满的情况下会反压到上游filebeat,  filebeat再反压到input停止日志收集。这种反压传导机制和flink的有点像。

通过logstash的PQ机制以及反压机制就可以保证在极端情况下整个pipeline的at least once语义。

下面是filebeat-->logstash-->kafka的简单架构示意图:


为了方便logstash的运维管理， 我们将其部署在k8s集群， 由于logstash向外暴露的是tcp端口，我们通过nginx ingress + vip实现4层lvs。这样就可以根据其负载动态扩缩容了，而且可以提供一个统一的域名入口。

filebeat.yml 示例:

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /root/test*.log
  fields:
    topic_name: test
    kafka_cluster: cluster1
- type: log
  enabled: true
  paths:
    - /root/test2*.log
  fields:
    topic_name: test2
    kafka_cluster: cluster2


output.logstash:
  hosts: ["logstash.k8s.domain:5044"]


# 定义http endpoint, 用于获取metrics
http.enabled: true
http.host: your host
http.port: 5066

简单说明下上面的配置， 这个配置是filebeat使用logstash output写入logstash,  filebeat.inputs部分定义了两个log type收集目录，并且通过fields添加了对应的topic_name, kafka_cluster, 其中增加的kafka_cluster是logstash将日志分拣到不同集群使用的。

output.logstash中定义的logstash.k8s.domain:5044 其实是一个lvs域名端口，其后对应了多个logstash实例。这里我们没有使用filebeat的loadbalance设置，因为不是很灵活。

logstash.yml示例

queue.type: persisted
queue.max_bytes: 8gb
queue.max_events: 3000000
path.queue: /path/to/queuefile

logstash-pipeline.conf 示例

input {
    beats {
        port => "5044"
    }
}


filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}


output {
    stdout { codec => rubydebug {metadata => true}}
    if [fields][kafka_cluster] == "xxx" {  # 日志分拣
        kafka {
            codec => plain {
                format => '{ message:"%{message}"}'
            }


            bootstrap_servers => "xxx:9092,xxx:9092,xxx:9092"
            topic_id => "%{[fields][topic_name]}"
            compression_type => "lz4"
        }
    }


    if [fields][kafka_cluster] == "xxx" {   # 日志分拣
        kafka {
            codec => plain {
                format => '{ message:"%{message}"}'
            }
            bootstrap_servers => "xxx:9092,xxx:9092,xxx:9092"
            topic_id => "%{[fields][topic_name]}"
            jaas_path => "/root/logstash/kafka-jaas.conf"    # 支持SASL SCRAM
            sasl_mechanism => "SCRAM-SHA-256"
            security_protocol => "SASL_PLAINTEXT"
            compression_type => "lz4"
        }
    }
}

简单说明下上面的配置， 通过filter plugin过滤出日志， output kafka plugin中通过filebeat传过来的fields分拣出日志发往不同的kafka集群， 其中codec可以定义要输出到kafka的日志格式。


监控

filebeat的Monitor是要收费的， 免费的filebeat版本只能从http endpoint来获取metrics，需要自己解析后写入pushgateway， 然后通过prometheus --> grafana展示出来。logstash的Monitor也是收费的，类似filebeat，提供rest api来获取metrics,  需要自己解析后写入pushgateway， 然后通过prometheus --> grafana展示出来。

以下是filebeat监控截图， 主要就是将http endpoint中的metrics都展示出来:



总结

基于filebeat的agent非常轻量级，依赖少，资源占用少，支持丰富的inputs, modules, outputs, 非常易于扩展。  logstash作为汇聚层部署在k8s集群， 通过4层lvs提供统一域名端口， 实现实例弹性扩展。  filebeat， logstash都具有持久化与反压机制， 都支持at least once语义， 从而保障了日志收集的完整性。

参考资料

https://www.elastic.co/guide/en/beats/filebeat/current/index.html

https://www.elastic.co/guide/en/logstash/current/index.html

https://cloud.tencent.com/developer/article/1634020

# 运维 # kafka # 数据库 # 大数据 # flume
Logo
华为云开发者联盟

为开发者提供学习成长、分享交流、生态实践、资源工具等服务，帮助开发者快速成长。

更多推荐

cover

HAS2024：华为云以系统性创新加速千行万业智能化升级

avatar 华为云开发者联盟
cover

数仓的两种轻量级数据交换格式：json与jsonb

avatar 华为云开发者联盟
cover

如何基于香橙派AIpro对视频/图像数据进行预处理？

avatar 华为云开发者联盟