selfheal-lab-apps/harness/alerts.go

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"sort"
	"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
	firing map[string]string      // fingerprint -> alertname while Grafana reports it firing (judge: no recovery with an open alert)
}

func NewAlertRouter(a *App) *AlertRouter {
	return &AlertRouter{app: a, timers: map[string]*time.Timer{}, firing: map[string]string{}}
}

// Firing lists the alert names currently firing (deduplicated, sorted).
func (r *AlertRouter) Firing() []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	seen := map[string]bool{}
	var out []string
	for _, n := range r.firing {
		if !seen[n] {
			seen[n] = true
			out = append(out, n)
		}
	}
	sort.Strings(out)
	return out
}

func (r *AlertRouter) setFiring(fp, name string, on bool) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if on {
		r.firing[fp] = name
	} else {
		delete(r.firing, fp)
	}
}

// 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++
		r.setFiring(al.Fingerprint, name, al.Status == "firing")
		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")