selfheal-lab-apps/harness/main.go

// 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"
	"strconv"
	"strings"
	"sync"
	"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
	AlertDedupeS  int // layer-2 alerts with the same name inside this window share one diagnosis (0 = off)
	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"),
		AlertDedupeS:  envInt("ALERT_DEDUPE_S", 300),
		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
	busyLegacy sync.Map // run/scenario -> gceInstance chosen for legacy:busy (gcp.go)
}

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("GET /proposals/{token}", app.handleProposalGet)
	mux.HandleFunc("POST /proposals/{token}/approve", app.handleProposalApprove)
	mux.HandleFunc("POST /proposals/{token}/reject", app.handleProposalReject)
	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)
}

func envInt(k string, d int) int {
	if v := os.Getenv(k); v != "" {
		if n, err := strconv.Atoi(v); err == nil {
			return n
		}
	}
	return d
}