分布式系统智能可观测性建设实战

分布式系统可观测性的三个维度
┌─────────────────────────┐     ┌─────────────────────────┐
│ 日志(Logging)           │────▶│ 指标(Metrics)           │
└─────────────────────────┘     └─────────────────────────┘
        ▲                                  ▲
        │                                  │
        │                                  │
┌─────────────────────────┐     ┌─────────────────────────┐
│ 链路追踪(Tracing)       │◀────│ 智能分析与可视化        │
└─────────────────────────┘     └─────────────────────────┘

分布式系统的复杂性
┌─────────────────────┐     ┌─────────────────────┐     ┌─────────────────────┐
│ 服务A               │────▶│ 服务B               │────▶│ 服务C               │
└─────────────────────┘     └─────────────────────┘     └─────────────────────┘
        │                          │                          │
        ▼                          ▼                          ▼
┌─────────────────────┐     ┌─────────────────────┐     ┌─────────────────────┐
│ 服务D               │────▶│ 服务E               │◀────│ 服务F               │
└─────────────────────┘     └─────────────────────┘     └─────────────────────┘

# 使用OpenTelemetry进行分布式追踪的示例代码
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.instrumentation.requests import RequestsInstrumentor

# 初始化追踪提供者
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

# 添加处理器
exporter = ConsoleSpanExporter()
processor = BatchSpanProcessor(exporter)
trace.get_tracer_provider().add_span_processor(processor)

# 自动检测HTTP请求
RequestsInstrumentor().instrument()

# 手动创建和使用span
def service_a():
    with tracer.start_as_current_span("service_a_operation") as span:
        # 添加属性
        span.set_attribute("http.method", "GET")
        span.set_attribute("http.url", "/api/service-a")
        
        # 调用其他服务
        result = service_b()
        
        # 添加事件
        span.add_event("service_a_operation_completed", {"result": result})
        
        return result

def service_b():
    with tracer.start_as_current_span("service_b_operation") as span:
        # 模拟服务调用
        import requests
        response = requests.get("https://example.com")
        
        span.set_attribute("external.http.status_code", response.status_code)
        
        return response.status_code

# 执行服务
service_a()

可观测性数据存储架构
┌─────────────────────┐     ┌─────────────────────┐     ┌─────────────────────┐
│ 数据采集层           │────▶│ 数据处理层           │────▶│ 数据存储层           │
└─────────────────────┘     └─────────────────────┘     └─────────────────────┘
                                                              │
                                                              ▼
┌─────────────────────┐     ┌─────────────────────┐     ┌─────────────────────┐
│ 应用层              │◀────│ 分析引擎            │◀────│ 查询引擎             │
└─────────────────────┘     └─────────────────────┘     └─────────────────────┘

# 使用Elasticsearch进行多维度数据关联分析的示例代码
from elasticsearch import Elasticsearch
from elasticsearch_dsl import Search

# 连接Elasticsearch
es = Elasticsearch(["http://localhost:9200"])

# 创建搜索对象
s = Search(using=es, index=["logs-*", "metrics-*", "traces-*"])

# 多维度数据关联查询
def search_correlated_data(trace_id):
    # 构建查询
    s = s.query("bool", 
        must=[
            {"term": {"trace_id": trace_id}}
        ],
        filter=[
            {"range": {"@timestamp": {"gte": "now-1h", "lt": "now"}}}
        ]
    )
    
    # 执行查询
    response = s.execute()
    
    # 处理结果
    correlated_data = {
        "logs": [],
        "metrics": [],
        "traces": []
    }
    
    for hit in response:
        if hit.meta.index.startswith("logs-"):
            correlated_data["logs"].append({
                "timestamp": hit["@timestamp"],
                "message": hit["message"],
                "service": hit.get("service", "unknown"),
                "level": hit.get("level", "info")
            })
        elif hit.meta.index.startswith("metrics-"):
            correlated_data["metrics"].append({
                "timestamp": hit["@timestamp"],
                "name": hit["metric_name"],
                "value": hit["value"],
                "service": hit.get("service", "unknown")
            })
        elif hit.meta.index.startswith("traces-"):
            correlated_data["traces"].append({
                "timestamp": hit["@timestamp"],
                "span_name": hit["span_name"],
                "duration": hit["duration"],
                "service": hit.get("service", "unknown")
            })
    
    return correlated_data

