docker run --name prometheus -d -p 127.0.0.1:9090:9090 prom/prometheus

LOGIN_COUNT = Counter(
 'user_login_total', # 指標名稱 (Metric Name)
 '用戶登錄總次數(shù)及狀態(tài)統(tǒng)計', # 指標描述 (Help Text)
 ['user_type', 'login_status', 'ip_region'] # 標簽/維度 (Labels/Dimensions)
 )

pip install prometheus-client

from prometheus_client import start_http_server, Counter
 import random
 import time

 # 定義一個計數(shù)器類型的指標（累計值，適合記錄次數(shù)）
 # 標簽說明：
 # - user_type: 用戶類型（普通用戶/管理員）
 # - login_status: 登錄狀態(tài)（成功/失敗）
 # - ip_region: IP所屬地區(qū)（模擬分布式用戶場景）
 LOGIN_COUNT = Counter(
 'user_login_total', # 指標名（遵循Prometheus命名規(guī)范：小寫+下劃線）
 '用戶登錄總次數(shù)及狀態(tài)統(tǒng)計', # 指標描述
 ['user_type', 'login_status', 'ip_region'] # 標簽（用于多維度篩選）
 )

 def simulate_login():
 """模擬用戶登錄行為（每3秒產(chǎn)生一次登錄事件）"""
 # 模擬用戶類型（70%普通用戶，30%管理員）
 user_type = random.choices(['normal', 'admin'], weights=[0.7, 0.3])[0]
 
 # 模擬登錄狀態(tài)（90%成功，10%失敗）
 login_status = random.choices(['success', 'failed'], weights=[0.9, 0.1])[0]
 
 # 模擬IP地區(qū)（國內(nèi)主要城市）
 ip_region = random.choice(['beijing', 'shanghai', 'guangzhou', 'shenzhen', 'hangzhou'])
 
 # 記錄一次登錄事件（計數(shù)器+1）
 LOGIN_COUNT.labels(
 user_type=user_type,
 login_status=login_status,
 ip_reginotallow=ip_region
 ).inc() # 每次調(diào)用+1

 if __name__ == '__main__':
 # 啟動HTTP服務(wù)，暴露指標在9091端口
 start_http_server(9091)
 print("用戶登錄監(jiān)控Exporter運行在 http://localhost:9091/metrics")
 
 # 持續(xù)模擬登錄事件
 while True:
 simulate_login()
 time.sleep(3) # 每3秒模擬一次登錄

def simulate_login():
 # 隨機生成用戶類型（70%普通用戶，30%管理員）
 user_type = random.choices(['normal', 'admin'], weights=[0.7, 0.3])[0]
 
 # 隨機生成登錄結(jié)果（90%成功，10%失敗）
 login_status = random.choices(['success', 'failed'], weights=[0.9, 0.1])[0]
 
 # 隨機選擇IP地區(qū)（國內(nèi)主要城市）
 ip_region = random.choice(['beijing', 'shanghai', 'guangzhou', 'shenzhen', 'hangzhou'])
 
 # 記錄登錄事件
 LOGIN_COUNT.labels(
 user_type=user_type,
 login_status=login_status,
 ip_reginotallow=ip_region
 ).inc() # 計數(shù)器+1

LOGIN_COUNT = Counter(
 'user_login_total', # 指標名稱
 '用戶登錄總次數(shù)及狀態(tài)統(tǒng)計', # 指標描述
 ['user_type', 'login_status', 'ip_region'] # 三維度標簽
 )

if __name__ == '__main__':
 start_http_server(9091) # 啟動指標暴露服務(wù)
 print("服務(wù)運行在 http://localhost:9091/metrics")
 
 while True:
 simulate_login()
 time.sleep(3)

python ~/docker/prometheus/custom_exporter.py

Exporter運行在 http://localhost:9091/metrics

