package main import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "log" "net/http" "sync" "time" ) // AlertRouter implements design 5.2 "알림 수신기": Grafana webhook -> t_alert, then // // signature present -> layer 1: the platform mechanism is already acting; the harness observes and, if the alert // is still firing after escalate_after_s, hands it to layer 2 // no signature -> layer 2: POST the alert to the AI VM /diagnose and record the diagnosis type AlertRouter struct { app *App mu sync.Mutex timers map[string]*time.Timer // fingerprint -> escalation timer } func NewAlertRouter(a *App) *AlertRouter { return &AlertRouter{app: a, timers: map[string]*time.Timer{}} } // Grafana webhook payload (https://grafana.com/docs/grafana/latest/alerting/configure-notifications/manage-contact-points/integrations/webhook-notifier/) type grafanaWebhook struct { Receiver string `json:"receiver"` Status string `json:"status"` Alerts []struct { Status string `json:"status"` Labels map[string]string `json:"labels"` Annotations map[string]string `json:"annotations"` StartsAt time.Time `json:"startsAt"` EndsAt time.Time `json:"endsAt"` Fingerprint string `json:"fingerprint"` GeneratorURL string `json:"generatorURL"` } `json:"alerts"` } // Layer-1 runbook: signature -> what the harness does. "observe" = a platform mechanism owns the recovery // (kubelet / MIG autohealing / Argo CD selfHeal, design 4.3); any other value must be a catalogue action_id of layer 1. var runbook = map[string]string{ "pod_crashloop": "observe", "legacy_unhealthy": "observe", "gitops_drift": "observe", } func (r *AlertRouter) handleWebhook(w http.ResponseWriter, req *http.Request) { body, err := io.ReadAll(io.LimitReader(req.Body, 1<<20)) if err != nil { writeErr(w, 400, err) return } var wh grafanaWebhook if err := json.Unmarshal(body, &wh); err != nil { writeErr(w, 400, fmt.Errorf("payload: %w", err)) return } cur := r.app.runs.current() var runID, sid *string if cur != nil { runID = &cur.RunID if cur.ScenarioID != "" { sid = &cur.ScenarioID } } bg := context.Background() handled := 0 for _, al := range wh.Alerts { name := al.Labels["alertname"] mAlerts.WithLabelValues(name, al.Status).Inc() sig := al.Labels["signature"] payload := map[string]any{"alertname": name, "status": al.Status, "labels": al.Labels, "annotations": al.Annotations, "startsAt": al.StartsAt, "fingerprint": al.Fingerprint, "layer": al.Labels["layer"], "signature": sig} r.app.store.Event(bg, runID, sid, "alert", "grafana", payload) handled++ if al.Status != "firing" { r.cancelTimer(al.Fingerprint) continue } if sig != "" { action := runbook[sig] if action == "" { action = "observe" } r.app.store.Event(bg, runID, sid, "action", "platform", map[string]any{"action_id": action, "signature": sig, "alertname": name, "note": "layer 1: platform mechanism owns the recovery; harness observes"}) if action != "observe" { res, err := r.app.executeAction(bg, action, targetFromLabels(al.Labels), "runbook", runID, sid) r.app.store.Event(bg, runID, sid, "action", "runbook", map[string]any{"action_id": action, "result": res, "error": errStr(err)}) } r.startEscalation(al.Fingerprint, payload, runID, sid) continue } go r.diagnose(payload, runID, sid) } writeJSON(w, 200, map[string]any{"received": handled, "run_id": runID, "scenario_id": sid}) } func targetFromLabels(l map[string]string) map[string]any { t := map[string]any{} for _, k := range []string{"namespace", "pod", "deployment", "node", "instance"} { if v := l[k]; v != "" { t[k] = v } } return t } func (r *AlertRouter) escalateAfter(sid *string) time.Duration { if sid != nil { if sc, ok := r.app.scen[*sid]; ok && sc.EscalateAfterS > 0 { return time.Duration(sc.EscalateAfterS) * time.Second } } return 5 * time.Minute } func (r *AlertRouter) startEscalation(fp string, payload map[string]any, runID, sid *string) { r.mu.Lock() defer r.mu.Unlock() if _, exists := r.timers[fp]; exists { return } d := r.escalateAfter(sid) r.timers[fp] = time.AfterFunc(d, func() { r.mu.Lock() delete(r.timers, fp) r.mu.Unlock() payload["escalated_after_s"] = d.Seconds() r.app.store.Event(context.Background(), runID, sid, "note", "runner", map[string]any{"phase": "escalate", "fingerprint": fp, "alertname": payload["alertname"]}) r.diagnose(payload, runID, sid) }) } func (r *AlertRouter) cancelTimer(fp string) { r.mu.Lock() defer r.mu.Unlock() if t, ok := r.timers[fp]; ok { t.Stop() delete(r.timers, fp) } } // diagnose asks the AI VM (design 6.3 strict schema). Until WP-12 the endpoint is absent; the failure is recorded as an event. func (r *AlertRouter) diagnose(alert map[string]any, runID, sid *string) { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) defer cancel() reqBody := map[string]any{"alert": alert, "run_id": runID, "scenario_id": sid} b, _ := json.Marshal(reqBody) req, _ := http.NewRequestWithContext(ctx, "POST", r.app.cfg.AIDiagnoseURL, bytes.NewReader(b)) req.Header.Set("Content-Type", "application/json") resp, err := http.DefaultClient.Do(req) if err != nil { r.app.store.Event(ctx, runID, sid, "diagnosis", "ai", map[string]any{"error": err.Error(), "alertname": alert["alertname"], "root_cause_label": "unavailable"}) log.Printf("diagnose: %v", err) return } defer resp.Body.Close() rb, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) var diag map[string]any if err := json.Unmarshal(rb, &diag); err != nil || resp.StatusCode != 200 { r.app.store.Event(ctx, runID, sid, "diagnosis", "ai", map[string]any{"error": fmt.Sprintf("status %d: %s", resp.StatusCode, truncate(string(rb), 500)), "alertname": alert["alertname"], "root_cause_label": "unavailable"}) return } diag["alertname"] = alert["alertname"] r.app.store.Event(ctx, runID, sid, "diagnosis", "ai", diag) } func errStr(err error) string { if err == nil { return "" } return err.Error() } var errNoActiveRun = errors.New("no active run")