flume+kafka nginx日志系统搭建完整流程

心已赠人 2022-11-22 05:16 315阅读 0赞

> 设计 Apache Flume + Zookeeper + Fafka中间件 + 持久化存储 。用 Apache Flume 获取 nginx 日志，设置 sources 为 tail -f access-log，设置 sink 为 kafka 集群 ，从kafka 中取数据 结果存入 redis/mongo/es ，提供 API查询

### Nginx 日志 format 修改 ###

首先，我们对 要收集的 Nginx 日志产生的数据进行定制

先来看看，Nginx 日志格式参数的配置

# http 模块
    http { 
        # Nginx 默认的 main 配置 这里不使用，改为 下面定制的 mylogformat
        #log_format main '$remote_addr - $remote_user [$time_local] "$request" '
        # '$status $body_bytes_sent "$http_referer" '
        # '"$http_user_agent" "$http_x_forwarded_for"';
    
    
        log_format  mylogformat '$remote_addr [$time_local] "$request_method" '
                          '"$uri" "$request_uri" "$request_time" '
                '$status $body_bytes_sent "$http_referer" '
                '"$http_x_forwarded_for" "$http_user_agent" "$upstream_status" '
                          '"$upstream_addr" "$upstream_response_time"';
        access_log  /var/log/nginx/access1.log  mylogformat;
    
    }

这里 format 使用了定制的 mylogformat 格式，看下下面是产生的一条 access日志：

117.78.38.154 [02/Nov/2020:10:20:23 +0800] "POST" "/api/user/analyse_his" "/api/user/analyse_his" "0.034" 200 60 "-" "42.244.62.137" "android" "200" "127.0.0.1:8008" "0.030"

我们把其中各个值 和 format 的参数对应成表格 ：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 1]

好了，上面就定制了我们需要产生的格式的 Nginx 日志，这样他就安静的躺在那里等待被收集了，下面来安装 zookeeper + Kafka。

### 基于 docker zookeeper + 3节点 kafka 集群 的安装搭建 ###

#### 安装 zookeeper ####

docker run --name zookeeper -d -p 2181:2181 -v /etc/localtime:/etc/localtime library/zookeeper
    # 查看 2181 端口占用
    lsof -i:2181

#### 安装 kafka 集群 ####

# ifconfig 查看内网ip
    
    119.3.216.248
    
    # 基于 hub.docker.com 上 star 最多的 wurstmeister/kafka image
    # 搭建3节点集群 注意 port 
     
    docker run -d --name kafka1 -p 9092:9092 --link zookeeper --env KAFKA_ZOOKEEPER_CONNECT=119.3.216.248:2181 --env KAFKA_ADVERTISED_HOST_NAME=119.3.216.248 --env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://119.3.216.248:9092 --env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 --env KAFKA_ADVERTISED_PORT=9092  wurstmeister/kafka
    
    docker run -d --name kafka2 -p 9093:9093 --link zookeeper --env KAFKA_ZOOKEEPER_CONNECT=119.3.216.248:2181 --env KAFKA_ADVERTISED_HOST_NAME=119.3.216.248 --env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://119.3.216.248:9093 --env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9093 --env KAFKA_ADVERTISED_PORT=9093  wurstmeister/kafka
    
    docker run -d --name kafka3 -p 9094:9094 --link zookeeper --env KAFKA_ZOOKEEPER_CONNECT=119.3.216.248:2181 --env KAFKA_ADVERTISED_HOST_NAME=119.3.216.248 --env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://119.3.216.248:9094 --env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9094 --env KAFKA_ADVERTISED_PORT=9094  wurstmeister/kafka

进入 kafka 发送消息 测试

$KAFKA_HOME/bin/kafka-console-producer.sh --broker-list 119.3.216.248:9092 --topic sun
    > Hello leon!

kafka 消费者接收消息

$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic sun --from-beginning
     
     Hello leon!

kafka 集群加入 zookeeper 中的 WARN 信息