# 查询特定trace_id的关联数据
correlated_data = search_correlated_data("abc123xyz456")
print("关联日志数量:", len(correlated_data["logs"]))
print("关联指标数量:", len(correlated_data["metrics"]))
print("关联链路数量:", len(correlated_data["traces"]))

# 使用NetworkX构建和可视化服务依赖图的示例代码
import networkx as nx
import matplotlib.pyplot as plt

# 创建服务依赖图
def build_service_dependency_graph(traces_data):
    G = nx.DiGraph()
    
    # 提取服务依赖关系
    for trace in traces_data:
        # 假设每个trace包含调用链信息
        if "call_chain" in trace and len(trace["call_chain"]) > 1:
            for i in range(len(trace["call_chain"]) - 1):
                source = trace["call_chain"][i]["service"]
                target = trace["call_chain"][i+1]["service"]
                
                # 添加节点
                if source not in G.nodes:
                    G.add_node(source)
                if target not in G.nodes:
                    G.add_node(target)
                
                # 添加或更新边
                if G.has_edge(source, target):
                    G[source][target]["weight"] += 1
                else:
                    G.add_edge(source, target, weight=1)
    
    return G

# 可视化服务依赖图
def visualize_service_dependency_graph(G):
    plt.figure(figsize=(12, 8))
    
    # 使用spring布局
    pos = nx.spring_layout(G, k=0.3)
    
    # 获取边权重作为宽度
    edges = G.edges()
    weights = [G[u][v]["weight"] for u, v in edges]
    
    # 绘制节点和边
    nx.draw_networkx_nodes(G, pos, node_size=500, node_color="lightblue")
    nx.draw_networkx_edges(G, pos, edgelist=edges, width=weights, edge_color="gray")
    nx.draw_networkx_labels(G, pos, font_size=10)
    
    # 添加边权重标签
    edge_labels = {(u, v): G[u][v]["weight"] for u, v in edges}
    nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels, font_size=8)
    
    plt.title("Service Dependency Graph")
    plt.axis("off")
    plt.tight_layout()
    plt.show()

# 示例数据（实际使用时应从追踪系统获取）
sample_traces = [
    {"call_chain": [{"service": "api-gateway"}, {"service": "user-service"}, {"service": "database"}]},
    {"call_chain": [{"service": "api-gateway"}, {"service": "order-service"}, {"service": "payment-service"}]},
    {"call_chain": [{"service": "api-gateway"}, {"service": "product-service"}, {"service": "database"}]},
    {"call_chain": [{"service": "api-gateway"}, {"service": "user-service"}, {"service": "cache"}]},
    {"call_chain": [{"service": "api-gateway"}, {"service": "order-service"}, {"service": "notification-service"}]},
]

# 构建和可视化服务依赖图
service_graph = build_service_dependency_graph(sample_traces)
visualize_service_dependency_graph(service_graph)

# 使用LSTM进行服务性能异常检测的示例代码
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
from sklearn.preprocessing import MinMaxScaler

# 准备数据
def prepare_data(data, look_back=60):
    # 数据标准化
    scaler = MinMaxScaler(feature_range=(0, 1))
    scaled_data = scaler.fit_transform(data.reshape(-1, 1))
    
    # 创建训练数据
    X, Y = [], []
    for i in range(look_back, len(scaled_data)):
        X.append(scaled_data[i-look_back:i, 0])
        Y.append(scaled_data[i, 0])
    
    # 转换为numpy数组并调整形状
    X = np.array(X)
    Y = np.array(Y)
    X = np.reshape(X, (X.shape[0], X.shape[1], 1))
    
    return X, Y, scaler

