From 5daf13b6690b7dab2d04d9e685af68494a8a1eb2 Mon Sep 17 00:00:00 2001 From: seonNoh Date: Thu, 24 Sep 2026 02:06:08 +0900 Subject: [PATCH] feat: add harness runner (scenario executor, alert receiver, action executor, judge, revert) --- harness/Dockerfile | 15 ++ harness/actions.go | 201 +++++++++++++++++++++ harness/alerts.go | 180 +++++++++++++++++++ harness/aws.go | 63 +++++++ harness/gcp.go | 172 ++++++++++++++++++ harness/go.mod | 93 ++++++++++ harness/go.sum | 222 ++++++++++++++++++++++++ harness/inject.go | 113 ++++++++++++ harness/judge.go | 91 ++++++++++ harness/k8s.go | 225 ++++++++++++++++++++++++ harness/main.go | 206 ++++++++++++++++++++++ harness/proposals.go | 65 +++++++ harness/run.go | 405 +++++++++++++++++++++++++++++++++++++++++++ harness/scenario.go | 116 +++++++++++++ harness/sshx.go | 69 ++++++++ harness/store.go | 314 +++++++++++++++++++++++++++++++++ 16 files changed, 2550 insertions(+) create mode 100644 harness/Dockerfile create mode 100644 harness/actions.go create mode 100644 harness/alerts.go create mode 100644 harness/aws.go create mode 100644 harness/gcp.go create mode 100644 harness/go.mod create mode 100644 harness/go.sum create mode 100644 harness/inject.go create mode 100644 harness/judge.go create mode 100644 harness/k8s.go create mode 100644 harness/main.go create mode 100644 harness/proposals.go create mode 100644 harness/run.go create mode 100644 harness/scenario.go create mode 100644 harness/sshx.go create mode 100644 harness/store.go diff --git a/harness/Dockerfile b/harness/Dockerfile new file mode 100644 index 0000000..43674c4 --- /dev/null +++ b/harness/Dockerfile @@ -0,0 +1,15 @@ +# harness runner (self-healing lab, experiment design 5). Built with Cloud Build into the public Artifact Registry. +FROM golang:1.27.1-alpine AS build +WORKDIR /src +ENV CGO_ENABLED=0 GOFLAGS=-mod=mod +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +RUN go build -trimpath -ldflags="-s -w" -o /out/harness . + +FROM alpine:3.22 +RUN apk add --no-cache ca-certificates && adduser -D -u 10001 harness +USER 10001 +COPY --from=build /out/harness /usr/local/bin/harness +EXPOSE 8080 +ENTRYPOINT ["/usr/local/bin/harness"] diff --git a/harness/actions.go b/harness/actions.go new file mode 100644 index 0000000..734b4ef --- /dev/null +++ b/harness/actions.go @@ -0,0 +1,201 @@ +package main + +import ( + "context" + "crypto/subtle" + "encoding/json" + "errors" + "fmt" + "net/http" + "time" +) + +// Catalogue of design 2.3. Exec = the harness can carry it out; otherwise it is a platform-internal (layer 1) or +// human-only (layer 3) action that the harness records but does not perform. +type CatalogEntry struct { + ActionID string `json:"action_id"` + SLOImpact string `json:"slo_impact"` // small | large + Reversible bool `json:"reversible"` + Layer string `json:"layer"` + Exec bool `json:"harness_executes"` + Description string `json:"description"` +} + +var catalog = []CatalogEntry{ + {"k8s.container_restart", "small", true, "1", false, "kubelet restarts the container on liveness failure"}, + {"k8s.pod_recreate", "small", true, "1", false, "ReplicaSet recreates the pod"}, + {"gitops.self_heal", "small", true, "1", false, "Argo CD selfHeal restores the Git state"}, + {"mig.autoheal_recreate", "small", true, "1", false, "MIG autohealing recreates the instance"}, + {"vm.service_restart", "small", true, "1", false, "compose restart:always / systemd Restart=on-failure"}, + {"k8s.rollout_restart", "small", true, "1/2", true, "kubectl rollout restart deployment/"}, + {"k8s.rollout_undo", "large", true, "2", true, "kubectl rollout undo deployment/"}, + {"k8s.scale", "small", true, "2", true, "change replicas"}, + {"k8s.node_drain", "large", true, "2", true, "cordon + drain a worker (undo: uncordon)"}, + {"k8s.node_uncordon", "small", true, "2", true, "uncordon a worker (revert of k8s.node_drain)"}, + {"gce.instance_reset", "large", true, "2", true, "gcloud compute instances reset "}, + {"mig.recreate_instance", "small", true, "2", true, "recreate one legacy MIG instance"}, + {"mig.resize", "small", true, "2", true, "MIG target_size (restore 2 before the session ends)"}, + {"db.replay_resume", "small", true, "2", true, "SELECT pg_wal_replay_resume() on db-replica"}, + {"net.route_change", "large", true, "2", false, "route/firewall change through Terraform"}, + {"db.promote", "large", false, "3", false, "pg_promote() + DSN switch (human)"}, + {"db.data_fix", "large", false, "3", false, "data fix SQL / restore from backup (human)"}, + {"tf.instance_lifecycle", "large", false, "3", false, "instance create/delete/spec change through Terraform (human)"}, + {"escalate", "-", true, "3", false, "no catalogue action fits; hand over to a human"}, +} + +func catalogEntry(id string) *CatalogEntry { + for i := range catalog { + if catalog[i].ActionID == id { + return &catalog[i] + } + } + return nil +} + +func (a *App) handleCatalog(w http.ResponseWriter, r *http.Request) { writeJSON(w, 200, catalog) } + +type ActionRequest struct { + ActionID string `json:"action_id"` + Target map[string]any `json:"target"` + Token string `json:"token"` + RunID *string `json:"run_id"` + ScenarioID *string `json:"scenario_id"` + Note string `json:"note"` +} + +// POST /actions/execute (design 5.2): token = human token (layer 3) or an approved proposal token (layer 2). +func (a *App) handleActionExecute(w http.ResponseWriter, r *http.Request) { + var req ActionRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeErr(w, 400, err) + return + } + ce := catalogEntry(req.ActionID) + if ce == nil { + writeErr(w, 400, fmt.Errorf("action_id %q is not in the catalogue", req.ActionID)) + return + } + actor, err := a.authorize(r.Context(), req) + if err != nil { + writeErr(w, 401, err) + return + } + runID, sid := req.RunID, req.ScenarioID + if cur := a.runs.current(); cur != nil && runID == nil { + runID = &cur.RunID + if cur.ScenarioID != "" { + sid = &cur.ScenarioID + } + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() + if !ce.Exec { + a.store.Event(ctx, runID, sid, "action", actor, map[string]any{"action_id": req.ActionID, "target": req.Target, "recorded_only": true, "note": req.Note, "layer": ce.Layer}) + mActs.WithLabelValues(req.ActionID, actor, "recorded").Inc() + writeJSON(w, 202, map[string]any{"action_id": req.ActionID, "recorded_only": true, "layer": ce.Layer, "note": "not executed by the harness (platform-internal or human-only); recorded in events"}) + return + } + res, err := a.executeAction(ctx, req.ActionID, req.Target, actor, runID, sid) + result := "ok" + if err != nil { + result = "error" + } + mActs.WithLabelValues(req.ActionID, actor, result).Inc() + a.store.Event(ctx, runID, sid, "action", actor, map[string]any{"action_id": req.ActionID, "target": req.Target, "result": res, "error": errStr(err), "note": req.Note}) + if req.Token != "" && actor == "ai-approved" { + _ = a.store.MarkProposalExecuted(ctx, req.Token) + } + if err != nil { + writeJSON(w, 500, map[string]any{"action_id": req.ActionID, "actor": actor, "error": err.Error(), "result": res}) + return + } + writeJSON(w, 200, map[string]any{"action_id": req.ActionID, "actor": actor, "result": res}) +} + +func (a *App) authorize(ctx context.Context, req ActionRequest) (string, error) { + if req.Token == "" { + return "", errors.New("token required") + } + if human, err := readFileTrim(a.cfg.HumanTokFile); err == nil && subtle.ConstantTimeCompare([]byte(human), []byte(req.Token)) == 1 { + return "human", nil + } + p, err := a.store.GetProposal(ctx, req.Token) + if err != nil { + return "", errors.New("invalid token") + } + switch { + case p.ApprovedAt == nil: + return "", errors.New("proposal not approved") + case p.ExecutedAt != nil: + return "", errors.New("proposal already executed") + case time.Now().After(p.ExpiresAt): + return "", errors.New("proposal expired") + case p.ActionID != req.ActionID: + return "", fmt.Errorf("token was approved for %s, not %s", p.ActionID, req.ActionID) + } + return "ai-approved", nil +} + +func str(m map[string]any, k string) string { + v, _ := m[k].(string) + return v +} + +func num(m map[string]any, k string) int64 { + switch v := m[k].(type) { + case float64: + return int64(v) + case int: + return int64(v) + case int64: + return v + } + return 0 +} + +// executeAction performs one executable catalogue action. +func (a *App) executeAction(ctx context.Context, id string, t map[string]any, actor string, runID, sid *string) (map[string]any, error) { + if t == nil { + t = map[string]any{} + } + ns := str(t, "namespace") + if ns == "" { + ns = "demo" + } + switch id { + case "k8s.rollout_restart": + return map[string]any{"deployment": str(t, "deployment")}, a.rolloutRestart(ctx, ns, str(t, "deployment")) + case "k8s.rollout_undo": + return map[string]any{"deployment": str(t, "deployment")}, a.rolloutUndo(ctx, ns, str(t, "deployment")) + case "k8s.scale": + return map[string]any{"deployment": str(t, "deployment"), "replicas": num(t, "replicas")}, a.scaleDeployment(ctx, ns, str(t, "deployment"), int32(num(t, "replicas"))) + case "k8s.node_drain": + n, err := a.drain(ctx, str(t, "node")) + return map[string]any{"node": str(t, "node"), "evicted": n}, err + case "k8s.node_uncordon": + return map[string]any{"node": str(t, "node")}, a.cordon(ctx, str(t, "node"), false) + case "gce.instance_reset": + return a.gcpOp(ctx, "instance_reset", str(t, "instance"), "", 0) + case "mig.recreate_instance": + return a.gcpOp(ctx, "mig_recreate", str(t, "instance"), migName(t), 0) + case "mig.resize": + return a.gcpOp(ctx, "mig_resize", "", migName(t), num(t, "size")) + case "db.replay_resume": + host := str(t, "host") + if host == "" { + host = "db-replica" + } + out, err := a.psqlExec(ctx, host, "postgres", "SELECT pg_wal_replay_resume()") + return map[string]any{"host": host, "output": out}, err + case "escalate": + return map[string]any{"escalated": true}, nil + } + return nil, fmt.Errorf("action %s is not executable by the harness", id) +} + +func migName(t map[string]any) string { + if m := str(t, "mig"); m != "" { + return m + } + return env("LEGACY_MIG", "selfheal-legacy") +} diff --git a/harness/alerts.go b/harness/alerts.go new file mode 100644 index 0000000..94f12fd --- /dev/null +++ b/harness/alerts.go @@ -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") diff --git a/harness/aws.go b/harness/aws.go new file mode 100644 index 0000000..b0daa88 --- /dev/null +++ b/harness/aws.go @@ -0,0 +1,63 @@ +package main + +import ( + "context" + "fmt" + "io" + "net/http" + "net/url" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/sts" +) + +// GCE metadata ID token (audience = the AWS role ARN, format=full so azp/sub are present) -> AssumeRoleWithWebIdentity +// (Terraform design 6.11; the trust policy accepts aud in [role ARN, SA unique id]). +func gceIDToken(ctx context.Context, audience string) (string, error) { + u := "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/identity?format=full&audience=" + url.QueryEscape(audience) + req, _ := http.NewRequestWithContext(ctx, "GET", u, nil) + req.Header.Set("Metadata-Flavor", "Google") + resp, err := (&http.Client{Timeout: 10 * time.Second}).Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + b, _ := io.ReadAll(resp.Body) + if resp.StatusCode != 200 { + return "", fmt.Errorf("metadata identity: %s: %s", resp.Status, string(b)) + } + return string(b), nil +} + +func (a *App) awsCreds(ctx context.Context) (aws.Config, error) { + if a.cfg.AWSRoleARN == "" { + return aws.Config{}, fmt.Errorf("AWS_ROLE_ARN not set") + } + tok, err := gceIDToken(ctx, a.cfg.AWSRoleARN) + if err != nil { + return aws.Config{}, err + } + anon := aws.Config{Region: a.cfg.AWSRegion, Credentials: aws.AnonymousCredentials{}} + out, err := sts.NewFromConfig(anon).AssumeRoleWithWebIdentity(ctx, &sts.AssumeRoleWithWebIdentityInput{ + RoleArn: aws.String(a.cfg.AWSRoleARN), RoleSessionName: aws.String("harness-runner"), WebIdentityToken: aws.String(tok), DurationSeconds: aws.Int32(3600), + }) + if err != nil { + return aws.Config{}, fmt.Errorf("assume-role-with-web-identity: %w", err) + } + c := out.Credentials + return aws.Config{Region: a.cfg.AWSRegion, Credentials: credentials.NewStaticCredentialsProvider(*c.AccessKeyId, *c.SecretAccessKey, *c.SessionToken)}, nil +} + +func (a *App) awsCallerIdentity(ctx context.Context) (string, error) { + cfg, err := a.awsCreds(ctx) + if err != nil { + return "", err + } + id, err := sts.NewFromConfig(cfg).GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{}) + if err != nil { + return "", err + } + return aws.ToString(id.Arn), nil +} diff --git a/harness/gcp.go b/harness/gcp.go new file mode 100644 index 0000000..2cd6af2 --- /dev/null +++ b/harness/gcp.go @@ -0,0 +1,172 @@ +package main + +import ( + "context" + "fmt" + "sort" + "strconv" + "strings" + "time" + + "google.golang.org/api/compute/v1" +) + +// GCP calls use the VM service account through ADC (metadata server); no key files (Terraform design 6.11). +func (a *App) computeSvc(ctx context.Context) (*compute.Service, error) { + if a.cfg.GCPProject == "" { + return nil, fmt.Errorf("GCP_PROJECT not set") + } + return compute.NewService(ctx) +} + +type gceInstance struct { + Name, Zone, IP, Status string +} + +func (a *App) gcpInstances(ctx context.Context, prefix string) ([]gceInstance, error) { + svc, err := a.computeSvc(ctx) + if err != nil { + return nil, err + } + var out []gceInstance + err = svc.Instances.AggregatedList(a.cfg.GCPProject).Filter(fmt.Sprintf("name eq ^%s.*", prefix)).Pages(ctx, func(l *compute.InstanceAggregatedList) error { + for _, s := range l.Items { + for _, in := range s.Instances { + ip := "" + if len(in.NetworkInterfaces) > 0 { + ip = in.NetworkInterfaces[0].NetworkIP + } + z := in.Zone[strings.LastIndex(in.Zone, "/")+1:] + out = append(out, gceInstance{Name: in.Name, Zone: z, IP: ip, Status: in.Status}) + } + } + return nil + }) + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) + return out, err +} + +func (a *App) gcpListInstances(ctx context.Context, prefix string) ([]string, error) { + l, err := a.gcpInstances(ctx, prefix) + if err != nil { + return nil, err + } + names := make([]string, 0, len(l)) + for _, i := range l { + names = append(names, i.Name+"("+i.Status+")") + } + return names, nil +} + +// resolveInstance turns "legacy:N" (N-th MIG instance by name) or a pet VM short name (db-primary -> selfheal-db-primary) into an instance. +func (a *App) resolveInstance(ctx context.Context, ref string) (gceInstance, error) { + if strings.HasPrefix(ref, "legacy:") { + n, _ := strconv.Atoi(strings.TrimPrefix(ref, "legacy:")) + l, err := a.gcpInstances(ctx, "legacy-") + if err != nil { + return gceInstance{}, err + } + if n < 0 || n >= len(l) { + return gceInstance{}, fmt.Errorf("legacy:%d out of range (%d instances)", n, len(l)) + } + return l[n], nil + } + name := ref + if !strings.HasPrefix(name, "selfheal-") { + name = "selfheal-" + name + } + l, err := a.gcpInstances(ctx, name) + if err != nil { + return gceInstance{}, err + } + for _, i := range l { + if i.Name == name { + return i, nil + } + } + return gceInstance{}, fmt.Errorf("instance %s not found", name) +} + +// waitOp polls a zone / region / global operation until DONE (scope = zone name, "region:" or ""). +func (a *App) waitOp(ctx context.Context, svc *compute.Service, scope string, op *compute.Operation) error { + for i := 0; i < 150; i++ { + var cur *compute.Operation + var err error + switch { + case strings.HasPrefix(scope, "region:"): + cur, err = svc.RegionOperations.Get(a.cfg.GCPProject, strings.TrimPrefix(scope, "region:"), op.Name).Context(ctx).Do() + case scope != "": + cur, err = svc.ZoneOperations.Get(a.cfg.GCPProject, scope, op.Name).Context(ctx).Do() + default: + cur, err = svc.GlobalOperations.Get(a.cfg.GCPProject, op.Name).Context(ctx).Do() + } + if err != nil { + return err + } + if cur.Status == "DONE" { + if cur.Error != nil && len(cur.Error.Errors) > 0 { + return fmt.Errorf("operation %s: %s", cur.Name, cur.Error.Errors[0].Message) + } + return nil + } + time.Sleep(2 * time.Second) + } + return fmt.Errorf("operation %s timed out", op.Name) +} + +// gcpOp executes one catalogue/injection operation. instance is "legacy:N" or a pet VM short name; mig is the MIG name. +func (a *App) gcpOp(ctx context.Context, op, instance, mig string, size int64) (map[string]any, error) { + svc, err := a.computeSvc(ctx) + if err != nil { + return nil, err + } + res := map[string]any{"op": op} + switch op { + case "instance_delete", "instance_stop", "instance_start", "instance_reset": + in, err := a.resolveInstance(ctx, instance) + if err != nil { + return nil, err + } + res["instance"], res["zone"] = in.Name, in.Zone + var o *compute.Operation + switch op { + case "instance_delete": + o, err = svc.Instances.Delete(a.cfg.GCPProject, in.Zone, in.Name).Context(ctx).Do() + case "instance_stop": + o, err = svc.Instances.Stop(a.cfg.GCPProject, in.Zone, in.Name).Context(ctx).Do() + case "instance_start": + o, err = svc.Instances.Start(a.cfg.GCPProject, in.Zone, in.Name).Context(ctx).Do() + case "instance_reset": + o, err = svc.Instances.Reset(a.cfg.GCPProject, in.Zone, in.Name).Context(ctx).Do() + } + if err != nil { + return nil, err + } + return res, a.waitOp(ctx, svc, in.Zone, o) + case "mig_recreate": + in, err := a.resolveInstance(ctx, instance) + if err != nil { + return nil, err + } + res["instance"] = in.Name + req := &compute.RegionInstanceGroupManagersRecreateRequest{Instances: []string{fmt.Sprintf("zones/%s/instances/%s", in.Zone, in.Name)}} + region := a.region(in.Zone) + o, err := svc.RegionInstanceGroupManagers.RecreateInstances(a.cfg.GCPProject, region, mig, req).Context(ctx).Do() + if err != nil { + return nil, err + } + return res, a.waitOp(ctx, svc, "region:"+region, o) + case "mig_resize": + region := a.regionOfMIG() + res["mig"], res["size"] = mig, size + o, err := svc.RegionInstanceGroupManagers.Resize(a.cfg.GCPProject, region, mig, size).Context(ctx).Do() + if err != nil { + return nil, err + } + return res, a.waitOp(ctx, svc, "region:"+region, o) + } + return nil, fmt.Errorf("unknown gcp op %q", op) +} + +func (a *App) region(zone string) string { return zone[:strings.LastIndex(zone, "-")] } +func (a *App) regionOfMIG() string { return env("GCP_REGION", "asia-northeast1") } diff --git a/harness/go.mod b/harness/go.mod new file mode 100644 index 0000000..0f2a173 --- /dev/null +++ b/harness/go.mod @@ -0,0 +1,93 @@ +module github.com/selfheal-lab/selfheal-lab-apps/harness + +go 1.27 + +require ( + github.com/aws/aws-sdk-go-v2 v1.47.0 + github.com/aws/aws-sdk-go-v2/credentials v1.20.5 + github.com/aws/aws-sdk-go-v2/service/sts v1.51.0 + github.com/jackc/pgx/v5 v5.11.0 + github.com/prometheus/client_golang v1.24.1 + golang.org/x/crypto v0.57.0 + google.golang.org/api v0.299.0 + gopkg.in/yaml.v3 v3.0.1 + k8s.io/api v0.37.0 + k8s.io/apimachinery v0.37.0 + k8s.io/client-go v0.37.0 + sigs.k8s.io/yaml v1.6.0 +) + +require ( + cloud.google.com/go/auth v0.23.3 // indirect + cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect + cloud.google.com/go/compute/metadata v0.9.1 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.3 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.3 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.3 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.3 // indirect + github.com/aws/smithy-go v1.28.1 // indirect + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect + github.com/emicklei/go-restful/v3 v3.13.0 // indirect + github.com/felixge/httpsnoop v1.1.0 // indirect + github.com/fxamacker/cbor/v2 v2.9.1 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/go-openapi/jsonpointer v1.0.0 // indirect + github.com/go-openapi/jsonreference v1.0.0 // indirect + github.com/go-openapi/swag v0.27.1 // indirect + github.com/go-openapi/swag/cmdutils v0.27.1 // indirect + github.com/go-openapi/swag/conv v0.27.1 // indirect + github.com/go-openapi/swag/fileutils v0.27.1 // indirect + github.com/go-openapi/swag/jsonutils v0.27.1 // indirect + github.com/go-openapi/swag/loading v0.27.1 // indirect + github.com/go-openapi/swag/mangling v0.27.1 // indirect + github.com/go-openapi/swag/netutils v0.27.1 // indirect + github.com/go-openapi/swag/pools v0.27.1 // indirect + github.com/go-openapi/swag/stringutils v0.27.1 // indirect + github.com/go-openapi/swag/typeutils v0.27.1 // indirect + github.com/go-openapi/swag/yamlutils v0.27.1 // indirect + github.com/google/gnostic-models v0.7.0 // indirect + github.com/google/s2a-go v0.1.10 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/googleapis/enterprise-certificate-proxy v0.3.22 // indirect + github.com/googleapis/gax-go/v2 v2.24.1 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.70.1 // indirect + github.com/prometheus/procfs v0.21.1 // indirect + github.com/x448/float16 v0.8.4 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 // indirect + go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect + go.yaml.in/yaml/v2 v2.4.4 // indirect + go.yaml.in/yaml/v3 v3.0.4 // indirect + golang.org/x/net v0.59.0 // indirect + golang.org/x/oauth2 v0.37.0 // indirect + golang.org/x/sync v0.23.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/term v0.46.0 // indirect + golang.org/x/text v0.42.0 // indirect + golang.org/x/time v0.16.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260921155816-b14227669459 // indirect + google.golang.org/grpc v1.84.0 // indirect + google.golang.org/protobuf v1.36.12 // indirect + gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect + gopkg.in/inf.v0 v0.9.1 // indirect + k8s.io/klog/v2 v2.140.0 // indirect + k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad // indirect + k8s.io/utils v0.0.0-20260626114624-be93311217bd // indirect + sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect + sigs.k8s.io/randfill v1.0.0 // indirect + sigs.k8s.io/structured-merge-diff/v6 v6.4.2 // indirect +) diff --git a/harness/go.sum b/harness/go.sum new file mode 100644 index 0000000..538cbed --- /dev/null +++ b/harness/go.sum @@ -0,0 +1,222 @@ +cloud.google.com/go/auth v0.23.3 h1:UMK+oBtuNGMCR/6i6mmySUItqjOazpJrbmZyhGbGBWo= +cloud.google.com/go/auth v0.23.3/go.mod h1:fClbry28fo7XkxhSeT6AQtAVAp6Jy0fW9N99PoPNPFM= +cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc= +cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c= +cloud.google.com/go/compute/metadata v0.9.1 h1:CTE1OWBQ0vnF5uHwdFAQJvMQ0Fi/KRcqqKTo9V0F8Ik= +cloud.google.com/go/compute/metadata v0.9.1/go.mod h1:NtnlvB6X3t4R6xSWyVX/ZWk493PCxGQlhI/iqxh4M8I= +github.com/aws/aws-sdk-go-v2 v1.47.0 h1:0jsHallhJCeaU0Ko48c/3FK1ctOQ7NpzggxriJOQ8MQ= +github.com/aws/aws-sdk-go-v2 v1.47.0/go.mod h1:bttEH6JqnUL8LepvDVfdrds/fZ5bCIxzpe3abyUrhDU= +github.com/aws/aws-sdk-go-v2/credentials v1.20.5 h1:wklUVvHMc9xTQ3rcp49/ISpiMnhbCicJcA6n6S8m7J8= +github.com/aws/aws-sdk-go-v2/credentials v1.20.5/go.mod h1:fyEdrn6ccLFOkoK84j5bQyGTxp9zPt5l2XMhxf4DVZs= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.3 h1:Hp/VgjP0BysR3OgLlR057Vz2LcbbVnoWeJ+3qWiS/fY= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.3/go.mod h1:nwGV5qw7F1IZPgxCvA/ph8N2TAuz+BkRG/bXn808qMA= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.3 h1:MUaM4f+kj1ZIBPZfUS8cxP1GKXXZtHJjAthy93AN7SM= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.3/go.mod h1:6YmVmEVRI5ZZzRjCSsb9SryKH0hAlMRdgA7kG9aDvBU= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.3 h1:fuSCw4Z2qfRCztMPO3GXJNSiEp6Wee+WOLwrHHUMy9c= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.3/go.mod h1:6SxcHheD1pPR5+kWm1wGvjlL/YqUsh267sAfEmN4K7A= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19 h1:bAdDl/HkGCcGPoe25ToSHEw23VIxt6CT5fLcg111BKg= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19/go.mod h1:KaUzbLxv4CeSxh6ZCl9B4m7CuFenS8kUEaDs+f/DQr4= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.3 h1:bON1rJf67TSTDCKg816AAIE4xSTtoo9tl0XRkO72R+I= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.3/go.mod h1:c5BBpjJcQXpfeq9iASyVKA3T6vX6B6LEXY4mL/gklDY= +github.com/aws/aws-sdk-go-v2/service/sts v1.51.0 h1:Zpnqa6XtrNzXZnwbdCqHOXpXhMsa01ql/pcRQ1sb4hk= +github.com/aws/aws-sdk-go-v2/service/sts v1.51.0/go.mod h1:/8JRcdTt//hG0Q4BTmGbuOplT7ABe+5rdtqUHqXvYIM= +github.com/aws/smithy-go v1.28.1 h1:R/nXH00c8qcfCzQVELtRw+eLQWtzv+VAIEFJ1/xxXlQ= +github.com/aws/smithy-go v1.28.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/emicklei/go-restful/v3 v3.13.0 h1:C4Bl2xDndpU6nJ4bc1jXd+uTmYPVUwkD6bFY/oTyCes= +github.com/emicklei/go-restful/v3 v3.13.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc= +github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeORc= +github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE= +github.com/fxamacker/cbor/v2 v2.9.1 h1:2rWm8B193Ll4VdjsJY28jxs70IdDsHRWgQYAI80+rMQ= +github.com/fxamacker/cbor/v2 v2.9.1/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-openapi/jsonpointer v1.0.0 h1:kR9tHqY0CtZaOPVFm622dPVNhrvYpwr4uCxgL3h1H8s= +github.com/go-openapi/jsonpointer v1.0.0/go.mod h1:Z3rw7dWu1p9IgitXCFamSlA5lmDiklEB6vkaxcNZW5Y= +github.com/go-openapi/jsonreference v1.0.0 h1:jlmTr6torcd1YgDQvSfNmRtKzYDO4FGBkrAdlAVWnpY= +github.com/go-openapi/jsonreference v1.0.0/go.mod h1:jtwdyGbJk0Xhe5Y+rwtglQP6Sb1WZST4rT32LWB+sv0= +github.com/go-openapi/swag v0.27.1 h1:VotvOLWW8q/EAxB0YdsBBGC8XYyeL1YwBj2ungAGPNg= +github.com/go-openapi/swag v0.27.1/go.mod h1:GTkJPwHfhJp6MWr4/rCh64HVI3Ofu+tcsbfjfHmTxpE= +github.com/go-openapi/swag/cmdutils v0.27.1 h1:I7sYqaWVl5mq0NEmNQkAmFDyNin9ufvMX/p2zwtQaOE= +github.com/go-openapi/swag/cmdutils v0.27.1/go.mod h1:Sm1MVFMkF6guJJ+pQqHnQA3N0j9qALV3NxzDSv6bETM= +github.com/go-openapi/swag/conv v0.27.1 h1:8wi9ZG+olmY1wXphl93EWniPtbSPkXM/feH7FgjsvrU= +github.com/go-openapi/swag/conv v0.27.1/go.mod h1:QbqMivkpKhC3g1B1GGGOJ6ANewI3S62dbzYu3Duowqs= +github.com/go-openapi/swag/fileutils v0.27.1 h1:QQqBSoi5mW4XpU85nS0mLcA+zAE6vLzrb0QkmLKf9oM= +github.com/go-openapi/swag/fileutils v0.27.1/go.mod h1:VvJFZLTZS0AI854gEQz5tk7dBESdLjiNUMSZ/th2ry8= +github.com/go-openapi/swag/jsonutils v0.27.1 h1:SVgK3i4USzCU5mibOOS/l4ea2h9UQXy7J7RNLTjuXjU= +github.com/go-openapi/swag/jsonutils v0.27.1/go.mod h1:tdlEpZqdcQ17uj6J4YdK9vd8It5qWMwjWXOs0tjpRlk= +github.com/go-openapi/swag/jsonutils/fixtures_test v0.27.1 h1:mJu3COL9WEaZVp/Kf2PRMi7tPszPEJfSr/OO75ynCs8= +github.com/go-openapi/swag/jsonutils/fixtures_test v0.27.1/go.mod h1:mofwUWx70wvskwESqRJ//k/9kURmCgyJl5m5Ppoh5kY= +github.com/go-openapi/swag/loading v0.27.1 h1:/DxUgDXKbBX4bcn7r9uEXfJyzN5XpiJmZplzQTjrRCY= +github.com/go-openapi/swag/loading v0.27.1/go.mod h1:jvGh3iA2+zyUUycB5fgJWzeHnhrpvGnJJM0RVE9ZShE= +github.com/go-openapi/swag/mangling v0.27.1 h1:yC9D0HyUE8gbP+BfmGx9+AA89ikwZTMjESK3OnnoaqA= +github.com/go-openapi/swag/mangling v0.27.1/go.mod h1:jtBE2+V+3pILxOR7Vgce+Cwp6A2PgZbvVqfNntbVs0w= +github.com/go-openapi/swag/netutils v0.27.1 h1:mICMFoS82F5TZ4Zy3cqmcQk+BFeCp3Uyq3Np7GI0/qU= +github.com/go-openapi/swag/netutils v0.27.1/go.mod h1:J+WYyFMLtvtCGqa6jLv+YNUmIKI3ZRQRrvfNDMoQoEQ= +github.com/go-openapi/swag/pools v0.27.1 h1:9LeadcMyb2GJCbXX5hVQDbZ2Lq9TL4dCs/nx1j5DO0E= +github.com/go-openapi/swag/pools v0.27.1/go.mod h1:kVQefhSK5RWuRe7BXsL8htgBPAMpN7HDGpGEknqugeE= +github.com/go-openapi/swag/stringutils v0.27.1 h1:ZXePZ0r2p1qSjo8tD3Un4vFj8+FqlCkczxDrJIhYUp8= +github.com/go-openapi/swag/stringutils v0.27.1/go.mod h1:lzRN95CxXmA03XcDWHLOb6nOMcxCqR5rGY0lOgsfRoM= +github.com/go-openapi/swag/typeutils v0.27.1 h1:KSTdFlfnse4r6dP9IrEnwMldjE+zs71UeEB3//PtVXc= +github.com/go-openapi/swag/typeutils v0.27.1/go.mod h1:Srm0xFNRZ1Y+vCxJclo5qzx8aj+1pAKda/YfFPrG0dQ= +github.com/go-openapi/swag/yamlutils v0.27.1 h1:ftxv6xvXb1E3zohUc+okZ9nSqNb9StQX/FXnKZ98sQA= +github.com/go-openapi/swag/yamlutils v0.27.1/go.mod h1:bnxFIB1qewGRiZHypXGZ3fNgf13/0HfRgnS/iZBDrOo= +github.com/go-openapi/testify/enable/yaml/v2 v2.6.0 h1:gGHwAJ0R/5jU8BEGDbfRNR3hL68dAVi84WuOApp29B0= +github.com/go-openapi/testify/enable/yaml/v2 v2.6.0/go.mod h1:tY+St1SGq4NFl0QIqdTY4aEdbChAHxhyB77XQi9iJCo= +github.com/go-openapi/testify/v2 v2.6.0 h1:5PKH2HE7YJ/LuRPQGvSxBRlFXNQhSetBLlGAgUEu3ug= +github.com/go-openapi/testify/v2 v2.6.0/go.mod h1:SgsVHtfooshd0tublTtJ50FPKhujf47YRqauXXOUxfw= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo= +github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/s2a-go v0.1.10 h1:EMp+aOuXN6l8cE/gjF5Bt+vyZxsUuyCWe9chDWR/+uU= +github.com/google/s2a-go v0.1.10/go.mod h1:pz4tyvwXvJLLbyrkh6FW1eS2zPUXMaTmyNhYtyP2tNw= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/googleapis/enterprise-certificate-proxy v0.3.22 h1:NU4XpII6jD+Dxcot94fqjE+AfJoE/lQP9q3faYGzC/c= +github.com/googleapis/enterprise-certificate-proxy v0.3.22/go.mod h1:L3D/IQExI6LqEjBdXcZQ1WluSgigQmSwBboFstVPM4w= +github.com/googleapis/gax-go/v2 v2.24.1 h1:AtqTN21IXMMWo99LiEVAiBfNNQmO40d8xUfZI640mc0= +github.com/googleapis/gax-go/v2 v2.24.1/go.mod h1:bWeBei0NVwaNZKb2y1HUBS7gLXIF3/Tu3pq7j8D2Tb0= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= +github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= +github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8= +github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU= +github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY= +github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc= +github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI= +github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= +github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= +github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 h1:8tvICD4vSTOOsNrsI4Ljf6C+6UKvpTEH5XY3JMoyPoo= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0/go.mod h1:z9+yiacE0IHRqM4qFfkbt/JYlmYXgss8GY/jXoNuPJI= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= +go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues= +golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= +golang.org/x/oauth2 v0.37.0 h1:JUlcxA8oAtauLfiH8FX2/FkAWHAdi0QtGCGc+hofE98= +golang.org/x/oauth2 v0.37.0/go.mod h1:IxwZNxUULJmpBFf9K/9NTMSIfZZuvuTy1gGxhigP/58= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE= +golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/api v0.299.0 h1:b3K+ydSMd0kh6TQI6bJyApRQfqQX2MfSOaVkpM59mJw= +google.golang.org/api v0.299.0/go.mod h1:zlR3GVA8b2R5nv5Ij9UWe37StVB3cxDD7DBFi4ZFsHw= +google.golang.org/genproto v0.0.0-20260715232425-e75dac1f907d h1:C9v1o0/4quuhOAfmRXA2j+we0PqZIp8traLdeogF3Ms= +google.golang.org/genproto v0.0.0-20260715232425-e75dac1f907d/go.mod h1:Wz2wFJntZFmLGo7pLDXZ3wYk5hyc0Mb+SkHhDDXT+lU= +google.golang.org/genproto/googleapis/api v0.0.0-20260715232425-e75dac1f907d h1:QwnJwPte4XXAkhPu26LTDIahnsMSUV0kK8HkxbC+Pc4= +google.golang.org/genproto/googleapis/api v0.0.0-20260715232425-e75dac1f907d/go.mod h1:WRrQ7/7N19PypuT0fxLOL5Lq0waoiRri4FbtHDEKrGE= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260921155816-b14227669459 h1:b0xCahf3FK2m2Cv0p4vTozGPWncCvLfwV86UNg8xWU8= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260921155816-b14227669459/go.mod h1:OaIUM3+LpYcK2GXM4FTmhWoIq371Owdr+Cc7/BsYHHc= +google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0= +google.golang.org/grpc v1.84.0/go.mod h1:ljCht0DrxQrXBDRTZp52Qxh3Ffk8CdYm2sj4O2QN2C0= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnfEbYzo= +gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M= +gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= +gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +k8s.io/api v0.37.0 h1:Z//Vj9N7RA/yS2sDmxyeo7h+RR4zbUrd2vrd3Z0TbB4= +k8s.io/api v0.37.0/go.mod h1:LKXgcJWMc+f4OLbP5SFR8rulEg07zZhpi/zMULiBImk= +k8s.io/apimachinery v0.37.0 h1:Np2AbDtf8x6RDHiD8T9LbKJ9gaegeVNa8yNm5FuGKm0= +k8s.io/apimachinery v0.37.0/go.mod h1:RN3nhprFSCxOi5Selxd7oMTXOe/c+ZbcE7Im+TS2zkE= +k8s.io/client-go v0.37.0 h1:nsN31fy8wBySuZ+QRnKmrjRSQLOG2rvoGN0tKd12zhQ= +k8s.io/client-go v0.37.0/go.mod h1:FcGqw+Ll/gNQiq+nPGY1Oyt9y7SgDh1d3MW3RFDEbn0= +k8s.io/klog/v2 v2.140.0 h1:Tf+J3AH7xnUzZyVVXhTgGhEKnFqye14aadWv7bzXdzc= +k8s.io/klog/v2 v2.140.0/go.mod h1:o+/RWfJ6PwpnFn7OyAG3QnO47BFsymfEfrz6XyYSSp0= +k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad h1:oXImqH8mQNk7PmvzKhmN3ddJoY6OnyM225MXwGHPm0A= +k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad/go.mod h1:0/mqHCVhlumdJ3BhCfnjSZQE037nAhNodh1/hK0T8/I= +k8s.io/utils v0.0.0-20260626114624-be93311217bd h1:Ea7fgQ5we8Y9T0OX5o0dAHzQOBRI07D/dEYRaB9ZZEs= +k8s.io/utils v0.0.0-20260626114624-be93311217bd/go.mod h1:xDxuJ0whA3d0I4mf/C4ppKHxXynQ+fxnkmQH0vTHnuk= +sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg= +sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg= +sigs.k8s.io/randfill v1.0.0 h1:JfjMILfT8A6RbawdsK2JXGBR5AQVfd+9TbzrlneTyrU= +sigs.k8s.io/randfill v1.0.0/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY= +sigs.k8s.io/structured-merge-diff/v6 v6.4.2 h1:qdOxHwrl2Kaag1aQEarlYcOA9vSyGCp3CIki3aW8c4Q= +sigs.k8s.io/structured-merge-diff/v6 v6.4.2/go.mod h1:M3W8sfWvn2HhQDIbGWj3S099YozAsymCo/wrT5ohRUE= +sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs= +sigs.k8s.io/yaml v1.6.0/go.mod h1:796bPqUfzR/0jLAl6XjHl3Ck7MiyVv8dbTdyT3/pMf4= diff --git a/harness/inject.go b/harness/inject.go new file mode 100644 index 0000000..84e0344 --- /dev/null +++ b/harness/inject.go @@ -0,0 +1,113 @@ +package main + +import ( + "context" + "fmt" + "net" + "strings" + "time" + + "github.com/jackc/pgx/v5" +) + +// runStep executes one injection/revert step; the returned map is recorded as injections.params and reused by the revert. +func (a *App) runStep(ctx context.Context, sc *Scenario, st Step, runID string, prev map[string]any) (map[string]any, error) { + res := map[string]any{"method": st.Method} + var err error + switch st.Method { + case "none": + case "chaos": + name := strings.ToLower(sc.ID) + "-" + runID[len(runID)-6:] + var ref map[string]any + ref, err = a.applyChaos(ctx, st.Manifest, name) + for k, v := range ref { + res[k] = v + } + case "chaos_delete": + err = a.deleteChaos(ctx, prev) + case "kubectl_scale": + res["namespace"], res["deployment"], res["replicas"] = st.Namespace, st.Deployment, st.Replicas + err = a.scaleDeployment(ctx, st.Namespace, st.Deployment, st.Replicas) + case "ssh": + res["host"], res["command"] = st.Host, st.Command + var out string + out, err = a.sshRun(ctx, st.Host, st.Command) + res["output"] = truncate(out, 2000) + case "gcloud": + var r map[string]any + r, err = a.gcpOp(ctx, st.Op, st.Instance, st.MIG, st.Size) + for k, v := range r { + res[k] = v + } + case "psql": + res["host"], res["database"], res["sql"] = st.Host, st.Database, st.SQL + var out string + out, err = a.psqlExec(ctx, st.Host, st.Database, st.SQL) + res["output"] = truncate(out, 2000) + default: + err = fmt.Errorf("unknown step method %q", st.Method) + } + if err != nil { + res["error"] = err.Error() + return res, err + } + if st.WaitS > 0 { + select { + case <-time.After(time.Duration(st.WaitS) * time.Second): + case <-ctx.Done(): + return res, ctx.Err() + } + } + return res, nil +} + +// psqlExec runs SQL on db-primary / db-replica as the harness role over hostssl (pg_hba: harness from 10.20.0.0/20). +func (a *App) psqlExec(ctx context.Context, host, database, sqlText string) (string, error) { + ip := host + if net.ParseIP(host) == nil { + in, err := a.resolveInstance(ctx, host) + if err != nil { + return "", err + } + ip = in.IP + } + pw, err := readFileTrim(a.cfg.DBPassFile) + if err != nil { + return "", fmt.Errorf("db password: %w", err) + } + if database == "" { + database = "postgres" + } + cfg, err := pgx.ParseConfig(fmt.Sprintf("postgres://harness@%s:5432/%s?sslmode=require&connect_timeout=10", ip, database)) + if err != nil { + return "", err + } + cfg.Password = pw + conn, err := pgx.ConnectConfig(ctx, cfg) + if err != nil { + return "", err + } + defer conn.Close(ctx) + rows, err := conn.Query(ctx, sqlText) + if err != nil { + return "", err + } + defer rows.Close() + var sb strings.Builder + for rows.Next() { + vals, err := rows.Values() + if err != nil { + return sb.String(), err + } + sb.WriteString(fmt.Sprint(vals...)) + sb.WriteString("\n") + } + return sb.String(), rows.Err() +} + +func truncate(s string, n int) string { + if len(s) > n { + return s[:n] + "...(truncated)" + } + return s +} diff --git a/harness/judge.go b/harness/judge.go new file mode 100644 index 0000000..6d52e8e --- /dev/null +++ b/harness/judge.go @@ -0,0 +1,91 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "io" + "math" + "net/http" + "net/url" + "strconv" + "time" +) + +// Mimir instant query (Prometheus HTTP API under /api/v1/query; single tenant, no X-Scope-OrgID). +// Returns NaN when the result vector is empty. +func (a *App) mimirScalar(ctx context.Context, expr string) (float64, error) { + u := a.cfg.MimirURL + "/api/v1/query?query=" + url.QueryEscape(expr) + req, _ := http.NewRequestWithContext(ctx, "GET", u, nil) + resp, err := (&http.Client{Timeout: 15 * time.Second}).Do(req) + if err != nil { + return math.NaN(), err + } + defer resp.Body.Close() + b, _ := io.ReadAll(resp.Body) + if resp.StatusCode != 200 { + return math.NaN(), fmt.Errorf("mimir %s: %s", resp.Status, truncate(string(b), 300)) + } + var r struct { + Status string `json:"status"` + Data struct { + ResultType string `json:"resultType"` + Result []struct { + Value []any `json:"value"` + } `json:"result"` + } `json:"data"` + } + if err := json.Unmarshal(b, &r); err != nil { + return math.NaN(), err + } + if r.Status != "success" || len(r.Data.Result) == 0 || len(r.Data.Result[0].Value) < 2 { + return math.NaN(), nil + } + s, _ := r.Data.Result[0].Value[1].(string) + return strconv.ParseFloat(s, 64) +} + +// SLOState is one evaluation of experiment design 4.2 (probe 12 consecutive successes = min over 3m; p95 and 5xx ratio over 5m). +type SLOState struct { + At time.Time `json:"at"` + ProbeMin3m float64 `json:"probe_min_3m"` + P95 float64 `json:"p95_s"` + ErrorRatio float64 `json:"error_ratio"` + ProbeOK bool `json:"probe_ok"` + LatencyOK bool `json:"latency_ok"` + ErrorsOK bool `json:"errors_ok"` + Recovered bool `json:"recovered"` + MissingData []string `json:"missing_data,omitempty"` + QueryErrors []string `json:"query_errors,omitempty"` +} + +const ( + qProbeMin3m = `min_over_time(probe_success{job="blackbox",target="alb"}[3m])` + qP95 = `histogram_quantile(0.95, sum by (le) (rate(demo_http_request_duration_seconds_bucket{route=~"orders|price"}[5m])))` + qErrRatio = `sum(rate(demo_http_requests_total{route=~"orders|price",status=~"5.."}[5m])) / sum(rate(demo_http_requests_total{route=~"orders|price"}[5m]))` +) + +func (a *App) evaluateSLO(ctx context.Context) SLOState { + st := SLOState{At: time.Now().UTC()} + get := func(name, expr string) (float64, bool) { + v, err := a.mimirScalar(ctx, expr) + if err != nil { + st.QueryErrors = append(st.QueryErrors, name+": "+err.Error()) + return math.NaN(), false + } + if math.IsNaN(v) { + st.MissingData = append(st.MissingData, name) + return v, false + } + return v, true + } + var ok bool + st.ProbeMin3m, ok = get("probe", qProbeMin3m) + st.ProbeOK = ok && st.ProbeMin3m >= 1 // the probe series must exist: no data is not recovery + st.P95, ok = get("p95", qP95) + st.LatencyOK = !ok || st.P95 < 0.5 // no traffic in the window (no k6) counts as within SLO + st.ErrorRatio, ok = get("errors", qErrRatio) + st.ErrorsOK = !ok || st.ErrorRatio < 0.01 + st.Recovered = st.ProbeOK && st.LatencyOK && st.ErrorsOK && len(st.QueryErrors) == 0 + return st +} diff --git a/harness/k8s.go b/harness/k8s.go new file mode 100644 index 0000000..767f125 --- /dev/null +++ b/harness/k8s.go @@ -0,0 +1,225 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strings" + "time" + + corev1 "k8s.io/api/core/v1" + policyv1 "k8s.io/api/policy/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/dynamic" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "sigs.k8s.io/yaml" +) + +// k8s clients are rebuilt per call so a refreshed token file (make token + make vm-sync) is picked up without restart. +func (a *App) restConfig() (*rest.Config, error) { + tok, err := readFileTrim(a.cfg.K8sTokenFile) + if err != nil { + return nil, fmt.Errorf("k8s token: %w", err) + } + ca, err := os.ReadFile(a.cfg.K8sCAFile) + if err != nil { + return nil, fmt.Errorf("k8s ca: %w", err) + } + return &rest.Config{Host: a.cfg.K8sAPI, BearerToken: tok, TLSClientConfig: rest.TLSClientConfig{CAData: ca}, Timeout: 30 * time.Second}, nil +} + +func (a *App) k8sClients() (kubernetes.Interface, dynamic.Interface, error) { + rc, err := a.restConfig() + if err != nil { + return nil, nil, err + } + cs, err := kubernetes.NewForConfig(rc) + if err != nil { + return nil, nil, err + } + dc, err := dynamic.NewForConfig(rc) + if err != nil { + return nil, nil, err + } + return cs, dc, nil +} + +func (a *App) k8sPodCount(ctx context.Context, ns string) (int, error) { + cs, _, err := a.k8sClients() + if err != nil { + return 0, err + } + l, err := cs.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{}) + if err != nil { + return 0, err + } + return len(l.Items), nil +} + +// chaos-mesh.org resources are named like their kind in lower case (podchaos, networkchaos, stresschaos ...) +func chaosGVR(kind, version string) schema.GroupVersionResource { + return schema.GroupVersionResource{Group: "chaos-mesh.org", Version: version, Resource: strings.ToLower(kind)} +} + +// applyChaos creates the CR from the scenario manifest with a run-unique name; returns kind/name for revert. +func (a *App) applyChaos(ctx context.Context, manifest, name string) (map[string]any, error) { + _, dc, err := a.k8sClients() + if err != nil { + return nil, err + } + var obj unstructured.Unstructured + if err := yaml.Unmarshal([]byte(manifest), &obj.Object); err != nil { + return nil, fmt.Errorf("manifest: %w", err) + } + gv, err := schema.ParseGroupVersion(obj.GetAPIVersion()) + if err != nil { + return nil, err + } + if gv.Group != "chaos-mesh.org" { + return nil, fmt.Errorf("only chaos-mesh.org manifests are allowed, got %s", obj.GetAPIVersion()) + } + ns := obj.GetNamespace() + if ns == "" { + ns = "chaos-mesh" + } + obj.SetNamespace(ns) + obj.SetName(name) + gvr := chaosGVR(obj.GetKind(), gv.Version) + created, err := dc.Resource(gvr).Namespace(ns).Create(ctx, &obj, metav1.CreateOptions{}) + if err != nil { + return nil, fmt.Errorf("create %s/%s: %w", obj.GetKind(), name, err) + } + return map[string]any{"kind": obj.GetKind(), "version": gv.Version, "namespace": ns, "name": created.GetName()}, nil +} + +func (a *App) deleteChaos(ctx context.Context, ref map[string]any) error { + _, dc, err := a.k8sClients() + if err != nil { + return err + } + kind, _ := ref["kind"].(string) + version, _ := ref["version"].(string) + ns, _ := ref["namespace"].(string) + name, _ := ref["name"].(string) + if kind == "" || name == "" { + return fmt.Errorf("no chaos reference to delete") + } + err = dc.Resource(chaosGVR(kind, version)).Namespace(ns).Delete(ctx, name, metav1.DeleteOptions{}) + if err != nil && !strings.Contains(err.Error(), "not found") { + return err + } + return nil +} + +func (a *App) scaleDeployment(ctx context.Context, ns, name string, replicas int32) error { + cs, _, err := a.k8sClients() + if err != nil { + return err + } + patch := fmt.Sprintf(`{"spec":{"replicas":%d}}`, replicas) + _, err = cs.AppsV1().Deployments(ns).Patch(ctx, name, types.MergePatchType, []byte(patch), metav1.PatchOptions{}) + return err +} + +func (a *App) rolloutRestart(ctx context.Context, ns, name string) error { + cs, _, err := a.k8sClients() + if err != nil { + return err + } + patch := fmt.Sprintf(`{"spec":{"template":{"metadata":{"annotations":{"kubectl.kubernetes.io/restartedAt":%q}}}}}`, time.Now().UTC().Format(time.RFC3339)) + _, err = cs.AppsV1().Deployments(ns).Patch(ctx, name, types.StrategicMergePatchType, []byte(patch), metav1.PatchOptions{}) + return err +} + +// rolloutUndo copies the pod template of the previous ReplicaSet revision into the deployment (what kubectl rollout undo does). +func (a *App) rolloutUndo(ctx context.Context, ns, name string) error { + cs, _, err := a.k8sClients() + if err != nil { + return err + } + dep, err := cs.AppsV1().Deployments(ns).Get(ctx, name, metav1.GetOptions{}) + if err != nil { + return err + } + sel := metav1.FormatLabelSelector(dep.Spec.Selector) + rss, err := cs.AppsV1().ReplicaSets(ns).List(ctx, metav1.ListOptions{LabelSelector: sel}) + if err != nil { + return err + } + cur := dep.Annotations["deployment.kubernetes.io/revision"] + var bestRev int64 = -1 + var best int = -1 + for i, rs := range rss.Items { + rev := rs.Annotations["deployment.kubernetes.io/revision"] + if rev == cur || rev == "" { + continue + } + var n int64 + fmt.Sscanf(rev, "%d", &n) + if n > bestRev { + bestRev, best = n, i + } + } + if best < 0 { + return fmt.Errorf("no previous revision for %s/%s", ns, name) + } + tpl := rss.Items[best].Spec.Template + delete(tpl.Labels, "pod-template-hash") + body, _ := json.Marshal(map[string]any{"spec": map[string]any{"template": tpl}}) + _, err = cs.AppsV1().Deployments(ns).Patch(ctx, name, types.StrategicMergePatchType, body, metav1.PatchOptions{}) + return err +} + +func (a *App) cordon(ctx context.Context, node string, unschedulable bool) error { + cs, _, err := a.k8sClients() + if err != nil { + return err + } + patch := fmt.Sprintf(`{"spec":{"unschedulable":%t}}`, unschedulable) + _, err = cs.CoreV1().Nodes().Patch(ctx, node, types.MergePatchType, []byte(patch), metav1.PatchOptions{}) + return err +} + +// drain = cordon + evict every non-DaemonSet, non-mirror pod on the node (PDBs are honoured by the eviction API) +func (a *App) drain(ctx context.Context, node string) (int, error) { + if err := a.cordon(ctx, node, true); err != nil { + return 0, err + } + cs, _, err := a.k8sClients() + if err != nil { + return 0, err + } + pods, err := cs.CoreV1().Pods("").List(ctx, metav1.ListOptions{FieldSelector: "spec.nodeName=" + node}) + if err != nil { + return 0, err + } + n := 0 + for _, p := range pods.Items { + if isDaemonOrMirror(&p) || p.Status.Phase == corev1.PodSucceeded || p.Status.Phase == corev1.PodFailed { + continue + } + ev := &policyv1.Eviction{ObjectMeta: metav1.ObjectMeta{Name: p.Name, Namespace: p.Namespace}} + if err := cs.CoreV1().Pods(p.Namespace).EvictV1(ctx, ev); err != nil { + return n, fmt.Errorf("evict %s/%s: %w", p.Namespace, p.Name, err) + } + n++ + } + return n, nil +} + +func isDaemonOrMirror(p *corev1.Pod) bool { + if _, ok := p.Annotations["kubernetes.io/config.mirror"]; ok { + return true + } + for _, o := range p.OwnerReferences { + if o.Kind == "DaemonSet" { + return true + } + } + return false +} diff --git a/harness/main.go b/harness/main.go new file mode 100644 index 0000000..3ecfacd --- /dev/null +++ b/harness/main.go @@ -0,0 +1,206 @@ +// harness runner: scenario executor, alert receiver, action executor, judge and reverter of the +// self-healing lab (experiment design 5). One binary, HTTP on :8080, state in the measure DB (5.3). +package main + +import ( + "context" + "encoding/json" + "errors" + "log" + "net/http" + "os" + "strings" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +type Config struct { + Addr string + MeasureDSN string + MimirURL string // e.g. http://10.20.0.10:9009/prometheus + ALBDNS string + K8sAPI string + K8sTokenFile string + K8sCAFile string + AWSRoleARN string + AWSRegion string + GCPProject string + SSHUser string + SSHKeyFile string + DBPassFile string // harness role password on db-primary / db-replica + HumanTokFile string + AIDiagnoseURL string + ScenarioDir string + JudgeEvery time.Duration + ObserveMax time.Duration +} + +func env(k, d string) string { + if v := os.Getenv(k); v != "" { + return v + } + return d +} + +func loadConfig() Config { + return Config{ + Addr: env("HARNESS_ADDR", ":8080"), + MeasureDSN: env("MEASURE_DSN", "postgres://harness:harness@measure-db:5432/harness?sslmode=disable"), + MimirURL: env("MIMIR_URL", "http://10.20.0.10:9009/prometheus"), + ALBDNS: env("ALB_DNS", ""), + K8sAPI: env("K8S_API", "https://10.10.0.10:6443"), + K8sTokenFile: env("K8S_TOKEN_FILE", "/run/secrets/k8s-token"), + K8sCAFile: env("K8S_CA_FILE", "/run/secrets/k8s-ca.crt"), + AWSRoleARN: env("AWS_ROLE_ARN", ""), + AWSRegion: env("AWS_REGION", "ap-northeast-1"), + GCPProject: env("GCP_PROJECT", ""), + SSHUser: env("SSH_USER", "harness"), + SSHKeyFile: env("SSH_KEY_FILE", "/run/secrets/vm-ssh-key"), + DBPassFile: env("DB_PASSWORD_FILE", "/run/secrets/db-password"), + HumanTokFile: env("HUMAN_TOKEN_FILE", "/run/secrets/harness-token"), + AIDiagnoseURL: env("AI_DIAGNOSE_URL", "http://10.20.0.12:8080/diagnose"), + ScenarioDir: env("SCENARIO_DIR", "/etc/harness/scenarios"), + JudgeEvery: 30 * time.Second, + ObserveMax: 30 * time.Minute, + } +} + +var ( + mAlerts = prometheus.NewCounterVec(prometheus.CounterOpts{Name: "harness_alerts_received_total", Help: "Grafana webhook alerts by name/status"}, []string{"alertname", "status"}) + mActs = prometheus.NewCounterVec(prometheus.CounterOpts{Name: "harness_actions_total", Help: "Executed catalogue actions by id/actor/result"}, []string{"action_id", "actor", "result"}) + mRuns = prometheus.NewGauge(prometheus.GaugeOpts{Name: "harness_runs_active", Help: "1 while a run is executing"}) +) + +type App struct { + cfg Config + store *Store + scen map[string]*Scenario + runs *RunManager + hooks *AlertRouter +} + +func main() { + log.SetFlags(log.LstdFlags | log.Lmicroseconds) + cfg := loadConfig() + prometheus.MustRegister(mAlerts, mActs, mRuns) + ctx := context.Background() + store, err := NewStore(ctx, cfg.MeasureDSN) + if err != nil { + log.Fatalf("measure db: %v", err) + } + scen, err := LoadScenarios(cfg.ScenarioDir) + if err != nil { + log.Fatalf("scenarios: %v", err) + } + if err := store.UpsertScenarios(ctx, scen); err != nil { + log.Fatalf("scenarios upsert: %v", err) + } + app := &App{cfg: cfg, store: store, scen: scen} + app.runs = NewRunManager(app) + app.hooks = NewAlertRouter(app) + log.Printf("harness runner up: %d scenarios, alb=%s, k8s=%s", len(scen), cfg.ALBDNS, cfg.K8sAPI) + + mux := http.NewServeMux() + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok\n")) }) + mux.HandleFunc("GET /selfcheck", app.handleSelfcheck) + mux.HandleFunc("GET /scenarios", app.handleScenarios) + mux.HandleFunc("POST /runs", app.handleRunCreate) + mux.HandleFunc("GET /runs", app.handleRunList) + mux.HandleFunc("GET /runs/{id}", app.handleRunGet) + mux.HandleFunc("POST /runs/{id}/finish", app.handleRunFinish) + mux.HandleFunc("POST /runs/{id}/abort", app.handleRunAbort) + mux.HandleFunc("POST /hooks/alert", app.hooks.handleWebhook) + mux.HandleFunc("POST /proposals", app.handleProposal) + mux.HandleFunc("POST /actions/execute", app.handleActionExecute) + mux.HandleFunc("GET /actions/catalog", app.handleCatalog) + mux.Handle("GET /metrics", promhttp.Handler()) + srv := &http.Server{Addr: cfg.Addr, Handler: logging(mux), ReadHeaderTimeout: 10 * time.Second} + log.Fatal(srv.ListenAndServe()) +} + +func logging(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/healthz" && r.URL.Path != "/metrics" { + log.Printf("%s %s from %s", r.Method, r.URL.Path, r.RemoteAddr) + } + h.ServeHTTP(w, r) + }) +} + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + _ = enc.Encode(v) +} + +func writeErr(w http.ResponseWriter, code int, err error) { + writeJSON(w, code, map[string]string{"error": err.Error()}) +} + +func readFileTrim(path string) (string, error) { + b, err := os.ReadFile(path) + if err != nil { + return "", err + } + s := strings.TrimSpace(string(b)) + if s == "" { + return "", errors.New(path + ": empty") + } + return s, nil +} + +// ---- self-check: every external dependency of the runner (WP-11 acceptance) ---- +func (a *App) handleSelfcheck(w http.ResponseWriter, r *http.Request) { + ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second) + defer cancel() + out := map[string]any{} + if err := a.store.Ping(ctx); err != nil { + out["measure_db"] = "error: " + err.Error() + } else { + out["measure_db"] = "ok" + } + if n, err := a.k8sPodCount(ctx, "chaos-mesh"); err != nil { + out["k8s"] = "error: " + err.Error() + } else { + out["k8s"] = map[string]any{"chaos_mesh_pods": n} + } + if arn, err := a.awsCallerIdentity(ctx); err != nil { + out["aws"] = "error: " + err.Error() + } else { + out["aws"] = map[string]any{"caller_arn": arn} + } + if names, err := a.gcpListInstances(ctx, "legacy-"); err != nil { + out["gcp"] = "error: " + err.Error() + } else { + out["gcp"] = map[string]any{"legacy_instances": names} + } + if v, err := a.mimirScalar(ctx, `max(probe_success{job="blackbox",target="alb"})`); err != nil { + out["mimir"] = "error: " + err.Error() + } else { + out["mimir"] = map[string]any{"probe_success_alb": v} + } + if a.cfg.ALBDNS != "" { + c := &http.Client{Timeout: 10 * time.Second} + resp, err := c.Get("http://" + a.cfg.ALBDNS + "/api/orders") + if err != nil { + out["alb"] = "error: " + err.Error() + } else { + resp.Body.Close() + out["alb"] = map[string]any{"status": resp.StatusCode} + } + } + writeJSON(w, 200, out) +} + +func (a *App) handleScenarios(w http.ResponseWriter, r *http.Request) { + list := make([]*Scenario, 0, len(a.scen)) + for _, s := range a.scen { + list = append(list, s) + } + sortScenarios(list) + writeJSON(w, 200, list) +} diff --git a/harness/proposals.go b/harness/proposals.go new file mode 100644 index 0000000..b07d1a4 --- /dev/null +++ b/harness/proposals.go @@ -0,0 +1,65 @@ +package main + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "net/http" + "time" +) + +// POST /proposals: the AI VM registers a layer-2 proposal (design 6.5). Response carries the approval token that the +// Discord approval hands back; approved=true in the body records the approval directly (used by tests and by the AI VM +// after the button round-trip). Execution then goes through /actions/execute with that token. +type ProposalRequest struct { + RunID *string `json:"run_id"` + ScenarioID *string `json:"scenario_id"` + ActionID string `json:"action_id"` + Target map[string]any `json:"target"` + Diagnosis map[string]any `json:"diagnosis"` + Approved bool `json:"approved"` + ApprovedBy string `json:"approved_by"` + TTLSeconds int `json:"ttl_seconds"` +} + +func (a *App) handleProposal(w http.ResponseWriter, r *http.Request) { + var req ProposalRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeErr(w, 400, err) + return + } + if catalogEntry(req.ActionID) == nil { + writeErr(w, 400, errors.New("action_id not in the catalogue")) + return + } + if cur := a.runs.current(); cur != nil && req.RunID == nil { + req.RunID = &cur.RunID + if cur.ScenarioID != "" { + req.ScenarioID = &cur.ScenarioID + } + } + ttl := 15 * time.Minute // design 2.1: no approval within 15 minutes -> layer 3 + if req.TTLSeconds > 0 { + ttl = time.Duration(req.TTLSeconds) * time.Second + } + b := make([]byte, 24) + _, _ = rand.Read(b) + token := "apr-" + hex.EncodeToString(b) + if err := a.store.InsertProposal(r.Context(), token, req.RunID, req.ScenarioID, req.ActionID, req.Target, ttl); err != nil { + writeErr(w, 500, err) + return + } + a.store.Event(r.Context(), req.RunID, req.ScenarioID, "proposal", "ai", map[string]any{"action_id": req.ActionID, "target": req.Target, "diagnosis": req.Diagnosis, "token_suffix": token[len(token)-6:]}) + if req.Approved { + by := req.ApprovedBy + if by == "" { + by = "unspecified" + } + _ = a.store.ApproveProposal(r.Context(), token, by) + a.store.Event(r.Context(), req.RunID, req.ScenarioID, "approval", by, map[string]any{"action_id": req.ActionID, "token_suffix": token[len(token)-6:]}) + } + writeJSON(w, 201, map[string]any{"token": token, "expires_in_s": ttl.Seconds(), "approved": req.Approved}) +} + +// POST /proposals/{token}/approve is not needed while the AI VM marks approval itself; kept out to keep the surface small. diff --git a/harness/run.go b/harness/run.go new file mode 100644 index 0000000..e563c61 --- /dev/null +++ b/harness/run.go @@ -0,0 +1,405 @@ +package main + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "log" + "net/http" + "strings" + "sync" + "time" +) + +// RunManager executes one run at a time: for each scenario baseline -> inject -> observe (judge every 30s, +// alerts/diagnoses/actions arrive through the HTTP handlers) -> revert -> verdict -> cooldown (design 5.2, 8). +type RunManager struct { + app *App + mu sync.Mutex + active *activeRun +} + +type activeRun struct { + RunID string + Campaign string + ScenarioID string // "" between scenarios + Scenario *Scenario + TInject time.Time + cancel context.CancelFunc +} + +type RunRequest struct { + Campaign string `json:"campaign"` + ScenarioIDs []string `json:"scenario_ids"` + BaselineS *int `json:"baseline_s"` // default 300 (design 8) + CooldownS *int `json:"cooldown_s"` // default 300 + ObserveTimeoutS *int `json:"observe_timeout_s"` // default 1800 + MinObserveS *int `json:"min_observe_s"` // default 180: earliest t_recover (12 probe successes after inject) + DriftBefore *int `json:"drift_before"` + GitSHA string `json:"git_sha"` + Notes string `json:"notes"` + SkipRevert bool `json:"skip_revert"` +} + +func NewRunManager(a *App) *RunManager { return &RunManager{app: a} } + +func (m *RunManager) current() *activeRun { + m.mu.Lock() + defer m.mu.Unlock() + if m.active == nil { + return nil + } + c := *m.active + return &c +} + +func newID(prefix string) string { + b := make([]byte, 4) + _, _ = rand.Read(b) + return prefix + time.Now().UTC().Format("20060102-150405") + "-" + hex.EncodeToString(b) +} + +func (a *App) handleRunCreate(w http.ResponseWriter, r *http.Request) { + var req RunRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeErr(w, 400, err) + return + } + if req.Campaign == "" || len(req.ScenarioIDs) == 0 { + writeErr(w, 400, errors.New("campaign and scenario_ids are required")) + return + } + for _, id := range req.ScenarioIDs { + if _, ok := a.scen[id]; !ok { + writeErr(w, 400, fmt.Errorf("unknown scenario %q", id)) + return + } + } + m := a.runs + m.mu.Lock() + if m.active != nil { + id := m.active.RunID + m.mu.Unlock() + writeJSON(w, 409, map[string]string{"error": "a run is active", "run_id": id}) + return + } + runID := newID("run-") + ctx, cancel := context.WithCancel(context.Background()) + m.active = &activeRun{RunID: runID, Campaign: req.Campaign, cancel: cancel} + m.mu.Unlock() + if err := a.store.CreateRun(r.Context(), runID, req.Campaign, req.DriftBefore, req.GitSHA, req.Notes); err != nil { + m.mu.Lock() + m.active = nil + m.mu.Unlock() + cancel() + writeErr(w, 500, err) + return + } + mRuns.Set(1) + go m.execute(ctx, runID, req) + writeJSON(w, 202, map[string]any{"run_id": runID, "scenario_ids": req.ScenarioIDs}) +} + +func (a *App) handleRunAbort(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + m := a.runs + m.mu.Lock() + if m.active == nil || m.active.RunID != id { + m.mu.Unlock() + writeErr(w, 404, errors.New("run not active")) + return + } + m.active.cancel() + m.mu.Unlock() + writeJSON(w, 202, map[string]string{"run_id": id, "status": "aborting"}) +} + +func (a *App) handleRunFinish(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + var body struct { + DriftAfter *int `json:"drift_after"` + } + _ = json.NewDecoder(r.Body).Decode(&body) + if err := a.store.FinishRun(r.Context(), id, "finished", body.DriftAfter); err != nil { + writeErr(w, 500, err) + return + } + writeJSON(w, 200, map[string]any{"run_id": id, "drift_after": body.DriftAfter}) +} + +func (a *App) handleRunList(w http.ResponseWriter, r *http.Request) { + runs, err := a.store.Runs(r.Context(), "") + if err != nil { + writeErr(w, 500, err) + return + } + writeJSON(w, 200, map[string]any{"active": a.runs.current(), "runs": runs}) +} + +func (a *App) handleRunGet(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + runs, err := a.store.Runs(r.Context(), id) + if err != nil || len(runs) == 0 { + writeErr(w, 404, errors.New("run not found")) + return + } + inj, _ := a.store.Injections(r.Context(), id) + ev, _ := a.store.Events(r.Context(), id, "") + vd, _ := a.store.Verdicts(r.Context(), id) + writeJSON(w, 200, map[string]any{"run": runs[0], "injections": inj, "events": ev, "verdicts": vd}) +} + +func dur(p *int, def time.Duration) time.Duration { + if p == nil { + return def + } + return time.Duration(*p) * time.Second +} + +func (m *RunManager) execute(ctx context.Context, runID string, req RunRequest) { + a := m.app + status := "finished" + defer func() { + m.mu.Lock() + m.active = nil + m.mu.Unlock() + mRuns.Set(0) + _ = a.store.FinishRun(context.Background(), runID, status, nil) + log.Printf("run %s %s", runID, status) + }() + baseline, cooldown := dur(req.BaselineS, 5*time.Minute), dur(req.CooldownS, 5*time.Minute) + observeMax, minObserve := dur(req.ObserveTimeoutS, a.cfg.ObserveMax), dur(req.MinObserveS, 3*time.Minute) + for i, sid := range req.ScenarioIDs { + sc := a.scen[sid] + if err := m.runScenario(ctx, runID, sc, baseline, observeMax, minObserve, req.SkipRevert); err != nil { + if errors.Is(err, context.Canceled) { + status = "aborted" + return + } + log.Printf("run %s scenario %s: %v", runID, sid, err) + } + if i < len(req.ScenarioIDs)-1 && !sleepCtx(ctx, cooldown) { + status = "aborted" + return + } + } +} + +func sleepCtx(ctx context.Context, d time.Duration) bool { + if d <= 0 { + return true + } + select { + case <-time.After(d): + return true + case <-ctx.Done(): + return false + } +} + +func (m *RunManager) setScenario(sc *Scenario, tInject time.Time) { + m.mu.Lock() + defer m.mu.Unlock() + if m.active == nil { + return + } + if sc == nil { + m.active.ScenarioID, m.active.Scenario, m.active.TInject = "", nil, time.Time{} + return + } + m.active.ScenarioID, m.active.Scenario, m.active.TInject = sc.ID, sc, tInject +} + +func (m *RunManager) runScenario(ctx context.Context, runID string, sc *Scenario, baseline, observeMax, minObserve time.Duration, skipRevert bool) error { + a := m.app + sid := sc.ID + bg := context.Background() + a.store.Event(bg, &runID, &sid, "note", "runner", map[string]any{"phase": "baseline", "seconds": baseline.Seconds()}) + if !sleepCtx(ctx, baseline) { + return ctx.Err() + } + // inject + m.setScenario(sc, time.Now().UTC()) + defer m.setScenario(nil, time.Time{}) + injCtx, cancelInj := context.WithTimeout(ctx, 5*time.Minute) + params, err := a.runStep(injCtx, sc, sc.Inject, runID, nil) + cancelInj() + tInject := time.Now().UTC() + injID, dbErr := a.store.InsertInjection(bg, runID, sid, sc.Inject.Method, params) + if dbErr != nil { + log.Printf("injection insert: %v", dbErr) + } + if err != nil { + a.store.Event(bg, &runID, &sid, "note", "runner", map[string]any{"phase": "inject", "error": err.Error()}) + _ = a.store.InsertVerdict(bg, Verdict{RunID: runID, ScenarioID: sid, Recovered: false, Notes: "injection failed: " + err.Error()}) + return err + } + m.setScenario(sc, tInject) + a.store.Event(bg, &runID, &sid, "note", "runner", map[string]any{"phase": "inject", "t_inject": tInject, "params": params}) + // observe: judge every 30s until recovered (after minObserve) or timeout + var tRecover *time.Time + var last SLOState + deadline := tInject.Add(observeMax) + for { + if !sleepCtx(ctx, a.cfg.JudgeEvery) { + break + } + last = a.evaluateSLO(ctx) + if last.Recovered && time.Since(tInject) >= minObserve { + t := time.Now().UTC() + tRecover = &t + a.store.Event(bg, &runID, &sid, "recovered", "judge", last) + break + } + if time.Now().After(deadline) { + a.store.Event(bg, &runID, &sid, "timeout", "judge", last) + break + } + } + aborted := ctx.Err() != nil + // revert (always, even on abort, unless the scenario is irreversible by design or skip_revert) + if !skipRevert && sc.Revert.Method != "none" { + rvCtx, cancelRv := context.WithTimeout(bg, 10*time.Minute) + rres, rerr := a.runStep(rvCtx, sc, sc.Revert, runID, params) + cancelRv() + if rerr != nil { + a.store.Event(bg, &runID, &sid, "note", "runner", map[string]any{"phase": "revert", "error": rerr.Error(), "result": rres}) + } else { + a.store.Event(bg, &runID, &sid, "note", "runner", map[string]any{"phase": "revert", "result": rres}) + } + } + if injID != 0 { + _ = a.store.ClearInjection(bg, injID) + } + // verdict + v := a.buildVerdict(bg, runID, sc, tInject, tRecover, last) + if aborted { + v.Notes += " aborted;" + } + if err := a.store.InsertVerdict(bg, v); err != nil { + log.Printf("verdict insert: %v", err) + } + if aborted { + return ctx.Err() + } + return nil +} + +// buildVerdict derives design 5.4 metrics from the events of this scenario. +func (a *App) buildVerdict(ctx context.Context, runID string, sc *Scenario, tInject time.Time, tRecover *time.Time, last SLOState) Verdict { + v := Verdict{RunID: runID, ScenarioID: sc.ID, Recovered: tRecover != nil} + if tRecover != nil { + s := tRecover.Sub(tInject).Seconds() + v.MTTRs = &s + } + evs, _ := a.store.Events(ctx, runID, sc.ID) + var tAlert, tProposal, tApproval *time.Time + layer := 1 + misjudge := 0 + var diagScore *float64 + var actionMatch *bool + for _, e := range evs { + var p map[string]any + _ = json.Unmarshal(e.Payload, &p) + switch e.Kind { + case "alert": + if tAlert == nil && e.TS.After(tInject) && p["status"] == "firing" { + t := e.TS + tAlert = &t + } + case "diagnosis": + if layer < 2 { + layer = 2 + } + label, _ := p["root_cause_label"].(string) + s := scoreDiagnosis(label, sc.GroundTruthLabel) + diagScore = &s + if s == 0 { + misjudge++ + } + if pa, ok := p["proposed_action"].(string); ok { + b := pa == sc.GroundTruthAction + actionMatch = &b + if sc.GroundTruthAction == "escalate" && pa != "escalate" { + misjudge++ + } + } + case "proposal": + if tProposal == nil { + t := e.TS + tProposal = &t + } + case "approval": + if tApproval == nil { + t := e.TS + tApproval = &t + } + case "action": + switch e.Actor { + case "human": + layer = 3 + case "ai-approved": + if layer < 2 { + layer = 2 + } + } + case "misjudge": + misjudge++ + } + } + if tAlert != nil { + s := tAlert.Sub(tInject).Seconds() + v.MTTDs = &s + } + if tProposal != nil && tApproval != nil { + s := tApproval.Sub(*tProposal).Seconds() + v.MTTAs = &s + } + v.LayerActual = &layer + v.DiagnosisScore = diagScore + v.ActionMatch = actionMatch + v.MisjudgeCount = misjudge + b, _ := json.Marshal(last) + v.Notes = "last_slo=" + string(b) + return v +} + +// design 9: exact label 1.0, same top category (network/compute/db/config) 0.5, else 0 +func scoreDiagnosis(got, want string) float64 { + if got == "" || got == "unknown" { + return 0 + } + if got == want { + return 1 + } + if category(got) != "" && category(got) == category(want) { + return 0.5 + } + return 0 +} + +func category(label string) string { + switch { + case contains(label, "network", "latency", "partition", "loss"): + return "network" + case contains(label, "pod", "process", "instance", "memory", "cpu"): + return "compute" + case contains(label, "db", "replication", "schema", "primary"): + return "db" + case contains(label, "drift", "config", "gitops"): + return "config" + } + return "" +} + +func contains(s string, subs ...string) bool { + for _, x := range subs { + if strings.Contains(s, x) { + return true + } + } + return false +} diff --git a/harness/scenario.go b/harness/scenario.go new file mode 100644 index 0000000..b7534a2 --- /dev/null +++ b/harness/scenario.go @@ -0,0 +1,116 @@ +package main + +import ( + "fmt" + "os" + "path/filepath" + "sort" + "strings" + + "gopkg.in/yaml.v3" +) + +// Scenario is one row of experiment design 7 (12 files under vm/harness/scenarios in the gitops repo). +type Scenario struct { + ID string `yaml:"id" json:"scenario_id"` + Path string `yaml:"path" json:"path"` // known | gray | irreversible | misjudge + Asset string `yaml:"asset" json:"asset"` // container | vm + ExpectedLayer int `yaml:"expected_layer" json:"expected_layer"` + GroundTruthLabel string `yaml:"ground_truth_label" json:"ground_truth_label"` + GroundTruthAction string `yaml:"ground_truth_action" json:"ground_truth_action"` // action_id | escalate | none + Description string `yaml:"description" json:"description"` + Inject Step `yaml:"inject" json:"inject"` + Revert Step `yaml:"revert" json:"revert"` + EscalateAfterS int `yaml:"escalate_after_s" json:"escalate_after_s"` // signature alerts: hand to layer 2 after this (design 4.3: 5min / 10min) + Notes map[string]any `yaml:"notes,omitempty" json:"notes,omitempty"` +} + +// Step is an injection or revert instruction. Exactly one method applies; the others stay empty. +type Step struct { + Method string `yaml:"method" json:"method"` // none | chaos | chaos_delete | kubectl_scale | ssh | gcloud | psql + // chaos: manifest of a Chaos Mesh CR (namespace/name are set by the runner) + Manifest string `yaml:"manifest,omitempty" json:"manifest,omitempty"` + // kubectl_scale + Namespace string `yaml:"namespace,omitempty" json:"namespace,omitempty"` + Deployment string `yaml:"deployment,omitempty" json:"deployment,omitempty"` + Replicas int32 `yaml:"replicas,omitempty" json:"replicas,omitempty"` + // ssh: host is an IP, a pet VM name (db-primary, db-replica) or legacy:N (N-th MIG instance, name order) + Host string `yaml:"host,omitempty" json:"host,omitempty"` + Command string `yaml:"command,omitempty" json:"command,omitempty"` + // gcloud: op = instance_delete | instance_stop | instance_start | instance_reset | mig_recreate | mig_resize + Op string `yaml:"op,omitempty" json:"op,omitempty"` + Instance string `yaml:"instance,omitempty" json:"instance,omitempty"` // pet VM name or legacy:N + MIG string `yaml:"mig,omitempty" json:"mig,omitempty"` + Size int64 `yaml:"size,omitempty" json:"size,omitempty"` + // psql: harness role over the network (hostssl, scram) + Database string `yaml:"database,omitempty" json:"database,omitempty"` + SQL string `yaml:"sql,omitempty" json:"sql,omitempty"` + // wait after the step (seconds) before continuing, e.g. tc rules that need time to bite + WaitS int `yaml:"wait_s,omitempty" json:"wait_s,omitempty"` +} + +func LoadScenarios(dir string) (map[string]*Scenario, error) { + files, err := filepath.Glob(filepath.Join(dir, "*.yaml")) + if err != nil { + return nil, err + } + if len(files) == 0 { + return nil, fmt.Errorf("no scenario files in %s", dir) + } + out := map[string]*Scenario{} + for _, f := range files { + b, err := os.ReadFile(f) + if err != nil { + return nil, err + } + var s Scenario + if err := yaml.Unmarshal(b, &s); err != nil { + return nil, fmt.Errorf("%s: %w", f, err) + } + if s.ID == "" { + return nil, fmt.Errorf("%s: missing id", f) + } + if _, dup := out[s.ID]; dup { + return nil, fmt.Errorf("%s: duplicate id %s", f, s.ID) + } + if s.Inject.Method == "" { + s.Inject.Method = "none" + } + if s.Revert.Method == "" { + s.Revert.Method = "none" + } + out[s.ID] = &s + } + return out, nil +} + +func sortScenarios(list []*Scenario) { + sort.Slice(list, func(i, j int) bool { return list[i].ID < list[j].ID }) +} + +// InjectionSpec is what goes into scenarios.injection_spec (jsonb): the step without its manifest body. +func (s *Scenario) InjectionSpec() map[string]any { + m := map[string]any{"method": s.Inject.Method} + switch s.Inject.Method { + case "chaos": + m["kind"] = manifestKind(s.Inject.Manifest) + case "kubectl_scale": + m["namespace"], m["deployment"], m["replicas"] = s.Inject.Namespace, s.Inject.Deployment, s.Inject.Replicas + case "ssh": + m["host"], m["command"] = s.Inject.Host, s.Inject.Command + case "gcloud": + m["op"], m["instance"], m["mig"], m["size"] = s.Inject.Op, s.Inject.Instance, s.Inject.MIG, s.Inject.Size + case "psql": + m["host"], m["database"], m["sql"] = s.Inject.Host, s.Inject.Database, s.Inject.SQL + } + return m +} + +func manifestKind(manifest string) string { + for _, l := range strings.Split(manifest, "\n") { + if strings.HasPrefix(strings.TrimSpace(l), "kind:") { + return strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(l), "kind:")) + } + } + return "" +} diff --git a/harness/sshx.go b/harness/sshx.go new file mode 100644 index 0000000..c578a72 --- /dev/null +++ b/harness/sshx.go @@ -0,0 +1,69 @@ +package main + +import ( + "bytes" + "context" + "fmt" + "net" + "os" + "time" + + "golang.org/x/crypto/ssh" +) + +// sshRun executes one command on a lab VM as the harness user (key from Secret Manager, design 6.11). +// Host resolution: IP as-is, "legacy:N" or pet VM short name via the compute API. +func (a *App) sshRun(ctx context.Context, host, command string) (string, error) { + ip := host + if net.ParseIP(host) == nil { + in, err := a.resolveInstance(ctx, host) + if err != nil { + return "", err + } + ip = in.IP + } + key, err := os.ReadFile(a.cfg.SSHKeyFile) + if err != nil { + return "", fmt.Errorf("ssh key: %w", err) + } + signer, err := ssh.ParsePrivateKey(key) + if err != nil { + return "", fmt.Errorf("ssh key parse: %w", err) + } + cfg := &ssh.ClientConfig{ + User: a.cfg.SSHUser, + Auth: []ssh.AuthMethod{ssh.PublicKeys(signer)}, + HostKeyCallback: ssh.InsecureIgnoreHostKey(), // lab VMs are recreated by autohealing; host keys are not pinned + Timeout: 15 * time.Second, + } + d := net.Dialer{Timeout: 15 * time.Second} + conn, err := d.DialContext(ctx, "tcp", ip+":22") + if err != nil { + return "", fmt.Errorf("dial %s: %w", ip, err) + } + c, chans, reqs, err := ssh.NewClientConn(conn, ip+":22", cfg) + if err != nil { + conn.Close() + return "", fmt.Errorf("ssh %s: %w", ip, err) + } + client := ssh.NewClient(c, chans, reqs) + defer client.Close() + sess, err := client.NewSession() + if err != nil { + return "", err + } + defer sess.Close() + var out bytes.Buffer + sess.Stdout, sess.Stderr = &out, &out + done := make(chan error, 1) + go func() { done <- sess.Run(command) }() + select { + case err := <-done: + if err != nil { + return out.String(), fmt.Errorf("%s: %w: %s", command, err, out.String()) + } + return out.String(), nil + case <-ctx.Done(): + return out.String(), ctx.Err() + } +} diff --git a/harness/store.go b/harness/store.go new file mode 100644 index 0000000..e68e669 --- /dev/null +++ b/harness/store.go @@ -0,0 +1,314 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// Store wraps the measure DB (experiment design 5.3). Schema is created at start-up; idempotent. +type Store struct{ pool *pgxpool.Pool } + +const measureSchema = ` +CREATE TABLE IF NOT EXISTS runs ( + run_id text PRIMARY KEY, + campaign text NOT NULL, + started_at timestamptz NOT NULL DEFAULT now(), + finished_at timestamptz, + drift_before int, + drift_after int, + git_sha text, + status text NOT NULL DEFAULT 'running', + notes text +); +CREATE TABLE IF NOT EXISTS scenarios ( + scenario_id text PRIMARY KEY, + path text NOT NULL, + asset text NOT NULL, + expected_layer int NOT NULL, + ground_truth_label text NOT NULL, + ground_truth_action text NOT NULL, + injection_spec jsonb NOT NULL +); +CREATE TABLE IF NOT EXISTS injections ( + id bigserial PRIMARY KEY, + run_id text NOT NULL REFERENCES runs(run_id), + scenario_id text NOT NULL, + t_inject timestamptz, + t_clear timestamptz, + method text NOT NULL, + params jsonb NOT NULL DEFAULT '{}'::jsonb +); +CREATE TABLE IF NOT EXISTS events ( + id bigserial PRIMARY KEY, + run_id text, + scenario_id text, + ts timestamptz NOT NULL DEFAULT now(), + kind text NOT NULL, -- alert | diagnosis | proposal | approval | action | recovered | timeout | misjudge | note + actor text NOT NULL, + payload jsonb NOT NULL DEFAULT '{}'::jsonb +); +CREATE INDEX IF NOT EXISTS events_run_idx ON events (run_id, scenario_id, ts); +CREATE TABLE IF NOT EXISTS verdicts ( + id bigserial PRIMARY KEY, + run_id text NOT NULL REFERENCES runs(run_id), + scenario_id text NOT NULL, + layer_actual int, + recovered boolean NOT NULL, + mttd_s double precision, + mtta_s double precision, + mttr_s double precision, + diagnosis_score double precision, + action_match boolean, + misjudge_count int NOT NULL DEFAULT 0, + notes text, + created_at timestamptz NOT NULL DEFAULT now() +); +CREATE TABLE IF NOT EXISTS proposals ( + token text PRIMARY KEY, + run_id text, + scenario_id text, + action_id text NOT NULL, + target jsonb NOT NULL DEFAULT '{}'::jsonb, + created_at timestamptz NOT NULL DEFAULT now(), + approved_at timestamptz, + approved_by text, + executed_at timestamptz, + expires_at timestamptz NOT NULL +);` + +func NewStore(ctx context.Context, dsn string) (*Store, error) { + var pool *pgxpool.Pool + var err error + for i := 0; i < 30; i++ { // measure-db container may still be starting + pool, err = pgxpool.New(ctx, dsn) + if err == nil { + if err = pool.Ping(ctx); err == nil { + break + } + pool.Close() + } + time.Sleep(2 * time.Second) + } + if err != nil { + return nil, err + } + if _, err := pool.Exec(ctx, measureSchema); err != nil { + return nil, fmt.Errorf("schema: %w", err) + } + return &Store{pool: pool}, nil +} + +func (s *Store) Ping(ctx context.Context) error { return s.pool.Ping(ctx) } + +func (s *Store) UpsertScenarios(ctx context.Context, m map[string]*Scenario) error { + for _, sc := range m { + spec, _ := json.Marshal(sc.InjectionSpec()) + _, err := s.pool.Exec(ctx, `INSERT INTO scenarios (scenario_id,path,asset,expected_layer,ground_truth_label,ground_truth_action,injection_spec) + VALUES ($1,$2,$3,$4,$5,$6,$7) ON CONFLICT (scenario_id) DO UPDATE SET path=EXCLUDED.path, asset=EXCLUDED.asset, + expected_layer=EXCLUDED.expected_layer, ground_truth_label=EXCLUDED.ground_truth_label, + ground_truth_action=EXCLUDED.ground_truth_action, injection_spec=EXCLUDED.injection_spec`, + sc.ID, sc.Path, sc.Asset, sc.ExpectedLayer, sc.GroundTruthLabel, sc.GroundTruthAction, spec) + if err != nil { + return err + } + } + return nil +} + +func (s *Store) CreateRun(ctx context.Context, id, campaign string, driftBefore *int, gitSHA, notes string) error { + _, err := s.pool.Exec(ctx, `INSERT INTO runs (run_id,campaign,drift_before,git_sha,notes) VALUES ($1,$2,$3,$4,$5)`, id, campaign, driftBefore, nz(gitSHA), nz(notes)) + return err +} + +func (s *Store) FinishRun(ctx context.Context, id, status string, driftAfter *int) error { + _, err := s.pool.Exec(ctx, `UPDATE runs SET finished_at=now(), status=$2, drift_after=COALESCE($3,drift_after) WHERE run_id=$1`, id, status, driftAfter) + return err +} + +func (s *Store) InsertInjection(ctx context.Context, runID, scenarioID, method string, params map[string]any) (int64, error) { + p, _ := json.Marshal(params) + var id int64 + err := s.pool.QueryRow(ctx, `INSERT INTO injections (run_id,scenario_id,t_inject,method,params) VALUES ($1,$2,now(),$3,$4) RETURNING id`, runID, scenarioID, method, p).Scan(&id) + return id, err +} + +func (s *Store) ClearInjection(ctx context.Context, id int64) error { + _, err := s.pool.Exec(ctx, `UPDATE injections SET t_clear=now() WHERE id=$1`, id) + return err +} + +func (s *Store) Event(ctx context.Context, runID, scenarioID *string, kind, actor string, payload any) { + p, _ := json.Marshal(payload) + if _, err := s.pool.Exec(ctx, `INSERT INTO events (run_id,scenario_id,kind,actor,payload) VALUES ($1,$2,$3,$4,$5)`, runID, scenarioID, kind, actor, p); err != nil { + fmt.Println("event insert failed:", err) + } +} + +type EventRow struct { + ID int64 `json:"id"` + RunID *string `json:"run_id"` + ScenarioID *string `json:"scenario_id"` + TS time.Time `json:"ts"` + Kind string `json:"kind"` + Actor string `json:"actor"` + Payload json.RawMessage `json:"payload"` +} + +func (s *Store) Events(ctx context.Context, runID, scenarioID string) ([]EventRow, error) { + rows, err := s.pool.Query(ctx, `SELECT id,run_id,scenario_id,ts,kind,actor,payload FROM events WHERE run_id=$1 AND ($2='' OR scenario_id=$2) ORDER BY ts, id`, runID, scenarioID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []EventRow + for rows.Next() { + var e EventRow + if err := rows.Scan(&e.ID, &e.RunID, &e.ScenarioID, &e.TS, &e.Kind, &e.Actor, &e.Payload); err != nil { + return nil, err + } + out = append(out, e) + } + return out, rows.Err() +} + +type Verdict struct { + RunID string `json:"run_id"` + ScenarioID string `json:"scenario_id"` + LayerActual *int `json:"layer_actual"` + Recovered bool `json:"recovered"` + MTTDs *float64 `json:"mttd_s"` + MTTAs *float64 `json:"mtta_s"` + MTTRs *float64 `json:"mttr_s"` + DiagnosisScore *float64 `json:"diagnosis_score"` + ActionMatch *bool `json:"action_match"` + MisjudgeCount int `json:"misjudge_count"` + Notes string `json:"notes"` +} + +func (s *Store) InsertVerdict(ctx context.Context, v Verdict) error { + _, err := s.pool.Exec(ctx, `INSERT INTO verdicts (run_id,scenario_id,layer_actual,recovered,mttd_s,mtta_s,mttr_s,diagnosis_score,action_match,misjudge_count,notes) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`, v.RunID, v.ScenarioID, v.LayerActual, v.Recovered, v.MTTDs, v.MTTAs, v.MTTRs, v.DiagnosisScore, v.ActionMatch, v.MisjudgeCount, nz(v.Notes)) + return err +} + +func (s *Store) Verdicts(ctx context.Context, runID string) ([]Verdict, error) { + rows, err := s.pool.Query(ctx, `SELECT run_id,scenario_id,layer_actual,recovered,mttd_s,mtta_s,mttr_s,diagnosis_score,action_match,misjudge_count,COALESCE(notes,'') FROM verdicts WHERE run_id=$1 ORDER BY id`, runID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Verdict + for rows.Next() { + var v Verdict + if err := rows.Scan(&v.RunID, &v.ScenarioID, &v.LayerActual, &v.Recovered, &v.MTTDs, &v.MTTAs, &v.MTTRs, &v.DiagnosisScore, &v.ActionMatch, &v.MisjudgeCount, &v.Notes); err != nil { + return nil, err + } + out = append(out, v) + } + return out, rows.Err() +} + +type RunRow struct { + RunID string `json:"run_id"` + Campaign string `json:"campaign"` + StartedAt time.Time `json:"started_at"` + FinishedAt *time.Time `json:"finished_at"` + DriftBefore *int `json:"drift_before"` + DriftAfter *int `json:"drift_after"` + GitSHA *string `json:"git_sha"` + Status string `json:"status"` + Notes *string `json:"notes"` +} + +func (s *Store) Runs(ctx context.Context, runID string) ([]RunRow, error) { + rows, err := s.pool.Query(ctx, `SELECT run_id,campaign,started_at,finished_at,drift_before,drift_after,git_sha,status,notes FROM runs WHERE ($1='' OR run_id=$1) ORDER BY started_at DESC LIMIT 100`, runID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []RunRow + for rows.Next() { + var r RunRow + if err := rows.Scan(&r.RunID, &r.Campaign, &r.StartedAt, &r.FinishedAt, &r.DriftBefore, &r.DriftAfter, &r.GitSHA, &r.Status, &r.Notes); err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +type InjectionRow struct { + ID int64 `json:"id"` + ScenarioID string `json:"scenario_id"` + TInject *time.Time `json:"t_inject"` + TClear *time.Time `json:"t_clear"` + Method string `json:"method"` + Params json.RawMessage `json:"params"` +} + +func (s *Store) Injections(ctx context.Context, runID string) ([]InjectionRow, error) { + rows, err := s.pool.Query(ctx, `SELECT id,scenario_id,t_inject,t_clear,method,params FROM injections WHERE run_id=$1 ORDER BY id`, runID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []InjectionRow + for rows.Next() { + var r InjectionRow + if err := rows.Scan(&r.ID, &r.ScenarioID, &r.TInject, &r.TClear, &r.Method, &r.Params); err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +// proposals (layer 2 approval tokens, design 6.5; the AI VM registers them, Discord approval marks them approved) +func (s *Store) InsertProposal(ctx context.Context, token string, runID, scenarioID *string, actionID string, target map[string]any, ttl time.Duration) error { + t, _ := json.Marshal(target) + _, err := s.pool.Exec(ctx, `INSERT INTO proposals (token,run_id,scenario_id,action_id,target,expires_at) VALUES ($1,$2,$3,$4,$5,now()+$6::interval)`, + token, runID, scenarioID, actionID, t, fmt.Sprintf("%d seconds", int(ttl.Seconds()))) + return err +} + +type ProposalRow struct { + Token string + RunID *string + ScenarioID *string + ActionID string + ApprovedAt *time.Time + ApprovedBy *string + ExecutedAt *time.Time + ExpiresAt time.Time +} + +func (s *Store) GetProposal(ctx context.Context, token string) (*ProposalRow, error) { + var p ProposalRow + err := s.pool.QueryRow(ctx, `SELECT token,run_id,scenario_id,action_id,approved_at,approved_by,executed_at,expires_at FROM proposals WHERE token=$1`, token). + Scan(&p.Token, &p.RunID, &p.ScenarioID, &p.ActionID, &p.ApprovedAt, &p.ApprovedBy, &p.ExecutedAt, &p.ExpiresAt) + if err != nil { + return nil, err + } + return &p, nil +} + +func (s *Store) ApproveProposal(ctx context.Context, token, by string) error { + _, err := s.pool.Exec(ctx, `UPDATE proposals SET approved_at=now(), approved_by=$2 WHERE token=$1 AND approved_at IS NULL`, token, by) + return err +} + +func (s *Store) MarkProposalExecuted(ctx context.Context, token string) error { + _, err := s.pool.Exec(ctx, `UPDATE proposals SET executed_at=now() WHERE token=$1`, token) + return err +} + +func nz(s string) *string { + if s == "" { + return nil + } + return &s +}