日志项目

从 Kafka 消费日志到邮件告警：一次完整的 Docker 化部署与排错实战

从 Kafka 消费日志到邮件告警：一次完整的 Docker 化部署与排错实战

关键词：Kafka 消费者、Python、Docker、QQ 邮箱 SMTP、Connection unexpectedly closed、systemctl restart 替代方案

适用场景：ELK 精简版日志分析系统｜自研告警平台｜CI/CD 日志监控
一、背景

在搭建轻量级日志分析系统时，采用的完整链路为：
Nginx → Filebeat → Kafka → Python 消费者 → MySQL + Redis + 邮件告警

整套流程在本地调试阶段一切正常，但部署到生产环境（CentOS 7 + Docker）后，出现两个典型的生产问题：

  1. 邮件发送失败：日志反复报错 [ERROR] 发送邮件失败: Connection unexpectedly closed
  2. 容器“静默”无日志：docker logs 命令无任何输出，但手动进入容器运行程序却一切正常
本文将完整复盘两个问题的排查过程，并给出可直接复用的解决方案和生产级配置。

二、问题 1：邮件发送失败 —— SMTP 连接被意外关闭

2.1 错误现象

消费者程序运行后，数据库写入、日志解析等核心逻辑均正常，但邮件告警模块持续报错，日志中反复出现：
[ERROR] 发送邮件失败: Connection unexpectedly closed

核心特征：业务逻辑无异常，问题仅集中在邮件发送模块，排除程序整体运行故障。

SMTP 连接意外关闭错误日志

2.2 原因分析

初始邮件发送代码采用 QQ 邮箱 587 端口 + STARTTLS 加密方式，核心代码如下：
server = smtplib.SMTP(‘smtp.qq.com’, 587)
server.starttls()
server.login(…)

在阿里云、腾讯云等云服务器环境中，587 端口常被云厂商安全组策略或运营商网络策略拦截，导致 TCP 连接刚建立就被强制关闭，最终表现为 Connection unexpectedly closed 错误。

2.3 正确配置方案

改用 QQ 邮箱 465 端口 + SSL 加密方式，这是云环境下最稳定的 SMTP 配置方案，同时采用异步发送方式，避免邮件发送阻塞 Kafka 消费主逻辑，核心配置代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import smtplib
from email.mime.text import MIMEText
from email.header import Header
import threading

# 邮件配置（请替换为实际值）
EMAIL_CONFIG = {
    'smtp_server': 'smtp.qq.com',           # QQ 邮箱 SMTP 服务器
    'smtp_port': 465,                       # 使用 SSL 加密端口
    'email': 'your_email@qq.com',           # 发件人邮箱（如：admin@qq.com）
    'password': 'your_authorization_code',  # 授权码（非登录密码，16位）
    'to_email': 'alert@example.com'         # 收件人邮箱（告警接收地址）
}

def send_email_async(subject, body):
    """
    异步发送邮件（不阻塞主线程）
    """
    def send():
        try:
            msg = MIMEText(body, 'plain', 'utf-8')
            msg['From'] = EMAIL_CONFIG['email']
            msg['To'] = EMAIL_CONFIG['to_email']
            msg['Subject'] = Header(subject, 'utf-8')

            # 使用 SSL 加密连接
            server = smtplib.SMTP_SSL(
                EMAIL_CONFIG['smtp_server'],
                EMAIL_CONFIG['smtp_port']
            )
            server.login(EMAIL_CONFIG['email'], EMAIL_CONFIG['password'])
            server.sendmail(EMAIL_CONFIG['email'], [EMAIL_CONFIG['to_email']], msg.as_string())
            server.quit()
            print(f"[ALERT] 邮件发送成功: {subject}")
        except Exception as e:
            print(f"[ERROR] 发送邮件失败: {e}")

    # 启动守护线程，异步执行
    threading.Thread(target=send, daemon=True).start()

2.4 验证方法

在生产服务器执行以下命令，测试 465 端口的网络连通性和 SSL 握手是否正常：
openssl s_client -connect smtp.qq.com:465 -quiet

若命令返回以下内容，说明网络通畅，SSL 握手成功，端口未被拦截：
220 smtp.qq.com Esmtp QQ Mail Server
三、问题 2：容器“静默”无日志？

