第一阶段：配置 AI 算力节点（192.168.227.161 机器）

登录到 161 机器，依次执行以下三条命令：

1. 以对外开放模式启动 Ollama 引擎容器

Bash

# 挂载数据卷并开放 0.0.0.0 允许跨机器调用
docker run -d --name ollama -p 11434:11434 -v ollama_data:/root/.ollama -e OLLAMA_HOST="0.0.0.0" ollama/ollama:0.6.8

2. 为引擎装载具体的“灵魂” (Qwen 模型)

Bash

# 让容器在后台下载通义千问模型
docker exec -it ollama ollama pull qwen

3. (可选) 启动配套的 OpenClaw 可视化面板

Bash

# 部署可视化界面，方便日常调试和对话
docker run -d --name openclaw -p 8080:8080 ghcr.io/phioranex/openclaw-docker:latest

第二阶段：配置监控业务节点（156 机器 / kafka2）

登录到 156 机器（你的 Kafka/Python 所在机器），进入到你的项目目录 /opt/monitor/celery_app/。

1. 安装必须的 Python 依赖库

Bash

pip install celery redis requests pymysql

2. 创建定时任务配置文件 celery_config.py 执行 vim celery_config.py，填入以下代码：

Python

from datetime import timedelta

# 定义 Celery 的定时调度计划
CELERYBEAT_SCHEDULE = {
    'check-traffic-every-minute': {
        'task': 'task.check_dynamic_traffic', # 指向我们要执行的动态告警任务
        'schedule': timedelta(minutes=1),     # 每隔 1 分钟执行一次
    },
    'check-static-traffic-every-minute': {
        'task': 'task.check_nginx_traffic',   # 指向静态阈值监控任务
        'schedule': timedelta(minutes=1),
    }
}

3. 覆盖核心逻辑代码 task.py 执行 vim task.py，将包含 AI 诊断和微信告警的全量代码填入（注意修改你的邮箱授权码、数据库密码和微信 Key）：

Python

# /opt/monitor/celery_app/task.py
from celery_app import app
import pymysql
import smtplib
from email.mime.text import MIMEText
from email.header import Header
from datetime import datetime, timedelta
import requests 

# --- 1. 邮件配置 ---
SMTP_SERVER = "smtp.qq.com"
SMTP_PORT = 465
SENDER_USER = "3187726341@qq.com"  
SENDER_PASS = "你的QQ授权码"  
RECEIVER = "3187726341@qq.com"     

# --- 2. AI 与微信告警配置 ---
OLLAMA_URL = "http://192.168.227.161:11434/api/generate"
WECHAT_WEBHOOK = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的企业微信机器人KEY"

# --- 3. 数据库配置 ---
DB_CONFIG = {
    'host': '127.0.0.1',
    'user': 'root',
    'password': '你的数据库密码',
    'database': 'log_monitor',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor
}

def send_alert_email(content):
    """发送邮件辅助函数 (审计留档)"""
    message = MIMEText(content, 'plain', 'utf-8')
    message['From'] = SENDER_USER
    message['To'] = RECEIVER
    message['Subject'] = Header("🚨 流量监控告警系统 🚨", 'utf-8')
    try:
        smtp_obj = smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT)
        smtp_obj.login(SENDER_USER, SENDER_PASS)
        smtp_obj.sendmail(SENDER_USER, [RECEIVER], message.as_string())
        smtp_obj.quit()
        print(f"✉️ [邮件系统] 告警邮件已发送")
    except Exception as e:
        print(f"❌ [邮件系统] 发送失败: {e}")

def fetch_recent_logs(cursor, limit=100):
    """捞取异常日志"""
    sql = "SELECT client_ip, status_code, request_path, log_time FROM traffic_logs ORDER BY log_time DESC LIMIT %s"
    cursor.execute(sql, (limit,))
    logs = cursor.fetchall()
    if not logs: return "暂无数据"
    return "\n".join([f"[{l['log_time']}] IP:{l['client_ip']} 请求:{l['request_path']} 状态:{l['status_code']}" for l in logs])

def analyze_with_ollama(log_data_str):
    """呼叫 AI 大模型进行诊断"""
    prompt = f"你是一个安全专家。请分析以下Nginx异常日志切片，判断是正常突增、恶意爬虫还是CC攻击，并给出建议。\n日志数据：\n{log_data_str}"
    payload = {"model": "qwen", "prompt": prompt, "stream": False}
    try:
        response = requests.post(OLLAMA_URL, json=payload, timeout=30)
        return response.json().get("response", "AI 分析失败。")
    except Exception as e:
        return f"🚨 AI 诊断服务连接异常: {e}"

