《游戏服务端编程实践》15.4 实战：构建游戏监控看板

// file: ch15/dashboard/dashboard.go
// tier: L1
// verify: go test ./ch15/dashboard/

// Package dashboard 提供游戏监控看板所需的最小指标聚合与阈值告警状态机。
//
// Aggregator 负责把多个分片的最新值合并成 Count/Sum/Min/Max；
// Evaluator 负责“超过阈值并持续一段时间才触发”，并在恢复时
// 发出一次恢复事件。时间由外部注入，测试可以精确控制推进。
package dashboard

import (
	"errors"
	"math"
	"sync"
)

var (
	// ErrInvalidSeries 表示指标名或分片名为空。
	ErrInvalidSeries = errors.New("dashboard: metric and shard names must not be empty")
	// ErrNonFinite 表示聚合值不是有限数。
	ErrNonFinite = errors.New("dashboard: non-finite value")
	// ErrInvalidRule 表示告警规则缺少必要字段或存在冲突。
	ErrInvalidRule = errors.New("dashboard: invalid alert rule")
)

// Aggregate 是一次快照中某个指标跨分片的聚合结果。
type Aggregate struct {
	Count int
	Sum   float64
	Min   float64
	Max   float64
}

// Avg 返回算术平均值；Count 为 0 时返回 0。
func (a Aggregate) Avg() float64 {
	if a.Count == 0 {
		return 0
	}
	return a.Sum / float64(a.Count)
}

// Aggregator 按“指标 + 分片”保存最新值，读取时再聚合。
//
// 保存最新值而不是累加历史，是为了让看板展示“当前状态”；
// 历史趋势应该由 Prometheus 这类时序库负责，而不是看板进程。
type Aggregator struct {
	mu     sync.RWMutex
	series map[string]map[string]float64
}

// NewAggregator 创建一个空聚合器。
func NewAggregator() *Aggregator {
	return &Aggregator{series: make(map[string]map[string]float64)}
}

// Set 覆盖某个分片的指标值。
func (a *Aggregator) Set(metric, shard string, value float64) error {
	if metric == "" || shard == "" {
		return ErrInvalidSeries
	}
	if math.IsNaN(value) || math.IsInf(value, 0) {
		return ErrNonFinite
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	shards := a.series[metric]
	if shards == nil {
		shards = make(map[string]float64)
		a.series[metric] = shards
	}
	shards[shard] = value
	return nil
}

// Delete 移除某个分片；指标下没有分片时一并移除指标。
func (a *Aggregator) Delete(metric, shard string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	shards := a.series[metric]
	if shards == nil {
		return
	}
	delete(shards, shard)
	if len(shards) == 0 {
		delete(a.series, metric)
	}
}

// Snapshot 返回所有指标的聚合副本。
func (a *Aggregator) Snapshot() map[string]Aggregate {
	a.mu.RLock()
	defer a.mu.RUnlock()
	out := make(map[string]Aggregate, len(a.series))
	for metric, shards := range a.series {
		agg := Aggregate{Min: math.Inf(1), Max: math.Inf(-1)}
		for _, value := range shards {
			agg.Count++
			agg.Sum += value
			if value < agg.Min {
				agg.Min = value
			}
			if value > agg.Max {
				agg.Max = value
			}
		}
		if agg.Count == 0 {
			agg.Min, agg.Max = 0, 0
		}
		out[metric] = agg
	}
	return out
}

// Metric 返回单个指标的聚合结果。
func (a *Aggregator) Metric(metric string) (Aggregate, bool) {
	agg, ok := a.Snapshot()[metric]
	return agg, ok
}

// file: ch15/dashboard/observe.go
// tier: L1
// verify: go test ./ch15/dashboard/

package dashboard

import (
	"math"
	"time"
)

// Observe 提交某个指标的一次观测，返回本次产生的状态迁移事件。
//
// 高于等于阈值时进入 pending；持续达到 For 才转为 firing，
// 且只通知一次。低于阈值时，firing 会发出一次恢复事件，
// pending 则静默复位，避免短时抖动产生告警。
func (e *Evaluator) Observe(metric string, value float64) []Event {
	if math.IsNaN(value) || math.IsInf(value, 0) {
		return nil
	}
	e.mu.Lock()
	indices := e.byMetric[metric]
	if len(indices) == 0 {
		e.mu.Unlock()
		return nil
	}
	now := e.now()
	var events []Event
	for _, idx := range indices {
		rule := e.rules[idx]
		rs := &e.states[idx]
		if value >= rule.Threshold {
			if rs.state == StateOK {
				rs.state = StatePending
				rs.since = now
			}
			if rs.state == StatePending && now.Sub(rs.since) >= rule.For {
				rs.state = StateFiring
				events = append(events, Event{
					Rule: rule.Name, Metric: metric, State: StateFiring, Value: value, At: now,
				})
			}
			continue
		}
		if rs.state == StateFiring {
			events = append(events, Event{
				Rule: rule.Name, Metric: metric, State: StateOK, Value: value, At: now,
			})
		}
		rs.state = StateOK
		rs.since = time.Time{}
	}
	notify := e.notify
	e.mu.Unlock()

	if notify != nil {
		for _, event := range events {
			notify(event)
		}
	}
	return events
}