# 创建LSTM模型
def create_lstm_model(input_shape):
    model = Sequential()
    model.add(LSTM(units=50, return_sequences=True, input_shape=input_shape))
    model.add(Dropout(0.2))
    model.add(LSTM(units=50, return_sequences=True))
    model.add(Dropout(0.2))
    model.add(LSTM(units=50))
    model.add(Dropout(0.2))
    model.add(Dense(units=1))
    
    model.compile(optimizer='adam', loss='mean_squared_error')
    
    return model

# 检测异常
def detect_anomalies(model, data, scaler, look_back=60, threshold=0.01):
    # 准备测试数据
    scaled_data = scaler.transform(data.reshape(-1, 1))
    
    # 创建测试数据
    test_data = []
    for i in range(look_back, len(scaled_data)):
        test_data.append(scaled_data[i-look_back:i, 0])
    
    test_data = np.array(test_data)
    test_data = np.reshape(test_data, (test_data.shape[0], test_data.shape[1], 1))
    
    # 预测
    predictions = model.predict(test_data)
    predictions = scaler.inverse_transform(predictions)
    
    # 计算误差
    actual = data[look_back:]
    mse = np.mean(np.power(actual.reshape(-1, 1) - predictions, 2), axis=1)
    
    # 检测异常
    anomalies = mse > threshold
    
    return predictions, mse, anomalies

# 示例数据（实际使用时应从监控系统获取）
# 创建模拟的服务响应时间数据，包含一些异常点
date_range = pd.date_range(start='2023-01-01', periods=1000, freq='5min')
normal_data = np.random.normal(loc=0.5, scale=0.1, size=1000)

# 添加一些异常点
anomaly_indices = [100, 200, 300, 500, 700, 900]
for idx in anomaly_indices:
    normal_data[idx] = np.random.uniform(low=2.0, high=3.0)

service_response_time = pd.Series(normal_data, index=date_range)

# 准备训练数据
look_back = 60
X_train, Y_train, scaler = prepare_data(service_response_time.values, look_back)

# 创建和训练模型
model = create_lstm_model((X_train.shape[1], 1))
model.fit(X_train, Y_train, epochs=50, batch_size=32)

# 检测异常
predictions, mse, anomalies = detect_anomalies(model, service_response_time.values, scaler, look_back)

# 可视化结果
plt.figure(figsize=(14, 8))

# 绘制实际值和预测值
plt.subplot(2, 1, 1)
plt.plot(service_response_time.index[look_back:], service_response_time.values[look_back:], label='Actual')
plt.plot(service_response_time.index[look_back:], predictions, label='Predicted')
plt.title('Service Response Time Prediction')
plt.legend()

# 绘制MSE和异常点
plt.subplot(2, 1, 2)
plt.plot(service_response_time.index[look_back:], mse, label='MSE')
plt.axhline(y=0.01, color='r', linestyle='--', label='Threshold')

# 标记异常点
anomaly_times = service_response_time.index[look_back:][anomalies]
anomaly_values = mse[anomalies]
plt.scatter(anomaly_times, anomaly_values, color='red', label='Anomalies')

plt.title('Anomaly Detection')
plt.legend()
plt.tight_layout()
plt.show()

# 输出异常点信息
print(f"检测到 {np.sum(anomalies)} 个异常点")
print("异常点时间:")
for i, time in enumerate(anomaly_times):
    print(f"- {time}: MSE = {anomaly_values[i]:.4f}")

# 使用Grafana API创建自定义仪表盘的示例代码
import requests
import json

# Grafana配置
GRAFANA_URL = "http://localhost:3000"
API_KEY = "your_api_key_here"