# HELP python_gc_objects_collected_total Objects collected during gc
 # TYPE python_gc_objects_collected_total counter
 python_gc_objects_collected_total{generatinotallow="0"} 255.0
 python_gc_objects_collected_total{generatinotallow="1"} 118.0
 python_gc_objects_collected_total{generatinotallow="2"} 0.0
 # HELP python_gc_objects_uncollectable_total Uncollectable object found during GC
 # TYPE python_gc_objects_uncollectable_total counter
 python_gc_objects_uncollectable_total{generatinotallow="0"} 0.0
 python_gc_objects_uncollectable_total{generatinotallow="1"} 0.0
 python_gc_objects_uncollectable_total{generatinotallow="2"} 0.0
 # HELP python_gc_collections_total Number of times this generation was collected
 # TYPE python_gc_collections_total counter
 python_gc_collections_total{generatinotallow="0"} 40.0
 python_gc_collections_total{generatinotallow="1"} 3.0
 python_gc_collections_total{generatinotallow="2"} 0.0
 # HELP python_info Python platform information
 # TYPE python_info gauge
 python_info{implementatinotallow="CPython",major="3",minor="9",patchlevel="18",versinotallow="3.9.18"} 1.0
 # HELP my_custom_metric 自定義測試指標
 # TYPE my_custom_metric gauge
 my_custom_metric{label1="test",label2="normal"} 1.3487897665970805

global:
 scrape_interval: 5s # 每15秒拉取一次數(shù)據(jù)

 scrape_configs:
 # 監(jiān)控Prometheus自身
 - job_name: 'prometheus'
 static_configs:
 - targets: ['localhost:9090']

 # 監(jiān)控本地的自定義Exporter（關(guān)鍵配置）
 - job_name: 'custom_exporter'
 static_configs:
 # 這里的IP不是localhost，而是你本地機器的實際IP（容器內(nèi)訪問宿主機的地址）
 # 如何獲取：在本地終端執(zhí)行 `ifconfig` 或 `ip addr` 查看（例如192.168.x.x）
 - targets: ['192.168.0.8:9091'] # 你的本地IP + Exporter端口

docker stop prometheus
 docker rm prometheus

docker run --name prometheus -d -p 127.0.0.1:9090:9090 -v /Users/cuihao/docker/prometheus:/etc/prometheus prom/prometheus --config.file=/etc/prometheus/prometheus.yml

user_login_total

groups:
 - name: login_anomaly_rules # 規(guī)則組名稱
 rules:
 # 規(guī)則1：5分鐘內(nèi)登錄失敗率超過20%（可能是攻擊或系統(tǒng)故障）
 - alert: HighLoginFailureRate
 expr: |
 sum(rate(user_login_total{login_status="failed"}[5m])) 
 / 
 sum(rate(user_login_total[5m])) 
 > 0.2 # 失敗率>20%
 for: 1m # 持續(xù)1分鐘觸發(fā)告警
 labels:
 severity: critical # 告警級別（緊急）
 annotations:
 summary: "登錄失敗率過高"
 description: "過去5分鐘登錄失敗率{{ $value | humanizePercentage }}，可能存在異常登錄行為"

 # 規(guī)則2：管理員賬號1分鐘內(nèi)登錄失敗次數(shù)>5次（可能被暴力破解）
 - alert: AdminLoginFailureSpike
 expr: |
 sum(rate(user_login_total{user_type="admin", login_status="failed"}[1m])) 
 > 5 # 1分鐘內(nèi)失敗>5次
 for: 30s # 持續(xù)30秒觸發(fā)
 labels:
 severity: warning
 annotations:
 summary: "管理員登錄失敗次數(shù)突增"
 description: "管理員賬號1分鐘內(nèi)登錄失敗{{ $value }}次，可能存在暴力破解風險"

global:
 scrape_interval: 15s

 rule_files:
 - "alert_rules.yml" # 加載剛才創(chuàng)建的告警規(guī)則文件

 scrape_configs:
 - job_name: 'custom_exporter'
 static_configs:
 - targets: ['192.168.0.16:9091'] # 你的Exporter地址

docker restart prometheus

# 先創(chuàng)建Alertmanager配置目錄
 mkdir -p /Users/cuihao/docker/prometheus/alertmanager

 # 啟動容器（映射9093端口，掛載配置目錄）
 docker run --name alertmanager -d -p 127.0.0.1:9093:9093 -v /Users/cuihao/docker/prometheus/alertmanager:/etc/alertmanager prom/alertmanager