3.1 现象复现

执行 docker run -d --network host --name log-consumer --restart=always log-consumer 启动容器后，出现以下异常现象：

  1. docker ps 查看容器状态，显示容器正常运行（Up 状态）
  2. docker logs log-consumer 或 docker logs -f log-consumer 无任何输出
  3. 手动进入容器执行 python3 consumer.py，程序正常运行，日志解析、数据库写入、邮件发送均无问题

容器静默无日志问题排查过程

3.2 根本原因

并非程序故障，而是 Kafka 消费者的正常行为：
Kafka 消费者配置了固定的 group_id='log-consumer-group'，当消费者首次运行并消费完 Kafka 主题中所有消息后，会自动提交消费偏移量（offset）。后续重启容器时，消费者会从上次提交的 offset 位置继续读取消息，若此时 Kafka 主题中无新的日志消息产生，消费者会进入安静等待状态，不会输出任何日志，并非程序卡死或运行异常。

3.3 验证与测试

通过手动触发一次新的 Nginx 请求，生成新的日志消息，即可验证消费者程序是否正常工作：
# 在服务器另一终端执行，触发新的Nginx访问请求
curl http://kafka1/
# 实时查看容器日志，验证是否捕获并处理新消息
docker logs -f log-consumer

若消费者程序正常，会立即在日志中输出以下内容，说明程序处于正常等待状态，仅需新消息触发即可：
[INFO] 处理 nginx-access 日志
[SUCCESS] 日志已存储: 36.251.161.209 -> /index.html
[ALERT] 错误告警: 36.251.161.209
[ALERT] 邮件发送成功: ALERT 服务器告警 - 错误
四、一键重启容器：替代 systemctl 的快捷方式

4.1 为什么需要自定义重启？

Docker 本身没有类似 systemctl restart 的原子重启操作，存在以下问题：

  1. 直接执行 docker restart log-consumer，仅重启容器进程，不会重新加载新的镜像和配置
  2. 开发和生产环境中，代码更新后需要执行「停止旧容器 → 删除旧容器 → 构建新镜像 → 启动新容器」四步操作，步骤繁琐

因此需要自定义一键重启脚本，实现类似 systemctl restart 的便捷操作。

4.2 创建重启脚本

创建全局可执行脚本 /usr/local/bin/restart-log-consumer.sh，包含完整的重启逻辑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#!/bin/bash

# 停止旧容器，忽略容器不存在的错误
docker stop log-consumer 2>/dev/null

# 删除旧容器，忽略容器不存在的错误
docker rm log-consumer 2>/dev/null

# 构建新镜像（基于当前目录下的 Dockerfile）
docker build -t log-consumer .

# 启动新容器，使用 host 网络模式，开启开机自启
docker run -d \
  --network host \
  --name log-consumer \
  --restart=always \
  log-consumer

echo "✅ log-consumer 容器已完成重启"

给脚本添加全局执行权限：
chmod +x /usr/local/bin/restart-log-consumer.sh

4.3 使用效果

配置完成后，只需在服务器任意目录执行一条命令，即可完成「停旧容器 → 删旧容器 → 建镜像 → 启新容器」的完整流程：
restart-log-consumer.sh

执行效果完全等同于传统系统服务的 systemctl restart log-consumer.service，大幅提升运维效率。
五、完整消费者核心逻辑（精简版）

整合邮件告警、日志解析、Kafka 消费、MySQL 写入的核心逻辑，精简版代码如下（可直接用于生产环境）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
-------
import json
import pymysql
import redis
from kafka import KafkaConsumer
from datetime import datetime

# ===================== 配置 =====================
# 数据库配置
MYSQL_CONFIG = {
    'host': 'localhost',
    'user': 'root',
    'password': 'your_mysql_password',  # ← 替换为实际密码
    'database': 'log_analysis'
}

# Redis 配置（用于缓存或去重）
REDIS_CLIENT = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True
)

# 邮件发送函数（见前文，此处仅引用）
# send_email_async(subject, body)  # 异步发送邮件