def send_wechat_alert(markdown_content):
    """推送企业微信告警"""
    payload = {"msgtype": "markdown", "markdown": {"content": markdown_content}}
    try:
        requests.post(WECHAT_WEBHOOK, json=payload, timeout=10)
        print("📱 [微信系统] 告警已推送到企业微信")
    except Exception as e:
        print(f"❌ [微信系统] 推送失败: {e}")

@app.task
def check_nginx_traffic():
    """任务一：静态阈值监控 (保留原有逻辑)"""
    print(f"[{datetime.now()}] 执行：静态阈值检查...")
    db = pymysql.connect(**DB_CONFIG)
    try:
        cursor = db.cursor()
        target_time = datetime.now() - timedelta(minutes=3)
        start_time = target_time.strftime('%Y-%m-%d %H:%M:00')
        end_time = target_time.strftime('%Y-%m-%d %H:%M:59')
        cursor.execute("SELECT SUM(bytes_sent) as total_flow FROM traffic_logs WHERE log_time >= %s AND log_time <= %s", (start_time, end_time))
        total_flow = cursor.fetchone()['total_flow'] or 0
        cursor.execute("SELECT * FROM monitor_rules WHERE is_active = 1")
        for rule in cursor.fetchall():
        if total_flow > rule['max_bytes']:
        msg = f"🔥 【阈值告警】\n规则：[{rule['rule_name']}]\n流量：{total_flow} Bytes\n上限：{rule['max_bytes']} Bytes"
        cursor.execute("INSERT INTO alerts (rule_id, alert_message, alert_time) VALUES (%s, %s, %s)", (rule['id'], msg, datetime.now()))
        send_alert_email(msg)
        print(f"🚨 [阈值触发]: {rule['rule_name']} - {total_flow}B")
        db.commit()
        finally:
        db.close()

        @app.task
        def check_dynamic_traffic():
        """任务二：动态趋势监控 + AI智能诊断"""
        print(f"[{datetime.now()}] 执行：动态趋势检查...")
        db = pymysql.connect(**DB_CONFIG)
        try:
        cursor = db.cursor()
        now_target = datetime.now() - timedelta(minutes=3)
        now_start = now_target.strftime('%Y-%m-%d %H:%M:00')
        now_end = now_target.strftime('%Y-%m-%d %H:%M:59')
        yes_start = (now_target - timedelta(days=1)).strftime('%Y-%m-%d %H:%M:00')
        yes_end = (now_target - timedelta(days=1)).strftime('%Y-%m-%d %H:%M:59')

        cursor.execute("SELECT SUM(bytes_sent) as flow FROM traffic_logs WHERE log_time >= %s AND log_time <= %s", (now_start, now_end))
        current_flow = cursor.fetchone()['flow'] or 0
        cursor.execute("SELECT SUM(bytes_sent) as flow FROM traffic_logs WHERE log_time >= %s AND log_time <= %s", (yes_start, yes_end))
        last_day_flow = cursor.fetchone()['flow'] or 0

        if last_day_flow > 0:
        growth_rate = (current_flow - last_day_flow) / last_day_flow
        if growth_rate > 0.5 and current_flow > 100:
        print("🔍 捞取日志中...")
        abnormal_logs = fetch_recent_logs(cursor, limit=100)
        print("🤖 呼叫 AI 诊断中...")
        ai_report = analyze_with_ollama(abnormal_logs)

        wechat_msg = (
        f"<font color=\"warning\">🔥【高危】Nginx 动态流量异常</font>\n\n"
        f"> **当前流量：** <font color=\"info\">{current_flow} Bytes</font>\n"
        f"> **昨日同期：** <font color=\"comment\">{last_day_flow} Bytes</font>\n"
        f"> **暴涨幅度：** <font color=\"warning\">{growth_rate*100:.2f}%</font>\n\n"
        f"**🤖 AI 专家实时诊断：**\n{ai_report}"
        )
        email_msg = f"📈 【趋势异常告警】\n当前流量：{current_flow} Bytes\n突增比例：{growth_rate*100:.2f}%\n\n🤖 [AI 智能诊断]\n{ai_report}"

        cursor.execute("INSERT INTO alerts (rule_id, alert_message, alert_time) VALUES (99, %s, %s)", (email_msg, datetime.now()))
        send_wechat_alert(wechat_msg)
        send_alert_email(email_msg)
        db.commit()
        finally:
        db.close()

