go-otel/internal/laya/decision.go

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
}