route:
 group_by: ['alertname'] # 按告警名稱分組
 group_wait: 10s # 組內(nèi)等待10秒再發(fā)送
 group_interval: 10s # 組內(nèi)間隔10秒發(fā)送
 repeat_interval: 1h # 重復(fù)告警間隔1小時
 receiver: 'webhook_handler' # 轉(zhuǎn)發(fā)到名為webhook_handler的接收者

 receivers:
 - name: 'webhook_handler'
 webhook_configs:
 - url: 'http://192.168.0.8:5008/alert' # 你的處理服務(wù)地址（本地5000端口）
 send_resolved: true # 告警解決后也發(fā)送通知

docker restart alertmanager

docker stop prometheus alertmanager

docker network create prometheus-network

# 將Prometheus加入網(wǎng)絡(luò)
 docker network connect prometheus-network prometheus

 # 將Alertmanager加入網(wǎng)絡(luò)
 docker network connect prometheus-network alertmanager

docker start prometheus alertmanager

# 進入Prometheus容器
 docker exec -it prometheus /bin/sh

 # 測試DNS解析（若容器內(nèi)有nslookup）
 nslookup alertmanager

 # 若沒有nslookup，用ping測試（能ping通說明解析正常）
 ping alertmanager

alerting:
 alertmanagers:
 - static_configs:
 - targets:
 - 'alertmanager:9093'

docker restart prometheus

