《游戏服务端架构·平台化与架构治理》11.1 AIOps 与异常检测

// file: internal/aiopsdetect/baseline.go
// tier: L1
// verify: go test ./internal/aiopsdetect/
package aiopsdetect

import (
	"errors"
	"math"
	"sort"
)

// ErrNoSamples means the seasonal bucket has never been observed. Callers
// should treat it as "learn more", not as an incident.
var ErrNoSamples = errors.New("aiopsdetect: no samples for bucket")

// Point is one sampled value. Bucket encodes the season, for example
// "sat-20" for Saturday 20:00 in the game's local time zone.
type Point struct {
	Bucket string
	Value  float64
}

// Baseline keeps one sample window per seasonal bucket. A separate bucket
// for weekday/hour is the simplest way to stop a weekend event from making
// every Tuesday look anomalous.
type Baseline struct {
	MinSamples int
	Samples    map[string][]float64
}

// NewBaseline returns a baseline that will not score a bucket until it has
// MinSamples observations. The default of 7 is one week of the same season.
func NewBaseline(minSamples int) *Baseline {
	if minSamples <= 0 {
		minSamples = 7
	}
	return &Baseline{
		MinSamples: minSamples,
		Samples:    make(map[string][]float64),
	}
}

// Add records one observation. Callers own retention: keeping the last N
// samples per bucket avoids unbounded growth and stale seasonality.
func (b *Baseline) Add(p Point) {
	b.Samples[p.Bucket] = append(b.Samples[p.Bucket], p.Value)
}

// Verdict is the decision the detector returns for one point.
type Verdict string

const (
	// ColdStart means there is not enough history to judge. Do not alert.
	ColdStart Verdict = "cold_start"
	// Normal means the point is inside the learned band.
	Normal Verdict = "normal"
	// Anomaly means the point crossed the configured robust threshold.
	Anomaly Verdict = "anomaly"
)

// Result carries enough context for a human to audit the decision.
type Result struct {
	Verdict Verdict
	Score   float64
	Median  float64
	Scale   float64
	Samples int
}

// Detector scores points against a Baseline using median and MAD. Median and
// MAD are used instead of mean and standard deviation because a single
// outlier in history would otherwise widen the band and hide real incidents.
type Detector struct {
	Baseline  *Baseline
	Threshold float64
	MinScale  float64
}

// NewDetector builds a detector. Threshold is a robust z-score; 3.5 is a
// common starting point and should be tuned against labeled incidents.
func NewDetector(b *Baseline, threshold float64) *Detector {
	if threshold <= 0 {
		threshold = 3.5
	}
	return &Detector{
		Baseline:  b,
		Threshold: threshold,
		MinScale:  1e-9,
	}
}

// Evaluate scores one point. It never returns Anomaly for a cold bucket:
// an empty history is a data gap, not evidence of a fault.
func (d *Detector) Evaluate(p Point) (Result, error) {
	if d.Baseline == nil {
		return Result{}, errors.New("aiopsdetect: nil baseline")
	}
	hist := d.Baseline.Samples[p.Bucket]
	if len(hist) == 0 {
		return Result{}, ErrNoSamples
	}

	res := Result{Samples: len(hist)}
	if len(hist) < d.Baseline.MinSamples {
		res.Verdict = ColdStart
		return res, nil
	}

	med := median(hist)
	scale := 1.4826 * mad(hist, med)
	if scale < d.MinScale {
		scale = d.MinScale
	}

	res.Median = med
	res.Scale = scale
	res.Score = math.Abs(p.Value-med) / scale
	res.Verdict = Normal
	if res.Score >= d.Threshold {
		res.Verdict = Anomaly
	}
	return res, nil
}

// median copies the input so the caller's history order is preserved.
func median(xs []float64) float64 {
	if len(xs) == 0 {
		return 0
	}
	cp := append([]float64(nil), xs...)
	sort.Float64s(cp)

	n := len(cp)
	if n%2 == 1 {
		return cp[n/2]
	}
	return (cp[n/2-1] + cp[n/2]) / 2
}

