监控与告警系统

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

var (
    // 订单总数
    orderTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "exchange_orders_total",
            Help: "Total number of orders",
        },
        []string{"symbol", "side", "type"},
    )

    // API请求总数
    apiRequestsTotal = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "exchange_api_requests_total",
            Help: "Total number of API requests",
        },
        []string{"method", "endpoint", "status"},
    )
)

// 使用示例
func PlaceOrder(order *Order) error {
    // 业务逻辑
    // ...

    // 记录指标
    orderTotal.WithLabelValues(order.Symbol, order.Side, order.Type).Inc()

    return nil
}

var (
    // 当前活跃订单数
    activeOrders = promauto.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "exchange_active_orders",
            Help: "Number of active orders",
        },
        []string{"symbol"},
    )

    // 用户在线数
    onlineUsers = promauto.NewGauge(
        prometheus.GaugeOpts{
            Name: "exchange_online_users",
            Help: "Number of online users",
        },
    )

    // 数据库连接池
    dbConnections = promauto.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "exchange_db_connections",
            Help: "Database connection pool stats",
        },
        []string{"state"}, // active, idle, waiting
    )
)

// 使用示例
func UpdateActiveOrders(symbol string, count int) {
    activeOrders.WithLabelValues(symbol).Set(float64(count))
}

func UpdateDBConnectionStats(stats *sql.DBStats) {
    dbConnections.WithLabelValues("active").Set(float64(stats.InUse))
    dbConnections.WithLabelValues("idle").Set(float64(stats.Idle))
    dbConnections.WithLabelValues("waiting").Set(float64(stats.WaitCount))
}

var (
    // API响应时间
    apiDuration = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "exchange_api_duration_seconds",
            Help:    "API request duration",
            Buckets: prometheus.DefBuckets, // [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10]
        },
        []string{"endpoint"},
    )

    // 订单处理延迟
    orderProcessingDuration = promauto.NewHistogram(
        prometheus.HistogramOpts{
            Name:    "exchange_order_processing_seconds",
            Help:    "Order processing duration",
            Buckets: []float64{0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1},
        },
    )
)

// 使用示例
func APIHandler(w http.ResponseWriter, r *http.Request) {
    start := time.Now()
    defer func() {
        duration := time.Since(start).Seconds()
        apiDuration.WithLabelValues(r.URL.Path).Observe(duration)
    }()

    // 处理请求
    // ...
}

var (
    // 撮合引擎延迟
    matchingLatency = promauto.NewSummaryVec(
        prometheus.SummaryOpts{
            Name:       "exchange_matching_latency_seconds",
            Help:       "Matching engine latency",
            Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001}, // P50, P90, P99
        },
        []string{"symbol"},
    )
)

func ProcessOrder(symbol string) {
    start := time.Now()
    defer func() {
        latency := time.Since(start).Seconds()
        matchingLatency.WithLabelValues(symbol).Observe(latency)
    }()

    // 撮合逻辑
    // ...
}

package metrics

import (
    "database/sql"
    "time"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)

var (
    // 交易量
    tradeVolume = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "exchange_trade_volume",
            Help: "Trading volume in quote currency",
        },
        []string{"symbol"},
    )

    // 手续费收入
    feeRevenue = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "exchange_fee_revenue",
            Help: "Fee revenue in USDT",
        },
        []string{"currency"},
    )

    // 充值提现
    deposits = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "exchange_deposits_total",
            Help: "Total deposits",
        },
        []string{"currency", "status"},
    )

    withdrawals = promauto.NewCounterVec(
        prometheus.CounterOpts{
            Name: "exchange_withdrawals_total",
            Help: "Total withdrawals",
        },
        []string{"currency", "status"},
    )

    // 资金总量
    totalBalance = promauto.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "exchange_total_balance",
            Help: "Total balance in USDT equivalent",
        },
        []string{"currency"},
    )
)

type MetricsCollector struct {
    db *sql.DB
}

func NewMetricsCollector(db *sql.DB) *MetricsCollector {
    mc := &MetricsCollector{db: db}

    // 启动定时采集
    go mc.collectPeriodically()

    return mc
}

