selfheal-lab-apps/harness/gcp.go

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 ref == "legacy:busy" {
		return a.resolveBusyLegacy(ctx)
	}
	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:<name>" 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") }

// resolveBusyLegacy picks the MIG instance that served the most legacy-api requests over the last 2 minutes. The ILB
// hashes the api pods' few keep-alive connections onto one instance (campaign 2, 2026-10-08: legacy:0 served 0 rps while
// the delay was on it), so a fault must go where the traffic is. The choice is cached per run+scenario so the revert
// hits the same instance after the traffic has moved away from it.
func (a *App) resolveBusyLegacy(ctx context.Context) (gceInstance, error) {
	key := ""
	if cur := a.runs.current(); cur != nil {
		key = cur.RunID + "/" + cur.ScenarioID
	}
	if key != "" {
		if v, ok := a.busyLegacy.Load(key); ok {
			return v.(gceInstance), nil
		}
	}
	name, err := a.mimirTopInstance(ctx, `topk(1, sum by (instance) (rate(legacy_http_requests_total{route!="healthz"}[2m])))`)
	if err != nil {
		return gceInstance{}, fmt.Errorf("legacy:busy: %w", err)
	}
	l, err := a.gcpInstances(ctx, "legacy-")
	if err != nil {
		return gceInstance{}, err
	}
	for _, i := range l {
		if i.Name == name {
			if key != "" {
				a.busyLegacy.Store(key, i)
			}
			return i, nil
		}
	}
	return gceInstance{}, fmt.Errorf("legacy:busy: instance %q from Mimir not in the MIG (%d instances)", name, len(l))
}