package domain import ( "fmt" "regexp" "strings" "time" ) // A worker's observation ring is bounded, lossy, and local: it holds distinct // failure messages with repeat counts and nothing else, and it disappears when // the process does. The debt ledger reported that gap about itself, because no // event carried any of it. // // These two events make it durable as incidents rather than as symptoms. Run // 11 saw the same 409 refusal 301 times; that is one incident with an // intensity of 301, not 301 pieces of evidence. Recurrence has to mean "this // happened on four independent leases", or one stuck loop makes everything // look chronic. const ( EventObservationIncidentOpened = "ObservationIncidentOpened" EventObservationIncidentClosed = "ObservationIncidentClosed" ) // ObservationCloseReason is why Orchestra finalized an incident. None of them // is "the message stopped appearing in the ring": the ring is a bounded // history, so absence proves eviction as easily as recovery. type ObservationCloseReason string const ( // ObservationCloseLeaseEnd and ObservationCloseEpochChange are the natural // boundaries of a lease-scoped incident. The work it was about is over. ObservationCloseLeaseEnd ObservationCloseReason = "lease_end" ObservationCloseEpochChange ObservationCloseReason = "epoch_change" // ObservationCloseWorkerRestart ends every incident of an incarnation. A // new process cannot continue the old one's symptom. ObservationCloseWorkerRestart ObservationCloseReason = "worker_restart" // ObservationCloseQuietTimeout is the only closer for an observation with // no lease to bound it, and it fires on last_seen going stale rather than // on the entry vanishing. ObservationCloseQuietTimeout ObservationCloseReason = "quiet_timeout" ) func (r ObservationCloseReason) Valid() bool { switch r { case ObservationCloseLeaseEnd, ObservationCloseEpochChange, ObservationCloseWorkerRestart, ObservationCloseQuietTimeout: return true } return false } // WorkerObservation is one entry of a worker's ring as reported on a // heartbeat. It is the input to the incident projection, never a stored event. type WorkerObservation struct { Message string `json:"message"` Count int `json:"count"` First time.Time `json:"first"` Last time.Time `json:"last"` } // ObservationIncident is one durable incident: a signature seen by one worker, // on one lease when there is one, from its first occurrence to the boundary // that ended it. type ObservationIncident struct { ID string `json:"observation_id"` WorkerID string `json:"worker_id"` Incarnation string `json:"incarnation,omitempty"` TaskID string `json:"task_id,omitempty"` LeaseEpoch string `json:"lease_epoch,omitempty"` // Signature is the message with its task ids, commit shas, paths and // durations replaced, so the same failure on two tasks shares it. Grouping // on the raw message would make every task its own kind of problem. Signature string `json:"signature"` Detail string `json:"detail,omitempty"` FirstSeen time.Time `json:"first_seen"` // LastSeen is the last actual occurrence. ClosedAt is when Orchestra // finalized the incident, which is later and often much later: an incident // stays open until its lease ends, and open means "not yet final evidence" // rather than "happening right now". LastSeen time.Time `json:"last_seen,omitempty"` ClosedAt time.Time `json:"closed_at,omitempty"` RepeatCount int `json:"repeat_count,omitempty"` CloseReason ObservationCloseReason `json:"close_reason,omitempty"` } // Key identifies an incident. Two workers reporting the same failure are two // incidents, and so are two leases of one task. func (i ObservationIncident) Key() string { return strings.Join([]string{i.WorkerID, i.TaskID, i.LeaseEpoch, i.Signature}, "\x00") } var ( // Case-insensitive: a task id appears upper-case in a message and // lower-case inside a pane name, and the live ledger's first run showed // pane names keeping their task, which would give the same failure a // different signature on every task. observationID = regexp.MustCompile(`(?i)\b[0-9A-HJKMNP-TV-Z]{26}\b`) observationSHA = regexp.MustCompile(`\b[0-9a-f]{7,64}\b`) observationDuration = regexp.MustCompile(`\b\d+(\.\d+)?(ns|us|µs|ms|s|m|h)(\d+(\.\d+)?(ns|us|µs|ms|s|m|h))*\b`) observationNumber = regexp.MustCompile(`\b\d+\b`) observationPath = regexp.MustCompile(`(/[\w.-]+){2,}`) ) // ObservationSignature collapses one message to the kind of failure it is. // "lease A not renewed" and "lease B not renewed" are the same problem seen // twice, which is the whole basis of counting recurrence across tasks. func ObservationSignature(message string) string { s := strings.TrimSpace(message) s = observationID.ReplaceAllString(s, "") s = observationPath.ReplaceAllString(s, "") s = observationDuration.ReplaceAllString(s, "") s = observationSHA.ReplaceAllString(s, "") s = observationNumber.ReplaceAllString(s, "") s = strings.Join(strings.Fields(s), " ") if len(s) > 200 { s = s[:200] } 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) } if w, _ := p["worker_id"].(string); strings.TrimSpace(w) == "" { return fmt.Errorf("%w: worker_id required", ErrInvalid) } if sig, _ := p["signature"].(string); strings.TrimSpace(sig) == "" { return fmt.Errorf("%w: signature required", ErrInvalid) } return nil } func ValidateObservationIncidentClosed(p map[string]any) error { if id, _ := p["observation_id"].(string); strings.TrimSpace(id) == "" { return fmt.Errorf("%w: observation_id required", ErrInvalid) } reason, _ := p["close_reason"].(string) if !ObservationCloseReason(reason).Valid() { return fmt.Errorf("%w: close_reason %q is not a close reason", ErrInvalid, reason) } if c, ok := p["repeat_count"].(float64); ok && c < 0 { return fmt.Errorf("%w: repeat_count cannot be negative", ErrInvalid) } return nil }