from flask import Flask, request, jsonify
 import os
 import sys
 import datetime
 import json
 from openai import OpenAI
 from dotenv import load_dotenv # 用于加載環(huán)境變量

 # 初始化Flask應(yīng)用
 app = Flask(__name__)

 # 加載環(huán)境變量（從.env文件讀取DEEPSEEK_API_KEY）
 try:
 # 嘗試加載環(huán)境變量，并添加日志
 load_result = load_dotenv() # 默認讀取當前目錄的.env文件
 script_dir = os.path.dirname(os.path.abspath(__file__))
 env_path = os.path.join(script_dir, ".env")
 
 # 配置日志和報告路徑（使用腳本所在目錄）
 error_log = os.path.join(script_dir, "error.log") # 錯誤日志路徑
 report_dir = os.path.join(script_dir, "alert_reports") # 分析報告保存目錄
 
 # 記錄環(huán)境變量加載情況
 with open(error_log, "a", encoding="utf-8") as f:
 if load_result:
 f.write(f"[{datetime.datetime.now()}] 成功加載環(huán)境變量文件: {env_path}\n")
 else:
 f.write(f"[{datetime.datetime.now()}] 未找到環(huán)境變量文件，使用系統(tǒng)環(huán)境變量: {env_path}\n")
 
 # 確保報告目錄存在
 os.makedirs(report_dir, exist_ok=True)
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 報告目錄準備就緒: {report_dir}\n")
 except Exception as e:
 # 記錄初始化錯誤
 error_msg = f"初始化文件系統(tǒng)時出錯: {str(e)}"
 print(f"[{datetime.datetime.now()}] {error_msg}") # 控制臺也輸出，方便調(diào)試
 # 嘗試寫入錯誤日志
 try:
 with open("error.log", "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] {error_msg}\n")
 except:
 pass
 raise # 拋出錯誤，終止啟動

 def init_deepseek_client():
 """初始化DeepSeek客戶端（兼容OpenAI SDK）"""
 try:
 # 驗證環(huán)境變量
 api_key = os.environ.get("DEEPSEEK_API_KEY")
 with open(error_log, "a", encoding="utf-8") as f:
 if api_key:
 f.write(f"[{datetime.datetime.now()}] 成功獲取DEEPSEEK_API_KEY (部分隱藏): {api_key[:4]}****\n")
 else:
 f.write(f"[{datetime.datetime.now()}] 未在環(huán)境變量中找到DEEPSEEK_API_KEY\n")
 
 if not api_key:
 raise ValueError("未在環(huán)境變量中找到DEEPSEEK_API_KEY，請檢查.env文件")
 
 # 初始化客戶端（DeepSeek兼容OpenAI SDK格式）
 client = OpenAI(
 api_key=api_key,
 base_url="https://api.deepseek.com/v1" # DeepSeek API基礎(chǔ)地址
 )
 
 # 驗證客戶端連接
 try:
 # 發(fā)送一個簡單的測試請求驗證連接
 response = client.chat.completions.create(
 model="deepseek-chat",
 messages=[{"role": "user", "content": "test"}],
 max_tokens=1
 )
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] DeepSeek客戶端初始化成功，API連接正常\n")
 except Exception as e:
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] DeepSeek API連接測試失敗: {str(e)}\n")
 raise
 
 return client
 except Exception as e:
 # 記錄初始化錯誤
 error_msg = f"DeepSeek客戶端初始化失敗：{str(e)}"
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] {error_msg}\n")
 raise # 拋出錯誤，避免服務(wù)啟動后無法調(diào)用API

 # 初始化DeepSeek客戶端（服務(wù)啟動時執(zhí)行）
 try:
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 開始初始化DeepSeek客戶端...\n")
 client = init_deepseek_client()
 except Exception as e:
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 初始化DeepSeek客戶端失敗，服務(wù)無法啟動：{str(e)}\n")
 raise


 def call_deepseek_analysis(alert_info):
 """調(diào)用DeepSeek分析告警信息，返回分析結(jié)果和報告路徑"""
 try:
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 開始處理告警分析...\n")
 
 # 1. 驗證并提取告警關(guān)鍵信息
 if not isinstance(alert_info, dict):
 raise ValueError(f"告警信息格式錯誤，預(yù)期字典類型，實際為: {type(alert_info)}")
 
 # 驗證必要字段
 required_fields = ["labels", "annotations"]
 for field in required_fields:
 if field not in alert_info:
 raise ValueError(f"告警信息缺少必要字段: {field}")
 
 alert_name = alert_info["labels"].get("alertname", "未知告警")
 severity = alert_info["labels"].get("severity", "未知級別")
 description = alert_info["annotations"].get("description", "無描述")
 start_time = alert_info.get("startsAt", datetime.datetime.now().isoformat())
 labels = alert_info["labels"]
 
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 提取到告警信息 - 名稱: {alert_name}, 級別: {severity}\n")

 # 2. 構(gòu)造提示詞（系統(tǒng)角色+用戶輸入）
 system_prompt = """

""" user_prompt = f"""

- 告警名稱：{alert_name}
 - 嚴重級別：{severity}
 - 告警描述：{description}
 - 發(fā)生時間：{start_time}
 - 關(guān)聯(lián)標簽：{json.dumps(labels, ensure_ascii=False)} # 包含用戶類型、地區(qū)等維度
 
 請基于以上信息分析并生成報告。
 """

 # 3. 調(diào)用DeepSeek模型
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 開始調(diào)用DeepSeek API進行分析...\n")
 
 response = client.chat.completions.create(
 model="deepseek-chat", # 使用的模型名稱
 messages=[
 {"role": "system", "content": system_prompt},
 {"role": "user", "content": user_prompt}
 ],
 stream=False,
 max_tokens=1000 # 限制最大返回長度
 )

 # 4. 提取分析結(jié)果
 if not response.choices or len(response.choices) == 0:
 raise ValueError("DeepSeek API返回結(jié)果為空")
 
 analysis_content = response.choices[0].message.content
 
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 成功獲取DeepSeek分析結(jié)果，長度: {len(analysis_content)}\n")

 # 5. 生成分析報告文件
 timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
 report_filename = f"analysis_{alert_name}_{timestamp}.txt".replace(" ", "_")
 report_path = os.path.join(report_dir, report_filename)
 
 with open(report_path, "w", encoding="utf-8") as f:
 f.write(f"=== 告警分析報告 ===\n")
 f.write(f"生成時間：{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
 f.write(f"告警名稱：{alert_name}\n")
 f.write(f"嚴重級別：{severity}\n")
 f.write(f"發(fā)生時間：{start_time}\n\n")
 f.write("=== 分析內(nèi)容 ===\n")
 f.write(analysis_content)
 
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 分析報告已保存至: {report_path}\n")

 return analysis_content, report_path

 except Exception as e:
 # 記錄調(diào)用錯誤
 error_msg = f"DeepSeek分析調(diào)用失敗：{str(e)}"
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] {error_msg}\n")
 return f"分析失敗：{str(e)}", None


 @app.route('/alert', methods=['POST'])
 def handle_alert():
 """接收Alertmanager的Webhook告警，調(diào)用分析并返回結(jié)果"""
 try:
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 收到新的告警請求\n")
 
 # 1. 解析Alertmanager發(fā)送的JSON數(shù)據(jù)
 try:
 alert_data = request.json
 except Exception as e:
 raise ValueError(f"解析JSON數(shù)據(jù)失敗: {str(e)}")
 
 if not alert_data:
 return jsonify({"status": "failed", "reason": "未收到有效數(shù)據(jù)"}), 400
 
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 成功解析告警數(shù)據(jù)，包含字段: {', '.join(alert_data.keys())}\n")

 # 2. 提取第一個觸發(fā)的告警
 if not alert_data.get("alerts"):
 return jsonify({"status": "failed", "reason": "告警數(shù)據(jù)中無alerts字段"}), 400
 
 if not isinstance(alert_data["alerts"], list) or len(alert_data["alerts"]) == 0:
 return jsonify({"status": "failed", "reason": "alerts字段不是有效的列表或為空"}), 400
 
 first_alert = alert_data["alerts"][0]
 alert_name = first_alert['labels'].get('alertname', '未知告警')
 print(f"[{datetime.datetime.now()}] 收到告警：{alert_name}") # 控制臺輸出
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 處理第一個告警: {alert_name}\n")

 # 3. 調(diào)用DeepSeek分析
 analysis_content, report_path = call_deepseek_analysis(first_alert)

 # 4. 返回結(jié)果
 return jsonify({
 "status": "success",
 "analysis": analysis_content,
 "report_path": report_path
 })

 except Exception as e:
 # 記錄請求處理錯誤
 error_msg = f"告警處理失敗：{str(e)}"
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] {error_msg}\n")
 return jsonify({"status": "failed", "reason": str(e)}), 500

 # 添加一個健康檢查接口，方便測試服務(wù)是否正常運行
 @app.route('/health', methods=['GET'])
 def health_check():
 try:
 return jsonify({
 "status": "healthy",
 "timestamp": datetime.datetime.now().isoformat(),
 "services": {
 "deepseek": "initialized" if 'client' in globals() else "not initialized",
 "report_dir": report_dir,
 "report_dir_writable": os.access(report_dir, os.W_OK)
 }
 })
 except Exception as e:
 return jsonify({
 "status": "unhealthy",
 "reason": str(e)
 }), 500


 if __name__ == '__main__':
 try:
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] 開始啟動alert_handler服務(wù)...\n")
 f.write(f"[{datetime.datetime.now()}] 服務(wù)將監(jiān)聽端口: 5008\n")
 
 # 啟動服務(wù)（監(jiān)聽所有IP的5008端口）
 app.run(host='0.0.0.0', port=5008, debug=True)
 except Exception as e:
 error_msg = f"服務(wù)啟動失敗：{str(e)}"
 print(f"[{datetime.datetime.now()}] {error_msg}")
 with open(error_log, "a", encoding="utf-8") as f:
 f.write(f"[{datetime.datetime.now()}] {error_msg}\n")
 sys.exit(1)

