more ml

cmd/spike-alerts/main.go

@@ -0,0 +1,177 @@
+// spike-alerts replays the deterministic spike rule over a bounded window of
+// normalized_events and, optionally, asks the Needle sidecar what kind of
+// failure each top pattern is. It only reads ClickHouse.
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+
+	"git.tornberg.me/go-otel/internal/features"
+)
+
+func main() {
+	addr := flag.String("addr", envOr("OTEL_CLICKHOUSE_ADDR", "127.0.0.1:9000"), "ClickHouse host:port")
+	database := flag.String("database", envOr("OTEL_CLICKHOUSE_DB", "otel"), "ClickHouse database")
+	user := flag.String("user", envOr("OTEL_CLICKHOUSE_USER", "default"), "ClickHouse user")
+	password := flag.String("password", os.Getenv("OTEL_CLICKHOUSE_PASSWORD"), "ClickHouse password")
+	since := flag.String("since", "", "inclusive UTC RFC3339 start of the range to replay")
+	until := flag.String("until", "", "exclusive UTC RFC3339 end of the range")
+	window := flag.Duration("window", 5*time.Minute, "observation window")
+	step := flag.Duration("step", 5*time.Minute, "replay step")
+	baselineWindows := flag.Int("baseline-windows", 36, "preceding windows used as baseline")
+	minExcess := flag.Int("min-excess", 5, "minimum excess errors to alert")
+	multiple := flag.Float64("multiple", 3, "pattern must be novel or >= multiple * its baseline")
+	cooldown := flag.Duration("cooldown", 30*time.Minute, "suppress repeats of a pattern for this long unless its count doubles")
+	needle := flag.String("needle", os.Getenv("OTEL_NEEDLE_ADDR"), "Needle sidecar URL; empty skips classification")
+	asJSON := flag.Bool("json", false, "emit JSON lines instead of text")
+	flag.Parse()
+	if *since == "" || *until == "" {
+		log.Fatal("-since and -until are required")
+	}
+	start, err := time.Parse(time.RFC3339, *since)
+	if err != nil {
+		log.Fatal(err)
+	}
+	end, err := time.Parse(time.RFC3339, *until)
+	if err != nil || !start.Before(end) {
+		log.Fatal("invalid range")
+	}
+	cfg := features.SpikeConfig{Window: *window, BaselineWindows: *baselineWindows, Multiple: *multiple, MinExcess: *minExcess}
+
+	events := load(*addr, *database, *user, *password, start, end)
+	log.Printf("loaded %d error events", len(events))
+
+	cache := map[string]classification{}
+	first := start.Add(time.Duration(*baselineWindows) * *window)
+	suppress := features.NewSuppressor(*cooldown)
+	alerts := 0
+	for at := first.Add(*window); !at.After(end); at = at.Add(*step) {
+		for _, spike := range suppress.Filter(features.DetectSpikes(events, at, cfg)) {
+			alerts++
+			var cls []classification
+			for _, p := range spike.Patterns {
+				c := classification{}
+				if *needle != "" {
+					if hit, ok := cache[p.Signature]; ok {
+						c = hit
+					} else {
+						c = classify(*needle, needleText(p.Example))
+						cache[p.Signature] = c
+					}
+				}
+				cls = append(cls, c)
+			}
+			if *asJSON {
+				_ = json.NewEncoder(os.Stdout).Encode(map[string]any{"spike": spike, "classifications": cls})
+				continue
+			}
+			printSpike(spike, cls)
+		}
+	}
+	log.Printf("%d alert(s) between %s and %s", alerts, first.Format(time.RFC3339), end.Format(time.RFC3339))
+}
+
+type classification struct {
+	Action     string  `json:"action"`
+	Confidence float64 `json:"confidence"`
+	Cached     bool    `json:"cached"`
+	Err        string  `json:"error,omitempty"`
+}
+
+func classify(base, text string) classification {
+	body, _ := json.Marshal(map[string]string{"text": text})
+	resp, err := (&http.Client{Timeout: 60 * time.Second}).Post(strings.TrimRight(base, "/")+"/classify", "application/json", bytes.NewReader(body))
+	if err != nil {
+		return classification{Action: "?", Err: err.Error()}
+	}
+	defer resp.Body.Close()
+	var out classification
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		return classification{Action: "?", Err: err.Error()}
+	}
+	return out
+}
+
+func needleText(e features.Event) string {
+	parts := []string{"Service: " + e.Service, "Kind: " + e.Kind}
+	if e.SpanName != "" {
+		parts = append(parts, "Title: "+e.SpanName)
+	}
+	if e.Message != "" {
+		parts = append(parts, "Message: "+e.Message)
+	}
+	if e.StatusCode > 0 {
+		parts = append(parts, fmt.Sprintf("Status: %d", e.StatusCode))
+	}
+	if e.Route != "" {
+		parts = append(parts, "Route: "+e.Route)
+	}
+	return strings.Join(parts, " | ")
+}
+
+func printSpike(s features.Spike, cls []classification) {
+	fmt.Printf("\n%s  %s  errors=%d (baseline %.1f/window, excess %.0f)\n", s.WindowEnd.Format("01-02 15:04"), s.Service, s.ErrorCount, s.BaselineMean, s.Excess)
+	for i, p := range s.Patterns {
+		tag := "rising"
+		if p.Novel {
+			tag = "NEW"
+		}
+		what := "-"
+		if cls[i].Action != "" {
+			what = fmt.Sprintf("%s (%.2f)", cls[i].Action, cls[i].Confidence)
+		}
+		fmt.Printf("  %-6s x%-4d base %-6.1f %-24s %s\n", tag, p.Count, p.BaselineMean, what, needleText(p.Example))
+	}
+}
+
+func load(addr, database, user, password string, start, end time.Time) []features.Event {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+	conn, err := clickhouse.Open(&clickhouse.Options{Addr: []string{addr}, Auth: clickhouse.Auth{Database: database, Username: user, Password: password}})
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer conn.Close()
+	// Errors only, and only columns present in every deployed schema version.
+	rows, err := conn.Query(ctx, fmt.Sprintf(`SELECT timestamp, service, release, kind, source, status_code, span_name, route, exception_type, message_normalized, pattern_signature
+		FROM %s.normalized_events FINAL
+		WHERE is_error = 1 AND timestamp >= toDateTime64('%s', 9, 'UTC') AND timestamp < toDateTime64('%s', 9, 'UTC')
+		ORDER BY timestamp LIMIT 2000000`, database, chTime(start), chTime(end)))
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer rows.Close()
+	var out []features.Event
+	for rows.Next() {
+		e := features.Event{IsError: true}
+		var status uint16
+		if err := rows.Scan(&e.Timestamp, &e.Service, &e.Release, &e.Kind, &e.Source, &status, &e.SpanName, &e.Route, &e.ExceptionType, &e.Message, &e.PatternSignature); err != nil {
+			log.Fatal(err)
+		}
+		e.StatusCode = int(status)
+		out = append(out, e)
+	}
+	if err := rows.Err(); err != nil {
+		log.Fatal(err)
+	}
+	return out
+}
+
+func chTime(t time.Time) string { return t.UTC().Format("2006-01-02 15:04:05.000000000") }
+func envOr(k, f string) string {
+	if v := os.Getenv(k); v != "" {
+		return v
+	}
+	return f
+}

