feat: add harness runner (scenario executor, alert receiver, action executor, judge, revert)
This commit is contained in:
seon committed
1 parent
8f5378b6ea
commit
5daf13b669
16 files changed
+2550
No files matched your search
@@ -0,0 +1,180 @@
|
||||
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")
|
||||
Reference in new issue
Block a user