python alert_handler.py

curl -X POST "http://localhost:5008/alert" \
 -H "Content-Type: application/json" \
 -d '{
 "version": "4",
 "groupKey": "{}:{alertname=\"HighLoginFailureRate\"}",
 "status": "firing",
 "receiver": "webhook_handler",
 "groupLabels": {
 "alertname": "HighLoginFailureRate"
 },
 "commonLabels": {
 "alertname": "HighLoginFailureRate",
 "severity": "critical"
 },
 "commonAnnotations": {
 "description": "登錄失敗率過高，當前失敗率為30%，超過閾值20%",
 "summary": "登錄失敗率異常升高"
 },
 "externalURL": "http://localhost:9093",
 "alerts": [
 {
 "status": "firing",
 "labels": {
 "alertname": "HighLoginFailureRate",
 "severity": "critical"
 },
 "annotations": {
 "description": "登錄失敗率過高，當前失敗率為30%，超過閾值20%",
 "summary": "登錄失敗率異常升高"
 },
 "startsAt": "2025-08-14T08:30:00.000Z",
 "endsAt": "0001-01-01T00:00:00Z",
 "generatorURL": "http://localhost:9090/graph?g0.expr=sum(rate(user_login_total%7Blogin_status%3D%22failed%22%7D%5B5m%5D))+%2F+sum(rate(user_login_total%5B5m%5D))+%3E+0.2&g0.tab=1"
 }
 ]
 }'