— 创建模型训练脚本
[root@fgedu ~]# vi /usr/local/bin/train_model.py
#!/usr/bin/env python3
# train_model.py
# from:www.itpux.com.qq113257174.wx:itpux-com
# web: http://www.fgedu.net.cn
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import pickle
# 加载训练数据
def load_training_data():
# 从Prometheus获取历史数据
# 这里使用模拟数据
np.random.seed(42)
normal_data = np.random.normal(50, 10, 1000)
anomaly_data = np.random.normal(150, 20, 50)
data = np.concatenate([normal_data, anomaly_data])
return data
# 训练模型
def train_model(data, contamination=0.05):
scaler = StandardScaler()
data_scaled = scaler.fit_transform(data.reshape(-1, 1))
model = IsolationForest(contamination=contamination, random_state=42)
model.fit(data_scaled)
return model, scaler
# 保存模型
def save_model(model, scaler, model_path=’model.pkl’, scaler_path=’scaler.pkl’):
with open(model_path, ‘wb’) as f:
pickle.dump(model, f)
with open(scaler_path, ‘wb’) as f:
pickle.dump(scaler, f)
# 主函数
def main():
# 加载训练数据
data = load_training_data()
# 训练模型
model, scaler = train_model(data)
# 保存模型
save_model(model, scaler)
print(‘模型训练完成并保存’)
if __name__ == ‘__main__’:
main()
— 执行模型训练
[root@fgedu ~]# python3 /usr/local/bin/train_model.py
— 输出结果
模型训练完成并保存

部署异常检测服务：

— 创建异常检测服务脚本
[root@fgedu ~]# vi /usr/local/bin/anomaly_detection_service.py
#!/usr/bin/env python3
# anomaly_detection_service.py
# from:www.itpux.com.qq113257174.wx:itpux-com
# web: http://www.fgedu.net.cn
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import pickle
import requests
import json
from datetime import datetime, timedelta
import time
# 加载模型
def load_model(model_path=’model.pkl’, scaler_path=’scaler.pkl’):
with open(model_path, ‘rb’) as f:
model = pickle.load(f)
with open(scaler_path, ‘rb’) as f:
scaler = pickle.load(f)
return model, scaler
# 获取Prometheus数据
def get_prometheus_data(query, start_time, end_time):
url = ‘http://192.168.1.100:9090/api/v1/query_range’
params = {
‘query’: query,
‘start’: start_time,
‘end’: end_time,
‘step’: ’60s’
}
response = requests.get(url, params=params)
data = response.json()
return data
# 处理Prometheus数据
def process_prometheus_data(data):
values = []
for result in data[‘data’][‘result’]:
for value in result[‘values’]:
values.append(float(value[1]))
return np.array(values)
# 异常检测
def detect_anomalies(model, scaler, data):
data_scaled = scaler.transform(data.reshape(-1, 1))
anomalies = model.predict(data_scaled)
return anomalies
# 发送告警
def send_alert(message):
# 发送邮件告警
import smtplib
from email.mime.text import MIMEText
msg = MIMEText(message)
msg[‘Subject’] = ‘DM数据库异常检测告警’
msg[‘From’] = ‘alert@fgedu.net.cn’
msg[‘To’] = ‘admin@fgedu.net.cn’
smtp = smtplib.SMTP(‘smtp.fgedu.net.cn’, 587)
smtp.send_message(msg)
smtp.quit()
# 主函数
def main():
# 加载模型
model, scaler = load_model()
while True:
# 获取过去5分钟的数据
end_time = datetime.now()
start_time = end_time – timedelta(minutes=5)
# 获取连接数数据
sessions_data = get_prometheus_data(
‘dm_sessions’,
start_time.isoformat(),
end_time.isoformat()
)
sessions_values = process_prometheus_data(sessions_data)
# 检测连接数异常
sessions_anomalies = detect_anomalies(model, scaler, sessions_values)
# 如果检测到异常，发送告警
if -1 in sessions_anomalies:
anomaly_indices = np.where(sessions_anomalies == -1)[0]
for idx in anomaly_indices:
message = f’检测到DM数据库连接数异常，时间点：{idx}，连接数：{sessions_values[idx]}’
print(message)
send_alert(message)
# 等待5分钟
time.sleep(300)
if __name__ == ‘__main__’:
main()
— 设置执行权限
[root@fgedu ~]# chmod +x /usr/local/bin/anomaly_detection_service.py
— 创建异常检测服务文件
[root@fgedu ~]# vi /etc/systemd/system/anomaly-detection.service
[Unit]
Description=Anomaly Detection Service
After=network.target
[Service]
Type=simple
User=root
ExecStart=/usr/bin/python3 /usr/local/bin/anomaly_detection_service.py
Restart=on-failure
[Install]
WantedBy=multi-user.target
— 启动异常检测服务
[root@fgedu ~]# systemctl daemon-reload
[root@fgedu ~]# systemctl start anomaly-detection
[root@fgedu ~]# systemctl enable anomaly-detection