• ZooKeeper+Kafka+ELK+Filebeat集群搭建实现大批量日志收集和展示

    大致流程：将nginx 服务器（web-filebeat）的日志通过filebeat收集之后，存储到缓存服务器kafka，之后logstash到kafka服务器上取出相应日志，经过处理后写入到elasticsearch服务器并在kibana上展示。

    一、集群环境准备

    1. 4c/8G/100G   10.10.200.33     Kafka+ZooKeeper+ES+Filebeat+ES-head
    2. 4c/8G/100G   10.10.200.34     Kafka+ZooKeeper+ES+Kibana
    3. 4c/8G/100G   10.10.200.35     Kafka+ZooKeeper+ES+Logstash

    二、搭建zookeeper集群

    前提条件：三台机器分别修改时区、关闭防火墙、安装JAVA环境变量、修改主机名

    1. [root@kf-zk-es-fb_es-head logs]# cat /etc/hosts
    2. 127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
    3. ::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
    4. 10.10.200.33 kf-zk-es-fb_es-head
    5. 10.10.200.34 kf-zk-es-kibana
    6. 10.10.200.35 kf-zk-es-logstash
    7. [root@kf-zk-es-fb_es-head logs]# java -version
    8. openjdk version "1.8.0_382"
    9. OpenJDK Runtime Environment (build 1.8.0_382-b05)
    10. OpenJDK 64-Bit Server VM (build 25.382-b05, mixed mode)
    11. [root@kf-zk-es-fb_es-head logs]# cat /etc/profile
    12. export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.382.b05-1.el7_9.x86_64/
    13. export JRE_HOME=$JAVA_HOME/jre
    14. export CLASSPATH=.:$JAVA_HOME/lib:$JRE_HOME/lib:$CLASSPATH
    15. export PATH=$PATH:$JRE_HOME/bin:$JAVA_HOME/bin

    2.1 安装zookeeper(三台机器同步)

    1. [root@kf-zk-es-fb_es-head ~]# wget http://dlcdn.apache.org/zookeeper/zookeeper-3.8.3/apache-zookeeper-3.8.3-bin.tar.gz
    2. [root@kf-zk-es-fb_es-head ~]# tar -zxvf apache-zookeeper-3.8.3-bin.tar.gz -C /usr/local/
    3. [root@kf-zk-es-fb_es-head ~]# mv /usr/local/apache-zookeeper-3.8.3-bin/  /usr/local/zookeeper-3.8.3
    4. [root@kf-zk-es-fb_es-head ~]# cp /usr/local/zookeeper-3.8.0/conf/zoo_sample.cfg /usr/local/zookeeper-3.8.0/conf/zoo.cfg
    5. 修改配置文件
    6. 本机IP设置成0.0.0.0
    7. 节点一：
    8. [root@kf-zk-es-fb_es-head ~]# vim /usr/local/zookeeper-3.8.3/conf/zoo.cfg
    9. tickTime=2000
    10. initLimit=10
    11. syncLimit=5
    12. dataDir=/usr/local/zookeeper-3.8.3/data
    13. dataLogDir=/usr/local/zookeeper-3.8.3/logs
    14. clientPort=2181
    15. server.1=0.0.0.0:2888:3888
    16. server.2=10.10.200.34:2888:3888
    17. server.3=10.10.200.35:2888:3888
    18. 节点二：
    19. [root@kf-zk-es-kibana ~]# egrep -v '^#|^$' /usr/local/zookeeper-3.8.3/conf/zoo.cfg
    20. tickTime=2000
    21. initLimit=10
    22. syncLimit=5
    23. dataDir=/usr/local/zookeeper-3.8.3/data
    24. dataLogDir=/usr/local/zookeeper-3.8.3/logs
    25. clientPort=2181
    26. server.1=10.10.200.33:2888:3888
    27. server.2=0.0.0.0:2888:3888
    28. server.3=10.10.200.35:2888:3888
    29. 节点三：
    30. [root@kf-zk-es-logstash ~]# egrep -v '^#|^$' /usr/local/zookeeper-3.8.3/conf/zoo.cfg
    31. tickTime=2000
    32. initLimit=10
    33. syncLimit=5
    34. dataDir=/usr/local/zookeeper-3.8.3/data
    35. dataLogDir=/usr/local/zookeeper-3.8.3/logs
    36. clientPort=2181
    37. server.1=10.10.200.33:2888:3888
    38. server.2=10.10.200.34:2888:3888
    39. server.3=0.0.0.0:2888:3888
    40.  
    41. 其中dataDir和dataLogDir需要手动创建，否则启动服务会报错目录完成后新建myid文件：
    42. 节点一：
    43. [root@kf-zk-es-fb_es-head data]# pwd
    44. /usr/local/zookeeper-3.8.3/data
    45. [root@kf-zk-es-fb_es-head data]# cat myid
    46. 1
    47. 节点二：
    48. [root@kf-zk-es-kibana data]# cat myid
    49. 2
    50. 节点三：
    51. [root@kf-zk-es-logstash data]# cat myid
    52. 3
    53. 主要是要跟各自的配置文件的server.1.2.3对应

    2.2 配置zookeeper启动脚本

    1. vim /etc/init.d/zookeeper
    2. #!/bin/bash
    3. #chkconfig:2345 20 90
    4. #description:Zookeeper Service Control Script
    5. ZK_HOME='/usr/local/zookeeper-3.5.7'
    6. case $1 in
    7. start)
    8. echo "---------- zookeeper 启动 ------------"
    9. $ZK_HOME/bin/zkServer.sh start
    10. ;;
    11. stop)
    12. echo "---------- zookeeper 停止 ------------"
    13. $ZK_HOME/bin/zkServer.sh stop
    14. ;;
    15. restart)
    16. echo "---------- zookeeper 重启 ------------"
    17. $ZK_HOME/bin/zkServer.sh restart
    18. ;;
    19. status)
    20. echo "---------- zookeeper 状态 ------------"
    21. $ZK_HOME/bin/zkServer.sh status
    22. ;;
    23. *)
    24. echo "Usage: $0 {start|stop|restart|status}"
    25. esac
    26.  
    27. 设置开机自启：
    28. chmod +x /etc/init.d/zookeeper
    29. chkconfig --add zookeeper

    2.3 启动zookeeper

    1. 节点一：
    2. [root@kf-zk-es-logstash data]# service start zookeeper
    3. [root@kf-zk-es-fb_es-head data]# service zookeeper status
    4. ---------- zookeeper 状态 ------------
    5. /bin/java
    6. ZooKeeper JMX enabled by default
    7. Using config: /usr/local/zookeeper-3.8.3/bin/../conf/zoo.cfg
    8. Client port found: 2181. Client address: localhost. Client SSL: false.
    9. Mode: follower   #从节点
    10.  
    11. 节点二：
    12. [root@kf-zk-es-logstash data]# service start zookeeper
    13. [root@kf-zk-es-kibana data]# service zookeeper status
    14. ---------- zookeeper 状态 ------------
    15. /bin/java
    16. ZooKeeper JMX enabled by default
    17. Using config: /usr/local/zookeeper-3.8.3/bin/../conf/zoo.cfg
    18. Client port found: 2181. Client address: localhost. Client SSL: false.
    19. Mode: leader   #主节点
    20.  
    21. 节点三：
    22. [root@kf-zk-es-logstash data]# service start zookeeper
    23. [root@kf-zk-es-logstash data]# service zookeeper status
    24. ---------- zookeeper 状态 ------------
    25. /bin/java
    26. ZooKeeper JMX enabled by default
    27. Using config: /usr/local/zookeeper-3.8.3/bin/../conf/zoo.cfg
    28. Client port found: 2181. Client address: localhost. Client SSL: false.
    29. Mode: follower   #从节点

    三、部署Kafka

    3.1 安装Kafka

    1. 以下所有步骤三节点都要执行，可下载好包scp过去
    2. [root@kf-zk-es-fb_es-head ~]# wget http://archive.apache.org/dist/kafka/2.8.2/kafka_2.13-2.8.2.tgz
    3. [root@kf-zk-es-fb_es-head ~]# tar xf kafka_2.13-2.8.2.tgz -C /usr/local/
    4. [root@kf-zk-es-fb_es-head ~]# mv kafka_2.13-2.8.2 kafka
    5. [root@kf-zk-es-fb_es-head ~]# cp /usr/local/kafka/config/server.properties /usr/local/kafka/config/server.properties_bak
    6.  
    7. 编辑配置文件：
    8. 节点一：
    9. [root@kf-zk-es-fb_es-head ~]# vim /usr/local/kafka/config/server.properties
    10. broker.id=1
    11. listeners=PLAINTEXT://10.10.200.33:9092
    12. num.network.threads=3
    13. num.io.threads=8
    14. socket.send.buffer.bytes=102400
    15. socket.receive.buffer.bytes=102400
    16. socket.request.max.bytes=104857600
    17. log.dirs=/tmp/kafka-logs
    18. num.partitions=1
    19. num.recovery.threads.per.data.dir=1
    20. offsets.topic.replication.factor=1
    21. transaction.state.log.replication.factor=1
    22. transaction.state.log.min.isr=1
    23. log.retention.hours=168
    24. log.segment.bytes=1073741824
    25. log.retention.check.interval.ms=300000
    26. zookeeper.connect=10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181
    27. zookeeper.connection.timeout.ms=18000
    28. group.initial.rebalance.delay.ms=0
    29.  
    30. 节点二：
    31. [root@kf-zk-es-kibana ~]# egrep -v '^#|^$' /usr/local/kafka/config/server.properties
    32. broker.id=2
    33. listeners=PLAINTEXT://10.10.200.34:9092
    34. num.network.threads=3
    35. num.io.threads=8
    36. socket.send.buffer.bytes=102400
    37. socket.receive.buffer.bytes=102400
    38. socket.request.max.bytes=104857600
    39. log.dirs=/tmp/kafka-logs
    40. num.partitions=1
    41. num.recovery.threads.per.data.dir=1
    42. offsets.topic.replication.factor=1
    43. transaction.state.log.replication.factor=1
    44. transaction.state.log.min.isr=1
    45. log.retention.hours=168
    46. log.segment.bytes=1073741824
    47. log.retention.check.interval.ms=300000
    48. zookeeper.connect=10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181
    49. zookeeper.connection.timeout.ms=18000
    50. group.initial.rebalance.delay.ms=0
    51.  
    52. 节点三：
    53. [root@kf-zk-es-logstash ~]# egrep -v '^#|^$' /usr/local/kafka/config/server.properties
    54. broker.id=3
    55. listeners=PLAINTEXT://10.10.200.35:9092
    56. num.network.threads=3
    57. num.io.threads=8
    58. socket.send.buffer.bytes=102400
    59. socket.receive.buffer.bytes=102400
    60. socket.request.max.bytes=104857600
    61. log.dirs=/tmp/kafka-logs
    62. num.partitions=1
    63. num.recovery.threads.per.data.dir=1
    64. offsets.topic.replication.factor=1
    65. transaction.state.log.replication.factor=1
    66. transaction.state.log.min.isr=1
    67. log.retention.hours=168
    68. log.segment.bytes=1073741824
    69. log.retention.check.interval.ms=300000
    70. zookeeper.connect=10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181
    71. zookeeper.connection.timeout.ms=18000
    72. group.initial.rebalance.delay.ms=0
    73.  
    74. 配置环境变量：
    75. [root@kf-zk-es-logstash ~]# tail -2 /etc/profile
    76. export KAFKA_HOME=/usr/local/kafka
    77. export PATH=$PATH:$KAFKA_HOME/bin
    78. [root@kf-zk-es-logstash ~]# source /etc/profile
    79.

    3.2 启动Kafka

    1. 三节点同时执行
    2. [root@kf-zk-es-logstash ~]# sh /usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
    3. [root@kf-zk-es-logstash ~]# netstat -lntup|grep 9092
    4. tcp6       0      0 10.10.200.35:9092       :::*                    LISTEN      7474/java

    3.3 Kafka常用命令

    1. #查看当前服务器中的所有topic
    2. kafka-topics.sh --list --zookeeper 10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181
    3. #查看某个topic的详情
    4. kafka-topics.sh --describe --zookeeper 10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181
    5. #发布消息
    6. kafka-console-producer.sh --broker-list 10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181 --topic test
    7. #消费消息
    8. kafka-console-consumer.sh --bootstrap-server 10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181 --topic test --from-beginning
    9. --from-beginning 会把主题中以往所有的数据都读取出来
    10. #修改分区数
    11. kafka-topics.sh --zookeeper 10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181 --alter --topic test --partitions 6
    12. #删除topic
    13. kafka-topics.sh --delete --zookeeper 10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181 --topic test

    3.4 Kafka命令创建topic

    1. [root@kf-zk-es-fb_es-head ~]# kafka-topics.sh --create --zookeeper 10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181 --partitions 3 --replication-factor 1 --topic nginx_access
    2. Created topic test.
    3.  
    4. --zookeeper: 定义 zookeeper 集群服务器地址,如果有多个 IP 地址使用逗号分割,一般使用一个 IP 即可
    5. --replication-factor: 定义分区副本数,1 代表单副本,建议为 2
    6. --partitions: 定义分区数
    7. --topic: 定义 topic 名称
    8.  
    9. [root@kf-zk-es-fb_es-head ~]# kafka-topics.sh --create --zookeeper 10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181 --partitions 3 --replication-factor 1 --topic nginx_error
    10.  
    11. 查看topic信息
    12. [root@kf-zk-es-fb_es-head ~]# sh /usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper 10.10.200.33:2181
    13. Topic: nginx_access     PartitionCount: 3       ReplicationFactor: 1    Configs:
    14.         Topic: nginx_access     Partition: 0    Leader: 1       Replicas: 1     Isr: 1
    15.         Topic: nginx_access     Partition: 1    Leader: 2       Replicas: 2     Isr: 2
    16.         Topic: nginx_access     Partition: 2    Leader: 3       Replicas: 3     Isr: 3
    17. Topic: nginx_error      PartitionCount: 3       ReplicationFactor: 1    Configs:
    18.         Topic: nginx_error      Partition: 0    Leader: 2       Replicas: 2     Isr: 2
    19.         Topic: nginx_error      Partition: 1    Leader: 3       Replicas: 3     Isr: 3
    20.         Topic: nginx_error      Partition: 2    Leader: 1       Replicas: 1     Isr: 1
    21.  
    22.

    3.5 测试Kafka-topic

    1. [root@kf-zk-es-fb_es-head ~]# kafka-console-producer.sh --broker-list 10.10.200.33:9092,10.10.200.34:9092,10.10.200.35:9092 --topic nginx_access
    2. >1
    3. >hello
    4. >my
    5. >name
    6. >is
    7. >world
    8.  
    9. [root@kf-zk-es-kibana ~]# kafka-console-consumer.sh --bootstrap-server 10.10.200.33:9092,10.10.200.34:9092,10.10.200.35:9092 --topic nginx_access --from-beginning
    10. 1
    11. hello
    12. my
    13. name
    14. is
    15. world

    3.6 Kafka问题总结

    重启服务需要先杀掉进程。如果重新创建topic时报错，需要删掉/tmp/kafka-logs/meta.properties才能正常启动，另外配置文件中一定要写好zookeeper的连接属性：zookeeper.connect=10.10.200.33:2181,10.10.200.34:2181,10.10.200.35:2181

    4、搭建Elastic search并配置

    三台主机都安装并修改配置文件后启动

    1. 以节点一为例，另外两台机器同步执行：
    2. [root@kf-zk-es-fb_es-head ~]# wget http://dl.elasticsearch.cn/elasticsearch/elasticsearch-7.9.2-x86_64.rpm
    3. [root@kf-zk-es-fb_es-head ~]# rpm -ivh elasticsearch-7.9.2-x86_64.rpm
    4. [root@kf-zk-es-fb_es-head ~]# cp /etc/elasticsearch/elasticsearch.yml /etc/elasticsearch/elasticsearch.yml_bak
    5. [root@kf-zk-es-fb_es-head elasticsearch]# egrep -v '^#|^$' /etc/elasticsearch/elasticsearch.yml
    6. cluster.name: my-application
    7. node.name: node1
    8. path.data: /var/lib/elasticsearch
    9. path.logs: /var/log/elasticsearch
    10. network.host: 0.0.0.0
    11. http.port: 9200
    12. discovery.seed_hosts: ["10.10.200.33","10.10.200.34","10.10.200.35"]
    13. cluster.initial_master_nodes: ["node1","node2","node3"]

    5、启动Elastic search并设置开机自启

    1. 三台机同步执行
    2. [root@kf-zk-es-fb_es-head ~]# systemctl start elasticsearch
    3. [root@kf-zk-es-fb_es-head ~]# systemctl status elasticsearch
    4. ● elasticsearch.service - Elasticsearch
    5.    Loaded: loaded (/usr/lib/systemd/system/elasticsearch.service; disabled; vendor preset: disabled)
    6.    Active: active (running) since Fri 2023-11-10 09:27:14 CST; 8s ago
    7. [root@kf-zk-es-fb_es-head ~]# systemctl enable elasticsearch
    8. Created symlink from /etc/systemd/system/multi-user.target.wants/elasticsearch.service to /usr/lib/systemd/system/elasticsearch.service.

    6、页面访问

    7、部署logstash消费kafka数据写入到ES

    1. 下载地址：http://dl.elasticsearch.cn/logstash/logstash-7.9.2.rpm
    2. [root@kf-zk-es-logstash ~]# rpm -vih logstash-7.9.2.rpm
    3. [root@kf-zk-es-logstash ~]# vim /etc/logstash/conf.d/logstash.conf
    4. input {
    5.        kafka  {
    6.        codec => "json"
    7.        topics => ["nginx_access","nginx_error"]
    8.        bootstrap_servers => "10.10.200.33:9092,10.10.200.34:9092,10.10.200.35:9092"
    9.        max_poll_interval_ms => "3000000"
    10.        session_timeout_ms => "6000"
    11.        heartbeat_interval_ms => "2000"
    12.        auto_offset_reset => "latest"
    13.        group_id => "logstash"
    14.        type => "logs"
    15.      }
    16. }
    17. output {
    18.         elasticsearch {
    19.         hosts => ["http://10.10.200.33:9200","http://10.10.200.34:9200","http://10.10.200.35:9200"]
    20.         index => "%{[fields][log_topics]}-%{+YYYY-MM-dd}"
    21.         }
    22. }
    23.  
    24. [root@kf-zk-es-logstash ~]# cd /usr/share/logstash/
    25. [root@kf-zk-es-logstash logstash]# mkdir config
    26. [root@kf-zk-es-logstash logstash]# cp /etc/logstash/pipelines.yml config/
    27. [root@kf-zk-es-logstash logstash]# ln -s /usr/share/logstash/bin/logstash /usr/local/bin/
    28. [root@kf-zk-es-logstash logstash]# logstash -t
    29. 无报错则启动logstash
    30. [root@kf-zk-es-logstash logstash]# systemctl start logstash && systemctl enable logstash
    31. Created symlink from /etc/systemd/system/multi-user.target.wants/logstash.service to /etc/systemd/system/logstash.service.
    32. 查看日志是否有消费kafka信息
    33. [root@kf-zk-es-logstash logstash]# tail -f /var/log/logstash/logstash-plain.log

    9、部署Filebeat

    1. http://dl.elasticsearch.cn/filebeat/filebeat-7.9.2-x86_64.rpm
    2.  
    3. [root@kf-zk-es-fb_es-head ~]# rpm -vih filebeat-7.9.2-x86_64.rpm
    4. [root@kf-zk-es-fb_es-head ~]# vim /etc/filebeat/filebeat.yml
    5. 配置收集日志信息
    6. filebeat.inputs:
    7. - type: log
    8.   enabled: true
    9.   paths:
    10.   	#收集日志地址
    11.     - /usr/local/logs/access.log #此日志为nginx的路径日志，事先在本机安装了nginx
    12.   fields:
    13.     log_topics: nginx_access
    14.  
    15. - type: log
    16.   enabled: true
    17.   paths:
    18.   	#收集日志地址
    19.     - /usr/local/logs/error.log #此日志为nginx的路径日志，事先在本机安装了nginx
    20.   fields:
    21.     log_topics: nginx_error
    22.  
    23. output.kafka: #只新增kafka这的output，把elasticsearch的注释掉，否则报错
    24.   #配置Kafka地址
    25.     hosts: ["10.10.200.33:9092","10.10.200.34:9092","10.10.200.35:9092"]
    26.     #这个Topic 要和Kafka一致
    27.     topic: '%{[fields][log_topics]}'
    28.  
    29. filebeat.config.modules:
    30.   # Glob pattern for configuration loading
    31.   path: ${path.config}/modules.d/*.yml
    32.  
    33.   # Set to true to enable config reloading
    34.   reload.enabled: false
    35.  
    36. setup.template.settings:
    37.   index.number_of_shards: 1
    38.  
    39. setup.kibana:
    40.  
    41. 启动filebeat
    42. [root@kf-zk-es-fb_es-head ~]# systemctl start filebeat

    10、安装kibana展示日志信息

    1. 下载地址：http://dl.elasticsearch.cn/kibana/kibana-7.9.2-x86_64.rpm
    2. [root@kf-zk-es-kibana ~]# rpm -ivh kibana-7.9.2-x86_64.rpm
    3. [root@kf-zk-es-kibana ~]# vim /etc/kibana/kibana.yml
    4. server.port: 5601
    5. server.host: "0.0.0.0"
    6. elasticsearch.hosts: ["http://10.10.200.33:9200","http://10.10.200.34:9200","http://10.10.200.35:9200"]
    7. kibana.index: ".kibana"
    8. 启动服务
    9. [root@kf-zk-es-kibana ~]# systemctl start kibana && systemctl enable kibana
    10. Created symlink from /etc/systemd/system/multi-user.target.wants/kibana.service to /etc/systemd/system/kibana.service.
    11. [root@node2 ~]# journalctl -u kibana #查看启动正常

    确认端口已启动后网页访问：

    点击后创建index，如果提示：You'll need to index some data into Elasticsearch before you can create an index pattern. Learn how 的话，原因是因为没有任何数据，任何索引，可在机器后台执行以下命令模拟新建数据：

    1. [root@node1 ~]# curl -H "Content-Type: application/json" -XPOST 'http://10.10.200.33:9200/ruizhi-log-2023-11-13/test-log' -d '{"code":200,"message":"测试"}'
    2. {"_index":"ruizhi-log-2023-11-13","_type":"test-log","_id":"mfRRx4sBE7vjRIscdoqf","_version":1,"result":"created","_shards":{"total":2,"successful":1,"failed":0},"_seq_no":0,"_primary_term":1}

    之后，刷新网页，如下图：

    选择两个nginx，之后模拟访问nginx:

    刷新此页面(后台命令行追加些内容到日志文件也可以)

    可在后台kafka的消费消息命令输出中，查看到nginx的日志已被传递到kafka:

    kibana上也同步了日志，内容是一模一样的

  • 相关阅读:
    使用单元测试提高代码质量与可维护性
    如何写出易于维护的Vue代码（踩坑经验）
    【✨十五天搞定电工基础】一阶电路的暂态分析
    Linux分区指南
    Visual C++ 2010 Express和EasyX的使用介绍
    matplotlib笔记
    力扣数组2（C++）
    day57 集合 List Set Map
    最简单的RNN预测股票收盘价
    C#多线程
  • 原文地址：https://blog.csdn.net/weixin_44147924/article/details/134310173