欢迎光临
我的个人博客网站

filebeat+ELK配置及常用操作


背景介绍

最近工作涉及几台新服务器的日志需要接入ELK系统,配置思路如下:

使用Filebeat收集本地日志数据,Filebeat监视日志目录或特定的日志文件,再发送到消息队列到kafka,然后logstash去获取消费,利用filter功能过滤分析,最终存储到elasticsearch中。

filebeat和flume都具有日志收集功能,不过filebeat更轻量,使用go语言编写占用资源更少,可以有很高的并发,带有内部模块(auditd,Apache,Nginx,System和MySQL),可通过一个指定命令来简化通用日志格式的收集,解析和可视化;flume使用java开发,需要安装java环境,相对会比较重。

当然两者也存在区别:Filebeat收集数据的速度大于写入速度的时候可能出现数据丢失的现象,而flume会在收集数据和写入数据之间做出调整,保证能在两者之间提供一种平稳的数据状态。可以实时的将分析数据并将数据保存在数据库或者其他系统中,不会出现数据丢失的现象。

以下仅记录配置过程及常见的几种排错命令,安装篇会独立一篇做详细介绍。

配置信息

filebeat配置

我是直接yum install filebeat一键安装的,这里不做具体讲解官网有详细介绍:

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

安装完成后我们以配置采集/var/log/messages为例,配置如下

# egrep -v '#|^$' /etc/filebeat/filebeat.yml filebeat.inputs: - type: log   enabled: true   paths:     - /var/log/messages filebeat.config.modules:   path: ${path.config}/modules.d/*.yml   reload.enabled: false setup.template.settings:   index.number_of_shards: 1 setup.kibana: output.kafka:   hosts: ["10.114.102.30:9092", "10.114.102.31:9092", "10.114.102.32:9092", "10.114.102.33:9092", "10.114.102.34:9092"]   topic: T621_messages   partition.round_robin:     reachable_only: false   required_acks: 1   compression: gzip processors:   - add_host_metadata:       when.not.contains.tags: forwarded   - add_cloud_metadata: ~   - add_docker_metadata: ~   - add_kubernetes_metadata: ~ 

有几个参数需要注意的:

paths表示需要提取的日志的路径,将日志输出到kafka中,创建topic

  • required_acks

0:这意味着生产者producer不等待来自broker同步完成的确认继续发送下一条(批)消息。此选项提供最低的延迟但最弱的耐久性保证(当服务器发生故障时某些数据会丢失,如leader已死,但producer并不知情,发出去的信息broker就收不到)。

1:这意味着producer在leader已成功收到的数据并得到确认后发送下一条message。此选项提供了更好的耐久性为客户等待服务器确认请求成功(被写入死亡leader但尚未复制将失去了唯一的消息)。

-1:这意味着producer在follower副本确认接收到数据后才算一次发送完成。 此选项提供最好的耐久性,我们保证没有信息将丢失,只要至少一个同步副本保持存活。 三种机制,性能依次递减 (producer吞吐量降低),数据健壮性则依次递增。

  • json.keys_under_root: true
  • json.add_error_key: true
  • json.message_key: log

这三行是识别json格式日志的配置,若日志格式不为json格式,需要注释掉,否则收集到的日志为filebeat的错误日志。

kafka配置

kafka原来已经安装并配置好了,这里不再说明具体安装过程,后续会出一篇ELK完整搭建过程。

这里不做重点讲解,可直接查官网:https://kafka.apache.org/documentation/#quickstart

因为有5台配合zookeeper做了集群,选其中一台配置如下:

# egrep -v '#|^$' /home/kafka/kafka/config/server.properties broker.id=1		#按顺序写,不要乱 listeners=PLAINTEXT://0.0.0.0:9092		 #自己的ip advertised.listeners=PLAINTEXT://10.114.102.30:9092 num.network.threads=24 num.io.threads=8 socket.send.buffer.bytes=102400 socket.receive.buffer.bytes=102400 socket.request.max.bytes=104857600 log.dirs=/data/kafka num.partitions=8 num.recovery.threads.per.data.dir=1 offsets.topic.replication.factor=1 transaction.state.log.replication.factor=1 transaction.state.log.min.isr=1 log.retention.hours=48 log.segment.bytes=1073741824 log.retention.check.interval.ms=300000 zookeeper.connect=10.114.102.30:2181,10.114.102.31:2181,10.114.102.32:2181,10.114.102.33:2181,10.114.102.34:2181 zookeeper.connection.timeout.ms=6000 auto.create.topics.enable=false group.initial.rebalance.delay.ms=0 

注意:每台服务器除broker.id需要修改之外,其他属性保持一致。

logstash配置

logstash安装也是直接参考官网就可以了

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

不过有个地方要注意,kafka和logstash的版本兼容问题,以下是kafka使用的版本:

