【Project】基于kafka的高可用分布式日志监控与告警系统

底层学习笔记

kafka数据存储

kafka最高水位线机制

Kraft模式

zooker和kraft

系统部署与使用说明

本系统主要包含 Kafka、Filebeat、Nginx（搭配 Keepalived）以及 Python 消费者程序的配置与部署，用于处理和存储 Nginx 访问日志。以下是详细的配置、部署和使用说明。

1. Kafka 配置与启动

1.1 配置文件

Kafka 采用 KRaft 模式，涉及三个节点（kafka01、kafka02、kafka03），各节点配置文件如下：

• kafka01：/opt/kafka_2.13-3.6.1/config/kraft/server.properties

process.roles=broker,controller
node.id=1
controller.quorum.voters=1@192.168.100.150:9093,2@192.168.100.151:9093,3@192.168.100.152:9093
listeners=PLAINTEXT://kafka01:9092,CONTROLLER://kafka01:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://kafka01:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/var/lib/kafka
num.partitions=1
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=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

• kafka02：/opt/kafka_2.13-3.6.1/config/kraft/server.properties

process.roles=broker,controller
node.id=2
controller.quorum.voters=1@192.168.100.150:9093,2@192.168.100.151:9093,3@192.168.100.152:9093
listeners=PLAINTEXT://kafka02:9092,CONTROLLER://kafka02:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://kafka02:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/var/lib/kafka
num.partitions=1
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=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

• kafka03：/opt/kafka_2.13-3.6.1/config/kraft/server.properties

process.roles=broker,controller
node.id=3
controller.quorum.voters=1@192.168.100.150:9093,2@192.168.100.151:9093,3@192.168.100.152:9093
listeners=PLAINTEXT://kafka03:9092,CONTROLLER://kafka03:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://kafka03:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/var/lib/kafka
num.partitions=1
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=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

1.2 启动与关闭命令

# 启动 Kafka
/opt/kafka_2.13-3.6.1/bin/kafka-server-start.sh -daemon /opt/kafka_2.13-3.6.1/config/kraft/server.properties
# 关闭 Kafka（可选）
#/opt/kafka_2.13-3.6.1/bin/kafka-server-stop.sh

1.3 格式化存储

kafka-storage.sh format  --cluster-id BlQPRL0HSdaaMwvx_dVcAQ  --config /opt/kafka_2.13-3.6.1/config/kraft/server.properties --node-id 1

2. Filebeat 配置

Filebeat 用于收集 Nginx 访问日志并发送到 Kafka，配置文件如下：

filebeat.inputs:
- type: logenabled: truepaths:- /usr/local/nginx1/logs/access.log
output.kafka:hosts: ["kafka01:9092","kafka02:9092","kafka03:9092"]topic: nginxlogkeep_alive: 10s

3. Nginx 配置

3.1 Keepalived 配置

! Configuration File for keepalived
global_defs {router_id LVS_DEVEL
}
vrrp_script chk_nginx {script "/etc/keepalived/check_nginx.sh"interval 10weight -5fall 2rise 1
}
vrrp_instance VI_1 {state MASTERinterface ens33mcast_src_ip 192.168.100.150virtual_router_id 51priority 100advert_int 2authentication {auth_type PASSauth_pass BLUEKING_NGINX_HA}virtual_ipaddress {192.168.100.100/24}track_script {chk_nginx}
}

3.2 检查脚本

#!/bin/bash
process_name="nginx"
process_abs_path="/usr/local/nginx1/sbin/nginx"function is_nginx_running() {process_info=$(ps --no-header -C $process_name -o ppid,pid,args | awk '{printf $1 "|" $2 "|" ; for (i=3; i<=NF; i++) { printf "%s ", $i };printf "\n"}' | grep master)if [[ -z "$process_info" ]]; thenreturn 1elseprocess_pids=($(echo "$process_info" | awk -F'|' '{print $2}'))for _pid in "${process_pids[@]}"; doabs_path=$(readlink -f /proc/$_pid/exe)if [ "$abs_path" == "$(readlink -f "$process_abs_path")" ]; thenreturn 0fidonereturn 2fi
}err=0
for k in $(seq 1 3); dois_nginx_runningif [[ $? != "0" ]]; thenerr=$((err + 1))sleep 1elseerr=0breakfi
doneif [[ $err != "0" ]]; thenexit 1  # 仅返回失败状态，不停止 keepalived
elseexit 0
fi

4. Python 程序

4.1 依赖安装

pip install kafka-python==2.0.2 pymongo==4.13.0 mysql-connector-python==9.3.0 redis==4.5.5 kombu==5.3.1 celery==5.3.1

4.2 代码

https://github.com/Lenoud/nginx-monitor-kafka-python

5. 注意事项

• 确保各节点的网络连通性，特别是 Kafka 节点之间以及与 MySQL 数据库的连接。
• 检查各配置文件中的 IP 地址、端口号、用户名和密码等信息是否正确。
• 在启动 Kafka 之前，确保已完成存储格式化操作。
• 若需要修改日志存储策略或 Kafka 配置，可根据实际需求调整相应的配置文件。

完成整个系统的部署和配置，实现 Nginx 访问日志的收集、传输和存储。