# 创建仪表盘
def create_dashboard(dashboard_json):
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    
    url = f"{GRAFANA_URL}/api/dashboards/db"
    response = requests.post(url, headers=headers, data=json.dumps(dashboard_json))
    
    if response.status_code == 200:
        print("Dashboard created successfully!")
        return response.json()
    else:
        print(f"Failed to create dashboard: {response.status_code}")
        print(response.text)
        return None

# 创建服务监控仪表盘
def create_service_monitoring_dashboard(service_name):
    dashboard = {
        "dashboard": {
            "id": None,
            "uid": None,
            "title": f"{service_name} Monitoring",
            "tags": ["service", service_name],
            "timezone": "browser",
            "schemaVersion": 30,
            "version": 0,
            "panels": [
                # 响应时间面板
                {
                    "type": "timeseries",
                    "title": "Response Time",
                    "gridPos": {
                        "h": 8,
                        "w": 12,
                        "x": 0,
                        "y": 0
                    },
                    "targets": [
                        {
                            "datasource": "Prometheus",
                            "query": f"sum by (instance) (rate(http_request_duration_seconds_sum{{service='{service_name}'}}[5m])) / sum by (instance) (rate(http_request_duration_seconds_count{{service='{service_name}'}}[5m]))",
                            "refId": "A"
                        }
                    ],
                    "options": {
                        "tooltip": {
                            "mode": "single",
                            "sort": "none"
                        },
                        "legend": {
                            "showLegend": True,
                            "placement": "bottom"
                        }
                    }
                },
                # 请求率面板
                {
                    "type": "timeseries",
                    "title": "Request Rate",
                    "gridPos": {
                        "h": 8,
                        "w": 12,
                        "x": 12,
                        "y": 0
                    },
                    "targets": [
                        {
                            "datasource": "Prometheus",
                            "query": f"sum by (instance) (rate(http_requests_total{{service='{service_name}'}}[5m]))",
                            "refId": "A"
                        }
                    ],
                    "options": {
                        "tooltip": {
                            "mode": "single",
                            "sort": "none"
                        },
                        "legend": {
                            "showLegend": True,
                            "placement": "bottom"
                        }
                    }
                },
                # 错误率面板
                {
                    "type": "timeseries",
                    "title": "Error Rate",
                    "gridPos": {
                        "h": 8,
                        "w": 12,
                        "x": 0,
                        "y": 8
                    },
                    "targets": [
                        {
                            "datasource": "Prometheus",
                            "query": f"sum by (instance) (rate(http_requests_total{{service='{service_name}', status_code=~'5..'}}[5m])) / sum by (instance) (rate(http_requests_total{{service='{service_name}'}}[5m]))",
                            "refId": "A"
                        }
                    ],
                    "options": {
                        "tooltip": {
                            "mode": "single",
                            "sort": "none"
                        },
                        "legend": {
                            "showLegend": True,
                            "placement": "bottom"
                        }
                    }
                },
                # CPU使用率面板
                {
                    "type": "timeseries",
                    "title": "CPU Usage",
                    "gridPos": {
                        "h": 8,
                        "w": 12,
                        "x": 12,
                        "y": 8
                    },
                    "targets": [
                        {
                            "datasource": "Prometheus",
                            "query": f"sum by (instance) (rate(container_cpu_usage_seconds_total{{service='{service_name}'}}[5m]))",
                            "refId": "A"
                        }
                    ],
                    "options": {
                        "tooltip": {
                            "mode": "single",
                            "sort": "none"
                        },
                        "legend": {
                            "showLegend": True,
                            "placement": "bottom"
                        }
                    }
                }
            ],
            "time": {
                "from": "now-6h",
                "to": "now"
            }
        },
        "overwrite": False
    }
    
    return dashboard

# 创建用户服务监控仪表盘
service_dashboard = create_service_monitoring_dashboard("user-service")
result = create_dashboard(service_dashboard)

if result:
    print(f"Dashboard URL: {GRAFANA_URL}/d/{result['uid']}/{service_dashboard['dashboard']['title'].lower().replace(' ', '-')}")

