APM最佳实践：操作指南

# 自动检测会自动处理这些
@app.route('/api/orders')
def create_order():
    # 仅对关键业务逻辑添加手动跨度
    with tracer.start_as_current_span("order.validation") as span:
        span.set_attribute("order.value", order_total)
        if not validate_order(order_data):
            span.set_status(Status(StatusCode.ERROR))
            return 400

# 传出请求 - 注入上下文
headers = {}
propagate.inject(headers)
response = requests.post('http://service-b/process', headers=headers)

@RestController
public class OrderController {
    
    @PostMapping("/orders")
    public ResponseEntity<Order> createOrder(@RequestBody OrderRequest request) {
        // 自动检测自动捕获此端点
        // 添加自定义业务上下文
        Span.current().setAttributes(Attributes.of(
            stringKey("order.value"), String.valueOf(request.getTotal()),
            stringKey("user.tier"), request.getUserTier()
        ));
        
        return ResponseEntity.ok(processOrder(request));
    }
}

# 良好的 - 有界基数
span.set_attribute("user.tier", user.subscription_tier)  # 3-5个值
span.set_attribute("http.status_code", response.status_code)  # ~10个值

# 不良的 - 无界基数  
span.set_attribute("user.id", user.id)  # 数百万个值
span.set_attribute("request.timestamp", now())  # 无限个值

slos:
  - name: checkout_availability
    target: 99.9%
    window: 7d
  - name: checkout_latency  
    target: 95%  # 95%的请求在500ms以内
    window: 7d

order_processing_duration = Histogram(
    "order_processing_seconds",
    "Time to process orders", 
    ["payment_method", "order_size"]
)

with order_processing_duration.labels(
    payment_method=payment.method,
    order_size=get_size_bucket(order.total)
).time():
    process_order(order)

// 关键用户流程的合成检查
const syntheticCheck = async () => {
    const span = tracer.startSpan('synthetic.checkout_flow');
    try {
        await loginUser();
        await addItemToCart();
        await completePurchase();
        span.setStatus({code: SpanStatusCode.OK});
    } catch (error) {
        span.recordException(error);
        span.setStatus({code: SpanStatusCode.ERROR});
        throw error;
    } finally {
        span.end();
    }
};

# 事件驱动系统 - 通过消息传播上下文
def publish_order_event(order_data):
    headers = {}
    propagate.inject(headers)
    
    message = {
        'data': order_data,
        'trace_headers': headers  # 保持追踪上下文
    }
    kafka_producer.send('order-events', message)

FROM traces-apm*
| WHERE user.id == "user_12345" 
  AND @timestamp >= "2024-06-06T09:00:00" 
  AND @timestamp <= "2024-06-06T10:00:00"
| EVAL duration_ms = transaction.duration.us / 1000
| KEEP trace.id, duration_ms, transaction.name, service.name, transaction.result
| WHERE duration_ms > 2000
| SORT duration_ms DESC
| LIMIT 10

# APM检测的N+1查询问题
def get_user_orders_slow(user_id):
    user = User.query.get(user_id)
    orders = []
    for order_id in user.order_ids:  # 每次迭代 = 1次数据库查询
        orders.append(Order.query.get(order_id))
    return orders

# APM分析后的优化
def get_user_orders_fast(user_id):
    return Order.query.filter(Order.user_id == user_id).all()  # 单次查询

// Java - 使用Spring Cloud自动传播
@PostMapping("/orders")
public ResponseEntity<Order> createOrder(@RequestBody OrderRequest request) {
    Span.current().setAttributes(Attributes.of(
        stringKey("order.type"), request.getOrderType(),
        longKey("order.value"), request.getTotalValue()));
    
    // OpenFeign自动将上下文传播到下游服务
    return paymentClient.processPayment(request.getPaymentData());
}

// Go - 手动上下文提取和传播
func processHandler(w http.ResponseWriter, r *http.Request) {
    ctx := otel.GetTextMapPropagator().Extract(r.Context(), 
                                              propagation.HeaderCarrier(r.Header))
    ctx, span := tracer.Start(ctx, "process_payment")
    defer span.End()
    // 继续保持追踪上下文
}

# 使用关联ID的合成跨度进行大型机调用
with tracer.start_as_current_span("mainframe.account_lookup") as span:
    correlation_id = format(span.get_span_context().trace_id, '032x')
    
    logger.info("CICS调用已开始", extra={
        "correlation_id": correlation_id,
        "trace_id": span.get_span_context().trace_id
    })
    
    result = call_mainframe_service(account_data, correlation_id)
    span.set_attribute("account.status", result.status)

-- 在投诉时间段内查找慢请求
FROM traces-apm*
| WHERE user.id == "user_12345" AND @timestamp >= "2024-06-06T09:00:00"
| EVAL duration_ms = transaction.duration.us / 1000
| WHERE duration_ms > 2000
| STATS avg_duration = AVG(duration_ms) BY service.name, transaction.name
| SORT avg_duration DESC

-- 跨服务边界关联错误
FROM traces-apm*
| WHERE trace.id == "44b3c2c06e15d444a770b87daab45c0a"
| EVAL is_error = CASE(transaction.result == "error", 1, 0)
| STATS error_rate = SUM(is_error) / COUNT(*) * 100 BY service.name
| WHERE error_rate > 0

# 生产者 - 将上下文注入消息
headers = {}
propagate.inject(headers)
message = {
    'data': order_data,
    'trace_headers': headers  # 保持追踪上下文
}
await kafka_producer.send(''order-events', message)

# 消费者 - 提取并继续追踪
trace_headers = message.get('trace_headers', {})
context = propagate.extract(trace_headers)
with tracer.start_as_current_span("order.process", context=context):
    await process_order(message['data'])