func (mc *MetricsCollector) collectPeriodically() {
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        mc.collectTotalBalance()
        mc.collectActiveOrders()
    }
}

func (mc *MetricsCollector) collectTotalBalance() {
    rows, err := mc.db.Query(`
        SELECT currency, SUM(available + frozen) as total
        FROM account_balances
        GROUP BY currency
    `)
    if err != nil {
        return
    }
    defer rows.Close()

    for rows.Next() {
        var currency string
        var total float64
        rows.Scan(&currency, &total)

        totalBalance.WithLabelValues(currency).Set(total)
    }
}

func (mc *MetricsCollector) collectActiveOrders() {
    rows, err := mc.db.Query(`
        SELECT symbol, COUNT(*) as count
        FROM orders
        WHERE status IN ('pending', 'partial')
        GROUP BY symbol
    `)
    if err != nil {
        return
    }
    defer rows.Close()

    for rows.Next() {
        var symbol string
        var count int
        rows.Scan(&symbol, &count)

        activeOrders.WithLabelValues(symbol).Set(float64(count))
    }
}

// 记录成交
func (mc *MetricsCollector) RecordTrade(trade *Trade) {
    // 记录交易量
    tradeVolume.WithLabelValues(trade.Symbol).Add(trade.Amount)

    // 记录手续费
    feeRevenue.WithLabelValues("USDT").Add(trade.BuyFee + trade.SellFee)
}

// 记录充值
func (mc *MetricsCollector) RecordDeposit(currency, status string, amount float64) {
    deposits.WithLabelValues(currency, status).Inc()
}

// 记录提现
func (mc *MetricsCollector) RecordWithdrawal(currency, status string, amount float64) {
    withdrawals.WithLabelValues(currency, status).Inc()
}

package main

import (
    "log"
    "net/http"

    "github.com/prometheus/client_golang/prometheus/promhttp"
)

func main() {
    // 初始化指标采集器
    collector := NewMetricsCollector(db)

    // 暴露metrics端点
    http.Handle("/metrics", promhttp.Handler())

    log.Println("Metrics server listening on :9090")
    log.Fatal(http.ListenAndServe(":9090", nil))
}

package logger

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

var log *zap.Logger

func InitLogger() {
    config := zap.NewProductionConfig()

    // 自定义日志级别
    config.Level = zap.NewAtomicLevelAt(zapcore.InfoLevel)

    // 日志格式
    config.Encoding = "json"

    // 输出路径
    config.OutputPaths = []string{
        "stdout",
        "/var/log/exchange/app.log",
    }

    // 错误日志单独输出
    config.ErrorOutputPaths = []string{
        "stderr",
        "/var/log/exchange/error.log",
    }

    log, _ = config.Build()
}

func Info(msg string, fields ...zap.Field) {
    log.Info(msg, fields...)
}

func Error(msg string, fields ...zap.Field) {
    log.Error(msg, fields...)
}

func Fatal(msg string, fields ...zap.Field) {
    log.Fatal(msg, fields...)
}

// 使用示例
func PlaceOrder(order *Order) error {
    logger.Info("order placed",
        zap.String("order_id", order.OrderID),
        zap.Int64("user_id", order.UserID),
        zap.String("symbol", order.Symbol),
        zap.String("side", order.Side),
        zap.Float64("price", order.Price),
        zap.Float64("quantity", order.Quantity),
    )

    // 业务逻辑
    // ...

    return nil
}

package tracing

import (
    "github.com/opentracing/opentracing-go"
    "github.com/uber/jaeger-client-go"
    "github.com/uber/jaeger-client-go/config"
)

func InitTracing(serviceName string) (opentracing.Tracer, io.Closer, error) {
    cfg := &config.Configuration{
        ServiceName: serviceName,

        // 采样策略
        Sampler: &config.SamplerConfig{
            Type:  "probabilistic",
            Param: 0.1, // 采样10%的请求
        },

        // 报告配置
        Reporter: &config.ReporterConfig{
            LogSpans:           true,
            LocalAgentHostPort: "jaeger-agent:6831",
        },
    }

    tracer, closer, err := cfg.NewTracer()
    if err != nil {
        return nil, nil, err
    }

    opentracing.SetGlobalTracer(tracer)

    return tracer, closer, nil
}

