15 - 监控系统设计

package main

import (
    "fmt"
    "net/http"
    "time"
    
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
)

// ==================== 自定义 Exporter ====================

// Metrics 指标定义
var (
    // Counter: 单调递增
    requestTotal = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "http_requests_total",
            Help: "Total number of HTTP requests",
        },
        []string{"method", "endpoint", "status"},
    )
    
    // Gauge: 可增可减
    activeConnections = prometheus.NewGauge(
        prometheus.GaugeOpts{
            Name: "active_connections",
            Help: "Number of active connections",
        },
    )
    
    // Histogram: 分布统计
    requestDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "http_request_duration_seconds",
            Help:    "HTTP request latencies",
            Buckets: []float64{0.001, 0.01, 0.1, 0.5, 1, 2, 5},
        },
        []string{"method", "endpoint"},
    )
    
    // Summary: 分位数统计
    requestSize = prometheus.NewSummaryVec(
        prometheus.SummaryOpts{
            Name:       "http_request_size_bytes",
            Help:       "HTTP request sizes",
            Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
        },
        []string{"method", "endpoint"},
    )
)

func init() {
    // 注册指标
    prometheus.MustRegister(requestTotal)
    prometheus.MustRegister(activeConnections)
    prometheus.MustRegister(requestDuration)
    prometheus.MustRegister(requestSize)
}

// MetricsMiddleware 指标采集中间件
func MetricsMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        start := time.Now()
        
        // 增加活跃连接数
        activeConnections.Inc()
        defer activeConnections.Dec()
        
        // 包装 ResponseWriter 以获取状态码
        rw := &responseWriter{ResponseWriter: w, statusCode: 200}
        
        // 处理请求
        next.ServeHTTP(rw, r)
        
        // 记录指标
        duration := time.Since(start).Seconds()
        
        requestTotal.WithLabelValues(
            r.Method,
            r.URL.Path,
            fmt.Sprintf("%d", rw.statusCode),
        ).Inc()
        
        requestDuration.WithLabelValues(
            r.Method,
            r.URL.Path,
        ).Observe(duration)
        
        requestSize.WithLabelValues(
            r.Method,
            r.URL.Path,
        ).Observe(float64(r.ContentLength))
    })
}

type responseWriter struct {
    http.ResponseWriter
    statusCode int
}

func (rw *responseWriter) WriteHeader(code int) {
    rw.statusCode = code
    rw.ResponseWriter.WriteHeader(code)
}

// ==================== 业务指标 ====================

// BusinessMetrics 业务指标
type BusinessMetrics struct {
    orderTotal   prometheus.Counter
    orderAmount  prometheus.Gauge
    userRegister prometheus.Counter
}

func NewBusinessMetrics() *BusinessMetrics {
    return &BusinessMetrics{
        orderTotal: prometheus.NewCounter(
            prometheus.CounterOpts{
                Name: "business_order_total",
                Help: "Total number of orders",
            },
        ),
        orderAmount: prometheus.NewGauge(
            prometheus.GaugeOpts{
                Name: "business_order_amount_yuan",
                Help: "Total order amount in yuan",
            },
        ),
        userRegister: prometheus.NewCounter(
            prometheus.CounterOpts{
                Name: "business_user_register_total",
                Help: "Total number of user registrations",
            },
        ),
    }
}

func (m *BusinessMetrics) RecordOrder(amount float64) {
    m.orderTotal.Inc()
    m.orderAmount.Add(amount)
}

func (m *BusinessMetrics) RecordUserRegister() {
    m.userRegister.Inc()
}

// ==================== 自定义 Collector ====================

// CustomCollector 自定义采集器
type CustomCollector struct {
    queueSizeDesc *prometheus.Desc
    queueLatency  *prometheus.Desc
}

func NewCustomCollector() *CustomCollector {
    return &CustomCollector{
        queueSizeDesc: prometheus.NewDesc(
            "queue_size",
            "Current queue size",
            []string{"queue_name"},
            nil,
        ),
        queueLatency: prometheus.NewDesc(
            "queue_latency_seconds",
            "Queue processing latency",
            []string{"queue_name"},
            nil,
        ),
    }
}

