老杨跟几个大厂的朋友聊了一下AI在运维领域他们咋玩的

# 检查系统负载
uptime
 09:15:01 up 45 days,  2:31,  2 users,  load average: 0.15, 0.18, 0.12

# 检查磁盘使用率
df -h
Filesystem      Size  Used Avail Use% Mounted on
/dev/vda1        40G   28G   10G  74% /
/dev/vdb1       100G   85G   10G  90% /data

# 检查内存使用
free -m
              total        used        free      shared  buff/cache   available
Mem:           7821        3256        1234         125        3331        4142

# 安装node_exporter
wget https://github.com/prometheus/node_exporter/releases/download/v1.6.1/node_exporter-1.6.1.linux-amd64.tar.gz
tar xvfz node_exporter-1.6.1.linux-amd64.tar.gz
./node_exporter --web.listen-address=":9100"

import pandas as pd
from sklearn.ensemble import IsolationForest

# 读取监控数据
data = pd.read_csv('server_metrics.csv')

# 训练异常检测模型
model = IsolationForest(contamination=0.1)
model.fit(data[['cpu_usage', 'memory_usage', 'disk_usage']])

# 预测异常
anomalies = model.predict(data[['cpu_usage', 'memory_usage', 'disk_usage']])

# 查找错误日志
grep -i "error" /var/log/application.log
2024-09-01 09:15:23 ERROR: Database connection failed
2024-09-01 09:16:45 ERROR: API timeout after 30 seconds
...

# 统计错误数量
grep -c "ERROR" /var/log/application.log
1248

# docker-compose.yml
version: '3'
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.8.0
    environment:
      - discovery.type=single-node
    ports:
      - "9200:9200"
  
  logstash:
    image: docker.elastic.co/logstash/logstash:8.8.0
    volumes:
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    ports:
      - "5044:5044"
  
  kibana:
    image: docker.elastic.co/kibana/kibana:8.8.0
    ports:
      - "5601:5601"

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
import jieba

# 中文日志分词
def tokenize_chinese(text):
    return ' '.join(jieba.cut(text))

# 日志聚类分析
logs = ["数据库连接超时", "API调用失败", "内存不足警告", ...]
logs_tokenized = [tokenize_chinese(log) for log in logs]

vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(logs_tokenized)

kmeans = KMeans(n_clusters=5)
clusters = kmeans.fit_predict(X)

import numpy as np
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_squared_error

# 构造时间序列特征
def create_features(data, window_size=24):
    features = []
    targets = []
    
    for i in range(window_size, len(data)):
        features.append(data[i-window_size:i])
        targets.append(data[i])
    
    return np.array(features), np.array(targets)

# 训练预测模型
cpu_data = [12.3, 15.6, 18.9, ...]  # CPU使用率时间序列
X, y = create_features(cpu_data)

model = RandomForestRegressor(n_estimators=100)
model.fit(X, y)

# 预测未来1小时的CPU使用率
future_cpu = model.predict(X[-1:])
print(f"预计1小时后CPU使用率: {future_cpu[0]:.2f}%")

# 基于历史数据预测流量增长
import pandas as pd
from prophet import Prophet

# 历史流量数据
df = pd.DataFrame({
    'ds': ['2024-01-01', '2024-01-02', ...],
    'y': [1000, 1200, 1150, ...]  # 每日QPS
})

# 训练Prophet模型
model = Prophet()
model.fit(df)

# 预测未来30天流量
future = model.make_future_dataframe(periods=30)
forecast = model.predict(future)

# 计算所需服务器数量
max_qps = forecast['yhat'].max()
servers_needed = int(max_qps / 1000) + 1  # 假设每台服务器能处理1000 QPS

#!/bin/bash
# 自动重启异常服务

SERVICE="nginx"
if ! systemctl is-active --quiet $SERVICE; then
    echo "$(date): $SERVICE is down, restarting..."
    systemctl restart $SERVICE
    sleep 5
    
    if systemctl is-active --quiet $SERVICE; then
        echo "$(date): $SERVICE restarted successfully"
    else
        echo "$(date): Failed to restart $SERVICE, alerting admin"
        # 发送告警
    fi
fi

import gym
from stable_baselines3 import PPO

# 定义运维环境
class OpsEnvironment(gym.Env):
    def __init__(self):
        self.action_space = gym.spaces.Discrete(4)  # 重启、扩容、降级、不处理
        self.observation_space = gym.spaces.Box(low=0, high=100, shape=(3,))  # CPU、内存、磁盘
    
    def step(self, action):
        # 执行动作并返回奖励
        if action == 0:  # 重启服务
            reward = self._restart_service()
        elif action == 1:  # 自动扩容
            reward = self._scale_up()
        # ...
        
        return self._get_state(), reward, done, info

# 训练智能运维代理
env = OpsEnvironment()
model = PPO("MlpPolicy", env, verbose=1)
model.learn(total_timesteps=10000)

# 使用训练好的模型
obs = env.reset()
action, _states = model.predict(obs)

from sklearn.ensemble import GradientBoostingClassifier
import pandas as pd

# 历史告警数据
alert_data = pd.DataFrame({
    'cpu_usage': [85, 92, 78, 95, ...],
    'memory_usage': [75, 88, 82, 91, ...], 
    'error_rate': [0.1, 0.5, 0.2, 1.2, ...],
    'is_critical': [0, 1, 0, 1, ...]  # 是否是关键告警
})

# 训练告警分类模型
X = alert_data[['cpu_usage', 'memory_usage', 'error_rate']]
y = alert_data['is_critical']

model = GradientBoostingClassifier()
model.fit(X, y)

# 实时告警分级
def classify_alert(cpu, memory, error_rate):
    prob = model.predict_proba([[cpu, memory, error_rate]])[0][1]
    if prob > 0.8:
        return "紧急"
    elif prob > 0.5:
        return "重要"
    else:
        return "一般"

from mlxtend.frequent_patterns import apriori, association_rules

# 告警关联分析
alerts_df = pd.DataFrame({
    'database_slow': [1, 1, 0, 1, ...],
    'api_timeout': [1, 1, 0, 1, ...],
    'high_cpu': [0, 1, 1, 1, ...],
})

# 找出频繁告警组合
frequent_itemsets = apriori(alerts_df, min_support=0.3, use_colnames=True)
rules = association_rules(frequent_itemsets, metric="confidence", min_threshold=0.8)

print(rules[['antecedents', 'consequents', 'confidence']])