发送消息测试 
    bash-4.4# $KAFKA_HOME/bin/kafka-console-producer.sh --broker-list 119.3.216.248:9092 --topic sun
    >Hello leon!
    [2020-11-02 05:59:09,465] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 3 : { sun=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
    [2020-11-02 05:59:09,577] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 4 : { sun=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
    >
    $KAFKA_HOME/bin/kafka-console-producer.sh --broker-list 119.3.216.248:9092 --topic test>
    >^Cbash-4.4# $KAFKA_HOME/bin/kafka-console-producer.sh --broker-list 119.3.216.248:9092 --topic test
    >^Cbash-4.4# $KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic sun --from-beginning
    Hello leon!

#### Kafka Eagle 安装 ####

安装kafka-manager 来管理 kafka

docker pull sheepkiller/kafka-manager
    
    
    lsof -i:9001
    lsof -i:9000
    
    
    docker run -d --name kafka-manager --link zookeeper:zookeeper --link kafka1:kafka1 -p 9001:9000 --restart=always --env ZK_HOSTS=zookeeper:2181 sheepkiller/kafka-manager
    
    
    http://127.0.0.1:9001/

配置 kafka-manager，添加 cluster zookeeper 所在机器 Ip + port

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 2]

这样 zookeeper 和 kafka 的3节点集群 就搭建完了，下面安装 flume 进行数据采集

### flume 采集 Nginx 日志 写入 kafka 集群 ###

安装 jdk-1.8
    
    Yum install -y java 
    Jdk 路径： 
    
    /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.262.b10-0.el7_8.x86_64/jre/bin/java

#### 安装 启动 flume ####

wget http://mirror.bit.edu.cn/apache/flume/1.8.0/apache-flume-1.8.0-bin.tar.gz
    
    cp flume-env.sh.template flume-env.sh
    vim flume-env.sh
    export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.262.b10-0.el7_8.x86_64/jre

#### 写入kafka 的 sink ####

# conf 目录 配置 写入kafka 的 sink - flume2kafka.conf
    a1.sources = r1
    a1.sinks = k1
    a1.channels = c1
    
    # 这里监听 /root/nginxlog 这个目录 
    # 描述和配置source组件：r1
    a1.sources.r1.type = spooldir
    a1.sources.r1.spoolDir = /root/nginxlog
    a1.sources.r1.fileHeader = true
    
    
    # 描述和配置sink组件：k1
    a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
    a1.sinks.k1.kafka.topic = ftoktopic
    # kafka 集群
    a1.sinks.k1.kafka.bootstrap.servers = 127.0.0.1:9092 127.0.0.1:9093 127.0.0.1:9094
    a1.sinks.k1.kafka.flumeBatchSize = 20
    a1.sinks.k1.kafka.producer.acks = 1
    a1.sinks.k1.kafka.producer.linger.ms = 1
    a1.sinks.ki.kafka.producer.compression.type = snappy
    
    
    # 描述和配置channel组件，此处使用是内存缓存的方式
    a1.channels.c1.type = memory
    a1.channels.c1.capacity = 1000
    a1.channels.c1.transactionCapacity = 100
    
    
    # 描述和配置source channel sink之间的连接关系
    a1.sources.r1.channels = c1
    a1.sinks.k1.channel = c1

#### 启动 flume 消费 ####

cd /root/apache-flume-1.8.0-bin
    
    bin/flume-ng  agent -c conf -f /root/apache-flume-1.8.0-bin/conf/access_2_kafka.conf -n a1 -Dflume.root.logger=INFO,console
    
    screen -S accesstopic bin/flume-ng  agent -c conf -f /root/apache-flume-1.8.0-bin/conf/access_2_kafka.conf -n a1 -Dflume.root.logger=INFO,console

#### kafka 检查写入情况 ####

往 /root/nginxlog 写入 文件 
    
    1.txt 
    test 
    1
    2
    3
    
    kafka 消费窗口出现
    >
    test
    1
    2
    3

至此，flume 收集文件 sink 写入 kafka 成功

### nginx 日志采集 ###

对nginx 切割，scp 到 flume 目录 写入 kafka，这里将 nginx 产生的日志 每日切割 并传到 flume 消费 的目录中 ，完成定时写入kafka 的目的

yum -y install sshpass
    
    
    /root/access.sh
    
    
    #/bin/bash
    nginxLog="/var/log/nginx/access1.log"
    logBakFile="/var/log/nginx/"$(date "+%F")/"access1.log"
    
    
    if [ ! -d ${logBakFile%/*} ]
    then
        mkdir -p ${logBakFile%/*}
    fi
    
    
    mv $nginxLog $logBakFile 2>&1 >/dev/null
    
    flumeFile="/root/nginxlog/"$(date "+%F")"access1.log"
    
    
    sshpass -p ssh密码 scp $logBakFile root@机器ip:$flumeFile
    
    
    /usr/local/nginx/sbin/nginx -s reload
    
    #每晚23:59 切割当天accesslog 并传给flume
    59 23 * * * bash /root/access.sh > /dev/null 2>&1

#### flume端 配置的 spooldir 目录 ####

\[root@k8s-node1 nginxlog\]\# ls

2020-11-03access1.log.COMPLETED

#### tail 方式收集日志 ####

这里也可以在 flume 的 agent 中 配置 exec 类型的任务，这样就不需要把 nginx 日志做切割，每天 传到 spooldir 目录了。

# Describe/configure the source
    a1.sources.r1.type = exec
    a1.sources.r1.channels = c1
    a1.sources.r1.command = tail -F /tmp/test

以上，就完成了 flume 消费日志，产生数据到 kafka ，后面的客户端消费日志，分析等操作就是业务相关的，自行发挥即可。

### 存储日志 ###

经过业务处理最终，存储的数据我这里存储在es 中，其中一条 access\_log 如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 3]  
可以看到，以后我们要进行日志分析，无论是对于 http\_status 还是以 response\_time 超时作为基准 都能很方便的使用。

### 效果 ###

基于 vue-admin 做了个简单的 UI 后台

根据 时间段 ，错误类型（ time-out/504/500等 ） 查看 error\_log / acces\_log /具体业务错误

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 4]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 5]  
kafka  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 6]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/539c17156a6b4965bd3fed734d3a476e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/57934fdaaa2c4987b877841d56231e31.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/812ce3accb6e42c2aa33a20b279c5447.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/350acbab6d834569aa02df03c829ba80.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221120/e2f25a842de84575b66134dcc32af337.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221120/2cc3044fb80f4f01874bbbe4dfb9ca66.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4MDE4Mjgz_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221120/291465b2382a4dbeb8a86fc2cadd2e0b.png