// Describe 实现 Collector 接口
func (c *CustomCollector) Describe(ch chan<- *prometheus.Desc) {
    ch <- c.queueSizeDesc
    ch <- c.queueLatency
}

// Collect 实现 Collector 接口
func (c *CustomCollector) Collect(ch chan<- prometheus.Metric) {
    // 模拟从队列获取指标
    queueSize := c.getQueueSize("order_queue")
    queueLatency := c.getQueueLatency("order_queue")
    
    ch <- prometheus.MustNewConstMetric(
        c.queueSizeDesc,
        prometheus.GaugeValue,
        float64(queueSize),
        "order_queue",
    )
    
    ch <- prometheus.MustNewConstMetric(
        c.queueLatency,
        prometheus.GaugeValue,
        queueLatency,
        "order_queue",
    )
}

func (c *CustomCollector) getQueueSize(queueName string) int {
    // 实际实现：从 Redis、RabbitMQ 等获取队列长度
    return 100
}

func (c *CustomCollector) getQueueLatency(queueName string) float64 {
    // 实际实现：计算队列处理延迟
    return 0.5
}

// ==================== Push Gateway ====================

import (
    "github.com/prometheus/client_golang/prometheus/push"
)

// PushMetrics 推送指标到 Push Gateway
func PushMetrics() {
    // 创建临时指标
    jobDuration := prometheus.NewGauge(
        prometheus.GaugeOpts{
            Name: "job_duration_seconds",
            Help: "Job duration",
        },
    )
    
    // 记录指标
    start := time.Now()
    doJob()
    duration := time.Since(start).Seconds()
    
    jobDuration.Set(duration)
    
    // 推送到 Push Gateway
    if err := push.New("http://localhost:9091", "batch_job").
        Collector(jobDuration).
        Grouping("instance", "batch-server-1").
        Push(); err != nil {
        fmt.Printf("Could not push to Pushgateway: %v\n", err)
    }
}

func doJob() {
    time.Sleep(2 * time.Second)
}

// ==================== Main ====================

func main() {
    // 注册自定义 Collector
    customCollector := NewCustomCollector()
    prometheus.MustRegister(customCollector)
    
    // 业务指标
    businessMetrics := NewBusinessMetrics()
    prometheus.MustRegister(
        businessMetrics.orderTotal,
        businessMetrics.orderAmount,
        businessMetrics.userRegister,
    )
    
    // HTTP 服务
    mux := http.NewServeMux()
    
    mux.HandleFunc("/api/order", func(w http.ResponseWriter, r *http.Request) {
        // 模拟下单
        businessMetrics.RecordOrder(99.99)
        w.Write([]byte("Order created"))
    })
    
    // Prometheus metrics endpoint
    mux.Handle("/metrics", promhttp.Handler())
    
    // 应用中间件
    handler := MetricsMiddleware(mux)
    
    fmt.Println("Server started at :8080")
    fmt.Println("Metrics available at http://localhost:8080/metrics")
    http.ListenAndServe(":8080", handler)
}

// ==================== 结构化日志 ====================

package logging

import (
    "os"
    
    "go.uber.org/zap"
    "go.uber.org/zap/zapcore"
)

// Logger 日志实例
var Logger *zap.Logger

func init() {
    // 配置 Encoder
    encoderConfig := zapcore.EncoderConfig{
        TimeKey:        "timestamp",
        LevelKey:       "level",
        NameKey:        "logger",
        CallerKey:      "caller",
        MessageKey:     "message",
        StacktraceKey:  "stacktrace",
        LineEnding:     zapcore.DefaultLineEnding,
        EncodeLevel:    zapcore.LowercaseLevelEncoder,
        EncodeTime:     zapcore.ISO8601TimeEncoder,
        EncodeDuration: zapcore.SecondsDurationEncoder,
        EncodeCaller:   zapcore.ShortCallerEncoder,
    }
    
    // JSON 格式输出（便于 ELK 解析）
    core := zapcore.NewCore(
        zapcore.NewJSONEncoder(encoderConfig),
        zapcore.AddSync(os.Stdout),
        zap.InfoLevel,
    )
    
    Logger = zap.New(core, zap.AddCaller(), zap.AddStacktrace(zapcore.ErrorLevel))
}

