遥测管道

工作机制

服务通过 SDK 输出日志、指标和链路，通常使用 OpenTelemetry。收集器接收信号后，由处理器完成批量、脱敏、富化和采样，再由导出器发送到一个或多个后端。缓冲与重试吸收后端故障，管道统一资源属性，使各服务的遥测数据保持一致语义。

适用场景与取舍

当多个服务都在输出遥测，并且需要统一 schema、脱敏、采样或多后端路由时，应引入管道；单个小应用可以直接导出。与逐服务配置相比，集中管道无需重新部署每个应用就能调整路由和成本策略，但增加了一跳，本身也必须被监控。管道是关键基础设施和潜在单点，需要容量规划、背压处理以及自身的遥测。

具体示例

一条 OpenTelemetry Collector 管道批量处理 span，删除 Authorization 头，对成功链路采样百分之一但保留全部错误，同时导出到链路后端和冷存储。上线后可观测性支出下降四成，错误可见性没有损失。

常见误解与关联概念

遥测管道不只是装一次代理，而是包含 schema、路由和失败行为的配置系统；队列溢出时收集器并不保证投递。管道也不只服务链路，日志和指标同样需要路由与脱敏。采样在管道内执行，分布式追踪依赖一致的上下文传播，黄金信号定义哪些指标必须完整抵达后端。

代码示例

import promClient from 'prom-client';
import express from 'express';

const registry = new promClient.Registry();
promClient.collectDefaultMetrics({ register: registry, prefix: 'svc_' });

const ordersTotal = new promClient.Counter({
  name: 'orders_created_total',
  help: 'Total orders created',
  labelNames: ['tenant', 'channel'],
  registers: [registry],
});

const requestDuration = new promClient.Histogram({
  name: 'http_request_duration_seconds',
  help: 'Request latency distribution',
  buckets: [0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5],
  labelNames: ['route', 'method'],
  registers: [registry],
});

async function handleCheckout(req, res) {
  const end = requestDuration.startTimer({ route: '/checkout', method: 'POST' });
  try {
    ordersTotal.inc({ tenant: 'acme', channel: 'web' });
    res.json({ ok: true });
  } finally {
    end();
  }
}

const app = express();
app.post('/checkout', handleCheckout);
app.get('/metrics', async (req, res) => {
  res.set('content-type', registry.contentType);
  res.end(await registry.metrics());
});

app.listen(3000);

package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"net/http"
)

type requestIDKey struct{}

func validRequestID(id string) bool {
	if len(id) == 0 || len(id) > 128 {
		return false
	}
	for _, char := range id {
		switch {
		case char >= 'a' && char <= 'z':
		case char >= 'A' && char <= 'Z':
		case char >= '0' && char <= '9':
		case char == '-' || char == '_':
		default:
			return false
		}
	}
	return true
}

func RequestIDMiddleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		id := r.Header.Get("X-Request-ID")
		if !validRequestID(id) {
			buffer := make([]byte, 8)
			if _, err := rand.Read(buffer); err != nil {
				http.Error(w, "cannot create request id", http.StatusInternalServerError)
				return
			}
			id = hex.EncodeToString(buffer)
		}
		w.Header().Set("X-Request-ID", id)
		ctx := context.WithValue(r.Context(), requestIDKey{}, id)
		next.ServeHTTP(w, r.WithContext(ctx))
	})
}

func RequestIDFrom(ctx context.Context) string {
	id, _ := ctx.Value(requestIDKey{}).(string)
	return id
}

func main() {
	handler := RequestIDMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		fmt.Fprintln(w, "request", RequestIDFrom(r.Context()))
	}))
	http.Handle("/", handler)
	_ = http.ListenAndServe(":8080", nil)
}