diff --git a/internal/laya/client.go b/internal/laya/client.go new file mode 100644 index 0000000..68950f9 --- /dev/null +++ b/internal/laya/client.go @@ -0,0 +1,289 @@ +// Package laya talks to the Laya decision API (https://laya.k6n.net) - a fast +// non-autoregressive "System 1" model that answers typed questions about a +// state in one forward pass. It is used here as an incident triage brain: give +// it a compact textual state of a firing spike and it returns a calibrated +// choice (restart / ignore / notify / escalate) with probabilities. +// +// The API is described by https://laya.k6n.net/openapi.json. Only the parts +// this repo uses are modelled; the wire format is deliberately loose on the +// response side (answers are a map) because the model decides which questions +// get answers. +package laya + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +// Question kinds understood by the API. +const ( + KindChoice = "choice" // pick one label from Criteria + KindScore = "score" // pick a level index from Levels + KindNoul = "noul" // a single 0..1 non-choice answer +) + +// Question is one typed question. Use the ChoiceQuestion, ScoreQuestion or +// NoulQuestion constructors rather than building it by hand, because +// "criteria" is a JSON object for choice questions and a JSON array for score +// ones - both are rendered by MarshalJSON. +type Question struct { + kind string + instructions string + criteria map[string]string // choice: label -> what that label means + levels []string // score: level descriptions, index 0 first +} + +func (q Question) MarshalJSON() ([]byte, error) { + out := map[string]any{"type": q.kind, "instructions": q.instructions} + switch q.kind { + case KindChoice: + if len(q.criteria) == 0 { + return nil, errors.New("laya: choice question needs criteria") + } + out["criteria"] = q.criteria + case KindScore: + if len(q.levels) == 0 { + return nil, errors.New("laya: score question needs criteria levels") + } + out["criteria"] = q.levels + } + return json.Marshal(out) +} + +func (q Question) Kind() string { return q.kind } + +// ChoiceQuestion asks the model to pick one of criteria. The label -> text map +// matters: the model reads the descriptions, not just the labels, so "restart" +// should say what restarting means here. +func ChoiceQuestion(instructions string, criteria map[string]string) Question { + return Question{kind: KindChoice, instructions: instructions, criteria: criteria} +} + +// ScoreQuestion asks for a level between 0 and len(levels)-1. +func ScoreQuestion(instructions string, levels []string) Question { + return Question{kind: KindScore, instructions: instructions, levels: levels} +} + +// NoulQuestion asks a single open question answered with one 0..1 value. +func NoulQuestion(instructions string) Question { + return Question{kind: KindNoul, instructions: instructions} +} + +// PredictRequest is the body of POST /v1/predict. State is a string, a map or +// a list; this repo always sends a rendered text state. +type PredictRequest struct { + State any `json:"state"` + Questions map[string]Question `json:"questions"` + Lang string `json:"lang,omitempty"` + MaxLen int `json:"max_len,omitempty"` + BatchSize int `json:"batch_size,omitempty"` // batch endpoint only +} + +// BatchPredictRequest is the body of POST /v1/predict/batch: several states +// evaluated against the same questions, packed into shared forward passes. +type BatchPredictRequest struct { + States []any `json:"states"` + Questions map[string]Question `json:"questions"` + Lang string `json:"lang,omitempty"` + BatchSize int `json:"batch_size,omitempty"` + MaxLen int `json:"max_len,omitempty"` +} + +// Answer is one answered question. Only the fields matching Type are set. +type Answer struct { + Type string `json:"type"` + Choice string `json:"choice,omitempty"` + Probabilities map[string]float64 `json:"probabilities,omitempty"` + Score float64 `json:"score,omitempty"` + Legend map[string]string `json:"legend,omitempty"` + Noul float64 `json:"noul,omitempty"` + Confidence float64 `json:"confidence,omitempty"` + // AnswerConfidence is the model's confidence in the answer it picked, + // as opposed to Confidence, which measures how decisive the forward pass + // was overall. + AnswerConfidence float64 `json:"answer_confidence,omitempty"` +} + +// Probability returns the model's probability for one choice label, 0 if the +// label was not offered or not answered. +func (a Answer) Probability(label string) float64 { return a.Probabilities[label] } + +// Margin is how far the chosen answer leads the next best one. The label +// probabilities are diffuse - even a correct answer rarely exceeds 0.5 - so +// the lead over the runner-up is a better confidence signal than the raw +// probability. +func (a Answer) Margin() float64 { + best, second := 0.0, 0.0 + for _, p := range a.Probabilities { + switch { + case p > best: + second, best = best, p + case p > second: + second = p + } + } + return best - second +} + +type Usage struct { + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` +} + +type Response struct { + Model string `json:"model"` + Answers map[string]Answer `json:"answers"` + Usage Usage `json:"usage"` +} + +// Health is the body of GET /healthz. +type Health struct { + Status string `json:"status"` + Model string `json:"model"` + Device string `json:"device"` + Version string `json:"version"` +} + +// Client is a Laya API client. BaseURL defaults to https://laya.k6n.net. +type Client struct { + BaseURL string + HTTP *http.Client + Lang string + MaxLen int + BatchSize int // forward-pass chunking for the batch endpoint; 0 = server default +} + +// NewClient returns a client for base ("https://laya.k6n.net" when empty). +func NewClient(base string) *Client { + return &Client{BaseURL: strings.TrimRight(base, "/"), HTTP: &http.Client{Timeout: 30 * time.Second}, Lang: "en"} +} + +func (c *Client) url(path string) string { + base := c.BaseURL + if base == "" { + base = "https://laya.k6n.net" + } + return base + path +} + +func (c *Client) client() *http.Client { + if c.HTTP != nil { + return c.HTTP + } + return &http.Client{Timeout: 30 * time.Second} +} + +// Health reports the model behind the API, for a setup probe. +func (c *Client) Health(ctx context.Context) (Health, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.url("/healthz"), nil) + if err != nil { + return Health{}, err + } + resp, err := c.client().Do(req) + if err != nil { + return Health{}, unwrap(err) + } + defer resp.Body.Close() + if err := checkStatus(resp); err != nil { + return Health{}, err + } + var h Health + if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<16)).Decode(&h); err != nil { + return Health{}, fmt.Errorf("laya: decode health: %w", err) + } + return h, nil +} + +// Predict evaluates one state against the questions in a single pass. +func (c *Client) Predict(ctx context.Context, state any, questions map[string]Question) (Response, error) { + body := PredictRequest{State: state, Questions: questions, Lang: c.Lang, MaxLen: c.MaxLen} + return c.post(ctx, "/v1/predict", body) +} + +// PredictBatch evaluates several states against the same questions. Results +// are returned in request order, one entry per state. +func (c *Client) PredictBatch(ctx context.Context, states []any, questions map[string]Question) ([]Response, error) { + body := BatchPredictRequest{States: states, Questions: questions, Lang: c.Lang, BatchSize: c.BatchSize, MaxLen: c.MaxLen} + var out struct { + Results []Response `json:"results"` + } + if err := c.postInto(ctx, "/v1/predict/batch", body, &out); err != nil { + return nil, err + } + if len(out.Results) != len(states) { + return nil, fmt.Errorf("laya: batch returned %d results for %d states", len(out.Results), len(states)) + } + return out.Results, nil +} + +func (c *Client) post(ctx context.Context, path string, body any) (Response, error) { + var out Response + if err := c.postInto(ctx, path, body, &out); err != nil { + return Response{}, err + } + return out, nil +} + +func (c *Client) postInto(ctx context.Context, path string, body, out any) error { + raw, err := json.Marshal(body) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.url(path), bytes.NewReader(raw)) + if err != nil { + return errors.New("laya: invalid request") + } + req.Header.Set("Content-Type", "application/json") + resp, err := c.client().Do(req) + if err != nil { + return unwrap(err) + } + defer resp.Body.Close() + if err := checkStatus(resp); err != nil { + return err + } + return json.NewDecoder(io.LimitReader(resp.Body, 1<<22)).Decode(out) +} + +// unwrap drops the URL the transport puts in its error, which may carry a +// token; the returned error must be safe to log. +func unwrap(err error) error { + var ue interface{ Unwrap() error } + if errors.As(err, &ue) { + return errors.New(scrub(ue.Unwrap().Error())) + } + return errors.New(scrub(err.Error())) +} + +// scrub removes anything that looks like a credential from a URL in an error. +func scrub(msg string) string { + for _, part := range strings.Fields(msg) { + if i := strings.Index(part, "://"); i >= 0 { + rest := part[i+3:] + if at := strings.IndexByte(rest, '@'); at >= 0 { + host := rest[at+1:] + msg = strings.ReplaceAll(msg, part, "laya://"+host) + } + } + } + return msg +} + +func checkStatus(resp *http.Response) error { + if resp.StatusCode/100 == 2 { + return nil + } + detail, _ := io.ReadAll(io.LimitReader(resp.Body, 500)) + msg := strings.TrimSpace(string(detail)) + if msg == "" { + msg = http.StatusText(resp.StatusCode) + } + return fmt.Errorf("laya: HTTP %d: %s", resp.StatusCode, scrub(msg)) +} diff --git a/internal/laya/decision.go b/internal/laya/decision.go new file mode 100644 index 0000000..bb7c310 --- /dev/null +++ b/internal/laya/decision.go @@ -0,0 +1,579 @@ +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 +} diff --git a/internal/laya/laya_test.go b/internal/laya/laya_test.go new file mode 100644 index 0000000..7f17c09 --- /dev/null +++ b/internal/laya/laya_test.go @@ -0,0 +1,344 @@ +package laya + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// fake serves the Laya API with a fixed answer per action label, and records +// the last state and questions it was asked. +type fake struct { + choice string + prob map[string]float64 + lastState string + lastQs map[string]json.RawMessage + status int + batchSizes int +} + +func (f *fake) server(t *testing.T) *httptest.Server { + t.Helper() + mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, Health{Status: "ok", Model: "test-model"}) + }) + mux.HandleFunc("/v1/predict", func(w http.ResponseWriter, r *http.Request) { + var req struct { + State json.RawMessage `json:"state"` + Questions map[string]json.RawMessage `json:"questions"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Errorf("decode request: %v", err) + } + f.lastState = string(req.State) + f.lastQs = req.Questions + f.respond(w, map[string]Answer{ + "action": {Type: KindChoice, Choice: f.choice, Probabilities: f.prob}, + "severity": {Type: KindChoice, Choice: "high", Probabilities: map[string]float64{"high": 0.8}}, + "urgency": {Type: KindScore, Score: 3}, + }) + }) + mux.HandleFunc("/v1/predict/batch", func(w http.ResponseWriter, r *http.Request) { + var req struct { + States []json.RawMessage `json:"states"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Errorf("decode batch request: %v", err) + } + f.batchSizes = len(req.States) + results := make([]Response, 0, len(req.States)) + for range req.States { + results = append(results, Response{Model: "test-model", Answers: map[string]Answer{ + "action": {Type: KindChoice, Choice: f.choice, Probabilities: f.prob}, + }}) + } + writeJSON(w, map[string]any{"results": results}) + }) + mux.HandleFunc("/boom", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte("upstream exploded")) + }) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + return srv +} + +func (f *fake) respond(w http.ResponseWriter, answers map[string]Answer) { + if f.status != 0 { + w.WriteHeader(f.status) + return + } + writeJSON(w, Response{Model: "test-model", Answers: answers, Usage: Usage{InputTokens: 128}}) +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} + +func TestPredictSendsTypedQuestionsAndReadsAnswers(t *testing.T) { + f := &fake{choice: "restart", prob: map[string]float64{"restart": 0.9, "notify": 0.1}} + srv := f.server(t) + c := NewClient(srv.URL) + + resp, err := c.Predict(context.Background(), "checkout-api is 500ing", Questions()) + if err != nil { + t.Fatalf("Predict: %v", err) + } + if got := resp.Answers["action"].Choice; got != "restart" { + t.Errorf("choice = %q, want restart", got) + } + if got := resp.Answers["urgency"].Score; got != 3 { + t.Errorf("urgency score = %v, want 3", got) + } + if got := resp.Usage.InputTokens; got != 128 { + t.Errorf("input tokens = %d, want 128", got) + } + // The choice question must carry a criteria object and the score question an + // array, from the same "criteria" key. + if q := string(f.lastQs["action"]); !strings.Contains(q, `"criteria":{"escalate"`) { + t.Errorf("action question = %s, want a criteria object", q) + } + if q := string(f.lastQs["urgency"]); !strings.Contains(q, `"criteria":[`) { + t.Errorf("urgency question = %s, want a criteria array", q) + } + if f.lastState != `"checkout-api is 500ing"` { + t.Errorf("state = %s, want the JSON-encoded state", f.lastState) + } +} + +func TestPredictBatchKeepsOrder(t *testing.T) { + srv := (&fake{choice: "notify", prob: map[string]float64{"notify": 1}}).server(t) + got, err := NewClient(srv.URL).PredictBatch(context.Background(), []any{"a", "b", "c"}, Questions()) + if err != nil { + t.Fatalf("PredictBatch: %v", err) + } + if len(got) != 3 { + t.Fatalf("got %d results, want 3", len(got)) + } + for i, r := range got { + if r.Answers["action"].Choice != "notify" { + t.Errorf("result %d choice = %q", i, r.Answers["action"].Choice) + } + } +} + +func TestPredictErrorsAreSafeToLog(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + _, err := NewClient(srv.URL).Predict(context.Background(), "x", Questions()) + if err == nil { + t.Fatal("expected an error") + } + if !strings.Contains(err.Error(), "503") { + t.Errorf("error = %v, want the status code", err) + } +} + +func TestScrubRemovesCredentialsFromURL(t *testing.T) { + got := scrub(`Post "https://user:hunter2@laya.k6n.net/v1/predict": timeout`) + if strings.Contains(got, "hunter2") { + t.Errorf("scrub(%q) leaked the password: %s", "user:hunter2@", got) + } + if !strings.Contains(got, "laya.k6n.net") { + t.Errorf("scrub(%q) dropped the host: %s", "user:hunter2@", got) + } +} + +func TestHealth(t *testing.T) { + srv := (&fake{}).server(t) + h, err := NewClient(srv.URL).Health(context.Background()) + if err != nil { + t.Fatalf("Health: %v", err) + } + if h.Model != "test-model" || h.Status != "ok" { + t.Errorf("health = %+v, want status ok from test-model", h) + } +} + +// --- state rendering and gates --- + +func nowFixture() time.Time { return time.Date(2026, 5, 1, 12, 0, 0, 0, time.UTC) } + +func healthyPod() Pod { + return Pod{Namespace: "shop", Deployment: "checkout-api", Name: "checkout-api-7d9", Found: true, Ready: true, Phase: "Running"} +} + +func engineFor(t *testing.T, policy Policy, choice string, prob float64) (*Engine, *fake) { + t.Helper() + f := &fake{choice: choice, prob: map[string]float64{choice: prob}} + srv := f.server(t) + policy.Now = nowFixture + return NewEngine(NewClient(srv.URL), policy), f +} + +func TestRenderStatesPodEvents(t *testing.T) { + exit := int32(137) + in := Input{ + AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 412, Window: 5 * time.Minute, + BaselineMean: 3, NewErrors: true, Clients: 42, + Patterns: []Pattern{{What: "HTTP 500 | /api/cart | panic", Category: "service-defect", Status: 500, Count: 300, Baseline: 2, New: true, Clients: 42}}, + Pods: []Pod{{ + Namespace: "shop", Deployment: "checkout-api", Name: "checkout-api-7d9", Found: true, Phase: "Running", + Restarts: 3, PodRestarts: 9, Reason: "OOMKilled", ExitCode: &exit, LastTerminatedAt: nowFixture().Add(-30 * time.Second), + LastEvents: []string{"panic: runtime error: invalid memory address"}, + }}, + Deploys: []Deploy{{Namespace: "shop", Workload: "checkout-api", Age: 4 * time.Minute, Ready: 1, Desired: 3}}, + } + state := in.Render(nowFixture()) + for _, want := range []string{ + "shop/checkout-api pod checkout-api-7d9", + "OOMKilled", + "exit code 137", + "its container restarted 30s ago and is still coming up", + "9 restarts across the Deployment", + "panic: runtime error: invalid memory address", + "rollout in progress started 4m0s ago", + "137x normal", + } { + if !strings.Contains(state, want) { + t.Errorf("state is missing %q:\n%s", want, state) + } + } +} + +func TestDecideExecutesRestartOnlyWhenPodsSupportIt(t *testing.T) { + e, f := engineFor(t, Policy{Actions: []Action{ActionRestart, ActionIgnore, ActionNotify, ActionEscalate}, MinProbability: 0.5}, "restart", 0.82) + v := e.Decide(context.Background(), Input{ + AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 100, Window: 5 * time.Minute, BaselineMean: 2, Clients: 30, + Patterns: []Pattern{{What: "HTTP 500 | /api/cart", Category: "service-defect", Count: 90, Baseline: 1, New: true, Clients: 30}}, + Pods: []Pod{{Namespace: "shop", Deployment: "checkout-api", Name: "p1", Found: true, Ready: false, Phase: "Running"}}, + }) + if !v.Executable() || v.Action != ActionRestart { + t.Fatalf("action = %q executable=%v gates=%v, want an executed restart", v.Action, v.Executable(), v.Gates) + } + if len(v.Targets) != 1 || v.Targets[0] != "shop/checkout-api" { + t.Errorf("targets = %v, want shop/checkout-api", v.Targets) + } + if !strings.Contains(f.lastState, "pod p1") { + t.Errorf("model was not given the pod state:\n%s", f.lastState) + } +} + +func TestDecideRefusesRestartDuringCrashLoop(t *testing.T) { + e, _ := engineFor(t, Policy{Actions: []Action{ActionRestart}, MinProbability: 0.5}, "restart", 0.95) + v := e.Decide(context.Background(), Input{ + AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 300, Window: 5 * time.Minute, BaselineMean: 2, + Patterns: []Pattern{{What: "HTTP 500 | panic", Category: "service-defect", Count: 300, New: true}}, + Pods: []Pod{{Namespace: "shop", Deployment: "checkout-api", Name: "p1", Found: true, Phase: "Running", Restarts: 5, LastTerminatedAt: nowFixture().Add(-20 * time.Second)}}, + }) + if v.Action != ActionNotify { + t.Errorf("action = %q, want notify: a crash loop needs a new release, not a restart", v.Action) + } + if len(v.Gates) == 0 || !strings.Contains(strings.Join(v.Gates, " "), "crash-looping") { + t.Errorf("gates = %v, want the crash-loop reason", v.Gates) + } +} + +// A short 503 burst while every pod is being replaced is the rollout's own, so +// a restart must never be fired into it. +func TestDecideRefusesRestartDuringRollout(t *testing.T) { + e, _ := engineFor(t, Policy{Actions: []Action{ActionRestart, ActionIgnore}, MinProbability: 0.5}, "restart", 0.9) + v := e.Decide(context.Background(), Input{ + AlertID: "a1", Service: "checkout-api", Priority: "medium", ErrorCount: 40, Window: 5 * time.Minute, BaselineMean: 2, + Patterns: []Pattern{{What: "HTTP 503 | /api/cart", Category: "capacity-saturation", Status: 503, Count: 40}}, + Pods: []Pod{{Namespace: "shop", Deployment: "checkout-api", Name: "p1", Found: true, Phase: "Pending", LastTerminatedAt: nowFixture().Add(-40 * time.Second)}}, + Deploys: []Deploy{{Namespace: "shop", Workload: "checkout-api", Age: 2 * time.Minute, Ready: 1, Desired: 3}}, + }) + if v.Action == ActionRestart { + t.Fatalf("action = restart during a rollout, want it withheld (gates=%v)", v.Gates) + } + if !strings.Contains(strings.Join(v.Gates, " "), "replaced by a rollout") { + t.Errorf("gates = %v, want the rollout reason", v.Gates) + } +} + +func TestDecideRefusesIgnoreOfANewError(t *testing.T) { + e, _ := engineFor(t, Policy{Actions: []Action{ActionIgnore}, MinProbability: 0.5}, "ignore", 0.9) + v := e.Decide(context.Background(), Input{ + AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 200, Window: 5 * time.Minute, BaselineMean: 2, NewErrors: true, + Patterns: []Pattern{{What: "HTTP 500 | /api/cart", Count: 200, New: true}}, + Pods: []Pod{healthyPod()}, + }) + if v.Action != ActionNotify { + t.Errorf("action = %q, want notify: a new error was never confirmed harmless", v.Action) + } +} + +func TestDecideAllowsIgnoreOfKnownNoise(t *testing.T) { + e, _ := engineFor(t, Policy{Actions: []Action{ActionIgnore}, MinProbability: 0.5}, "ignore", 0.88) + v := e.Decide(context.Background(), Input{ + AlertID: "a1", Service: "blog", Priority: "low", ErrorCount: 1800, Window: 5 * time.Minute, BaselineMean: 20, KnownNoise: true, + Patterns: []Pattern{{What: "HTTP 404 | /old-post", Status: 404, Count: 1800}}, + }) + if v.Action != ActionIgnore || !v.Executable() { + t.Errorf("action = %q executable = %v, want an executable ignore", v.Action, v.Executable()) + } +} + +func TestDecideWithholdsActionsOutsideTheAllowlist(t *testing.T) { + e, _ := engineFor(t, Policy{Actions: []Action{ActionNotify}, MinProbability: 0.5}, "ignore", 0.99) + v := e.Decide(context.Background(), Input{ + AlertID: "a1", Service: "blog", Priority: "low", ErrorCount: 100, Window: 5 * time.Minute, BaselineMean: 20, + Patterns: []Pattern{{What: "HTTP 404", Status: 404, Count: 100}}, + Pods: []Pod{healthyPod()}, + }) + if v.Action != ActionNotify || v.Requested != ActionIgnore || !v.Changed() { + t.Errorf("action = %q requested = %q changed = %v, want ignore downgraded to notify", v.Action, v.Requested, v.Changed()) + } +} + +func TestDecideDowngradesALowConfidenceChoice(t *testing.T) { + e, _ := engineFor(t, Policy{Actions: []Action{ActionRestart, ActionIgnore}, MinProbability: 0.6}, "restart", 0.41) + v := e.Decide(context.Background(), Input{ + AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 90, Window: 5 * time.Minute, BaselineMean: 2, + Patterns: []Pattern{{What: "HTTP 500", Count: 90, New: true}}, + Pods: []Pod{{Namespace: "shop", Deployment: "checkout-api", Found: true, Ready: false, Phase: "Running"}}, + }) + if v.Action != ActionNotify { + t.Errorf("action = %q, want notify: 41%% is a guess", v.Action) + } +} + +func TestRestartBudgetIsEnforced(t *testing.T) { + e, _ := engineFor(t, Policy{Actions: []Action{ActionRestart}, MinProbability: 0.5, MaxRestartsPerHour: 1}, "restart", 0.9) + in := Input{ + AlertID: "a1", Service: "checkout-api", Priority: "high", ErrorCount: 90, Window: 5 * time.Minute, BaselineMean: 2, + Patterns: []Pattern{{What: "HTTP 500", Count: 90, New: true}}, + Pods: []Pod{{Namespace: "shop", Deployment: "checkout-api", Found: true, Ready: false, Phase: "Running"}}, + } + if v := e.Decide(context.Background(), in); v.Action != ActionRestart { + t.Fatalf("first decision = %q, want restart", v.Action) + } + second := e.Decide(context.Background(), in) + if second.Action != ActionNotify { + t.Errorf("second decision = %q, want notify: the hourly restart budget is spent", second.Action) + } +} + +func TestDecideDegradesToNotifyWhenTheModelIsDown(t *testing.T) { + f := &fake{status: http.StatusServiceUnavailable} + srv := f.server(t) + e := NewEngine(NewClient(srv.URL), Policy{Actions: []Action{ActionRestart}, Now: nowFixture}) + v := e.Decide(context.Background(), Input{AlertID: "a1", Service: "checkout-api", ErrorCount: 90, Window: 5 * time.Minute}) + if v.Action != ActionNotify || v.Executable() || v.Error == "" { + t.Errorf("action = %q executable = %v error = %q, want a plain notification", v.Action, v.Executable(), v.Error) + } +} + +func TestDecideRejectsAnUnknownAction(t *testing.T) { + f := &fake{choice: "delete-the-cluster", prob: map[string]float64{"delete-the-cluster": 1}} + srv := f.server(t) + e := NewEngine(NewClient(srv.URL), Policy{Actions: []Action{ActionRestart}, Now: nowFixture}) + v := e.Decide(context.Background(), Input{AlertID: "a1", Service: "x", ErrorCount: 1, Window: time.Minute}) + if v.Error == "" || v.Executable() { + t.Errorf("error = %q executable = %v, want an unknown action to be rejected", v.Error, v.Executable()) + } +} diff --git a/internal/laya/live_test.go b/internal/laya/live_test.go new file mode 100644 index 0000000..03d010a --- /dev/null +++ b/internal/laya/live_test.go @@ -0,0 +1,59 @@ +package laya + +import ( + "context" + "os" + "testing" + "time" +) + +// TestLiveModel is a smoke test against the real Laya API, so the state +// rendering and the questions stay in step with what the model actually +// understands. It is opt-in: LAYA_URL=https://laya.k6n.net go test ./internal/laya -run TestLiveModel +func TestLiveModel(t *testing.T) { + base := os.Getenv("LAYA_URL") + if base == "" { + t.Skip("set LAYA_URL to run against the live model") + } + cases := []struct { + name string + in Input + }{ + { + name: "crash loop after a release", + in: Input{ + Service: "checkout-api", Priority: "high", ErrorCount: 412, Window: 5 * time.Minute, BaselineMean: 3, NewErrors: true, Clients: 42, + Patterns: []Pattern{{What: "HTTP 500 | /api/cart | panic: runtime error: invalid memory address", Category: "service-defect", Status: 500, Count: 300, Baseline: 2, New: true, Clients: 42, Services: []string{"checkout-api", "cart-service"}}}, + Pods: []Pod{{Namespace: "shop", Deployment: "checkout-api", Name: "checkout-api-7d9", Found: true, Phase: "Running", Restarts: 3, PodRestarts: 9, Reason: "CrashLoopBackOff", LastTerminatedAt: time.Now().Add(-30 * time.Second), LastEvents: []string{"panic: runtime error: invalid memory address", "SIGTERM received"}}}, + Deploys: []Deploy{{Namespace: "shop", Workload: "checkout-api", Release: "v2.31", Age: 4 * time.Minute, Ready: 1, Desired: 3}}, + }, + }, + { + name: "503 burst while a pod is being replaced", + in: Input{ + Service: "checkout-api", Priority: "medium", ErrorCount: 40, Window: 5 * time.Minute, BaselineMean: 2, Clients: 8, + Patterns: []Pattern{{What: "HTTP 503 | /api/cart | no healthy upstream", Category: "capacity-saturation", Status: 503, Count: 40, Baseline: 1, Clients: 8}}, + Pods: []Pod{{Namespace: "shop", Deployment: "checkout-api", Name: "checkout-api-2k1", Found: true, Phase: "Pending", LastTerminatedAt: time.Now().Add(-40 * time.Second), LastEvents: []string{"terminated"}}}, + Deploys: []Deploy{{Namespace: "shop", Workload: "checkout-api", Age: 2 * time.Minute, Ready: 1, Desired: 3}}, + }, + }, + { + name: "crawler 404s, nobody affected", + in: Input{ + Service: "blog", Priority: "low", ErrorCount: 1800, Window: 5 * time.Minute, BaselineMean: 20, KnownNoise: true, + Patterns: []Pattern{{What: "HTTP 404 | /old-post-2019 | UA looks like a crawler", Status: 404, Count: 1800, Baseline: 15}}, + Pods: []Pod{{Namespace: "shop", Deployment: "blog", Name: "blog-6f2", Found: true, Ready: true, Phase: "Running", LastEvents: []string{"GET /old-post-2019 404"}}}, + }, + }, + } + // Engine defaults, as deployed: the label probabilities are diffuse, so the + // gate is the lead over the runner-up. + engine := NewEngine(NewClient(base), Policy{Actions: Actions, Now: time.Now}) + for _, tc := range cases { + v := engine.Decide(context.Background(), tc.in) + t.Logf("%s:\n%s\n -> %s\n", tc.name, v.State, v.Explain()) + if v.Error != "" { + t.Errorf("%s: %v", tc.name, v.Error) + } + } +} diff --git a/internal/topology/podlogs.go b/internal/topology/podlogs.go index c5a4e22..404281b 100644 --- a/internal/topology/podlogs.go +++ b/internal/topology/podlogs.go @@ -16,6 +16,38 @@ type PodLog struct { Lines []string // most recent log lines, oldest first Found bool // a pod exists and was queried (Lines may still be empty: a quiet pod) Err string // set when the lookup itself failed; Found is false + + // The pod's own lifecycle facts, which explain errors the log lines alone + // cannot: a 503 burst while a replica is being replaced, a crash loop the + // kubelet is already retrying, or a pod that runs but never becomes ready. + Ready bool // container passing its readiness probe + Phase string // Pending | Running | Succeeded | Failed | Unknown + Restarts int32 // restarts of the selected pod's containers + PodRestarts int32 // restarts summed over every pod of the Deployment + Reason string // CrashLoopBackOff, OOMKilled, ImagePullBackOff, ... ("" while healthy) + ExitCode *int32 // exit code of the last container termination + Message string // message of the last container termination + LastTerminatedAt time.Time // when the last container terminated +} + +// rollingWindow is how recently a container must have died for the resulting +// connection errors to be attributed to its replacement still coming up. +const rollingWindow = 5 * time.Minute + +// CrashLooping reports whether the pod is stuck restarting instead of serving. +// Restarting the Deployment then only repeats what the kubelet is already +// doing, so the answer is a new release, not a restart. +func (p PodLog) CrashLooping() bool { + if p.Reason == "CrashLoopBackOff" { + return true + } + return p.Restarts > 0 && !p.LastTerminatedAt.IsZero() +} + +// Restarting reports whether a pod of this Deployment is right now being +// replaced, which is what makes a short 503 burst expected rather than new. +func (p PodLog) Restarting() bool { + return !p.LastTerminatedAt.IsZero() && time.Since(p.LastTerminatedAt) < rollingWindow } // podLogTimeout bounds one lookup so a slow or unreachable API server never @@ -64,6 +96,7 @@ func (r *Registry) PodLogs(ctx context.Context, namespace, deployment string, ta return live[i].CreationTimestamp.After(live[j].CreationTimestamp.Time) }) pod := live[0] + state := podState(pod, live) container := "" if len(pod.Spec.Containers) > 0 { container = pod.Spec.Containers[0].Name @@ -71,7 +104,7 @@ func (r *Registry) PodLogs(ctx context.Context, namespace, deployment string, ta tailLines := int64(tail) stream, err := r.client.CoreV1().Pods(namespace).GetLogs(pod.Name, &corev1.PodLogOptions{Container: container, TailLines: &tailLines}).Stream(ctx) if err != nil { - return PodLog{Pod: pod.Name, Found: true, Err: err.Error()} + return withState(state, PodLog{Pod: pod.Name, Found: true, Err: err.Error()}) } defer stream.Close() var lines []string @@ -83,5 +116,53 @@ func (r *Registry) PodLogs(ctx context.Context, namespace, deployment string, ta if len(lines) > tail { lines = lines[len(lines)-tail:] } - return PodLog{Pod: pod.Name, Lines: lines, Found: true} + return withState(state, PodLog{Pod: pod.Name, Lines: lines, Found: true}) +} + +// withState keeps the query result but carries the lifecycle facts along. +func withState(state, result PodLog) PodLog { + result.Ready, result.Phase, result.Restarts, result.PodRestarts = state.Ready, state.Phase, state.Restarts, state.PodRestarts + result.Reason, result.ExitCode, result.Message, result.LastTerminatedAt = state.Reason, state.ExitCode, state.Message, state.LastTerminatedAt + return result +} + +// podState reads the lifecycle facts off the selected pod and totals the +// restarts of its siblings, so one unhealthy replica is not mistaken for a +// healthy deployment. +func podState(selected corev1.Pod, all []corev1.Pod) PodLog { + state := PodLog{Found: true, Ready: len(selected.Status.ContainerStatuses) > 0, Phase: string(selected.Status.Phase)} + // The most recent termination across the pod's containers is the + // interesting one: a pod being replaced is explained by its last death. + keepTermination := func(term *corev1.ContainerStateTerminated) { + if term == nil || !term.FinishedAt.Time.After(state.LastTerminatedAt) { + return + } + state.LastTerminatedAt = term.FinishedAt.Time + state.Message = term.Message + code := term.ExitCode + state.ExitCode = &code + switch { + case term.Reason != "": + state.Reason = term.Reason + case term.Message != "": + state.Reason = "Error" + } + } + for i := range selected.Status.ContainerStatuses { + cs := &selected.Status.ContainerStatuses[i] + state.Restarts += cs.RestartCount + if !cs.Ready { + state.Ready = false + } + keepTermination(cs.LastTerminationState.Terminated) + } + if selected.Status.Reason != "" { + state.Reason = selected.Status.Reason + } + for _, p := range all { + for _, cs := range p.Status.ContainerStatuses { + state.PodRestarts += cs.RestartCount + } + } + return state }