《后端架构设计·演进与治理》6.3 从信号到诊断：一次跨服务排障复盘

// file: internal/platform/obs/correlate.go
// tier: L2
// verify: go test ./internal/platform/obs/ -count=1
package obs

import (
	"fmt"
	"math"
	"sort"
	"strings"
	"time"
)

// EventKind is one observability signal or change record.
type EventKind string

const (
	EventLog    EventKind = "log"
	EventMetric EventKind = "metric"
	EventSpan   EventKind = "span"
	EventChange EventKind = "change"
)

// Event is a normalized record. Only anomalies and changes are fed to the
// diagnoser; normal telemetry stays in the warehouse.
type Event struct {
	Time    time.Time
	Kind    EventKind
	Service string
	Name    string
	TraceID string
	Value   float64
	Attrs   map[string]string
}

// Correlator stores a small incident slice and answers join questions.
type Correlator struct{ events []Event }

func NewCorrelator() *Correlator  { return &Correlator{} }
func (c *Correlator) Add(e Event) { c.events = append(c.events, e) }

// Events returns a stable copy sorted by time; equal timestamps keep insertion
// order, so replaying the same input produces the same timeline.
func (c *Correlator) Events() []Event {
	out := append([]Event(nil), c.events...)
	sort.SliceStable(out, func(i, j int) bool { return out[i].Time.Before(out[j].Time) })
	return out
}

// ByTrace joins direct trace IDs and metric exemplars.
func (c *Correlator) ByTrace(traceID string) []Event {
	if traceID == "" {
		return nil
	}
	var out []Event
	for _, e := range c.Events() {
		if e.TraceID == traceID || e.Attrs["trace_id"] == traceID {
			out = append(out, e)
		}
	}
	return out
}

// ExemplarTraces joins a metric to traces through its trace_id attribute.
func (c *Correlator) ExemplarTraces(metric string, since time.Time) []string {
	seen := make(map[string]bool)
	var out []string
	for _, e := range c.Events() {
		if e.Kind != EventMetric || e.Name != metric || e.Time.Before(since) {
			continue
		}
		id := e.Attrs["trace_id"]
		if id != "" && !seen[id] {
			seen[id] = true
			out = append(out, id)
		}
	}
	sort.Strings(out)
	return out
}

// TimeToCorrelate measures alert-to-first-span for one trace.
func (c *Correlator) TimeToCorrelate(alert time.Time, traceID string) (time.Duration, bool) {
	for _, e := range c.Events() {
		if e.TraceID == traceID && e.Kind == EventSpan && !e.Time.Before(alert) {
			return e.Time.Sub(alert), true
		}
	}
	return 0, false
}

// Hypothesis is one candidate cause; Weight is the prior, not the final score.
type Hypothesis struct {
	Name     string
	Keywords []string
	Weight   float64
}

type Candidate struct {
	Name     string
	Score    float64
	Events   int
	Signals  map[EventKind]int
	Evidence []string
}

type Diagnosis struct {
	Candidates []Candidate
	Ambiguous  bool
	Reason     string
}

// Diagnoser ranks causes. HalfLife decays evidence away from the onset;
// minEvidence suppresses one-off coincidences.
type Diagnoser struct {
	hypotheses  []Hypothesis
	halfLife    time.Duration
	lookback    time.Duration
	minEvidence int
}

func NewDiagnoser(halfLife time.Duration) *Diagnoser {
	if halfLife <= 0 {
		halfLife = 10 * time.Minute
	}
	return &Diagnoser{halfLife: halfLife, lookback: 30 * time.Minute, minEvidence: 2}
}

func (d *Diagnoser) Add(h Hypothesis) {
	if h.Weight <= 0 {
		h.Weight = 1
	}
	d.hypotheses = append(d.hypotheses, h)
}

