总体思路

还是之前的场景，我需要把各个业务的线上服务器日志接入到统一的日志处理平台中。具体会用 Kafka 做中间件，所以需要解决的就是如何把日志传到 Kafka。原先的考虑是利用系统自带的 rsyslog，这样我只需要自动配置一下 rsyslog 的处理发送规则就可以了，免去了安装和维护的麻烦。但是系统自带的 rsyslog 版本太低，所以到头来还是要更新维护，那就不如直接用更强大且更好用的 Logstash 了。

需要注意的有两点：

  • 不要即时推送日志，以免增加服务器负担
  • 能够妥善处理 logrotate 的情况

幸运的是，这 Logstash 都考虑到了，我们只需要简单配置一下即可。

安装 Java

因为大部分线上服务器跑的是 Ruby，所以需要先安装一下 Java

 

sudo add-apt-repository -y ppa:webupd8team/java

sudo apt-get update

sudo apt-get -y install oracle-java8-installer

安装 Logstash

在 ELK 指南中介绍了用 apt-get 进行安装的方法，这里介绍如何手动下载安装

  • 下载到服务器 wget https://download.elastic.co/logstash/logstash/logstash-all-plugins-2.3.4.tar.gz
  • 解压 tar -xvzf logstash-all-plugins-2.3.4.tar.gz
  • 进入 Logstash 文件夹并创建配置文件夹（个人习惯） cd logstash-2.3.4; mkdir confs

之后所有的配置文件均可放在 confs 文件夹中。

配置 Logstash 到 Kafka

这里 Logstash 相当于 producer

Input 读取文件

Logstash 使用一个名叫 FileWatch 的 Ruby Gem 库来监听文件变化。这个库支持 glob 展开文件路径，而且会记录一个叫 .sincedb 的数据库文件来跟踪被监听的日志文件的当前读取位置。通过记录下来的 inode, major number, minor number 和 pos 就可以保证不漏过每一条日志。一个可能的配置文件是这样的

 

input {

file {

path => ["/data/home/service/project/current/log/logstash_production.log", "/data/home/service/project/current/log/logstash_production.log.1"]

codec => "json"

add_field => { "topic" => "djiservice"}

stat_interval => 1800

}

}

这里说一下 File rotation 的情况，为了处理被 rotate 的情况，最好把 rotate 之后的文件名也加到 path 中（如上面所示），这里注意，如果 start_position 被设为 beginning，被 rotate 的文件因为会被认为是新文件，而重新导入。如果用默认值 end，那么在最后一次读之后到被 rotate 结束前生成的日志不会被采集。

其他一些配置的设定原因

  • add_field 添加一个 topic 字段，用作之后导入 elasticsearch 的索引标识
  • stat_interval 单位是秒，这里 30 分钟进行一次检测，不过测试的时候需要去掉这个配置
  • codec 因为已经处理成 logstash 兼容格式，就直接以 json 解析

Filter 内容定制

Filter 主要是对数据进行一些处理，比如说我用的是：

 

filter {

mutate {

remove_field => ["format"]

}

geoip {

source => "ip"

fields => ["location", "city_name", "country_name", "country_code2","country_code3", "region_name", "continent_code"]

}

}

这里做的操作一是移除无效的域，二是把 ip 转换为地理位置，方便后期的处理。

Output 输出到 Kafka

因为 Logstash 自带 Kafka 插件，直接配置上即可，比如：

 

output {

kafka {

topic_id => "test"

bootstrap_servers => "kafka_url:port"

}

}

下面是基本的配置及其设定原因

  • topic_id 指定 topic 来进行发送
  • bootstrap_servers 这里是 Kafka 的接入地址

其他一些需要注意的配置

  • acks 可以选的值为 0, 1, all，这里解释一下，0 表示不需要 server 返回就认为请求已完成；1 表示需要 leader 返回才认为请求完成；all 表示需要所有的服务器返回才认为请求完成
  • batch_size 单位是字节，如果是发送到同一分区，会攒够这个大小才发送一次请求
  • block_on_buffer_full 这个设置在缓冲区慢了之后阻塞还是直接报错
  • buffer_memory 发送给服务器之前的缓冲区大小，单位是字节
  • client_id 可以在这里设定有意义的名字，就不一定要用 ip 和 端口来区分
  • compression_type 压缩方式，默认是 none，其他可选的是 gzip 和 snappy

利用 Logstash 从 Kafka 导出数据到 Elasticsearch

这一步就比较简单了，先从 Kafka 中读取，然后写入到 elasticsearch，这里 Logstash 作为 consumer

 

output {

input {

kafka {

zk_connect => "localhost:2181"

topic_id => "log"

}

}

stdout {

codec => rubydebug

}

elasticsearch {

hosts => ["url:port"]

user => "name"

password => "password"

index => "%{service}-%{+YYYY-MM-dd}"

}

}

这样可以按照配置的服务名称和日期来切割。

至此，我们完成了从 Logstash 到 Kafka 再到 Elasticsearch 的连接，下一步就可以用 kibana 来展示日志的监控分析结果了。

 

转自：https://wdxtub.com/2016/08/18/logstash-kafka-guide/

Logo
Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一） 概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

avatar Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

avatar Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...

avatar Kafka开源项目指南