diff --git a/internal/domain/observation.go b/internal/domain/observation.go index 4630ea5..8ed1a21 100644 --- a/internal/domain/observation.go +++ b/internal/domain/observation.go @@ -118,6 +118,20 @@ func ObservationSignature(message string) string { return s } +// ObservationTaskID reads the task a failure was about out of the message +// itself. The ring is a history: it holds entries from tasks that ended long +// ago, so the worker's currently active task is the wrong answer for most of +// them, and attributing an old failure to whatever is running now would be a +// fabricated association. +func ObservationTaskID(message string) string { + if m := observationTaskID.FindString(message); m != "" { + return strings.ToUpper(m) + } + return "" +} + +var observationTaskID = regexp.MustCompile(`(?i)\b[0-9A-HJKMNP-TV-Z]{26}\b`) + func ValidateObservationIncidentOpened(p map[string]any) error { if id, _ := p["observation_id"].(string); strings.TrimSpace(id) == "" { return fmt.Errorf("%w: observation_id required", ErrInvalid) diff --git a/internal/operations/observations.go b/internal/operations/observations.go index a3f6d00..eead7a1 100644 --- a/internal/operations/observations.go +++ b/internal/operations/observations.go @@ -36,11 +36,25 @@ type ObservationTracker struct { // restarting. Reported 34, evicted, reported 3 again means 37 occurrences, // not 3. In-memory: a restart loses the accumulation, never the incident. counts map[string]int + // seen is the high-water mark per worker and signature, kept after an + // incident closes. The ring is a history, so a closed incident's entry + // keeps being reported for as long as it survives eviction; without this, + // every heartbeat after a quiet timeout opened the same incident again and + // manufactured recurrence out of one old failure. Live on the first real + // run: three signatures, four incidents each, none of them a new event. + seen map[string]watermark // incarnations is the last incarnation seen per worker, which is what makes // a restart detectable at all. incarnations map[string]string } +// watermark is the last occurrence this tracker accounted for one worker's +// signature, whether or not its incident is still open. +type watermark struct { + last time.Time + count int +} + // WorkerReport is one heartbeat's worth of attributed observations. The // coordinator attributes them, because the ring carries only messages: the // worker's active task and that task's current lease epoch are what bind an @@ -63,6 +77,7 @@ func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { } if t.counts == nil { t.counts, t.incarnations = map[string]int{}, map[string]string{} + t.seen = map[string]watermark{} } at := r.At if at.IsZero() { @@ -96,12 +111,27 @@ func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { if signature == "" { continue } + // The task comes from the message, because the ring outlives the work + // it describes. Only a failure that names no task is attributed to the + // lease running now. + taskID, epoch := domain.ObservationTaskID(o.Message), "" + if taskID == "" { + taskID, epoch = r.TaskID, r.LeaseEpoch + } else if taskID == r.TaskID { + epoch = r.LeaseEpoch + } candidate := domain.ObservationIncident{ WorkerID: r.WorkerID, Incarnation: r.Incarnation, Signature: signature, - TaskID: r.TaskID, LeaseEpoch: r.LeaseEpoch, + TaskID: taskID, LeaseEpoch: epoch, } + mark := t.seen[r.WorkerID+"\x00"+signature] existing, isOpen := open[candidate.Key()] if !isOpen { + // Nothing new: this is a closed incident's entry still sitting in + // the ring. Presence is not occurrence. + if !firstOr(o.Last, at).After(mark.last) && o.Count <= mark.count { + continue + } candidate.ID = domain.NewID() candidate.Detail = o.Message candidate.FirstSeen = firstOr(o.First, at) @@ -112,6 +142,7 @@ func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { } appended = append(appended, e) t.counts[candidate.ID] = o.Count + t.seen[r.WorkerID+"\x00"+signature] = watermark{last: candidate.LastSeen, count: o.Count} continue } // Open already: accumulate, append nothing. A count lower than the last @@ -125,6 +156,7 @@ func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { existing.LastSeen = last t.Store.NoteObservation(existing) // last_seen is durable at close } + t.seen[r.WorkerID+"\x00"+signature] = watermark{last: existing.LastSeen, count: o.Count} } // The boundaries. A lease that ended, an epoch that changed, and a @@ -133,7 +165,10 @@ func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { if inc.WorkerID != r.WorkerID { return false } - if inc.TaskID == "" { + // Only an incident bound to a live lease has a lease boundary to close + // it. One read out of the ring about a task that already finished has + // no such boundary, so it ends the way a worker-level incident does. + if inc.LeaseEpoch == "" { return at.Sub(inc.LastSeen) > QuietTimeout } return inc.TaskID != r.TaskID || inc.LeaseEpoch != r.LeaseEpoch @@ -156,9 +191,9 @@ func (t *ObservationTracker) closeWhere(at time.Time, reason domain.ObservationC inc.CloseReason = reason if inc.CloseReason == "" { switch { - case inc.TaskID == "": + case inc.LeaseEpoch == "": inc.CloseReason = domain.ObservationCloseQuietTimeout - case inc.LeaseEpoch != "": + case inc.TaskID != "": inc.CloseReason = domain.ObservationCloseEpochChange default: inc.CloseReason = domain.ObservationCloseLeaseEnd diff --git a/internal/operations/observations_test.go b/internal/operations/observations_test.go index fc5874e..8b27153 100644 --- a/internal/operations/observations_test.go +++ b/internal/operations/observations_test.go @@ -218,3 +218,78 @@ func TestSignatureCollapsesAPaneName(t *testing.T) { t.Fatalf("one failure has two signatures:\n%s\n%s", a, b) } } + +// The first live run manufactured recurrence out of one old failure: a quiet +// timeout closed the incident, the entry stayed in the ring because the ring +// is a bounded history rather than a set of live conditions, and every later +// heartbeat opened it again. Four incidents, one failure, no new occurrence. +func TestAClosedIncidentDoesNotReopenFromAStaleRingEntry(t *testing.T) { + tr, s := tracker(t) + at := time.Unix(1700000000, 0).UTC() + entry := ring("heartbeat: connection refused", 4, at) + if _, err := tr.Ingest(WorkerReport{WorkerID: "w", Incarnation: "b1", Observations: entry, At: at}); err != nil { + t.Fatal(err) + } + // Long enough to close on quiet timeout, with the entry still reported. + quiet := at.Add(QuietTimeout + time.Minute) + for i := 0; i < 4; i++ { + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "b1", Observations: entry, + At: quiet.Add(time.Duration(i) * time.Minute), + }); err != nil { + t.Fatal(err) + } + } + opened, closed := 0, 0 + for _, e := range s.Events(0) { + switch e.Type { + case domain.EventObservationIncidentOpened: + opened++ + case domain.EventObservationIncidentClosed: + closed++ + } + } + if opened != 1 || closed != 1 { + t.Fatalf("opened=%d closed=%d for one failure that never happened again", opened, closed) + } + + // A real new occurrence, which the ring shows by advancing the entry. + later := quiet.Add(time.Hour) + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "b1", Observations: ring("heartbeat: connection refused", 5, later), At: later, + }); err != nil { + t.Fatal(err) + } + reopened := 0 + for _, e := range s.Events(0) { + if e.Type == domain.EventObservationIncidentOpened { + reopened++ + } + } + if reopened != 2 { + t.Fatalf("a genuine new occurrence did not open an incident: opened=%d", reopened) + } +} + +// The ring outlives the work it describes, so the task comes from the message +// rather than from whatever the worker happens to be running now. +func TestTheTaskComesFromTheMessageNotTheCurrentLease(t *testing.T) { + tr, s := tracker(t) + at := time.Unix(1700000000, 0).UTC() + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "b1", TaskID: "06G4XAFH1MBPC35VSJN7V3NS14", LeaseEpoch: "now", + Observations: ring("renew lease 06G4WW6TND26M16CZA6WE5T458: 409 conflict", 3, at), At: at, + }); err != nil { + t.Fatal(err) + } + open := s.OpenObservations() + if len(open) != 1 { + t.Fatalf("open = %+v", open) + } + if open[0].TaskID != "06G4WW6TND26M16CZA6WE5T458" { + t.Fatalf("the failure was attributed to the wrong task: %q", open[0].TaskID) + } + if open[0].LeaseEpoch != "" { + t.Fatalf("an old failure inherited the current lease's epoch: %q", open[0].LeaseEpoch) + } +}