From 32d51589c428624cc7ef3cbe31847e7b22baf04a Mon Sep 17 00:00:00 2001 From: rldyourmnd Date: Sun, 30 Aug 2026 01:01:18 +0500 Subject: [PATCH] feat(fleet): make draining a member an operation, not a remembered sequence Taking a member out of service worked, but only if you knew three things in the right order, and getting it wrong looked like it had worked. The member's gate is owned by the pressure publisher, which reasserts it every eleven seconds. Closing the gate -- or writing scheduler.instance by hand -- without first stopping gha-pressure-gate.timer is undone within one cycle: observed live, set at 17:02:27 and read back open at 17:02:52. `gha-fleet drain-member` does the sequence: stop the timer, close the gate through its owner, then wait for the jobs already running to finish. It never stops a worker. If they outlast the deadline it reports that the member is still occupied and names what is on it, and exits non-zero so a caller scripting a reboot sees it -- ending someone's build is not this command's decision. `--restore` hands the member back, republishing from live pressure rather than forcing the gate open, so a member genuinely under pressure stays closed on its own merits. Without --apply it reports what it would do and what the member is carrying. The restore result says gate_republished and reports the scheduler value the publisher actually wrote, which is "manual": hysteresis holds a just-closed member shut for a cycle or two. Claiming the gate was open there would have been the more convenient field and the false one. Proven twice on gha-runner-3: gate all -> manual with the timer stopped, then republished and open again at t+45s and t+60s under its owner. Closes #262. Claude-Session: https://claude.ai/code/session_01NpzpgiRaxi5mHVTMoRYndt --- cmd/gha-fleet/main.go | 168 +++++++++++++++- internal/memberdrain/drain.go | 296 +++++++++++++++++++++++++++++ internal/memberdrain/drain_test.go | 292 ++++++++++++++++++++++++++++ 3 files changed, 755 insertions(+), 1 deletion(-) create mode 100644 internal/memberdrain/drain.go create mode 100644 internal/memberdrain/drain_test.go diff --git a/cmd/gha-fleet/main.go b/cmd/gha-fleet/main.go index 86f2d3f4..3c298a41 100644 --- a/cmd/gha-fleet/main.go +++ b/cmd/gha-fleet/main.go @@ -38,6 +38,7 @@ import ( "github.com/NDDev-OpenNetwork/github-actions/internal/imageplan" "github.com/NDDev-OpenNetwork/github-actions/internal/incusplan" "github.com/NDDev-OpenNetwork/github-actions/internal/incusreconcile" + "github.com/NDDev-OpenNetwork/github-actions/internal/memberdrain" "github.com/NDDev-OpenNetwork/github-actions/internal/observabilitydashboards" "github.com/NDDev-OpenNetwork/github-actions/internal/observabilityrules" "github.com/NDDev-OpenNetwork/github-actions/internal/pressuregate" @@ -79,6 +80,8 @@ func run(args []string, stdout, stderr io.Writer) int { return runPreflight(args[1:], stdout, stderr) case "publish-pressure": return runPublishPressure(args[1:], stdout, stderr) + case "drain-member": + return runDrainMember(args[1:], stdout, stderr) case "reconcile-incus": return runReconcileIncus(args[1:], stdout, stderr) case "reconcile-image": @@ -1514,6 +1517,169 @@ func runPublishPressure(args []string, stdout, stderr io.Writer) int { return writeJSONOrFail(stdout, stderr, result) } +// systemdUnits is the drain's control over the timer that owns the member's +// gate. The pressure publisher reasserts the gate every eleven seconds, so this +// is what makes a close hold rather than last one cycle. +type systemdUnits struct{} + +func (systemdUnits) Stop(ctx context.Context, unit string) error { + if output, err := exec.CommandContext(ctx, "systemctl", "stop", unit).CombinedOutput(); err != nil { + return fmt.Errorf("systemctl stop %s: %v: %s", unit, err, strings.TrimSpace(string(output))) + } + return nil +} + +func (systemdUnits) Start(ctx context.Context, unit string) error { + if output, err := exec.CommandContext(ctx, "systemctl", "start", unit).CombinedOutput(); err != nil { + return fmt.Errorf("systemctl start %s: %v: %s", unit, err, strings.TrimSpace(string(output))) + } + return nil +} + +func (systemdUnits) IsActive(ctx context.Context, unit string) (bool, error) { + // `is-active` exits non-zero for every inactive state, so the exit code + // alone cannot tell "the unit is stopped" from "systemctl is missing". The + // word it prints can. + output, err := exec.CommandContext(ctx, "systemctl", "is-active", unit).CombinedOutput() + state := strings.TrimSpace(string(output)) + switch state { + case "active", "activating", "reloading": + return true, nil + case "inactive", "deactivating", "failed", "unknown": + return false, nil + } + return false, fmt.Errorf("systemctl is-active %s: %v: %s", unit, err, state) +} + +// pressureGate closes and reopens the member's gate through the publisher that +// owns it. Writing scheduler.instance directly is overwritten within one cycle. +type pressureGate struct { + client pressurepublish.Client + memberName string + statePath string + policy pressuregate.Policy +} + +func (g pressureGate) publish(ctx context.Context, sample pressuregate.Sample, reason string) (string, error) { + now := time.Now().UTC() + sample.ObservedAt = now + result, err := pressurepublish.Reconcile(ctx, g.client, pressurepublish.Options{ + MemberName: g.memberName, StatePath: g.statePath, Policy: g.policy, + Sample: sample, Apply: true, Now: now, ForceCloseReason: reason, + }) + if err != nil { + return "", err + } + return result.Scheduler, nil +} + +func (g pressureGate) ForceClose(ctx context.Context, reason string) (string, error) { + return g.publish(ctx, pressuregate.Sample{}, reason) +} + +// Reopen does not force the gate open. It publishes from live pressure, so a +// member that is genuinely under pressure stays closed on its own merits. +func (g pressureGate) Reopen(ctx context.Context) (string, error) { + host, err := hostprobe.Collect(ctx) + if err != nil { + return "", fmt.Errorf("collect pressure: %w", err) + } + if !host.Pressure.Available { + return "", fmt.Errorf("Linux PSI is unavailable") + } + return g.publish(ctx, pressuregate.Sample{ + CPUSomeAvg10: host.Pressure.CPU.Some.Avg10, + MemoryFullAvg10: host.Pressure.Memory.Full.Avg10, + IOFullAvg10: host.Pressure.IO.Full.Avg10, + OOMKillsTotal: host.Memory.OOMKillsTotal, + }, "") +} + +func runDrainMember(args []string, stdout, stderr io.Writer) int { + flags := flag.NewFlagSet("drain-member", flag.ContinueOnError) + flags.SetOutput(stderr) + configPath := flags.String("config", "", "exact platform configuration path") + statePath := flags.String("state-path", "/var/lib/gha-fleet/pressure-gate.json", "private pressure gate state") + incusSocket := flags.String("incus-socket", "/var/lib/incus/unix.socket", "local Incus unix socket") + timerUnit := flags.String("timer-unit", memberdrain.DefaultTimerUnit, "the timer that owns this member's gate") + reason := flags.String("reason", "", "why the member is being taken out of service, published as the gate's close reason") + restore := flags.Bool("restore", false, "hand the member back: republish from live pressure and start the timer") + timeout := flags.Duration("timeout", memberdrain.DefaultTimeout, "how long to wait for running jobs to finish") + poll := flags.Duration("poll", memberdrain.DefaultPoll, "how often to re-read what the member is carrying") + apply := flags.Bool("apply", false, "stop the timer and publish, rather than reporting what would happen") + if err := flags.Parse(args); err != nil { + return 2 + } + if flags.NArg() != 0 || *configPath == "" { + fmt.Fprintln(stderr, "gha-fleet: drain-member requires --config and no positional arguments") + return 2 + } + if *restore && *reason != "" { + fmt.Fprintln(stderr, "gha-fleet: drain-member --restore takes no --reason") + return 2 + } + cfg, err := config.Load(*configPath) + if err != nil { + fmt.Fprintf(stderr, "gha-fleet: %v\n", err) + return 1 + } + if !cfg.Incus.Cluster.Enabled || cfg.Incus.Cluster.MemberName == "" { + fmt.Fprintln(stderr, "gha-fleet: drain-member requires an Incus cluster member config") + return 1 + } + if !cfg.Pressure.Required { + fmt.Fprintln(stderr, "gha-fleet: drain-member requires an enabled pressure_admission policy") + return 1 + } + // The gate is per-member and published by the member itself, so draining + // one host from another would write somebody else's state. + hostname, err := os.Hostname() + if err != nil || hostname != cfg.Platform.Host || hostname != cfg.Incus.Cluster.MemberName { + fmt.Fprintf(stderr, "gha-fleet: drain host %q differs from platform/member %q/%q\n", hostname, cfg.Platform.Host, cfg.Incus.Cluster.MemberName) + return 1 + } + ctx, cancel := context.WithTimeout(context.Background(), *timeout+2*time.Minute) + defer cancel() + client, err := incusclient.ConnectIncusUnixWithContext(ctx, *incusSocket, &incusclient.ConnectionArgs{}) + if err != nil { + fmt.Fprintf(stderr, "gha-fleet: connect Incus: %v\n", err) + return 1 + } + // Cluster members are not project-scoped; workers are. The gate is read and + // written on the unscoped connection, the occupancy on the worker project. + deps := memberdrain.Deps{ + Client: client.UseProject(cfg.Incus.Project), + Units: systemdUnits{}, + Gate: pressureGate{ + client: client, memberName: cfg.Incus.Cluster.MemberName, + statePath: *statePath, policy: cfg.Pressure, + }, + } + options := memberdrain.Options{ + MemberName: cfg.Incus.Cluster.MemberName, Reason: *reason, + TimerUnit: *timerUnit, Timeout: *timeout, Poll: *poll, Apply: *apply, + } + var result memberdrain.Result + if *restore { + result, err = memberdrain.Restore(ctx, deps, options) + } else { + result, err = memberdrain.Drain(ctx, deps, options) + } + if err != nil { + fmt.Fprintf(stderr, "gha-fleet: drain-member: %v\n", err) + return 1 + } + if writeJSONOrFail(stdout, stderr, result) != 0 { + return 1 + } + // A drain that ran out of time left somebody's job running and the member + // closed. That is not a success, and a caller scripting a reboot must see it. + if result.TimedOut { + return 1 + } + return 0 +} + func runValidate(args []string, stdout, stderr io.Writer) int { flags := flag.NewFlagSet("validate", flag.ContinueOnError) flags.SetOutput(stderr) @@ -1879,5 +2045,5 @@ func runCapacity(args []string, stdout, stderr io.Writer) int { } func printUsage(writer io.Writer) { - fmt.Fprintln(writer, "usage: gha-fleet [options]") + fmt.Fprintln(writer, "usage: gha-fleet [options]") } diff --git a/internal/memberdrain/drain.go b/internal/memberdrain/drain.go new file mode 100644 index 00000000..7509252b --- /dev/null +++ b/internal/memberdrain/drain.go @@ -0,0 +1,296 @@ +// Package memberdrain turns taking an Incus cluster member out of service from +// a sequence somebody remembers into a single operation. +// +// The sequence is not obvious and getting it wrong looks like it worked. The +// member's admission gate is owned by the pressure publisher, which reasserts +// it every eleven seconds, so closing the gate without first stopping +// gha-pressure-gate.timer is undone within one cycle -- observed live: set at +// 17:02:27, read back open at 17:02:52. Setting scheduler.instance by hand has +// the same fate for the same reason. +// +// A drain never stops a running worker. It closes the gate so no new work is +// placed, then waits for the jobs already there to finish on their own. If they +// outlast the deadline the drain reports that it is still occupied and by what; +// it does not decide to end someone's build. +package memberdrain + +import ( + "context" + "fmt" + "sort" + "time" + + "github.com/lxc/incus/v7/shared/api" +) + +// Client lists the instances the cluster is carrying. Scoped to the worker +// project by the caller, so what comes back is workers rather than the whole +// cluster's containers. +type Client interface { + GetInstances(api.InstanceType) ([]api.Instance, error) +} + +// Units is the systemd control a drain needs. Stopping the pressure timer is +// what makes the closed gate hold; starting it again is what hands the member +// back to its owner. +type Units interface { + Stop(ctx context.Context, unit string) error + Start(ctx context.Context, unit string) error + IsActive(ctx context.Context, unit string) (bool, error) +} + +// Gate closes and reopens the member's admission gate through the component +// that owns it, rather than around it. +type Gate interface { + // Each returns the scheduler value the member now carries, so the drain + // reports what is true rather than what it asked for. + ForceClose(ctx context.Context, reason string) (string, error) + Reopen(ctx context.Context) (string, error) +} + +// Deps are the three effects a drain has, injected so the decisions above can +// be tested without a cluster. +type Deps struct { + Client Client + Units Units + Gate Gate + Sleep func(context.Context, time.Duration) error + Now func() time.Time +} + +// Options describe one drain or restore. +type Options struct { + MemberName string + Reason string + TimerUnit string + Timeout time.Duration + Poll time.Duration + Apply bool +} + +// Occupant is one instance still held by the member. +type Occupant struct { + Name string `json:"name"` + Status string `json:"status"` +} + +// Result is what the operation did and what it found, in the shape the other +// fleet commands report. +type Result struct { + MemberName string `json:"member_name"` + Action string `json:"action"` + Reason string `json:"reason,omitempty"` + TimerUnit string `json:"timer_unit"` + TimerStopped bool `json:"timer_stopped"` + TimerRunning bool `json:"timer_running"` + GateClosed bool `json:"gate_closed"` + // Republished says the gate was handed back to its owner. It is not a + // promise that the gate is open: hysteresis holds a just-closed member shut + // until recovery has lasted, so a restore normally reports scheduler + // "manual" and the publisher opens it a cycle or two later. Observed live: + // republished at t+0, open at t+45s. + GateRepublished bool `json:"gate_republished"` + SchedulerInstance string `json:"scheduler_instance,omitempty"` + Occupants []Occupant `json:"occupants"` + WaitedSecs int `json:"waited_seconds"` + Drained bool `json:"drained"` + TimedOut bool `json:"timed_out"` + Applied bool `json:"applied"` +} + +const ( + // DefaultTimerUnit owns the member's gate. + DefaultTimerUnit = "gha-pressure-gate.timer" + // DefaultTimeout is generous on purpose. A drain that gives up early is + // worse than one that takes a while: the operator's alternative is to stop + // a job that was going to finish. + DefaultTimeout = 45 * time.Minute + // DefaultPoll is well under the publisher's eleven-second cycle, so the + // member is seen to empty about as soon as it does. + DefaultPoll = 5 * time.Second +) + +// Occupancy reports the instances the member is still carrying. Anything not +// stopped counts: a container that is starting is about to run a job, and a +// drain that ignored it would hand back a member with work landing on it. +func Occupancy(client Client, memberName string) ([]Occupant, error) { + if client == nil || memberName == "" { + return nil, fmt.Errorf("member occupancy requires a client and a member name") + } + instances, err := client.GetInstances(api.InstanceTypeAny) + if err != nil { + return nil, fmt.Errorf("list instances: %w", err) + } + occupants := make([]Occupant, 0, len(instances)) + for _, instance := range instances { + if instance.Location != memberName { + continue + } + if instance.Status == "Stopped" { + continue + } + occupants = append(occupants, Occupant{Name: instance.Name, Status: instance.Status}) + } + sort.Slice(occupants, func(i, j int) bool { return occupants[i].Name < occupants[j].Name }) + return occupants, nil +} + +func (d Deps) valid() error { + if d.Client == nil || d.Units == nil || d.Gate == nil { + return fmt.Errorf("drain requires a client, unit control and a gate") + } + return nil +} + +func (o Options) normalised() (Options, error) { + if o.MemberName == "" { + return Options{}, fmt.Errorf("drain requires a member name") + } + if o.TimerUnit == "" { + o.TimerUnit = DefaultTimerUnit + } + if o.Timeout <= 0 { + o.Timeout = DefaultTimeout + } + if o.Poll <= 0 { + o.Poll = DefaultPoll + } + if o.Poll > o.Timeout { + return Options{}, fmt.Errorf("drain poll interval %s exceeds its timeout %s", o.Poll, o.Timeout) + } + return o, nil +} + +func (d Deps) now() time.Time { + if d.Now != nil { + return d.Now() + } + return time.Now().UTC() +} + +func (d Deps) sleep(ctx context.Context, interval time.Duration) error { + if d.Sleep != nil { + return d.Sleep(ctx, interval) + } + timer := time.NewTimer(interval) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +// Drain closes the member to new work and waits for what is already running to +// finish. Without Apply it reports what it would do and what the member is +// carrying, and changes nothing. +func Drain(ctx context.Context, deps Deps, options Options) (Result, error) { + if err := ctx.Err(); err != nil { + return Result{}, err + } + if err := deps.valid(); err != nil { + return Result{}, err + } + options, err := options.normalised() + if err != nil { + return Result{}, err + } + if options.Reason == "" { + return Result{}, fmt.Errorf("drain requires a reason, which is published as the gate's close reason") + } + result := Result{ + MemberName: options.MemberName, Action: "drain", Reason: options.Reason, + TimerUnit: options.TimerUnit, Applied: options.Apply, + } + if !options.Apply { + occupants, err := Occupancy(deps.Client, options.MemberName) + if err != nil { + return Result{}, err + } + running, err := deps.Units.IsActive(ctx, options.TimerUnit) + if err != nil { + return Result{}, err + } + result.Occupants = occupants + result.Drained = len(occupants) == 0 + result.TimerRunning = running + return result, nil + } + + // Order matters. The timer stops first, because a gate closed while the + // publisher is still running is reopened on its next cycle. + if err := deps.Units.Stop(ctx, options.TimerUnit); err != nil { + return Result{}, fmt.Errorf("stop %s: %w", options.TimerUnit, err) + } + result.TimerStopped = true + scheduler, err := deps.Gate.ForceClose(ctx, options.Reason) + if err != nil { + return Result{}, fmt.Errorf("close the gate on %s: %w", options.MemberName, err) + } + result.GateClosed = true + result.SchedulerInstance = scheduler + + started := deps.now() + deadline := started.Add(options.Timeout) + for { + occupants, err := Occupancy(deps.Client, options.MemberName) + if err != nil { + return Result{}, err + } + result.Occupants = occupants + result.WaitedSecs = int(deps.now().Sub(started) / time.Second) + if len(occupants) == 0 { + result.Drained = true + return result, nil + } + if !deps.now().Before(deadline) { + // The member is closed and no new work lands on it. What is left is + // somebody's build, and ending it is not this command's decision. + result.TimedOut = true + return result, nil + } + if err := deps.sleep(ctx, options.Poll); err != nil { + return Result{}, err + } + } +} + +// Restore hands the member back: the gate is published from live pressure again +// and the timer that owns it is started, in that order, so the member is never +// left open with nothing maintaining it. +func Restore(ctx context.Context, deps Deps, options Options) (Result, error) { + if err := ctx.Err(); err != nil { + return Result{}, err + } + if err := deps.valid(); err != nil { + return Result{}, err + } + options, err := options.normalised() + if err != nil { + return Result{}, err + } + result := Result{ + MemberName: options.MemberName, Action: "restore", + TimerUnit: options.TimerUnit, Applied: options.Apply, + } + running, err := deps.Units.IsActive(ctx, options.TimerUnit) + if err != nil { + return Result{}, err + } + result.TimerRunning = running + if !options.Apply { + return result, nil + } + scheduler, err := deps.Gate.Reopen(ctx) + if err != nil { + return Result{}, fmt.Errorf("republish the gate on %s: %w", options.MemberName, err) + } + result.GateRepublished = true + result.SchedulerInstance = scheduler + if err := deps.Units.Start(ctx, options.TimerUnit); err != nil { + return Result{}, fmt.Errorf("start %s: %w", options.TimerUnit, err) + } + result.TimerRunning = true + return result, nil +} diff --git a/internal/memberdrain/drain_test.go b/internal/memberdrain/drain_test.go new file mode 100644 index 00000000..e92b33c5 --- /dev/null +++ b/internal/memberdrain/drain_test.go @@ -0,0 +1,292 @@ +package memberdrain + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/lxc/incus/v7/shared/api" +) + +type fakeClient struct { + batches [][]api.Instance + calls int +} + +func (f *fakeClient) GetInstances(api.InstanceType) ([]api.Instance, error) { + if f.calls >= len(f.batches) { + f.calls++ + return f.batches[len(f.batches)-1], nil + } + batch := f.batches[f.calls] + f.calls++ + return batch, nil +} + +type fakeUnits struct { + events *[]string + active bool + stopErr error +} + +func (f *fakeUnits) Stop(context.Context, string) error { + if f.stopErr != nil { + return f.stopErr + } + *f.events = append(*f.events, "timer-stop") + return nil +} + +func (f *fakeUnits) Start(context.Context, string) error { + *f.events = append(*f.events, "timer-start") + return nil +} + +func (f *fakeUnits) IsActive(context.Context, string) (bool, error) { return f.active, nil } + +type fakeGate struct { + events *[]string + reason string +} + +func (f *fakeGate) ForceClose(_ context.Context, reason string) (string, error) { + f.reason = reason + *f.events = append(*f.events, "gate-close") + return "manual", nil +} + +func (f *fakeGate) Reopen(context.Context) (string, error) { + *f.events = append(*f.events, "gate-reopen") + // Hysteresis holds a just-closed member shut, so a restore reports the + // value the publisher actually wrote, not the one it is heading for. + return "manual", nil +} + +func instance(name, member, status string) api.Instance { + return api.Instance{Name: name, Location: member, Status: status} +} + +func TestOccupancyCountsOnlyThisMemberAndIgnoresStopped(t *testing.T) { + client := &fakeClient{batches: [][]api.Instance{{ + instance("worker-a", "gha-runner-3", "Running"), + instance("worker-b", "gha-runner-4", "Running"), + instance("worker-c", "gha-runner-3", "Stopped"), + instance("worker-d", "gha-runner-3", "Starting"), + }}} + occupants, err := Occupancy(client, "gha-runner-3") + if err != nil { + t.Fatalf("occupancy: %v", err) + } + if len(occupants) != 2 { + t.Fatalf("expected two occupants, got %#v", occupants) + } + // A starting container is about to run a job. Treating it as empty would + // hand back a member with work landing on it. + if occupants[0].Name != "worker-a" || occupants[1].Name != "worker-d" { + t.Fatalf("unexpected occupants: %#v", occupants) + } +} + +func TestDrainStopsTheTimerBeforeClosingTheGate(t *testing.T) { + // The publisher reasserts the gate every eleven seconds, so a gate closed + // while it is still running is reopened on the next cycle. Order is the + // whole correctness of this operation. + var events []string + client := &fakeClient{batches: [][]api.Instance{{}}} + deps := Deps{ + Client: client, + Units: &fakeUnits{events: &events}, + Gate: &fakeGate{events: &events}, + Now: func() time.Time { return time.Unix(0, 0).UTC() }, + } + result, err := Drain(context.Background(), deps, Options{ + MemberName: "gha-runner-3", Reason: "kernel slab reboot", Apply: true, + }) + if err != nil { + t.Fatalf("drain: %v", err) + } + if len(events) != 2 || events[0] != "timer-stop" || events[1] != "gate-close" { + t.Fatalf("expected the timer to stop before the gate closed, got %v", events) + } + if !result.Drained || !result.TimerStopped || !result.GateClosed { + t.Fatalf("unexpected result: %#v", result) + } +} + +func TestDrainWaitsForRunningWorkToFinish(t *testing.T) { + var events []string + client := &fakeClient{batches: [][]api.Instance{ + {instance("worker-a", "gha-runner-3", "Running")}, + {instance("worker-a", "gha-runner-3", "Running")}, + {}, + }} + clock := time.Unix(0, 0).UTC() + slept := 0 + deps := Deps{ + Client: client, + Units: &fakeUnits{events: &events}, + Gate: &fakeGate{events: &events}, + Now: func() time.Time { return clock }, + Sleep: func(_ context.Context, d time.Duration) error { + slept++ + clock = clock.Add(d) + return nil + }, + } + result, err := Drain(context.Background(), deps, Options{ + MemberName: "gha-runner-3", Reason: "slab reboot", Apply: true, + Poll: 5 * time.Second, Timeout: time.Minute, + }) + if err != nil { + t.Fatalf("drain: %v", err) + } + if !result.Drained || result.TimedOut { + t.Fatalf("expected a completed drain, got %#v", result) + } + if slept != 2 { + t.Fatalf("expected to wait twice, waited %d times", slept) + } + if result.WaitedSecs != 10 { + t.Fatalf("expected ten seconds of waiting, got %d", result.WaitedSecs) + } +} + +func TestDrainTimesOutWithoutEndingAnybodysJob(t *testing.T) { + var events []string + client := &fakeClient{batches: [][]api.Instance{ + {instance("worker-a", "gha-runner-3", "Running")}, + }} + clock := time.Unix(0, 0).UTC() + deps := Deps{ + Client: client, + Units: &fakeUnits{events: &events}, + Gate: &fakeGate{events: &events}, + Now: func() time.Time { return clock }, + Sleep: func(_ context.Context, d time.Duration) error { + clock = clock.Add(d) + return nil + }, + } + result, err := Drain(context.Background(), deps, Options{ + MemberName: "gha-runner-3", Reason: "slab reboot", Apply: true, + Poll: 10 * time.Second, Timeout: 20 * time.Second, + }) + if err != nil { + t.Fatalf("drain: %v", err) + } + if !result.TimedOut || result.Drained { + t.Fatalf("expected a timed-out drain, got %#v", result) + } + if len(result.Occupants) != 1 || result.Occupants[0].Name != "worker-a" { + t.Fatalf("a timed-out drain must name what is still there, got %#v", result.Occupants) + } + // The gate stays closed and the job stays running: nothing in the event log + // stops an instance. + for _, event := range events { + if event == "instance-stop" { + t.Fatal("a drain must never end a running job") + } + } +} + +func TestDrainWithoutApplyChangesNothing(t *testing.T) { + var events []string + client := &fakeClient{batches: [][]api.Instance{ + {instance("worker-a", "gha-runner-3", "Running")}, + }} + deps := Deps{ + Client: client, + Units: &fakeUnits{events: &events, active: true}, + Gate: &fakeGate{events: &events}, + } + result, err := Drain(context.Background(), deps, Options{ + MemberName: "gha-runner-3", Reason: "dry run", + }) + if err != nil { + t.Fatalf("drain: %v", err) + } + if len(events) != 0 { + t.Fatalf("a drain without --apply must change nothing, did %v", events) + } + if result.Applied || result.TimerStopped || result.GateClosed { + t.Fatalf("unexpected result: %#v", result) + } + if !result.TimerRunning || len(result.Occupants) != 1 { + t.Fatalf("a dry run must still report what it found: %#v", result) + } +} + +func TestDrainRequiresAReason(t *testing.T) { + var events []string + deps := Deps{ + Client: &fakeClient{batches: [][]api.Instance{{}}}, + Units: &fakeUnits{events: &events}, + Gate: &fakeGate{events: &events}, + } + if _, err := Drain(context.Background(), deps, Options{MemberName: "gha-runner-3"}); err == nil { + t.Fatal("expected a drain with no reason to be refused: the reason is published as the close reason") + } +} + +func TestDrainReportsAFailureToStopTheTimerRatherThanClosingAnyway(t *testing.T) { + var events []string + deps := Deps{ + Client: &fakeClient{batches: [][]api.Instance{{}}}, + Units: &fakeUnits{events: &events, stopErr: errors.New("unit not found")}, + Gate: &fakeGate{events: &events}, + } + _, err := Drain(context.Background(), deps, Options{ + MemberName: "gha-runner-3", Reason: "slab reboot", Apply: true, + }) + if err == nil { + t.Fatal("expected the drain to fail when the timer cannot be stopped") + } + // Closing the gate with the publisher still running would read as a drain + // and be undone within eleven seconds. + for _, event := range events { + if event == "gate-close" { + t.Fatal("the gate must not be closed when the timer is still running") + } + } +} + +func TestRestoreRepublishesBeforeStartingTheTimer(t *testing.T) { + var events []string + deps := Deps{ + Client: &fakeClient{batches: [][]api.Instance{{}}}, + Units: &fakeUnits{events: &events}, + Gate: &fakeGate{events: &events}, + } + result, err := Restore(context.Background(), deps, Options{ + MemberName: "gha-runner-3", Apply: true, + }) + if err != nil { + t.Fatalf("restore: %v", err) + } + if len(events) != 2 || events[0] != "gate-reopen" || events[1] != "timer-start" { + t.Fatalf("expected the gate republished before the timer started, got %v", events) + } + if !result.GateRepublished || !result.TimerRunning { + t.Fatalf("unexpected result: %#v", result) + } + if result.SchedulerInstance != "manual" { + t.Fatalf("restore must report the value the publisher wrote, got %q", result.SchedulerInstance) + } +} + +func TestPollLongerThanTheTimeoutIsRefused(t *testing.T) { + var events []string + deps := Deps{ + Client: &fakeClient{batches: [][]api.Instance{{}}}, + Units: &fakeUnits{events: &events}, + Gate: &fakeGate{events: &events}, + } + _, err := Drain(context.Background(), deps, Options{ + MemberName: "gha-runner-3", Reason: "slab reboot", + Poll: time.Hour, Timeout: time.Minute, + }) + if err == nil { + t.Fatal("expected a poll interval longer than the timeout to be refused") + } +}