From 94f30165f978b2e1697fb32ecab86ce17c38d9c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mats=20T=C3=B6rnberg?= Date: Sat, 19 Sep 2026 22:00:19 +0200 Subject: [PATCH] more ml --- cmd/spike-alerts/main.go | 177 +++++++++++++++++++++++++++ internal/features/spike.go | 103 ++++++++++++++-- internal/features/spike_test.go | 38 ++++++ internal/kubernetes/watcher.go | 118 ++++++++++++++---- internal/kubernetes/watcher_test.go | 182 ++++++++++++++++++++++++++++ 5 files changed, 584 insertions(+), 34 deletions(-) create mode 100644 cmd/spike-alerts/main.go create mode 100644 internal/kubernetes/watcher_test.go diff --git a/cmd/spike-alerts/main.go b/cmd/spike-alerts/main.go new file mode 100644 index 0000000..efe4c8b --- /dev/null +++ b/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 +} diff --git a/internal/features/spike.go b/internal/features/spike.go index ebcf693..c4d92b9 100644 --- a/internal/features/spike.go +++ b/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 +} diff --git a/internal/features/spike_test.go b/internal/features/spike_test.go index fbe4baf..3571262 100644 --- a/internal/features/spike_test.go +++ b/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") + } +} diff --git a/internal/kubernetes/watcher.go b/internal/kubernetes/watcher.go index 85d15ef..cdfaccd 100644 --- a/internal/kubernetes/watcher.go +++ b/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) { diff --git a/internal/kubernetes/watcher_test.go b/internal/kubernetes/watcher_test.go new file mode 100644 index 0000000..183ef0a --- /dev/null +++ b/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") + } +}