// Diagnose scores every hypothesis against the incident slice.
func (d *Diagnoser) Diagnose(onset time.Time, events []Event) Diagnosis {
	var out []Candidate
	for _, h := range d.hypotheses {
		signals := make(map[EventKind]int)
		var score float64
		var evidence []string
		count := 0
		for _, e := range events {
			if e.Time.Before(onset.Add(-d.lookback)) || e.Time.After(onset.Add(d.lookback)) || !matches(h.Keywords, e) {
				continue
			}
			dt := e.Time.Sub(onset)
			if dt < 0 {
				dt = -dt
			}
			score += h.Weight * math.Exp(-float64(dt)/float64(d.halfLife))
			signals[e.Kind]++
			count++
			if len(evidence) < 5 {
				evidence = append(evidence, fmt.Sprintf("%s %s %s (%+.0fs)", e.Kind, e.Service, e.Name, e.Time.Sub(onset).Seconds()))
			}
		}
		if count < d.minEvidence {
			continue
		}
		score *= 1 + 0.15*float64(len(signals)-1) // independent signals are stronger
		out = append(out, Candidate{
			Name: h.Name, Score: math.Round(score*1000) / 1000,
			Events: count, Signals: signals, Evidence: evidence,
		})
	}
	sort.SliceStable(out, func(i, j int) bool {
		if out[i].Score != out[j].Score {
			return out[i].Score > out[j].Score
		}
		return out[i].Name < out[j].Name
	})
	diag := Diagnosis{Candidates: out}
	if len(out) >= 2 && out[0].Score > 0 && out[1].Score/out[0].Score >= 0.9 {
		diag.Ambiguous = true
		diag.Reason = fmt.Sprintf("top two within 10%%: %s=%.3f %s=%.3f", out[0].Name, out[0].Score, out[1].Name, out[1].Score)
	}
	return diag
}

func matches(keywords []string, e Event) bool {
	for _, kw := range keywords {
		if strings.Contains(e.Name, kw) {
			return true
		}
		for k, v := range e.Attrs {
			if k == kw || strings.Contains(v, kw) {
				return true
			}
		}
	}
	return false
}

// ManualClock is a fixed, advanceable clock for replay tests.
type ManualClock struct{ now time.Time }

func NewManualClock(t time.Time) *ManualClock  { return &ManualClock{now: t} }
func (c *ManualClock) Now() time.Time          { return c.now }
func (c *ManualClock) Advance(d time.Duration) { c.now = c.now.Add(d) }

// file: internal/platform/obs/correlate_test.go
// tier: L2
// verify: go test ./internal/platform/obs/ -count=1
package obs

import (
	"fmt"
	"strings"
	"testing"
	"time"
)

func incidentCorrelator() (*Correlator, time.Time) {
	onset := time.Date(2026, 10, 29, 10, 0, 0, 0, time.UTC)
	clock := NewManualClock(onset)
	c := NewCorrelator()
	add := func(offset time.Duration, e Event) {
		e.Time = clock.Now().Add(offset)
		c.Add(e)
	}
	add(-2*time.Minute, Event{Kind: EventChange, Service: "order", Name: "deploy.started", Attrs: map[string]string{"version": "v42"}})
	add(-80*time.Second, Event{Kind: EventChange, Service: "order", Name: "deploy.completed"})
	add(-5*time.Minute, Event{Kind: EventChange, Service: "order", Name: "config.apply"})
	add(-290*time.Second, Event{Kind: EventChange, Service: "order", Name: "feature.flag.checkout"})
	add(5*time.Second, Event{Kind: EventLog, Service: "alertmanager", Name: "alert.fired"})
	add(10*time.Second, Event{Kind: EventMetric, Service: "payment", Name: "payment.latency.p99", Attrs: map[string]string{"trace_id": "tr-1001"}})
	add(12*time.Second, Event{Kind: EventSpan, Service: "payment", Name: "payment.authorize", TraceID: "tr-1001"})
	add(12*time.Second, Event{Kind: EventLog, Service: "order", Name: "order.checkout.slow", TraceID: "tr-1001"})
	add(13*time.Second, Event{Kind: EventLog, Service: "payment", Name: "payment.timeout", TraceID: "tr-1001"})
	add(14*time.Second, Event{Kind: EventSpan, Service: "inventory", Name: "inventory.hold", TraceID: "tr-1001"})
	add(60*time.Second, Event{Kind: EventMetric, Service: "order", Name: "order.error.rate", Attrs: map[string]string{"trace_id": "tr-1001"}})
	return c, onset
}

func incidentDiagnoser() *Diagnoser {
	d := NewDiagnoser(10 * time.Minute)
	d.Add(Hypothesis{Name: "payment_dependency", Keywords: []string{"payment.authorize", "payment.timeout", "payment.latency"}, Weight: 1})
	d.Add(Hypothesis{Name: "deploy_regression", Keywords: []string{"deploy", "release"}, Weight: 0.8})
	d.Add(Hypothesis{Name: "config_change", Keywords: []string{"config.apply", "feature.flag"}, Weight: 0.6})
	d.Add(Hypothesis{Name: "db_saturation", Keywords: []string{"db.pool", "db.query"}, Weight: 0.9})
	return d
}