find /home/kafka/kafka/libs/ -name *kafka_* | head -1 | grep -o 'kafka[^n]*'    kafka_2.11-1.1.0.jar 

通过查找rpm包可以看到logstash用的是7.8.0

/home/cxhchusr/logstash-7.8.0.rpm 

conf.d目录下配置消费messages的文件如下

# egrep -v '#|^$' /etc/logstash/conf.d/T621_messages.conf  input {   kafka {     bootstrap_servers => "10.114.102.30:9092,10.114.102.31:9092,10.114.102.32:9092,10.114.102.33:9092,10.114.102.34:9092"     client_id => "T621_messages"     group_id => "T621_messages"     auto_offset_reset => "latest"     consumer_threads => 10     decorate_events => true     topics => ["T621_messages"]     decorate_events => true     type => syslog     } } filter{   grok {     match => { "message" => "%{SYSLOGLINE}" }   }   date {     match => [ "logdate", "YYYY-MM-dd HH:mm:ss.SSS" ]     target => "@timestamp"     timezone =>"+00:00"   }   mutate{     remove_field => "logdate"   } } output {   elasticsearch {     hosts => ["10.114.102.30:9200", "10.114.102.31:9200", "10.114.102.32:9200", "10.114.102.33:9200", "10.114.102.34:9200"]     index => "t621_messages-%{+YYYY.MM.dd}"     user => caixun     password => "******()90"     } }  

注意:es索引需要全部为小写。

最后启动即可,并加入开机自启动/etc/rc.local

nohup /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/ > /dev/null 2>&1 & 

kibana创建索引

logstash配置完成后即可在kibana创建索引

filebeat+ELK配置及常用操作

创建完成效果如下,表示接入成功:

filebeat+ELK配置及常用操作

常用排查命令

配置过程中除了在kafka上创建topic,还需查询topic是否创建成功、消费情况、以及消息处理情况。

以及es是否正常入库并创建了索引。下面列出几个ELK运维常用命令。

kafka常用运维指令

  • 查询当前topic列表

/home/kafka/kafka/bin/kafka-topics.sh –zookeeper 127.0.0.1:2181 –list

  • 创建topic

/home/kafka/kafka/bin/kafka-topics.sh –zookeeper 127.0.0.1:2181 –create –partitions 10 –replication-factor 1 –topic T621_messages

  • topic描述(某topic详细信息)

/home/kafka/kafka/bin/kafka-topics.sh –zookeeper 127.0.0.1:2181 –describe –topic T621_messages

  • topic消费情况(测试消息是否正常生产)

发送:/home/kafka/kafka/bin/kafka-console-producer.sh –broker-list 127.0.0.1:9092 –topic T621_messages

接收:/home/kafka/kafka/bin/kafka-console-consumer.sh –bootstrap-server 127.0.0.1:9092 –topic T621_messages –from-beginning

  • 查看topic堆积情况

/home/kafka/kafka/bin/kafka-consumer-groups.sh –bootstrap-server 127.0.0.1:9092 –list ##查看组列表

/home/kafka/kafka/bin/kafka-consumer-groups.sh –bootstrap-server 127.0.0.1:9092 –describe –group T621_messages ##偏移量

/data/kafka/kafka/bin/kafka-console-consumer.sh –bootstrap-server 127.0.0.1:9092 –topic T621_messages –from-beginning

ES常用运维指令

  • 查看ES集群健康情况

curl -u caixxx:”CAIxxx()90″ ‘10.114.102.30:9200/_cluster/health’

  • 查看索引存储情况

curl -u caixxx:”CAIxxx()90″ ‘10.114.102.30:9200/_cat/indices?v’

  • 查看帮助命令
# curl -u caixun:"CAIXUN()90" '10.114.102.30:9200/_cat' =^.^= /_cat/allocation /_cat/shards /_cat/shards/{index} /_cat/master /_cat/nodes /_cat/tasks /_cat/indices /_cat/indices/{index} /_cat/segments /_cat/segments/{index} /_cat/count /_cat/count/{index} /_cat/recovery /_cat/recovery/{index} /_cat/health /_cat/pending_tasks /_cat/aliases /_cat/aliases/{alias} /_cat/thread_pool /_cat/thread_pool/{thread_pools} /_cat/plugins /_cat/fielddata /_cat/fielddata/{fields} /_cat/nodeattrs /_cat/repositories /_cat/snapshots/{repository} /_cat/templates /_cat/transforms /_cat/transforms/{transform_id} 

过滤索引查看消息是否成功存储在es,有的话代表配置成功。

—- 钢铁 648403020@qq.com 2021.08.20

参考鸣谢

官方kafka:http://kafka.apache.org/

赞(0) 打赏
未经允许不得转载:张拓的天空 » filebeat+ELK配置及常用操作
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

专业的IT技术经验分享 更专业 更方便

联系我们本站主机

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