package laya import ( "context" "fmt" "sort" "strings" "time" ) // Action is what the decision engine wants done with a firing alert. These are // the labels offered to the model, so they must stay stable: a rename changes // what the model was trained to answer. type Action string const ( ActionRestart Action = "restart" // roll the Deployment's pods ActionIgnore Action = "ignore" // known noise, do not page ActionNotify Action = "notify" // tell the on-call ActionEscalate Action = "escalate" // tell a human who owns the service ) // Actions is the full ladder, least to most forceful. var Actions = []Action{ActionIgnore, ActionNotify, ActionEscalate, ActionRestart} // Pattern is one reduced error pattern of an alert, as the triage package // classifies it. It is deliberately plain data: the engine can be fed from // anywhere without importing the alerting pipeline. type Pattern struct { What string // "HTTP 500 | /api/cart | panic: ..." Category string // triage category, e.g. service-defect Status int // HTTP status, 0 when the error was not an HTTP one Count int Baseline float64 New bool // never seen in the baseline windows Clients int // distinct clients affected Services []string // every service failing in these requests Host string // ingress host, for edge errors Path string // request path Routes []string // "host/path -> service ns/name -> deployment ns/name" } // Pod is a pod behind one of the alert's deployments, with the facts that // explain errors the log lines alone cannot: a pod being replaced right now // (so a 503 burst is expected) or a crash loop (so a restart changes nothing). type Pod struct { Namespace, Deployment, Name string Found bool Ready bool Phase string Restarts int32 // restarts of this pod PodRestarts int32 // restarts across the Deployment Reason string // CrashLoopBackOff, OOMKilled, ... ExitCode *int32 Message string LastTerminatedAt time.Time LastEvents []string // last log lines, oldest first Err string } // Deploy is a rollout that overlaps the alert window. type Deploy struct { Namespace, Workload string Release string Age time.Duration // how long ago the rollout started Completed bool Ready, Desired int32 } // Input is everything known about a firing alert, rendered into the state text // the model reasons over. type Input struct { AlertID string Service string Priority string // high | medium | low ErrorCount int Window time.Duration BaselineMean float64 NewErrors bool Clients int Patterns []Pattern Pods []Pod Deploys []Deploy // KnownNoise marks a pattern a human has already marked as ignorable, e.g. // a scanner or a muted issue. The model is told so it need not guess. KnownNoise bool } // maxStateChars keeps the rendered state inside the model's context. The tail // (pods, deploys) is what the model most often misses, so patterns are cut // first, never the pod events. const maxStateChars = 4000 // Render turns an alert into a compact factual state. It states only what was // measured - the model supplies judgement, this supplies facts - and always // includes the pods' last events, because a 503 burst while a pod is being // replaced is expected and indistinguishable from an outage without them. func (in Input) Render(now time.Time) string { var b strings.Builder multiple := 0.0 if in.BaselineMean > 0 { multiple = float64(in.ErrorCount) / in.BaselineMean } fmt.Fprintf(&b, "Firing error spike in service %s (priority %s).\n", or(in.Service, "unknown"), or(in.Priority, "unknown")) fmt.Fprintf(&b, "%d errors in the last %s, normally about %.0f per window (%.0fx normal).\n", in.ErrorCount, in.Window.Round(time.Second), in.BaselineMean, multiple) if in.NewErrors { b.WriteString("At least one error is new: it was never seen in the baseline windows.\n") } if in.Clients > 0 { fmt.Fprintf(&b, "%d distinct clients affected.\n", in.Clients) } if in.KnownNoise { b.WriteString("A human has already marked this pattern as ignorable noise.\n") } b.WriteString("\nError patterns, most urgent first:\n") for i, p := range in.Patterns { fmt.Fprintf(&b, "%d. [%s] %s\n", i+1, or(p.Category, "unclassified"), or(p.What, "no message")) fmt.Fprintf(&b, " %d in the window, normally %.0f. %s. clients affected: %d.", p.Count, p.Baseline, newness(p.New), p.Clients) if p.Status > 0 { fmt.Fprintf(&b, " HTTP %d.", p.Status) } b.WriteString("\n") if len(p.Services) > 1 { fmt.Fprintf(&b, " Failing together: %s.\n", strings.Join(p.Services, ", ")) } for _, r := range p.Routes { fmt.Fprintf(&b, " Route: %s.\n", r) } } if len(in.Pods) == 0 { b.WriteString("\nPod state: unknown, the deployment behind these errors was not resolved.\n") } for _, p := range in.Pods { b.WriteString("\nPod events:\n") fmt.Fprintf(&b, "- %s/%s pod %s: %s\n", p.Namespace, p.Deployment, or(p.Name, "unknown"), podStateLine(p, now)) for _, line := range p.LastEvents { fmt.Fprintf(&b, " log: %s\n", clip(line, 160)) } } for _, d := range in.Deploys { state := "rollout in progress" if d.Completed { state = "rollout completed" } fmt.Fprintf(&b, "\nDeploy: %s/%s %s started %s ago, %d/%d replicas ready", d.Namespace, d.Workload, state, d.Age.Round(time.Second), d.Ready, d.Desired) if d.Release != "" { fmt.Fprintf(&b, " (release %s)", d.Release) } b.WriteString("\n") } return clip(b.String(), maxStateChars) } // podStateLine says what the pod is doing now, in the words an operator would // use: crash-looping, being replaced, up but not ready, or healthy. func podStateLine(p Pod, now time.Time) string { switch { case p.Err != "": return "could not be inspected (" + clip(p.Err, 80) + ")" case !p.Found: return "no pod running, the Deployment is scaled to 0 or has nothing ready" } var parts []string phase := or(p.Phase, "unknown") if p.Ready { parts = append(parts, "phase "+phase+", passing its readiness probe") } else { parts = append(parts, "phase "+phase+", NOT passing its readiness probe") } switch { case p.Restarting(now): parts = append(parts, fmt.Sprintf("its container restarted %s ago and is still coming up", since(p.LastTerminatedAt, now))) case p.CrashLooping(now): parts = append(parts, fmt.Sprintf("crash-looping: %d restarts", p.Restarts)) case p.Restarts > 0: parts = append(parts, fmt.Sprintf("%d restarts, none recent", p.Restarts)) } if p.PodRestarts > p.Restarts { parts = append(parts, fmt.Sprintf("%d restarts across the Deployment", p.PodRestarts)) } if p.Reason != "" { parts = append(parts, "last container termination: "+p.Reason) } if p.ExitCode != nil { parts = append(parts, fmt.Sprintf("exit code %d", *p.ExitCode)) } if p.Message != "" { parts = append(parts, clip(p.Message, 120)) } return strings.Join(parts, "; ") } // Restarting reports a pod that is right now being replaced, so the errors // around it are the rollout's, not a new defect. func (p Pod) Restarting(now time.Time) bool { return !p.LastTerminatedAt.IsZero() && !now.Before(p.LastTerminatedAt) && now.Sub(p.LastTerminatedAt) < rollingWindow } // CrashLooping reports a pod stuck restarting: a restart cannot fix it. func (p Pod) CrashLooping(now time.Time) bool { if p.Reason == "CrashLoopBackOff" { return true } return p.Restarts >= 2 && !p.LastTerminatedAt.IsZero() && now.Sub(p.LastTerminatedAt) < rollingWindow } func newness(isNew bool) string { if isNew { return "never seen before" } return "seen before" } func since(t, now time.Time) string { d := now.Sub(t) switch { case d < time.Minute: return fmt.Sprintf("%ds", int(d.Seconds())) case d < time.Hour: return fmt.Sprintf("%dm", int(d.Minutes())) default: return fmt.Sprintf("%.1fh", d.Hours()) } } func or(v, fallback string) string { if strings.TrimSpace(v) == "" { return fallback } return v } func clip(s string, n int) string { s = strings.TrimSpace(s) if len(s) <= n { return s } return strings.TrimSpace(s[:n-1]) + "…" } // Policy decides which of the model's answers may be executed. The model // proposes; the policy disposes. Nothing is acted on unless it is in Actions, // clears MinProbability, and survives the gates below. type Policy struct { // Actions allowed to be executed automatically. An empty list means // "decide and report, never act", which is the safe default. Actions []Action // MinProbability is the model's probability for its chosen action below // which the decision is downgraded to a notification. MinProbability float64 // MinMargin is the lead over the runner-up required to act. The label // probabilities are diffuse, so this is the more meaningful of the two. MinMargin float64 // MinClients: a pattern affecting fewer clients than this is never // auto-acted on, however confident the model is. MinClients int // MaxRestartsPerHour caps automatic restarts per deployment, so a flapping // service cannot be restarted into a loop. MaxRestartsPerHour int // Now is the clock, overridable in tests. Now func() time.Time } func (p Policy) now() time.Time { if p.Now != nil { return p.Now() } return time.Now() } func (p Policy) allows(a Action) bool { for _, allowed := range p.Actions { if allowed == a { return true } } return false } func (p Policy) minProbability() float64 { if p.MinProbability <= 0 { return 0.35 // below this the leading label barely beats the field } return p.MinProbability } func (p Policy) minMargin() float64 { if p.MinMargin <= 0 { return 0.05 } return p.MinMargin } // Verdict is the outcome for one alert: what the model asked for, what was // allowed, and why it differs. type Verdict struct { AlertID string Action Action // what is actually done Requested Action // what the model chose Probability float64 // model's probability for Requested Margin float64 // how far Requested leads the runner-up Confidence float64 // how decisive the forward pass was Probabilities map[string]float64 Severity string Urgency int Model string State string // the text the model was given, kept for the UI Gates []string // why the action was changed or withheld, in order // Targets are the deployments a restart was cleared for, after the // hourly budget. Executable means exactly this list may be acted on. Targets []string Tokens int At time.Time Error string // set when the model could not be reached } // Executable reports whether the action may be carried out automatically. func (v Verdict) Executable() bool { return v.Error == "" && v.Action != ActionNotify && v.Action != ActionEscalate } // Changed reports whether a gate altered the model's answer. func (v Verdict) Changed() bool { return v.Error == "" && v.Requested != "" && v.Action != v.Requested } // Explain renders the verdict as one line for a notification or a log. func (v Verdict) Explain() string { if v.Error != "" { return "decision unavailable: " + v.Error + " (notifying as usual)" } msg := fmt.Sprintf("laya %s: %s (%.0f%%, margin %.0f%%, decisiveness %.2f)", v.Action, v.Requested, v.Probability*100, v.Margin*100, v.Confidence) if v.Severity != "" { msg += ", severity " + v.Severity } for _, g := range v.Gates { msg += "; " + g } return msg } // Decider is the part of an Engine the alerting pipeline talks to, so the // pipeline can be tested without a model. type Decider interface { Decide(ctx context.Context, in Input) Verdict } // Engine renders an alert, asks Laya what to do, and applies the policy gates. type Engine struct { Client *Client Policy Policy // restarts counts automatic restarts per deployment within the hour. restarts map[string][]time.Time } func NewEngine(client *Client, policy Policy) *Engine { return &Engine{Client: client, Policy: policy, restarts: map[string][]time.Time{}} } // Questions are the typed questions asked of the model. The criteria spell // out what each action means here, because the model reads the descriptions // rather than the labels. func Questions() map[string]Question { return map[string]Question{ "action": ChoiceQuestion("What is the right operational response to this error spike right now?", map[string]string{ "restart": "Restart the pods of the deployment: only right when the pod itself is wedged or crash-looping and a fresh process would fix it", "ignore": "Ignore: known noise, an expected consequence of a deploy or a scanner, nothing for a human to do", "notify": "Notify the on-call: a real problem that needs a human to look, but not self-fixable", "escalate": "Escalate to the service owner: serious, ongoing, and likely to need a code change or a rollback", }), "severity": ChoiceQuestion("How severe is this for users of the service?", map[string]string{ "low": "cosmetic, a tiny fraction of requests, no user-visible effect", "medium": "degraded, some users are affected but the service still works", "high": "users are visibly broken, a large share of requests fail", "critical": "the service is down or unusable for most of its users", }), "urgency": ScoreQuestion("How soon does a human need to act? 0 = nobody needs to act at all, 4 = a human is needed right now.", []string{"0 nobody needs to act", "1 low priority", "2 today", "3 within the hour", "4 needed right now"}), } } // Decide renders the state, asks the model, and gates the answer. It never // returns an error: a decision that could not be made degrades to a plain // notification, which is what the pipeline would have done anyway. func (e *Engine) Decide(ctx context.Context, in Input) Verdict { now := e.Policy.now() v := Verdict{AlertID: in.AlertID, Requested: ActionNotify, Action: ActionNotify, At: now, State: in.Render(now)} if e == nil || e.Client == nil { return v } resp, err := e.Client.Predict(ctx, v.State, Questions()) if err != nil { v.Error = err.Error() return v } v.Model, v.Tokens = resp.Model, resp.Usage.InputTokens answer := resp.Answers["action"] v.Probabilities = answer.Probabilities v.Probability = answer.Probability(answer.Choice) v.Margin = answer.Margin() v.Confidence = answer.Confidence v.Severity = resp.Answers["severity"].Choice v.Urgency = int(resp.Answers["urgency"].Score + 0.5) requested := Action(answer.Choice) if !knownAction(requested) { v.Error = "model answered with an unknown action " + answer.Choice return v } v.Requested = requested v.Action = requested e.gate(ctx, &v, in, now) return v } func knownAction(a Action) bool { for _, k := range Actions { if k == a { return true } } return false } // gate is the safety envelope. Each check either leaves the action alone or // downgrades it to a notification, always recording why. func (e *Engine) gate(ctx context.Context, v *Verdict, in Input, now time.Time) { downgrade := func(to Action, why string) { if v.Action != to { v.Gates = append(v.Gates, why) v.Action = to } } // 1. The model has to lead the field by a real margin. Both the absolute // probability and the lead are checked: a diffuse answer that only just // wins is a guess, and must not move a cluster on its own. if v.Probability < e.Policy.minProbability() { downgrade(ActionNotify, fmt.Sprintf("model was only %.0f%% sure of %s", v.Probability*100, v.Requested)) } if v.Margin < e.Policy.minMargin() { downgrade(ActionNotify, fmt.Sprintf("%s only just beat the alternatives (margin %.0f%%)", v.Requested, v.Margin*100)) } // 2. Restarts are the one action that can make things worse, so the pod // state has to support it rather than merely allow it. if v.Action == ActionRestart { switch reason := restartRefused(in, now); { case reason != "": downgrade(ActionNotify, reason) case !e.Policy.allows(ActionRestart): downgrade(ActionNotify, "automatic restarts are not enabled") case !e.allowRestart(v, in, now): downgrade(ActionNotify, "restart budget exhausted for this deployment") } } // 3. Ignoring is only safe when the errors are explained: a human's mute, // a crawler, or a rollout that is still replacing pods. if v.Action == ActionIgnore { if reason := ignoreRefused(in); reason != "" { downgrade(ActionNotify, reason) } } // 4. Anything not on the allowlist is reported, never executed. if v.Action != ActionNotify && v.Action != ActionEscalate && !e.Policy.allows(v.Action) { downgrade(ActionNotify, "action "+string(v.Action)+" is not in the allowed set") } } // restartRefused explains why a restart would not help, using the pods' last // events. A pod the kubelet is already restarting will just be restarted again, // and an expected 503 burst during a rollout is not an incident at all. func restartRefused(in Input, now time.Time) string { var live, crashLooping, restarting int for _, p := range in.Pods { if p.Err != "" || !p.Found { continue } live++ switch { case p.CrashLooping(now): crashLooping++ case p.Restarting(now): restarting++ } } if live == 0 { return "no pod state to justify a restart" } if crashLooping == live { return "pods are already crash-looping, a restart repeats what the kubelet already does" } // Every pod died moments ago, or a rollout is replacing them: the 503s // around it are the rollout's own, and restarting would restart the // rollout as well. if restarting == live || (notReady(in) && rolloutOngoing(in, now)) { return "every pod is being replaced by a rollout, the 503s are the rollout's own" } return "" } // notReady reports whether no pod behind the alert is serving traffic. func notReady(in Input) bool { ready := false for _, p := range in.Pods { if p.Err == "" && p.Found && p.Ready { ready = true } } return !ready } // ignoreRefused explains why a spike may not be dropped silently. A brand new // error is never ignored: nobody has seen it before, so nobody can know it is // harmless. func ignoreRefused(in Input) string { if in.KnownNoise { return "" } if in.NewErrors { return "the error is new, so nobody has confirmed it is harmless" } if in.Priority == "high" && in.ErrorCount > 0 { return "a high-priority spike is never ignored without a human's confirmation" } for _, p := range in.Patterns { if p.New { return "pattern " + p.What + " is new, so it cannot be assumed harmless" } } return "" } func rolloutOngoing(in Input, now time.Time) bool { for _, d := range in.Deploys { if !d.Completed && d.Age > 0 && d.Age < rollingWindow { return true } } return false } // allowRestart enforces the hourly budget per deployment, so a service that // flaps cannot be restarted over and over. func (e *Engine) allowRestart(v *Verdict, in Input, now time.Time) bool { limit := e.Policy.MaxRestartsPerHour if limit <= 0 { limit = 3 } cleared := 0 for _, p := range in.Pods { key := p.Namespace + "/" + p.Deployment cut := now.Add(-time.Hour) kept := e.restarts[key][:0] for _, at := range e.restarts[key] { if at.After(cut) { kept = append(kept, at) } } e.restarts[key] = kept if len(kept) >= limit { continue } e.restarts[key] = append(kept, now) v.Targets = append(v.Targets, key) cleared++ } return cleared > 0 } // RollingWindow is how recently a container must have died for the errors // around it to be attributed to the pod still coming up. const rollingWindow = 5 * time.Minute // sortedDeploys is used by the pipeline to feed Deploys newest first. func sortedDeploys(in []Deploy) []Deploy { out := append([]Deploy(nil), in...) sort.SliceStable(out, func(i, j int) bool { return out[i].Age < out[j].Age }) return out }