internal/features/spike.go

@@ -66,12 +66,14 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike {
 	baseStart := start.Add(-time.Duration(cfg.BaselineWindows) * cfg.Window)
 
 	type agg struct {
-		cur, base int
-		example   Event
+		cur     int
+		base    []int // per baseline window
+		example Event
 	}
 	type svc struct {
-		cur, base int
-		patterns  map[string]*agg
+		cur      int
+		base     []int // per baseline window
+		patterns map[string]*agg
 	}
 	services := map[string]*svc{}
 	for _, e := range events {
@@ -90,7 +92,7 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike {
 		}
 		s := services[name]
 		if s == nil {
-			s = &svc{patterns: map[string]*agg{}}
+			s = &svc{base: make([]int, cfg.BaselineWindows), patterns: map[string]*agg{}}
 			services[name] = s
 		}
 		key := e.PatternSignature
@@ -99,7 +101,7 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike {
 		}
 		p := s.patterns[key]
 		if p == nil {
-			p = &agg{}
+			p = &agg{base: make([]int, cfg.BaselineWindows)}
 			s.patterns[key] = p
 		}
 		if inCur {
@@ -107,8 +109,12 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike {
 			p.cur++
 			p.example = e
 		} else {
-			s.base++
-			p.base++
+			idx := int(ts.Sub(baseStart) / cfg.Window)
+			if idx >= cfg.BaselineWindows {
+				idx = cfg.BaselineWindows - 1
+			}
+			s.base[idx]++
+			p.base[idx]++
 			if p.example.PatternSignature == "" && p.cur == 0 {
 				p.example = e
 			}
@@ -116,24 +122,24 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike {
 	}
 
 	var spikes []Spike
-	n := float64(cfg.BaselineWindows)
 	for name, s := range services {
-		baseMean := float64(s.base) / n
+		baseMean := trimmedMean(s.base, baselineTrim)
 		if s.cur == 0 {
 			continue
 		}
 		spike := Spike{Service: name, WindowStart: start, WindowEnd: at, ErrorCount: s.cur, BaselineMean: baseMean}
 		for sig, p := range s.patterns {
-			mean := float64(p.base) / n
+			mean := trimmedMean(p.base, baselineTrim)
+			seen := sum(p.base) > 0
 			excess := float64(p.cur) - mean
 			// A pattern only counts when it is novel or well above its own
 			// baseline; the service-wide total is masked by chronic noise.
-			if p.cur == 0 || excess <= 0 || (p.base > 0 && float64(p.cur) < cfg.Multiple*mean) {
+			if p.cur == 0 || excess <= 0 || (seen && float64(p.cur) < cfg.Multiple*mean) {
 				continue
 			}
 			spike.Excess += excess
 			spike.Patterns = append(spike.Patterns, PatternDelta{
-				Signature: sig, Count: p.cur, BaselineMean: mean, Excess: excess, Novel: p.base == 0, Example: p.example,
+				Signature: sig, Count: p.cur, BaselineMean: mean, Excess: excess, Novel: !seen, Example: p.example,
 			})
 		}
 		if spike.Excess < float64(cfg.MinExcess) {
@@ -153,3 +159,74 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike {
 	sort.Slice(spikes, func(i, j int) bool { return spikes[i].Excess > spikes[j].Excess })
 	return spikes
 }
+
+// baselineTrim is the fraction of the busiest baseline windows discarded
+// before averaging, so one earlier storm cannot inflate the baseline and hide
+// the next one.
+const baselineTrim = 0.2
+
+func sum(v []int) (t int) {
+	for _, x := range v {
+		t += x
+	}
+	return t
+}
+
+func trimmedMean(counts []int, trim float64) float64 {
+	if len(counts) == 0 {
+		return 0
+	}
+	sorted := append([]int(nil), counts...)
+	sort.Ints(sorted)
+	keep := len(sorted) - int(float64(len(sorted))*trim)
+	if keep < 1 {
+		keep = 1
+	}
+	return float64(sum(sorted[:keep])) / float64(keep)
+}
+
+// Suppressor stops the same incident from alerting in every consecutive
+// window. A pattern re-alerts once its cooldown has elapsed, or sooner when
+// its count at least doubles (an escalation).
+type Suppressor struct {
+	Cooldown time.Duration
+	last     map[string]suppressed
+}
+
+type suppressed struct {
+	at    time.Time
+	count int
+}
+
+func NewSuppressor(cooldown time.Duration) *Suppressor {
+	if cooldown <= 0 {
+		cooldown = 30 * time.Minute
+	}
+	return &Suppressor{Cooldown: cooldown, last: map[string]suppressed{}}
+}
+
+// Filter returns spikes reduced to the patterns that are new or escalated;
+// spikes with no remaining patterns are dropped.
+func (s *Suppressor) Filter(spikes []Spike) []Spike {
+	var out []Spike
+	for _, spike := range spikes {
+		var kept []PatternDelta
+		var excess float64
+		for _, p := range spike.Patterns {
+			key := spike.Service + "\x00" + p.Signature
+			prev, ok := s.last[key]
+			if ok && spike.WindowEnd.Sub(prev.at) < s.Cooldown && p.Count < 2*prev.count {
+				continue
+			}
+			s.last[key] = suppressed{at: spike.WindowEnd, count: p.Count}
+			kept = append(kept, p)
+			excess += p.Excess
+		}
+		if len(kept) == 0 {
+			continue
+		}
+		spike.Patterns, spike.Excess = kept, excess
+		out = append(out, spike)
+	}
+	return out
+}

internal/features/spike_test.go

@@ -39,3 +39,41 @@ func TestDetectSpikesRequiresMinimumExcess(t *testing.T) {
 		t.Fatalf("3 errors should stay under the floor: %+v", got)
 	}
 }
+
+func TestBaselineTrimIgnoresEarlierStorm(t *testing.T) {
+	at := time.Date(2026, 9, 19, 12, 0, 0, 0, time.UTC)
+	cfg := SpikeConfig{Window: 5 * time.Minute, BaselineWindows: 10}
+	var events []Event
+	add := func(sig string, n int, when time.Time) {
+		for i := 0; i < n; i++ {
+			events = append(events, Event{Timestamp: when.Add(time.Duration(i) * time.Millisecond), Service: "api", IsError: true, PatternSignature: sig})
+		}
+	}
+	// one earlier storm in 1 of 10 baseline windows, then a smaller repeat now
+	add("storm", 1000, at.Add(-6*cfg.Window))
+	add("storm", 60, at.Add(-cfg.Window))
+	got := DetectSpikes(events, at, cfg)
+	if len(got) != 1 || got[0].Patterns[0].BaselineMean != 0 {
+		t.Fatalf("storm baseline should be trimmed to 0, got %+v", got)
+	}
+}
+
+func TestSuppressorCooldownAndEscalation(t *testing.T) {
+	at := time.Date(2026, 9, 19, 12, 0, 0, 0, time.UTC)
+	spike := func(at time.Time, count int) []Spike {
+		return []Spike{{Service: "api", WindowEnd: at, Patterns: []PatternDelta{{Signature: "x", Count: count, Excess: float64(count)}}}}
+	}
+	s := NewSuppressor(30 * time.Minute)
+	if len(s.Filter(spike(at, 20))) != 1 {
+		t.Fatal("first alert should pass")
+	}
+	if len(s.Filter(spike(at.Add(5*time.Minute), 22))) != 0 {
+		t.Fatal("repeat inside cooldown should be suppressed")
+	}
+	if len(s.Filter(spike(at.Add(10*time.Minute), 45))) != 1 {
+		t.Fatal("2x escalation should re-alert")
+	}
+	if len(s.Filter(spike(at.Add(50*time.Minute), 46))) != 1 {
+		t.Fatal("alert after cooldown should pass")
+	}
+}

internal/kubernetes/watcher.go

@@ -28,8 +28,12 @@ const maxRollouts = 500
 type Watcher struct {
 	mu       sync.RWMutex
 	rollouts map[string]Rollout
-	enabled  bool
-	err      string
+	// revisions maps "namespace/deployment/revision" to the creation time of
+	// the ReplicaSet backing that revision - the cluster's own record of when
+	// each rollout started, so times survive watcher restarts.
+	revisions map[string]int64
+	enabled   bool
+	err       string
 }
 
 // NewFromEnv enables observation in-cluster, or with KUBECONFIG locally. It is
@@ -50,7 +54,7 @@ func NewFromEnv(ctx context.Context) (*Watcher, error) {
 		return nil, fmt.Errorf("create kubernetes client: %w", err)
 	}
 
-	w := &Watcher{rollouts: make(map[string]Rollout), enabled: true}
+	w := &Watcher{rollouts: make(map[string]Rollout), revisions: make(map[string]int64), enabled: true}
 	namespace := os.Getenv("OTEL_KUBERNETES_NAMESPACE")
 	selector := os.Getenv("OTEL_KUBERNETES_LABEL_SELECTOR")
 	factory := informers.NewSharedInformerFactoryWithOptions(client, 30*time.Second,
@@ -69,9 +73,14 @@ func NewFromEnv(ctx context.Context) (*Watcher, error) {
 		AddFunc:    func(obj interface{}) { w.observe(obj) },
 		UpdateFunc: func(_, obj interface{}) { w.observe(obj) },
 	})
+	rsInformer := factory.Apps().V1().ReplicaSets().Informer()
+	_, _ = rsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    func(obj interface{}) { w.observeReplicaSet(obj) },
+		UpdateFunc: func(_, obj interface{}) { w.observeReplicaSet(obj) },
+	})
 	go func() {
 		factory.Start(ctx.Done())
-		if !cache.WaitForCacheSync(ctx.Done(), informer.HasSynced) {
+		if !cache.WaitForCacheSync(ctx.Done(), informer.HasSynced, rsInformer.HasSynced) {
 			w.setError("deployment informer did not sync")
 			return
 		}
@@ -107,19 +116,42 @@ func (w *Watcher) observe(obj interface{}) {
 	}
 	key := deployment.Namespace + "/" + deployment.Name + "/" + revision
 	phase, reason, message, completed := rolloutStatus(deployment)
-	started := now
-	for _, condition := range deployment.Status.Conditions {
-		if condition.Type == appsv1.DeploymentProgressing && !condition.LastUpdateTime.IsZero() {
-			started = condition.LastUpdateTime.UnixNano()
-		}
-		if condition.Type == appsv1.DeploymentAvailable && !condition.LastUpdateTime.IsZero() && completed {
-			completed = condition.LastUpdateTime.UnixNano() != 0
+
+	// StartedAt is when the rollout began. The ReplicaSet for this revision was
+	// created at that moment, so prefer the cluster's own timestamp; it is also
+	// correct for rollouts first observed long after they happened. Fall back
+	// to first-seen via the Progressing condition, then to this event.
+	started := w.revisionCreated(deployment.Namespace, deployment.Name, revision)
+	if started == 0 {
+		if old, exists := w.rollouts[key]; exists && old.StartedAt != 0 {
+			started = old.StartedAt
+		} else {
+			for _, condition := range deployment.Status.Conditions {
+				if condition.Type == appsv1.DeploymentProgressing && !condition.LastUpdateTime.IsZero() {
+					started = condition.LastUpdateTime.UnixNano()
+				}
+			}
+			if started == 0 {
+				started = now
+			}
 		}
 	}
-	if old, exists := w.rollouts[key]; exists {
-		started = old.StartedAt
-		if old.CompletedAt != 0 {
-			completed = true
+	// CompletedAt prefers the Progressing condition's own update time for a
+	// finished rollout, and once set is never re-bumped: a completed rollout
+	// stays completed at the moment it completed.
+	completedAt := int64(0)
+	if completed {
+		for _, condition := range deployment.Status.Conditions {
+			if condition.Type == appsv1.DeploymentProgressing && condition.Status == corev1.ConditionTrue && condition.Reason == "NewReplicaSetAvailable" {
+				completedAt = condition.LastUpdateTime.UnixNano()
+			}
+		}
+		if completedAt == 0 {
+			if old, exists := w.rollouts[key]; exists && old.CompletedAt != 0 {
+				completedAt = old.CompletedAt
+			} else {
+				completedAt = now
+			}
 		}
 	}
 	rollout := Rollout{
@@ -130,11 +162,15 @@ func (w *Watcher) observe(obj interface{}) {
 		Message: message, StartedAt: started, LastSeen: now, Generation: deployment.Generation,
 		DesiredReplicas: desiredReplicas(deployment), UpdatedReplicas: deployment.Status.UpdatedReplicas,
 		AvailableReplicas: deployment.Status.AvailableReplicas, ReadyReplicas: deployment.Status.ReadyReplicas,
-	}
-	if completed {
-		rollout.CompletedAt = now
+		CompletedAt: completedAt,
 	}
 	w.mu.Lock()
+	// The informer re-delivers every deployment on each resync (~30s). Only an
+	// observable change advances LastSeen, so a quiet rollout keeps its real
+	// last-changed time instead of looking perpetually fresh.
+	if old, exists := w.rollouts[key]; exists && !rolloutChanged(old, rollout) {
+		rollout.LastSeen = old.LastSeen
+	}
 	w.rollouts[key] = rollout
 	if len(w.rollouts) > maxRollouts {
 		w.evictOldestLocked()
@@ -142,12 +178,47 @@ func (w *Watcher) observe(obj interface{}) {
 	w.mu.Unlock()
 }
 
+func (w *Watcher) observeReplicaSet(obj interface{}) {
+	rs, ok := obj.(*appsv1.ReplicaSet)
+	if !ok {
+		return
+	}
+	revision := rs.Annotations["deployment.kubernetes.io/revision"]
+	owner := ""
+	for _, ref := range rs.OwnerReferences {
+		if ref.Kind == "Deployment" && ref.Name != "" {
+			owner = ref.Name
+		}
+	}
+	if revision == "" || owner == "" || rs.CreationTimestamp.IsZero() {
+		return
+	}
+	w.mu.Lock()
+	w.revisions[rs.Namespace+"/"+owner+"/"+revision] = rs.CreationTimestamp.UnixNano()
+	w.mu.Unlock()
+}
+
+func (w *Watcher) revisionCreated(namespace, workload, revision string) int64 {
+	w.mu.RLock()
+	defer w.mu.RUnlock()
+	return w.revisions[namespace+"/"+workload+"/"+revision]
+}
+
+// rolloutChanged reports whether anything observable about the rollout moved.
+func rolloutChanged(a, b Rollout) bool {
+	return a.Phase != b.Phase || a.Reason != b.Reason || a.Message != b.Message ||
+		a.Service != b.Service || a.Release != b.Release || a.Image != b.Image ||
+		a.Generation != b.Generation ||
+		a.DesiredReplicas != b.DesiredReplicas || a.UpdatedReplicas != b.UpdatedReplicas ||
+		a.AvailableReplicas != b.AvailableReplicas || a.ReadyReplicas != b.ReadyReplicas
+}
+
 func (w *Watcher) evictOldestLocked() {
 	var oldest string
 	var ts int64
 	for key, item := range w.rollouts {
-		if oldest == "" || item.LastSeen < ts {
-			oldest, ts = key, item.LastSeen
+		if oldest == "" || item.StartedAt < ts {
+			oldest, ts = key, item.StartedAt
 		}
 	}
 	if oldest != "" {
@@ -163,7 +234,12 @@ func (w *Watcher) List() []Rollout {
 	for _, item := range w.rollouts {
 		result = append(result, item)
 	}
-	sort.Slice(result, func(i, j int) bool { return result[i].LastSeen > result[j].LastSeen })
+	sort.Slice(result, func(i, j int) bool {
+		if result[i].StartedAt != result[j].StartedAt {
+			return result[i].StartedAt > result[j].StartedAt
+		}
+		return result[i].LastSeen > result[j].LastSeen
+	})
 	return result
 }
 func (w *Watcher) Get(id string) (Rollout, bool) {

internal/kubernetes/watcher_test.go

@@ -0,0 +1,182 @@
+package kubernetes
+
+import (
+	"testing"
+	"time"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+// conditionTime is the cluster's own stamp for "this rollout completed",
+// pinned so assertions can compare nanoseconds exactly.
+var conditionTime = metav1.NewTime(time.Unix(1700000000, 0))
+
+func testDeployment(overrides func(*appsv1.Deployment)) *appsv1.Deployment {
+	created := metav1.NewTime(time.Now().Add(-2 * time.Hour))
+	deployReplicas := int32(3)
+	deployment := &appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:         "shop",
+			Name:              "checkout",
+			Generation:        3,
+			CreationTimestamp: created,
+			Annotations: map[string]string{
+				"deployment.kubernetes.io/revision": "7",
+			},
+		},
+		Spec: appsv1.DeploymentSpec{Replicas: &deployReplicas},
+		Status: appsv1.DeploymentStatus{
+			Replicas:          3,
+			UpdatedReplicas:   3,
+			AvailableReplicas: 3,
+			ReadyReplicas:     3,
+			Conditions: []appsv1.DeploymentCondition{
+				{
+					Type:               appsv1.DeploymentProgressing,
+					Status:             corev1.ConditionTrue,
+					Reason:             "NewReplicaSetAvailable",
+					Message:            "ReplicaSet \"checkout-abc\" has successfully progressed.",
+					LastUpdateTime:     conditionTime,
+					LastTransitionTime: conditionTime,
+				},
+				{
+					Type:               appsv1.DeploymentAvailable,
+					Status:             corev1.ConditionTrue,
+					LastUpdateTime:     conditionTime,
+					LastTransitionTime: conditionTime,
+				},
+			},
+		},
+	}
+	if overrides != nil {
+		overrides(deployment)
+	}
+	return deployment
+}
+
+func observeDeployment(w *Watcher, deployment *appsv1.Deployment) {
+	w.observe(deployment)
+}
+
+func TestObserveCompletedRolloutKeepsTimesAcrossResync(t *testing.T) {
+	w := &Watcher{rollouts: make(map[string]Rollout), revisions: make(map[string]int64)}
+
+	observeDeployment(w, testDeployment(nil))
+	first, ok := w.rollouts["shop/checkout/7"]
+	if !ok {
+		t.Fatal("rollout not recorded")
+	}
+	if first.Phase != "completed" {
+		t.Fatalf("phase = %q, want completed", first.Phase)
+	}
+	if first.CompletedAt == 0 {
+		t.Fatal("completed rollout missing CompletedAt")
+	}
+	// The Progressing condition's update time is the cluster's completion stamp,
+	// not "now".
+	if first.CompletedAt != conditionTime.UnixNano() {
+		t.Fatalf("CompletedAt = %d, want the condition time %d", first.CompletedAt, conditionTime.UnixNano())
+	}
+
+	// A resync ~30s later re-delivers the same deployment. A quiet rollout must
+	// keep its times instead of looking perpetually fresh.
+	time.Sleep(2 * time.Millisecond)
+	observeDeployment(w, testDeployment(nil))
+	second := w.rollouts["shop/checkout/7"]
+	if second.LastSeen != first.LastSeen {
+		t.Fatalf("LastSeen advanced from %d to %d on an unchanged resync", first.LastSeen, second.LastSeen)
+	}
+	if second.CompletedAt != first.CompletedAt {
+		t.Fatalf("CompletedAt re-bumped from %d to %d on an unchanged resync", first.CompletedAt, second.CompletedAt)
+	}
+
+	// Any observable change does advance LastSeen.
+	changed := testDeployment(func(d *appsv1.Deployment) {
+		d.Status.ReadyReplicas = 2
+	})
+	observeDeployment(w, changed)
+	third := w.rollouts["shop/checkout/7"]
+	if third.LastSeen == first.LastSeen {
+		t.Fatal("LastSeen did not advance after an observable change")
+	}
+}
+
+func TestObserveStartedAtPrefersReplicaSetCreation(t *testing.T) {
+	w := &Watcher{rollouts: make(map[string]Rollout), revisions: make(map[string]int64)}
+	rolloutStart := time.Now().Add(-48 * time.Hour)
+	w.observeReplicaSet(&appsv1.ReplicaSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:         "shop",
+			Name:              "checkout-abc",
+			CreationTimestamp: metav1.NewTime(rolloutStart),
+			OwnerReferences:   []metav1.OwnerReference{{Kind: "Deployment", Name: "checkout"}},
+			Annotations:       map[string]string{"deployment.kubernetes.io/revision": "7"},
+		},
+	})
+
+	observeDeployment(w, testDeployment(nil))
+	rollout := w.rollouts["shop/checkout/7"]
+	if rollout.StartedAt != rolloutStart.UnixNano() {
+		t.Fatalf("StartedAt = %d, want ReplicaSet creation %d", rollout.StartedAt, rolloutStart.UnixNano())
+	}
+}
+
+func TestObserveFirstSeenBeforeReplicaSetInformerRecovers(t *testing.T) {
+	w := &Watcher{rollouts: make(map[string]Rollout), revisions: make(map[string]int64)}
+	// Deployment events can arrive before the ReplicaSet cache has synced; the
+	// recorded start must be corrected once the ReplicaSet timestamp appears.
+	observeDeployment(w, testDeployment(nil))
+	stale := w.rollouts["shop/checkout/7"]
+
+	rolloutStart := time.Now().Add(-24 * time.Hour)
+	w.observeReplicaSet(&appsv1.ReplicaSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:         "shop",
+			Name:              "checkout-abc",
+			CreationTimestamp: metav1.NewTime(rolloutStart),
+			OwnerReferences:   []metav1.OwnerReference{{Kind: "Deployment", Name: "checkout"}},
+			Annotations:       map[string]string{"deployment.kubernetes.io/revision": "7"},
+		},
+	})
+	observeDeployment(w, testDeployment(nil))
+	rollout := w.rollouts["shop/checkout/7"]
+	if rollout.StartedAt != rolloutStart.UnixNano() {
+		t.Fatalf("StartedAt = %d, want corrected ReplicaSet creation %d", rollout.StartedAt, rolloutStart.UnixNano())
+	}
+	if stale.StartedAt == 0 {
+		t.Fatal("first observation should still record a fallback start")
+	}
+}
+
+func TestListSortsNewestRolloutFirst(t *testing.T) {
+	w := &Watcher{rollouts: make(map[string]Rollout), revisions: make(map[string]int64)}
+	old := testDeployment(nil)
+	old.Annotations["deployment.kubernetes.io/revision"] = "6"
+	old.CreationTimestamp = metav1.NewTime(time.Now().Add(-72 * time.Hour))
+	recent := testDeployment(nil)
+
+	observeDeployment(w, old)
+	observeDeployment(w, recent)
+
+	list := w.List()
+	if len(list) != 2 {
+		t.Fatalf("len(List) = %d, want 2", len(list))
+	}
+	if list[0].Revision != "7" || list[1].Revision != "6" {
+		t.Fatalf("List order = [%s, %s], want newest rollout first", list[0].Revision, list[1].Revision)
+	}
+}
+
+// Deletions and tombstones must not panic the handlers.
+func TestObserveIgnoresNonDeploymentObjects(t *testing.T) {
+	w := &Watcher{rollouts: make(map[string]Rollout), revisions: make(map[string]int64)}
+	w.observe(&unstructured.Unstructured{})
+	w.observeReplicaSet(runtime.Object(&unstructured.Unstructured{}))
+	if len(w.List()) != 0 {
+		t.Fatal("no rollouts should be recorded")
+	}
+}