func PlaceOrderHandler(w http.ResponseWriter, r *http.Request) {
    // 1. 从HTTP请求中提取span context
    spanCtx, _ := opentracing.GlobalTracer().Extract(
        opentracing.HTTPHeaders,
        opentracing.HTTPHeadersCarrier(r.Header),
    )

    // 2. 创建新的span
    span := opentracing.GlobalTracer().StartSpan(
        "place_order",
        opentracing.ChildOf(spanCtx),
    )
    defer span.Finish()

    // 3. 添加tags
    span.SetTag("user_id", getUserID(r))
    span.SetTag("symbol", r.FormValue("symbol"))

    // 4. 调用下游服务，传递span context
    ctx := opentracing.ContextWithSpan(r.Context(), span)
    err := placeOrder(ctx, order)

    if err != nil {
        // 记录错误
        span.SetTag("error", true)
        span.LogKV("error", err.Error())
    }
}

func placeOrder(ctx context.Context, order *Order) error {
    // 创建子span
    span, _ := opentracing.StartSpanFromContext(ctx, "check_balance")
    defer span.Finish()

    // 检查余额
    // ...

    // 调用撮合引擎
    matchSpan, _ := opentracing.StartSpanFromContext(ctx, "matching_engine")
    defer matchSpan.Finish()

    // 撮合逻辑
    // ...

    return nil
}

package alerting

import (
    "encoding/json"
    "net/http"
)

type AlertWebhook struct {
    notifier *Notifier
}

type Alert struct {
    Status      string            `json:"status"`
    Labels      map[string]string `json:"labels"`
    Annotations map[string]string `json:"annotations"`
    StartsAt    string            `json:"startsAt"`
    EndsAt      string            `json:"endsAt"`
}

type AlertMessage struct {
    Version  string  `json:"version"`
    GroupKey string  `json:"groupKey"`
    Status   string  `json:"status"`
    Alerts   []Alert `json:"alerts"`
}

func (aw *AlertWebhook) HandleWebhook(w http.ResponseWriter, r *http.Request) {
    var msg AlertMessage
    err := json.NewDecoder(r.Body).Decode(&msg)
    if err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    for _, alert := range msg.Alerts {
        // 根据severity决定通知方式
        severity := alert.Labels["severity"]

        switch severity {
        case "critical":
            // 电话 + 短信 + 企业微信
            aw.notifier.SendPhoneCall(alert)
            aw.notifier.SendSMS(alert)
            aw.notifier.SendWechat(alert)

        case "warning":
            // 邮件 + 企业微信
            aw.notifier.SendEmail(alert)
            aw.notifier.SendWechat(alert)

        default:
            // 仅企业微信
            aw.notifier.SendWechat(alert)
        }

        // 记录告警到数据库
        aw.saveAlert(alert)
    }

    w.WriteHeader(http.StatusOK)
}

type Notifier struct {
    // 通知渠道配置
}

func (n *Notifier) SendPhoneCall(alert Alert) {
    // 调用电话告警API
    log.Printf("Calling ops team: %s", alert.Annotations["summary"])
}

func (n *Notifier) SendSMS(alert Alert) {
    // 调用短信API
    log.Printf("SMS sent: %s", alert.Annotations["summary"])
}

func (n *Notifier) SendWechat(alert Alert) {
    // 调用企业微信API
    message := fmt.Sprintf(
        "【%s】\n告警: %s\n详情: %s\n时间: %s",
        alert.Labels["severity"],
        alert.Annotations["summary"],
        alert.Annotations["description"],
        alert.StartsAt,
    )

    // 发送企业微信消息
    log.Printf("Wechat message: %s", message)
}

func (n *Notifier) SendEmail(alert Alert) {
    // 发送邮件
    subject := fmt.Sprintf("[%s] %s", alert.Labels["severity"], alert.Annotations["summary"])
    body := alert.Annotations["description"]

    log.Printf("Email sent: %s", subject)
}