// mad is the median absolute deviation, a robust stand-in for sigma. The
// 1.4826 factor in Evaluate makes it comparable to a normal distribution.
func mad(xs []float64, med float64) float64 {
	if len(xs) == 0 {
		return 0
	}
	dev := make([]float64, len(xs))
	for i, v := range xs {
		dev[i] = math.Abs(v - med)
	}
	return median(dev)
}

// file: internal/aiopsdetect/suppress.go
// tier: L1
// verify: go test ./internal/aiopsdetect/
package aiopsdetect

import "time"

// Alert is the minimum input for suppression. Fingerprint should be stable
// across restarts, for example "metric:login-fail:shard-7".
type Alert struct {
	Fingerprint string
	At          time.Time
	Severity    string
}

// Suppressor applies three cheap rules before paging a human: silence
// windows, per-fingerprint cooldown, and a sliding-window rate limit.
// It deliberately does not merge different fingerprints; correlation belongs
// to the grouping layer above it.
type Suppressor struct {
	Cooldown     time.Duration
	Window       time.Duration
	MaxPerWindow int

	last     map[string]time.Time
	recent   map[string][]time.Time
	silenced map[string]time.Time
}

// NewSuppressor builds a suppressor. A zero MaxPerWindow disables the rate
// limit; a zero Cooldown disables duplicate suppression.
func NewSuppressor(cooldown, window time.Duration, maxPerWindow int) *Suppressor {
	if cooldown < 0 {
		cooldown = 0
	}
	if window < 0 {
		window = 0
	}
	if maxPerWindow < 0 {
		maxPerWindow = 0
	}
	return &Suppressor{
		Cooldown:     cooldown,
		Window:       window,
		MaxPerWindow: maxPerWindow,
		last:         make(map[string]time.Time),
		recent:       make(map[string][]time.Time),
		silenced:     make(map[string]time.Time),
	}
}

// Allow returns whether the alert should reach the paging path, plus a
// machine-readable reason. Alerts are expected in non-decreasing time order;
// callers replaying historical data must sort them first.
func (s *Suppressor) Allow(a Alert) (bool, string) {
	if a.Fingerprint == "" {
		return false, "empty_fingerprint"
	}
	if until, ok := s.silenced[a.Fingerprint]; ok && a.At.Before(until) {
		return false, "silenced"
	}
	if last, ok := s.last[a.Fingerprint]; ok && a.At.Sub(last) < s.Cooldown {
		return false, "cooldown"
	}
	if s.MaxPerWindow > 0 && s.Window > 0 {
		if len(s.prune(a.Fingerprint, a.At)) >= s.MaxPerWindow {
			return false, "rate_limited"
		}
	}

	s.last[a.Fingerprint] = a.At
	if s.Window > 0 {
		s.recent[a.Fingerprint] = append(s.recent[a.Fingerprint], a.At)
	}
	return true, "accepted"
}

// Silence suppresses one fingerprint for maintenance or a known migration.
// The window must be explicit and short; an open-ended silence hides the
// next real incident.
func (s *Suppressor) Silence(fingerprint string, until time.Time) {
	if s.silenced == nil {
		s.silenced = make(map[string]time.Time)
	}
	s.silenced[fingerprint] = until
}

// prune drops timestamps older than the sliding window and returns what is
// left. It reuses the backing array, so the map stays allocation-light on the
// hot path.
func (s *Suppressor) prune(fingerprint string, at time.Time) []time.Time {
	cut := at.Add(-s.Window)
	kept := s.recent[fingerprint][:0]
	for _, t := range s.recent[fingerprint] {
		if !t.Before(cut) {
			kept = append(kept, t)
		}
	}
	s.recent[fingerprint] = kept
	return kept
}