func TestDiagnosePaymentDependency(t *testing.T) {
	c, onset := incidentCorrelator()
	diag := incidentDiagnoser().Diagnose(onset, c.Events())
	if diag.Ambiguous || len(diag.Candidates) != 3 || diag.Candidates[0].Name != "payment_dependency" {
		t.Fatalf("diagnosis = %+v", diag)
	}
	for _, cand := range diag.Candidates {
		t.Logf("candidate=%s score=%.3f events=%d signals=%v", cand.Name, cand.Score, cand.Events, cand.Signals)
	}
	if got := diag.Candidates[0].Signals; got[EventMetric] != 1 || got[EventSpan] != 1 || got[EventLog] != 1 {
		t.Fatalf("payment signals = %v", got)
	}
}

func TestExemplarAndTimeline(t *testing.T) {
	c, onset := incidentCorrelator()
	traces := c.ExemplarTraces("payment.latency.p99", onset.Add(-time.Minute))
	if len(traces) != 1 || traces[0] != "tr-1001" || len(c.ByTrace("tr-1001")) != 6 {
		t.Fatalf("exemplars=%v joined=%d", traces, len(c.ByTrace("tr-1001")))
	}
	if d, ok := c.TimeToCorrelate(onset.Add(5*time.Second), "tr-1001"); !ok || d != 7*time.Second {
		t.Fatalf("alert to trace = %s", d)
	}
	events := c.Events()
	var at12 []string
	for _, e := range events {
		if e.Time.Equal(onset.Add(12 * time.Second)) {
			at12 = append(at12, e.Name)
		}
	}
	if len(events) != 11 || len(at12) != 2 || at12[0] != "payment.authorize" || at12[1] != "order.checkout.slow" {
		t.Fatalf("timeline: events=%d at12=%v", len(events), at12)
	}
	t.Logf("exemplars=%v joined=6 alert_to_span=7s events=%d stable=%v", traces, len(events), at12)
}

func TestDeterministicAndAmbiguous(t *testing.T) {
	c, onset := incidentCorrelator()
	d := incidentDiagnoser()
	first := fmt.Sprint(d.Diagnose(onset, c.Events()))
	if first != fmt.Sprint(d.Diagnose(onset, c.Events())) {
		t.Fatalf("diagnosis changed between runs")
	}
	base := time.Date(2026, 10, 29, 10, 0, 0, 0, time.UTC)
	events := []Event{
		{Time: base, Kind: EventLog, Service: "a", Name: "symptom.one"},
		{Time: base.Add(time.Second), Kind: EventLog, Service: "a", Name: "symptom.two"},
	}
	amb := NewDiagnoser(10 * time.Minute)
	amb.Add(Hypothesis{Name: "cause_a", Keywords: []string{"symptom"}, Weight: 1})
	amb.Add(Hypothesis{Name: "cause_b", Keywords: []string{"symptom"}, Weight: 1})
	diag := amb.Diagnose(base, events)
	if !diag.Ambiguous || !strings.Contains(diag.Reason, "within 10%") {
		t.Fatalf("ambiguity = %+v", diag)
	}
	t.Logf("stable_bytes=%d ambiguous=%v reason=%s", len(first), diag.Ambiguous, diag.Reason)
}

$ go version
go version go1.26.0 darwin/arm64
$ gofmt -l .
$ go vet ./...
$ go test ./internal/platform/obs/ -count=1 -v -run 'TestDiagnosePaymentDependency|TestExemplarAndTimeline|TestDeterministicAndAmbiguous'
=== RUN   TestDiagnosePaymentDependency
    correlate_test.go:51: candidate=payment_dependency score=3.825 events=3 signals=map[log:1 metric:1 span:1]
    correlate_test.go:51: candidate=deploy_regression score=1.355 events=2 signals=map[change:2]
    correlate_test.go:51: candidate=config_change score=0.734 events=2 signals=map[change:2]
--- PASS: TestDiagnosePaymentDependency (0.00s)
=== RUN   TestExemplarAndTimeline
    correlate_test.go:77: exemplars=[tr-1001] joined=6 alert_to_span=7s events=11 stable=[payment.authorize order.checkout.slow]
--- PASS: TestExemplarAndTimeline (0.00s)
=== RUN   TestDeterministicAndAmbiguous
    correlate_test.go:99: stable_bytes=417 ambiguous=true reason=top two within 10%: cause_a=1.998 cause_b=1.998
--- PASS: TestDeterministicAndAmbiguous (0.00s)
PASS
ok  	archlab/internal/platform/obs	0.237s