第三阶段：启动服务跑通闭环（156 机器）

所有的代码都准备好了，最后就是把心跳和打工人跑起来。 在 156 机器上，开两个独立的终端窗口，分别执行：

终端 A (启动 Worker，干活的进程):

Bash

cd /opt/monitor/celery_app/
celery -A task worker --loglevel=info

终端 B (启动 Beat，发号施令的心跳进程):

Bash

cd /opt/monitor/celery_app/
celery -A task beat -l info --config=celery_config

大功告成！ 你现在拥有一套业界领先的、带有 AI 智能诊断和微信推送能力的高并发日志监控架构了！

接着在161机器上，

docker stop ollama
docker rm ollama
docker run -d --name ollama -p 11434:11434 -v ollama_data:/root/.ollama -e OLLAMA_HOST="0.0.0.0" ollama/ollama:0.6.8
docker exec -it ollama ollama pull qwen

[root@localhost ~]# docker run -d --name ollama -p 11434:11434 -v ollama_data:/root/.ollama -e OLLAMA_HOST="0.0.0.0" ollama/ollama:0.6.8
391f70dd8441e362fb3375cbd6bd1b946fa734a78f7803db4805712658d21479
[root@localhost ~]# docker exec -it ollama ollama pull qwen
pulling manifest 
pulling 46bb65206e0e: 100% ▕███████████████▏ 2.3 GB                         
pulling 41c2cf8c272f: 100% ▕███████████████▏ 7.3 KB                         
pulling 1da0581fd4ce: 100% ▕███████████████▏  130 B                         
pulling f02dd72bb242: 100% ▕███████████████▏   59 B                         
pulling b861bd365e67: 100% ▕███████████████▏  483 B                         
verifying sha256 digest 
writing manifest 
success

一、 为什么不花钱？（本地部署的魅力）

平时我们听说用大模型（比如 ChatGPT、文心一言）要花钱，是因为我们调用的是云端 API，人家按你输入的字数（Token）扣费。 但是！你刚才执行的 ollama pull qwen，是把大模型的“权重文件（相当于它的脑子）”直接下载到了你自己的服务器硬盘上。

  • 算力是你自己的： 它在思考时，消耗的是你 161 机器的 CPU 和内存。
  • 完全断网可用： 哪怕你现在把 161 机器的外网网线拔了，你的 Nginx 监控告警系统依然能正常让 AI 分析日志。 这叫**“本地私有化部署”**，开源免费，没有任何隐藏消费。

二、 为什么要选 Qwen（通义千问）？

现在开源的免费模型有成百上千个（比如 Meta 的 Llama3、谷歌的 Gemma），为什么我们在 task.py 里偏偏指定要用 qwen 呢？主要有三个核心原因：

1. 中文语境的“天花板” Qwen 是阿里巴巴开源的大模型。相比于国外的 Llama3，它的中文原生理解能力极强。当你让它写微信告警或者分析日志时，它能输出非常地道、符合中国运维人员阅读习惯的中文报告，不会出现那种生硬的“机翻味”。

2. 运维与代码分析的“行家里手” 我们要 AI 做的事是**“看懂 Nginx 日志”**。阿里作为国内最大的云服务厂商，Qwen 在训练时“喂”了海量的代码、服务器日志和运维排错文档。它对 IP 聚集、403/502 状态码、CC 攻击特征的敏感度极高，非常契合我们这个【日志监控与异常诊断】的项目场景。

3. 极高的性价比（轻量级） Ollama 默认拉取的 qwen 通常是它的轻量级量化版本（比如 7B 参数量）。这意味着哪怕你的 161 机器没有极其昂贵的专业显卡（GPU），仅仅依靠普通 CPU 和十几 G 内存，它也能在几秒钟内快速阅读完 100 条日志并给出结论。这保证了你的告警系统不会因为 AI 思考太慢而发生延迟。

# kafka # 人工智能 # 分布式
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

AtomGit 6 月：AtomCode CodingPlan 上线 GLM-5.2！

avatar AtomGit开源社区
cover

直播预约｜JiuwenSwarm 驱动下的一人 AI 业务团队系统构建

avatar AtomGit开源社区
cover

行业 Agent 黑客松圆满收官：打通行业落地“最后一公里”

avatar AtomGit开源社区