# 使用大模型进行故障智能诊断的示例代码
from transformers import pipeline
import json

# 初始化大模型
nlp = pipeline("text-generation", model="gpt-4")

# 收集故障相关数据
def collect_failure_data(service_name, time_range):
    # 实际应用中，这里应该从日志系统、监控系统和追踪系统收集数据
    # 这里使用示例数据
    failure_data = {
        "service": service_name,
        "time_range": time_range,
        "error_logs": [
            {"timestamp": "2023-01-01T10:00:00", "message": "Connection refused to database"},
            {"timestamp": "2023-01-01T10:01:00", "message": "Timeout waiting for database response"},
            {"timestamp": "2023-01-01T10:02:00", "message": "Too many connections to database"}
        ],
        "metrics": [
            {"name": "response_time", "value": 10.5, "unit": "s", "threshold": 1.0},
            {"name": "error_rate", "value": 0.95, "unit": "", "threshold": 0.05},
            {"name": "database_connections", "value": 1000, "unit": "", "threshold": 500}
        ],
        "trace_data": [
            {"trace_id": "abc123", "status": "error", "error_type": "database_error", "service_path": ["api-gateway", "user-service", "database"]}
        ],
        "service_dependencies": ["api-gateway", "user-service", "database", "cache-service"]
    }
    
    return failure_data

# 构建故障诊断提示
def build_diagnosis_prompt(failure_data):
    prompt = f"""
我需要诊断一个分布式系统的故障问题。以下是相关信息：

服务名称：{failure_data['service']}
时间范围：{failure_data['time_range']}

错误日志：
"""
    
    for log in failure_data['error_logs']:
        prompt += f"- {log['timestamp']}: {log['message']}\n"
    
    prompt += "\n异常指标：\n"
    for metric in failure_data['metrics']:
        prompt += f"- {metric['name']}: {metric['value']} {metric['unit']} (阈值: {metric['threshold']})\n"
    
    prompt += "\n追踪数据：\n"
    for trace in failure_data['trace_data']:
        prompt += f"- Trace ID: {trace['trace_id']}, Status: {trace['status']}, Error: {trace['error_type']}\n"
        prompt += f"  服务路径: {' -> '.join(trace['service_path'])}\n"
    
    prompt += f"\n服务依赖：{' -> '.join(failure_data['service_dependencies'])}"
    
    prompt += "\n\n请分析这个故障的可能原因，并提供详细的故障诊断和修复建议。"
    
    return prompt

# 使用大模型进行故障诊断
def diagnose_failure(failure_data):
    # 构建提示
    prompt = build_diagnosis_prompt(failure_data)
    
    # 使用大模型进行推理
    result = nlp(prompt, max_length=2000, temperature=0.7)[0]["generated_text"]
    
    return result

# 收集故障数据
failure_data = collect_failure_data("user-service", "2023-01-01T10:00:00 to 2023-01-01T10:05:00")

# 进行故障诊断
diagnosis_result = diagnose_failure(failure_data)

# 输出诊断结果
print("故障智能诊断结果：")
print("="*80)
print(diagnosis_result)
print("="*80)

可观测性未来发展趋势
AI技术深度融合 → 多模态数据分析 → 边缘计算可观测性 → 云原生原生集成 → 自动化与自愈增强
  ↓                                 ↑
安全可观测性融合 ← 业务可观测性提升 ← 用户体验优化

参考资料关系图
┌─────────────────────────┐     ┌─────────────────────────┐
│ 分布式系统理论          │────▶│ 可观测性基础            │
└─────────────────────────┘     └─────────────────────────┘
        ▲                                  ▲
        │                                  │
        │                                  │
┌─────────────────────────┐     ┌─────────────────────────┐
│ 机器学习与AI            │────▶│ 可观测性实践与工具      │
└─────────────────────────┘     └─────────────────────────┘