// 使用示例
func ExampleUsage() {
    Logger.Info("User login",
        zap.String("user_id", "12345"),
        zap.String("ip", "192.168.1.100"),
        zap.Duration("duration", 150*time.Millisecond),
    )
    
    Logger.Error("Payment failed",
        zap.String("order_id", "ORD123"),
        zap.Float64("amount", 99.99),
        zap.Error(fmt.Errorf("insufficient balance")),
    )
}

// ==================== 链路追踪 ====================

package tracing

import (
    "context"
    "io"
    "time"
    
    "github.com/opentracing/opentracing-go"
    "github.com/opentracing/opentracing-go/ext"
    "github.com/uber/jaeger-client-go"
    jaegercfg "github.com/uber/jaeger-client-go/config"
)

// InitTracer 初始化 Tracer
func InitTracer(serviceName string) (opentracing.Tracer, io.Closer, error) {
    cfg := jaegercfg.Configuration{
        ServiceName: serviceName,
        Sampler: &jaegercfg.SamplerConfig{
            Type:  jaeger.SamplerTypeConst,
            Param: 1, // 采样率 100%
        },
        Reporter: &jaegercfg.ReporterConfig{
            LogSpans:            true,
            BufferFlushInterval: 1 * time.Second,
            LocalAgentHostPort:  "localhost:6831",
        },
    }
    
    tracer, closer, err := cfg.NewTracer()
    if err != nil {
        return nil, nil, err
    }
    
    opentracing.SetGlobalTracer(tracer)
    return tracer, closer, nil
}

// TracingMiddleware HTTP 追踪中间件
func TracingMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        tracer := opentracing.GlobalTracer()
        
        // 从 HTTP Header 提取 SpanContext
        spanCtx, _ := tracer.Extract(
            opentracing.HTTPHeaders,
            opentracing.HTTPHeadersCarrier(r.Header),
        )
        
        // 创建 Span
        span := tracer.StartSpan(
            r.URL.Path,
            ext.RPCServerOption(spanCtx),
        )
        defer span.Finish()
        
        // 设置标签
        ext.HTTPMethod.Set(span, r.Method)
        ext.HTTPUrl.Set(span, r.URL.String())
        
        // 将 Span 注入到 Context
        ctx := opentracing.ContextWithSpan(r.Context(), span)
        
        // 继续处理请求
        next.ServeHTTP(w, r.WithContext(ctx))
    })
}

// 业务代码示例
func CreateOrder(ctx context.Context, orderID string) error {
    // 创建子 Span
    span, ctx := opentracing.StartSpanFromContext(ctx, "CreateOrder")
    defer span.Finish()
    
    span.SetTag("order_id", orderID)
    
    // 调用库存服务
    if err := checkInventory(ctx, orderID); err != nil {
        ext.Error.Set(span, true)
        span.LogKV("event", "error", "message", err.Error())
        return err
    }
    
    // 调用支付服务
    if err := processPayment(ctx, orderID); err != nil {
        ext.Error.Set(span, true)
        return err
    }
    
    return nil
}

func checkInventory(ctx context.Context, orderID string) error {
    span, _ := opentracing.StartSpanFromContext(ctx, "CheckInventory")
    defer span.Finish()
    
    // 模拟 RPC 调用
    time.Sleep(50 * time.Millisecond)
    
    span.SetTag("inventory_available", true)
    return nil
}

func processPayment(ctx context.Context, orderID string) error {
    span, _ := opentracing.StartSpanFromContext(ctx, "ProcessPayment")
    defer span.Finish()
    
    // 模拟支付处理
    time.Sleep(100 * time.Millisecond)
    
    span.SetTag("payment_status", "success")
    return nil
}

// 避免高基数标签
//  不好：user_id 作为标签（百万级）
http_requests_total{user_id="12345"}

//  好：放在日志中
log.Info("request", zap.String("user_id", "12345"))

// HTTP Header
X-Trace-ID: abc123
X-Span-ID: def456
X-Parent-Span-ID: 789

// gRPC Metadata
metadata.New(map[string]string{
    "trace-id": "abc123",
    "span-id": "def456",
})

span := tracer.StartSpan("CreateOrder")
span.SetTag("order_id", "ORD123")
span.SetTag("amount", 99.99)
span.Finish()