def process_log_message(message):
    """
    处理单条 Kafka 日志消息
    """
    try:
        # 解析 Kafka 消息体（假设是 JSON 格式）
        log_data = json.loads(message.value.decode('utf-8'))
        ip = log_data.get('client_ip')
        url = log_data.get('request_uri')

        if not ip or not url:
            print(f"[ERROR] 日志数据不完整: {log_data}")
            return

        # 1. 写入 MySQL
        conn = pymysql.connect(**MYSQL_CONFIG)
        cursor = conn.cursor()
        sql = "INSERT INTO access_log (ip, url, created_at) VALUES (%s, %s, %s)"
        cursor.execute(sql, (ip, url, datetime.now()))
        conn.commit()
        conn.close()

        print(f"[SUCCESS] 日志已存储: {ip} -> {url}")

        # 2. 触发邮件告警（示例：所有访问请求均告警，可根据业务调整规则）
        send_email_async(
            subject="【系统告警】新访问",
            body=f"检测到新访问：IP={ip} → URL={url}"
        )

        print(f"[ALERT] 告警已发送: {ip}")

    except Exception as e:
        print(f"[ERROR] 处理日志失败: {e}")


def main():
    """
    程序主入口
    """
    print("=" * 50)
    print("启动 Kafka 日志消费者")
    print("=" * 50)

    # 初始化 Kafka 消费者
    consumer = KafkaConsumer(
        'nginx-logs',                          # 订阅的主题
        bootstrap_servers=['kafka1:9092'],     # Kafka 集群地址
        auto_offset_reset='latest',            # 从最新偏移量开始消费
        enable_auto_commit=True,               # 自动提交消费偏移量
        group_id='log-consumer-group'          # 消费组 ID
    )

    print("TARGET 开始监听日志...")
    print("按 Ctrl+C 停止消费者")

    try:
        # 循环消费 Kafka 消息
        for message in consumer:
            process_log_message(message)
    except KeyboardInterrupt:
        # 捕获手动停止信号，优雅退出
        print("\n✅ 消费者程序已手动停止")
    finally:
        consumer.close()


if __name__ == "__main__":
    main()

-------

六、总结与建议

核心问题总结

  1. 邮件发送失败：云环境下优先使用 QQ 邮箱 465 端口 + SMTP_SSL 加密，避免 587 端口被拦截，同时采用异步发送避免阻塞主逻辑
  2. 容器无日志：并非故障，是 Kafka 消费者在无新消息时的正常等待行为，可通过触发新请求验证
  3. 容器重启：通过自定义 Shell 脚本实现一键重启，替代 systemctl 操作，提升运维效率

生产环境建议

  1. 告警容灾：为邮件告警增加钉钉/企业微信 Webhook 作为备份，避免 SMTP 服务故障导致告警失联
  2. 邮件发送：云服务器慎用公网 SMTP 服务，推荐使用云厂商专属邮件服务（如阿里云 DirectMail），稳定性更高
  3. Kafka 集群：生产环境中 Kafka 集群至少部署 3 个节点，保障服务高可用，避免单节点故障导致日志链路中断
  4. 异常处理：程序中增加完善的异常捕获逻辑，避免单条日志处理失败导致整个消费者程序退出
  5. 日志持久化：将容器日志挂载到宿主机，或接入日志收集系统，避免容器日志丢失

七、附录：常用命令速查

整理生产环境中高频使用的运维命令，一键复制即可使用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 1. 查看容器运行状态
docker ps | grep log-consumer

# 2. 实时跟踪容器日志（核心排错命令）
docker logs -f log-consumer

# 3. 查看容器最近 20 行日志
docker logs --tail 20 log-consumer

# 4. 一键重启消费者容器（需先创建 restart.sh 脚本）
./restart-log-consumer.sh

# 5. 测试 QQ 邮箱 465 端口连通性（验证邮件告警是否可达）
openssl s_client -connect smtp.qq.com:465 -quiet

# 6. 触发 Nginx 新请求，验证消费者程序响应
curl http://kafka1/

# 7. 手动进入容器，调试程序
docker exec -it log-consumer /bin/bash

# 8. 重新构建 Docker 镜像（代码更新后）
docker build -t log-consumer .