diff --git a/docs/plans-executions/2026-09-18-1502-proxmox-provider.md b/docs/plans-executions/2026-09-18-1502-proxmox-provider.md index ca16638..d2e7308 100644 --- a/docs/plans-executions/2026-09-18-1502-proxmox-provider.md +++ b/docs/plans-executions/2026-09-18-1502-proxmox-provider.md @@ -9,7 +9,7 @@ Pairs with [docs/plans/2026-09-18-1502-proxmox-provider.md](../plans/2026-09-18- - [x] Phase 2 — Config block, registration, package skeleton - [x] Phase 3 — Extract `nocloud` and `keyedlock` from kvm2 - [x] Phase 4 — Seam, HTTP client, decorator, fake -- [ ] Phase 5 — Get, ListByTag, task helpers +- [x] Phase 5 — Get, ListByTag, task helpers - [ ] Phase 6 — Create - [ ] Phase 7 — Delete, Checker, CRD comment - [ ] Phase 8 — Contract and observability sweep, integration test, testing.md @@ -554,3 +554,106 @@ Worth noting: `fake: VM is running - destroy failed`, so Phase 7's stop-before-destroy order is enforced by the fake even though the real text is unknown. + +## Phase 5 — Get, ListByTag, task helpers + +`task.go` holds `parseUPID`, a shared `pollUntil` loop, `waitTask` and +`waitUnlocked`. `get.go` holds `Get`, the guest-agent read and `resolve` +(one resources call, QEMU entries only, which Delete will reuse). `list.go` +holds `ListByTag` and its candidate rule. The Get and ListByTag stubs are +gone, and `TestStubs_answerTransient` now covers only Create and Delete. + +The spec overrides the plan in the places the Phase 4 notes listed. Get +checks in entry 9's order, so a lock or `status: unknown` in the resources +entry is Transient before the name is ever read, and a running Get makes +four calls (`TestGet_runningMakesFourCalls`). The three agent messages +follow entry 7. The poll interval is `budgets.poll`, and a failed task logs +a V(1) `proxmox task failed` line with `upid` and `exitstatus`. + +Judgment calls, most of them spec gaps: + +- **`WARNINGS: ` counts as a successful task.** Proxmox reports a task + that finished but logged warnings with that exitstatus and treats it as + finished, not failed. Under the spec's old "anything but `OK` is + Transient" rule, a start that logged a warning would make Create retry + forever. Recon never produced one, so this row relies on how Proxmox is + known to behave, not on a capture. The spec's entry 10 text and entry 13 + table now say so. +- **A `suspended` lock still goes through the name and mark checks.** The + entry 9 table puts "lock suspended is Stopped" before the config rows. + Taken literally, a hibernated VM that reused our VMID would be reported + as ours and Stopped, and Delete would then skip it as foreign in a loop. + Get therefore reads the config, which it needs anyway for UID and + CreatedAt, applies the name and mark checks, and only then answers + Stopped. It skips `status/current` and the agent. Entry 9 now says this. +- **`TestGet_twoVMsWithOurNameIsPermanent` became + `TestGet_otherVMWithOurNameIsIgnored`.** Spec entries 2 and 9 address Get + by VMID, and only Create refuses a duplicate name. Get answering + Permanent would latch a working Proxy as Failed because of a VM someone + else created. +- **`TestGet_unmanagedDescriptionHasEmptyUID` tests a mark with no `uid` + line.** The plan's name read literally ("unmanaged") would contradict + "no mark is NotFound". +- **Agent errors outside the entry 7 table are forced to Transient**, even + a 403 or a `does not exist`. Otherwise `classify` would make them + Permanent or NotFound, and the spec says "anything else: Transient". +- **`waitUnlocked` returns the unlocked config**, which saves Create's + ownership check one read. When the VM disappears while it waits (a + failed clone), it returns the does-not-exist error and lets Create + decide. + +Pulled forward and left behind: + +- **`provider_test.go` now exists** with `newFixture`, `testBudgets`, + `addOurVM` and `assertProviderError`, because every Get and ListByTag + test needs them. The plan created this file in Phase 8. Phase 8 adds the + contract sweep to it. +- **`waitRunning` is left to Phase 6.** The plan's package layout puts it + in `task.go`, but Phase 5 lists no test for it and only Create uses it. + It should be built on `pollUntil`. +- **The fake gained two things.** `fakeVM.kind` makes an entry listed as + `lxc` invisible to the qemu endpoints. `hookOn` is a general per-method + hook, and `failWith` is now built on it. Tests use `hookOn` (through + `mutateOn` in `get_test.go`) to change a VM between two of Get's calls. + +A mutation check confirmed the new tests can fail. Each of these failed a +test: classifying agent errors instead of forcing Transient; dropping the +config lock check; dropping the `suspended` special case; ignoring an ended +ctx in ListByTag; treating `WARNINGS` as failure; and not mapping +`No QEMU guest agent configured` to Stopped. Removing the resources-entry +lock and `unknown` checks together failed only +`TestGet_unknownStatusIsTransient`. The fake reports a lock in the config +as well, so the config check catches every lock row, including a mid-clone +VM. The resources check stays because the spec orders it first and +because it spares a config read. Every mutation was reverted, and the +files were compared identical to backups. + +```bash +go test -race -count=2 ./internal/provider/proxmox/ # ok +grep -n 'not implemented' internal/provider/proxmox/*.go | grep -v _test # Create and Delete only +go build ./... && go vet ./... && go vet -tags=integration ./... +go test -race -count=1 ./internal/provider/... ./cmd/... # all ok +go test -short -count=1 ./... # all ok +go test -count=1 -v ./internal/provider/proxmox/ | grep -c '^=== RUN' # 289 before, 393 after +``` + +Lint ran with staticcheck disabled, because staticcheck panics on +Go 1.27: + +```bash +bin/golangci-lint run --disable staticcheck ./internal/provider/proxmox/... +``` + +Worth noting: + +- The lint findings in the new files are goconst on test strings and + logcheck on `get`, `agentAddress` and `waitTask` taking both a ctx and a + logger. That shape copies kvm2, which has five of the same findings. The + errcheck hits on `assertGetFails` were fixed by returning the error text. + The two unparam hits and the mapsloop hits predate this phase. +- `waitUnlocked` waits on any lock, `suspended` included. Phase 6's + ownership check should decide whether a hibernated VM at our VMID is + worth waiting out the clone budget for, or should be refused at once. +- Nothing in this phase talked to the host. Every Proxmox message a test + relies on comes from `testdata/` or from fake errors that + `TestFakeAPI_errorsMatchCaptures` ties to the captures. diff --git a/docs/proxmox-provider-spec.md b/docs/proxmox-provider-spec.md index 9e1dda4..84712ab 100644 --- a/docs/proxmox-provider-spec.md +++ b/docs/proxmox-provider-spec.md @@ -285,6 +285,11 @@ and starts it; the reconciler has no ID for it to `Get`. | `status: running`, any other `qmpstatus` (`paused`, `suspended`, `io-error`, …) | `status/current` | Stopped | | any other `status` | `status/current` | Stopped | +A `suspended` lock decides the state but not ownership: the config is still +read, and a foreign name or a missing mark is still `ErrNotFound`, so a +hibernated VM that reused our VMID is never reported as ours. Only +`status/current` and the agent are skipped for it. + `status/current` is read, not only the resources entry, because a paused VM shows as plain `running` in `/cluster/resources`. `UID` and `CreatedAt` come from the description on every call. A running `Get` makes four calls: @@ -301,9 +306,10 @@ contract §2.2 and §4.11. `backend-lock`. Clone, upload, resize, start, stop, destroy and volume delete each return a UPID; the provider polls `GET /nodes/{node}/tasks/{upid}/status` every second until `status` is -`stopped`, then reads `exitstatus`: `OK` is success, anything else is -`ErrTransient` carrying that text. The node comes from the UPID -(`UPID::…`). The handle is held within the call only: Create blocks +`stopped`, then reads `exitstatus`: `OK` is success, and so is +`WARNINGS: `, which Proxmox itself counts as a finished task (logged at +V(1)); anything else is `ErrTransient` carrying that text. The node comes +from the UPID (`UPID::…`). The handle is held within the call only: Create blocks until the VM runs, so no handle outlives it, and a failure surfaces as Create's own error with the task's message. Locks: a VM locked by a task is waited on in Create (the ownership check) and reported as `ErrTransient` by @@ -447,7 +453,7 @@ per-parameter map. One `classify(op string, err error)`: | 500 `unable to create VM : config file already exists` | handled by the allocator (entry 2); Transient if it escapes | the VMID is taken | | 500 `QEMU guest agent is not running`, `No QEMU guest agent configured`, `VM is not running` | handled by `Get` (entry 7); Transient if they escape | agent state | | any other 5xx, including pveproxy's 595 and 596 | Transient | unknown means retry | -| task `exitstatus` other than `OK` | Transient, with the text | the task failed; the next attempt retries | +| task `exitstatus` other than `OK` or `WARNINGS: ` | Transient, with the text | the task failed; the next attempt retries | | TLS verification failure (`x509.UnknownAuthorityError`, `x509.HostnameError`, `x509.CertificateInvalidError`) | Permanent | wrong URL or CA file; the admin must act | | token file missing (`fs.ErrNotExist`) | Transient | the Secret has not landed yet | | token file present but not `@!=` | Permanent | wrong content in the Secret | diff --git a/internal/provider/proxmox/fake_test.go b/internal/provider/proxmox/fake_test.go index 62242e6..52206d0 100644 --- a/internal/provider/proxmox/fake_test.go +++ b/internal/provider/proxmox/fake_test.go @@ -52,9 +52,12 @@ type fakeAPI struct { type fakeVM struct { node, name, status, qmpstatus, lock string - template bool - config map[string]string - ifaces []agentInterface + // kind is the resource type; empty is "qemu". Any other kind, such as + // "lxc", is listed but invisible to the qemu endpoints. + kind string + template bool + config map[string]string + ifaces []agentInterface } type fakeTask struct { @@ -125,9 +128,15 @@ func (f *fakeAPI) addTemplate(vmid int, name string) *fakeVM { } func (f *fakeAPI) failWith(method string, err error) { + f.hookOn(method, func() error { return err }) +} + +// hookOn runs fn at the start of every call to method, without f.mu held; +// a non-nil result fails the call. A hook that changes state takes f.mu. +func (f *fakeAPI) hookOn(method string, fn func() error) { f.mu.Lock() defer f.mu.Unlock() - f.hooks[method] = func() error { return err } + f.hooks[method] = fn } func (f *fakeAPI) blockOn(method string) chan struct{} { @@ -217,7 +226,7 @@ func (f *fakeAPI) enter(ctx context.Context, method string) error { // node has no config file on this one. func (f *fakeAPI) vm(node string, vmid int) (*fakeVM, error) { vm, ok := f.vms[vmid] - if !ok || vm.node != node { + if !ok || vm.node != node || vm.kind != "" { return nil, errVMMissing(node, vmid) } return vm, nil @@ -279,6 +288,9 @@ func (f *fakeAPI) ClusterResources(ctx context.Context) ([]vmResource, error) { for _, vmid := range slices.Sorted(maps.Keys(f.vms)) { vm := f.vms[vmid] r := vmResource{VMID: vmid, Name: vm.name, Node: vm.node, Type: "qemu", Status: vm.status, Lock: vm.lock, Tags: vm.config["tags"]} + if vm.kind != "" { + r.Type = vm.kind + } if vm.template { r.Template = 1 } diff --git a/internal/provider/proxmox/get.go b/internal/provider/proxmox/get.go new file mode 100644 index 0000000..b0a2379 --- /dev/null +++ b/internal/provider/proxmox/get.go @@ -0,0 +1,132 @@ +package proxmox + +import ( + "context" + "errors" + "fmt" + + "github.com/go-logr/logr" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// lockSuspended is the lock of a VM hibernated to disk: it is not a task in +// progress, and the VM stays that way until someone resumes it. +const lockSuspended = "suspended" + +// Get is spec entries 7 to 9: the VMID's resource entry, then its config +// for the name and the mark, then status/current and, for a running VM, the +// guest agent. A running Get makes four calls. +func (p *Provider) Get(ctx context.Context, id string) (*provider.Instance, error) { + vmid, name, err := parseProviderID(id) + if err != nil { + return nil, p.wrapErr("get", id, err) + } + log := p.logger(ctx).WithValues("vmid", vmid, "name", name) + inst, err := p.get(ctx, log, vmid, name) + if err != nil { + err = p.wrapErr("get", id, err) + if errors.Is(err, provider.ErrNotFound) { + log.V(1).Info("proxmox vm not found", "reason", err.Error()) + } + return nil, err + } + return inst, nil +} + +func (p *Provider) get(ctx context.Context, log logr.Logger, vmid int, name string) (*provider.Instance, error) { + r, ok, err := p.resolve(ctx, vmid) + if err != nil { + return nil, err + } + if !ok { + return nil, fmt.Errorf("%w: VM %d is not in the cluster's resources", provider.ErrNotFound, vmid) + } + log = log.WithValues("node", r.Node) + // A mid-clone entry has no name and status unknown, so the lock is + // checked before anything that would read it as absent or foreign. + if r.Lock != "" && r.Lock != lockSuspended { + return nil, transientf("VM %d is locked (%s)", vmid, r.Lock) + } + if r.Status == "unknown" { + return nil, transientf("VM %d has status unknown: its node %s is not answering", vmid, r.Node) + } + cfg, err := p.api.VMConfig(ctx, r.Node, vmid) + if err != nil { + return nil, fmt.Errorf("reading config of VM %d: %w", vmid, err) + } + if lock := cfg["lock"]; lock != "" && lock != lockSuspended { + return nil, transientf("VM %d is locked (%s)", vmid, lock) + } + if cfg["name"] != name { + return nil, fmt.Errorf("%w: VM %d is now named %q: the VMID was reused", provider.ErrNotFound, vmid, cfg["name"]) + } + mark := parseDescription(cfg["description"]) + if !mark.Managed { + return nil, fmt.Errorf("%w: VM %d carries no proxy-operator mark", provider.ErrNotFound, vmid) + } + inst := &provider.Instance{ID: formatProviderID(vmid, name), UID: mark.UID, CreatedAt: mark.CreatedAt} + + // A hibernated VM is Stopped whatever status/current says; it needs no + // further call. + if r.Lock == lockSuspended || cfg["lock"] == lockSuspended { + inst.State = provider.StateStopped + log.V(1).Info("proxmox vm fetched", "state", inst.State, "lock", lockSuspended, "uid", inst.UID) + return inst, nil + } + st, err := p.api.VMStatus(ctx, r.Node, vmid) + if err != nil { + return nil, fmt.Errorf("reading status of VM %d: %w", vmid, err) + } + if st.Status == "running" && (st.QMPStatus == "running" || st.QMPStatus == "") { + ip, state, err := p.agentAddress(ctx, log, r.Node, vmid) + if err != nil { + return nil, err + } + inst.IP = ip + inst.State = state + } else { + inst.State = mapState(st.Status, st.QMPStatus, "") + } + log.V(1).Info("proxmox vm fetched", "state", inst.State, "ip", inst.IP, "uid", inst.UID, "status", st.Status, "qmpstatus", st.QMPStatus) + return inst, nil +} + +// agentAddress asks the guest agent of a running VM for its address and +// turns its answers into a state (spec entry 7). Any answer it cannot place +// is Transient, whatever its status: an address read never latches or +// loses a Proxy. +func (p *Provider) agentAddress(ctx context.Context, log logr.Logger, node string, vmid int) (string, provider.InstanceState, error) { + ifaces, err := p.api.AgentNetworkInterfaces(ctx, node, vmid) + switch { + case err == nil: + ip := firstIPv4(ifaces) + log.V(2).Info("proxmox guest agent answered", "interfaces", len(ifaces), "ip", ip) + return ip, mapState("running", "running", ip), nil + case isAgentNotRunning(err): + return "", provider.StateProvisioning, nil + case isAgentNotConfigured(err): + // agent=1 was removed: this VM can never report an address, so it + // is replaced by one that can. + return "", provider.StateStopped, nil + default: + // Includes "VM is not running": it stopped since the status + // read, and the next Get sees that directly. + return "", "", fmt.Errorf("%w: reading the guest agent of VM %d: %w", provider.ErrTransient, vmid, err) + } +} + +// resolve finds a VMID in the cluster's resources. Only QEMU VMs count: a +// container at the VMID means ours is gone. +func (p *Provider) resolve(ctx context.Context, vmid int) (vmResource, bool, error) { + rs, err := p.api.ClusterResources(ctx) + if err != nil { + return vmResource{}, false, fmt.Errorf("listing cluster resources: %w", err) + } + for _, r := range rs { + if r.VMID == vmid && r.Type == "qemu" { + return r, true, nil + } + } + return vmResource{}, false, nil +} diff --git a/internal/provider/proxmox/get_test.go b/internal/provider/proxmox/get_test.go new file mode 100644 index 0000000..ca7de3b --- /dev/null +++ b/internal/provider/proxmox/get_test.go @@ -0,0 +1,476 @@ +package proxmox + +import ( + "errors" + "slices" + "strings" + "testing" + "time" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// mutateOn changes the fake's state under its lock at the start of every +// call to method: a VM changing between two of Get's calls. +func mutateOn(f *fakeAPI, method string, fn func()) { + f.hookOn(method, func() error { + f.mu.Lock() + defer f.mu.Unlock() + fn() + return nil + }) +} + +// capturedIfaces is the agent's answer from the Phase 0 guest: lo and ens18 +// with 192.168.0.150. +func capturedIfaces(t *testing.T) []agentInterface { + t.Helper() + return decodeData[struct { + Result []agentInterface `json:"result"` + }](t, "agent_network_get_interfaces").Result +} + +// assertGetFails runs Get on fx.id(), checks the error's class and returns +// its text. +func assertGetFails(t *testing.T, fx *fixture, class error) string { + t.Helper() + inst, err := fx.p.Get(fx.ctx, fx.id()) + if inst != nil { + t.Errorf("Get = %+v, want nil", inst) + } + assertProviderError(t, err, "get", class) + return err.Error() +} + +func TestGet_missingVMIsNotFound(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID+1, testName, "running") + assertGetFails(t, fx, provider.ErrNotFound) + lines := fx.logs.matching(`"msg"="proxmox vm not found"`) + if len(lines) != 1 || !strings.Contains(lines[0], `"vmid"=150042`) || !strings.Contains(lines[0], `"name"="`+testName+`"`) { + t.Errorf("not-found lines = %v, want one with vmid and name", lines) + } + if got := fx.fake.callLog(); !slices.Equal(got, []string{"ClusterResources"}) { + t.Errorf("calls = %v, want only the resources listing", got) + } +} + +func TestGet_containerAtVMIDIsNotFound(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").kind = "lxc" + assertGetFails(t, fx, provider.ErrNotFound) +} + +func TestGet_nameMismatchIsNotFound(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, "proxy-someoneelse00000", "running") + msg := assertGetFails(t, fx, provider.ErrNotFound) + if !strings.Contains(msg, "reused") { + t.Errorf("error = %s, want it to say the VMID was reused", msg) + } + if n := fx.fake.callCount("VMStatus"); n != 0 { + t.Errorf("VMStatus calls = %d, want 0 for a VM that is not ours", n) + } + if lines := fx.logs.matching(`"msg"="proxmox vm not found"`); len(lines) != 1 { + t.Errorf("not-found lines = %v, want one", lines) + } +} + +func TestGet_unmarkedIsNotFound(t *testing.T) { + t.Parallel() + tests := []struct { + name string + description string + }{ + {name: "no description"}, + {name: "managed false", description: "proxy-operator-managed=false\nproxy-operator-uid=" + testUID}, + {name: "admin notes only", description: "hand made, do not touch"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + vm := fx.fake.addVM(testVMID, testName, "running") + if tc.description != "" { + vm.config["description"] = tc.description + } + msg := assertGetFails(t, fx, provider.ErrNotFound) + if !strings.Contains(msg, "no proxy-operator mark") { + t.Errorf("error = %s, want it to name the missing mark", msg) + } + }) + } +} + +// TestGet_otherVMWithOurNameIsIgnored replaces the plan's +// TestGet_twoVMsWithOurNameIsPermanent: spec entries 2 and 9 address Get by +// VMID, and only Create treats a duplicate name as Permanent. +func TestGet_otherVMWithOurNameIsIgnored(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "stopped") + fx.addOurVM(testVMID+7, testName, "running") + if inst := fx.get(fx.id()); inst.State != provider.StateStopped || inst.ID != fx.id() { + t.Errorf("Get = %+v, want the VM at the ID's VMID, Stopped", inst) + } +} + +func TestGet_vmOnAnotherNodeIsFound(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + vm := fx.addOurVM(testVMID, testName, "running") + vm.node = "pv1-2" + vm.ifaces = capturedIfaces(t) + inst := fx.get(fx.id()) + if inst.State != provider.StateRunning || inst.IP != "192.168.0.150" { + t.Errorf("Get = %+v, want Running with 192.168.0.150 read from pv1-2", inst) + } + if lines := fx.logs.matching(`"msg"="proxmox vm fetched"`); len(lines) != 1 || !strings.Contains(lines[0], `"node"="pv1-2"`) { + t.Errorf("fetched lines = %v, want the resolved node", lines) + } +} + +func TestGet_runningWithoutIPIsProvisioning(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running") + if inst := fx.get(fx.id()); inst.State != provider.StateProvisioning || inst.IP != "" { + t.Errorf("Get = %+v, want Provisioning with no IP", inst) + } +} + +func TestGet_runningWithIPIsRunning(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").ifaces = capturedIfaces(t) + inst := fx.get(fx.id()) + if inst.State != provider.StateRunning || inst.IP != "192.168.0.150" || inst.ID != fx.id() { + t.Errorf("Get = %+v, want Running with 192.168.0.150", inst) + } + lines := fx.logs.matching(`"msg"="proxmox vm fetched"`) + if len(lines) != 1 { + t.Fatalf("fetched lines = %v, want one", lines) + } + for _, kv := range []string{`"provider"="proxmox-home"`, `"vmid"=150042`, `"name"="` + testName + `"`, `"node"="pv1-1"`, `"state"="Running"`, `"ip"="192.168.0.150"`, `"uid"="` + testUID + `"`} { + if !strings.Contains(lines[0], kv) { + t.Errorf("fetched line lacks %s: %s", kv, lines[0]) + } + } +} + +func TestGet_runningMakesFourCalls(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").ifaces = capturedIfaces(t) + fx.get(fx.id()) + want := []string{"ClusterResources", "VMConfig", "VMStatus", "AgentNetworkInterfaces"} + if got := fx.fake.callLog(); !slices.Equal(got, want) { + t.Errorf("calls = %v, want %v", got, want) + } +} + +func TestGet_agentNotRunningIsProvisioning(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").ifaces = capturedIfaces(t) + fx.fake.agentNotRunning = true + if inst := fx.get(fx.id()); inst.State != provider.StateProvisioning || inst.IP != "" { + t.Errorf("Get = %+v, want Provisioning while the agent is not running", inst) + } +} + +// TestGet_agentMessagesTable is spec entry 7's table, with the agent's +// answers taken from the captures. +func TestGet_agentMessagesTable(t *testing.T) { + t.Parallel() + tests := []struct { + name string + agentErr error + ifaces []agentInterface + wantState provider.InstanceState + wantClass error + }{ + {name: "200 with a routable IPv4", ifaces: capturedIfaces(t), wantState: provider.StateRunning}, + { + name: "200 without an IPv4", + ifaces: []agentInterface{{Name: "lo", IPAddresses: []agentIP{{Address: "127.0.0.1", Type: "ipv4"}}}, {Name: "ens18", IPAddresses: []agentIP{{Address: "fe80::1", Type: "ipv6"}}}}, + wantState: provider.StateProvisioning, + }, + {name: "agent not running", agentErr: loadAPIError(t, "err_agent_not_running"), wantState: provider.StateProvisioning}, + {name: "no agent configured", agentErr: loadAPIError(t, "err_agent_vm_stopped"), wantState: provider.StateStopped}, + {name: "VM stopped since the status read", agentErr: loadAPIError(t, "err_agent_vm_stopped_agent_enabled"), wantClass: provider.ErrTransient}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").ifaces = tc.ifaces + if tc.agentErr != nil { + fx.fake.failWith("AgentNetworkInterfaces", tc.agentErr) + } + if tc.wantClass != nil { + assertGetFails(t, fx, tc.wantClass) + return + } + inst := fx.get(fx.id()) + if inst.State != tc.wantState { + t.Errorf("Get state = %s, want %s", inst.State, tc.wantState) + } + if (inst.State == provider.StateRunning) != (inst.IP != "") { + t.Errorf("Get = %+v: an IP goes with Running and only with Running", inst) + } + }) + } +} + +func TestGet_agentOtherErrorIsTransient(t *testing.T) { + t.Parallel() + tests := []struct { + name string + err error + }{ + {name: "forbidden", err: loadAPIError(t, "err_forbidden")}, + {name: "VM vanished since the status read", err: loadAPIError(t, "err_vm_missing")}, + {name: "pveproxy 596", err: fakeErr(596, "Broken pipe")}, + {name: "other 500", err: fakeErr(500, "fake: guest-network-get-interfaces timed out")}, + {name: "not an api error", err: errors.New("read tcp: connection reset by peer")}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running") + fx.fake.failWith("AgentNetworkInterfaces", tc.err) + assertGetFails(t, fx, provider.ErrTransient) + }) + } +} + +func TestGet_ipFilterSkipsLoopbackAndLinkLocal(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").ifaces = []agentInterface{ + {Name: "lo", IPAddresses: []agentIP{{Address: "127.0.0.1", Type: "ipv4"}}}, + {Name: "ens18", IPAddresses: []agentIP{{Address: "169.254.10.2", Type: "ipv4"}, {Address: "fe80::be24:11ff:fe48:1146", Type: "ipv6"}}}, + {Name: "ens19", IPAddresses: []agentIP{{Address: "2001:db8::5", Type: "ipv6"}, {Address: "192.168.4.147", Type: "ipv4"}}}, + } + if inst := fx.get(fx.id()); inst.State != provider.StateRunning || inst.IP != "192.168.4.147" { + t.Errorf("Get = %+v, want Running with 192.168.4.147", inst) + } +} + +func TestGet_stoppedIsStopped(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "stopped") + if inst := fx.get(fx.id()); inst.State != provider.StateStopped || inst.IP != "" || inst.UID != testUID { + t.Errorf("Get = %+v, want Stopped with uid %s", inst, testUID) + } + if n := fx.fake.callCount("AgentNetworkInterfaces"); n != 0 { + t.Errorf("agent calls = %d, want 0 for a stopped VM", n) + } +} + +func TestGet_pausedIsStopped(t *testing.T) { + t.Parallel() + for _, qmp := range []string{"paused", "suspended", "io-error", "postmigrate", "internal-error"} { + t.Run(qmp, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").qmpstatus = qmp + if inst := fx.get(fx.id()); inst.State != provider.StateStopped { + t.Errorf("Get state with qmpstatus %s = %s, want Stopped", qmp, inst.State) + } + if n := fx.fake.callCount("AgentNetworkInterfaces"); n != 0 { + t.Errorf("agent calls = %d, want 0", n) + } + }) + } +} + +// TestGet_pausedIsStoppedFromStatusCurrent uses F14's paused VM: the +// resources listing shows plain running, only status/current tells. +func TestGet_pausedIsStoppedFromStatusCurrent(t *testing.T) { + t.Parallel() + st := decodeData[vmStatus](t, "status_current_paused") + if st.Status != "running" || st.QMPStatus != "paused" { + t.Fatalf("capture status_current_paused = %+v, want running and paused", st) + } + fx := newFixture(t, 1) + vm := fx.addOurVM(testVMID, testName, st.Status) + vm.qmpstatus = st.QMPStatus + vm.ifaces = capturedIfaces(t) + if r, _ := resourceByID(t, fx.fake, testVMID); r.Status != "running" { + t.Fatalf("resources status = %q, want running as F14 saw", r.Status) + } + if inst := fx.get(fx.id()); inst.State != provider.StateStopped || inst.IP != "" { + t.Errorf("Get = %+v, want Stopped with no IP", inst) + } +} + +func TestGet_prelaunchIsProvisioning(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + vm := fx.addOurVM(testVMID, testName, "running") + vm.qmpstatus = "prelaunch" + vm.ifaces = capturedIfaces(t) + if inst := fx.get(fx.id()); inst.State != provider.StateProvisioning || inst.IP != "" { + t.Errorf("Get = %+v, want Provisioning", inst) + } +} + +func TestGet_suspendedLockIsStopped(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "stopped").lock = lockSuspended + inst := fx.get(fx.id()) + if inst.State != provider.StateStopped || inst.UID != testUID { + t.Errorf("Get = %+v, want Stopped with uid", inst) + } + if got := fx.fake.callLog(); !slices.Equal(got, []string{"ClusterResources", "VMConfig"}) { + t.Errorf("calls = %v, want resources and config only", got) + } +} + +func TestGet_unknownStatusIsTransient(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").status = "unknown" + msg := assertGetFails(t, fx, provider.ErrTransient) + if !strings.Contains(msg, "pv1-1") { + t.Errorf("error = %s, want it to name the node", msg) + } + if n := fx.fake.callCount("VMConfig"); n != 0 { + t.Errorf("VMConfig calls = %d, want 0 while the node is unknown", n) + } +} + +func TestGet_lockedIsTransientNeverNotFound(t *testing.T) { + t.Parallel() + for _, lock := range []string{"backup", "migrate", "snapshot", "rollback", "create"} { + t.Run("resources lock "+lock, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").lock = lock + msg := assertGetFails(t, fx, provider.ErrTransient) + if !strings.Contains(msg, "locked ("+lock+")") { + t.Errorf("error = %s, want it to name the lock", msg) + } + }) + } + t.Run("lock taken between the resources and config reads", func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + vm := fx.addOurVM(testVMID, testName, "running") + mutateOn(fx.fake, "VMConfig", func() { vm.lock = "backup" }) + assertGetFails(t, fx, provider.ErrTransient) + if n := fx.fake.callCount("VMStatus"); n != 0 { + t.Errorf("VMStatus calls = %d, want 0", n) + } + }) +} + +// TestGet_midCloneNamelessEntryIsTransient drives a real fake clone: while +// it runs the entry has no name (F8), which must not read as NotFound. +func TestGet_midCloneNamelessEntryIsTransient(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.addTemplate(9001, "debian-13-template") + upid, err := fx.fake.Clone(fx.ctx, fakeNode, 9001, cloneParams{NewID: testVMID, Name: testName, Description: formatDescription(testUID, testCreatedAt)}) + if err != nil { + t.Fatalf("Clone error = %v", err) + } + if r, _ := resourceByID(t, fx.fake, testVMID); r.Name != "" || r.Lock != "clone" { + t.Fatalf("mid-clone entry = %+v, want nameless and locked", r) + } + assertGetFails(t, fx, provider.ErrTransient) + pollTask(t, fx.fake, upid) + if inst := fx.get(fx.id()); inst.State != provider.StateStopped || inst.UID != testUID { + t.Errorf("Get after the clone = %+v, want Stopped with uid", inst) + } +} + +func TestGet_vanishedBetweenCallsIsNotFound(t *testing.T) { + t.Parallel() + for _, method := range []string{"VMConfig", "VMStatus"} { + t.Run("before "+method, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running") + mutateOn(fx.fake, method, func() { delete(fx.fake.vms, testVMID) }) + assertGetFails(t, fx, provider.ErrNotFound) + }) + } +} + +func TestGet_unknownStateIsStopped(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + vm := fx.addOurVM(testVMID, testName, "running") + vm.status, vm.qmpstatus = "hibernating-v2", "hibernating-v2" + if inst := fx.get(fx.id()); inst.State != provider.StateStopped { + t.Errorf("Get state = %s, want Stopped for an unrecognised status", inst.State) + } +} + +func TestGet_readsUIDAndCreatedAtFromBackend(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + created := time.Date(2026, 9, 1, 8, 30, 0, 0, time.UTC) + fx.addOurVM(testVMID, testName, "stopped").config["description"] = formatDescription("uid-from-backend", created) + inst := fx.get(fx.id()) + if inst.UID != "uid-from-backend" || !inst.CreatedAt.Equal(created) { + t.Errorf("Get = uid %q createdAt %v, want uid-from-backend and %v", inst.UID, inst.CreatedAt, created) + } +} + +// TestGet_unmanagedDescriptionHasEmptyUID: a mark without a uid line is +// still ours, with an empty UID, which callers must tolerate. +func TestGet_unmanagedDescriptionHasEmptyUID(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "stopped").config["description"] = "proxy-operator-managed=true\r\nadmin note\r\n" + inst := fx.get(fx.id()) + if inst.UID != "" || !inst.CreatedAt.IsZero() || inst.State != provider.StateStopped { + t.Errorf("Get = %+v, want empty uid, zero createdAt, Stopped", inst) + } +} + +func TestGet_malformedIDIsPermanent(t *testing.T) { + t.Parallel() + for _, id := range []string{"", "vms/150042", "vms/0150042/" + testName, "vm/150042/" + testName, "vms/abc/" + testName, "vms/150042/a/b"} { + t.Run(id, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + _, err := fx.p.Get(fx.ctx, id) + assertProviderError(t, err, "get", provider.ErrPermanent) + if calls := fx.fake.callLog(); len(calls) != 0 { + t.Errorf("calls = %v, want none", calls) + } + }) + } +} + +func TestGet_resourcesFailureIsTransient(t *testing.T) { + t.Parallel() + tests := []struct { + name string + err error + }{ + {name: "pveproxy 595", err: fakeErr(595, "Connection refused")}, + {name: "other 500", err: fakeErr(500, "fake: cluster filesystem not ready")}, + {name: "network", err: errors.New("dial tcp 192.168.0.8:8006: connect: no route to host")}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running") + fx.fake.failWith("ClusterResources", tc.err) + assertGetFails(t, fx, provider.ErrTransient) + }) + } +} diff --git a/internal/provider/proxmox/list.go b/internal/provider/proxmox/list.go new file mode 100644 index 0000000..4bb634a --- /dev/null +++ b/internal/provider/proxmox/list.go @@ -0,0 +1,81 @@ +package proxmox + +import ( + "context" + "strings" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// proxyTag is the tag Create sets for the Proxmox UI; ListByTag uses it, +// with the name prefix provider.NameFromUID gives, only to pick candidates. +// The mark decides. +const ( + proxyTag = "proxy_operator" + proxyNamePrefix = "proxy-" +) + +// ListByTag is spec entry 12: one resources call covers every node, then +// one config read per candidate for the mark. GC reads only ID, UID and +// CreatedAt (contract §2.4), so no status or agent call is made. The +// resources call failing fails the list; one unreadable candidate is +// skipped, unless ctx ended, which fails the list rather than pass off a +// partial view as complete. +func (p *Provider) ListByTag(ctx context.Context) ([]provider.Instance, error) { + log := p.logger(ctx) + rs, err := p.api.ClusterResources(ctx) + if err != nil { + return nil, p.wrapErr("list", "", err) + } + var ( + out []provider.Instance + candidates, skipped int + ) + for _, r := range rs { + if !isCandidate(r) { + continue + } + candidates++ + vlog := log.WithValues("vmid", r.VMID, "name", r.Name, "node", r.Node) + cfg, err := p.api.VMConfig(ctx, r.Node, r.VMID) + if err != nil { + if ctx.Err() != nil { + return nil, p.wrapErr("list", "", err) + } + skipped++ + vlog.V(1).Info("skipping unreadable proxmox vm", "error", err.Error()) + continue + } + mark := parseDescription(cfg["description"]) + if !mark.Managed { + continue + } + inst := provider.Instance{ID: formatProviderID(r.VMID, r.Name), UID: mark.UID, CreatedAt: mark.CreatedAt} + vlog.V(2).Info("proxmox listed instance", "id", inst.ID, "uid", inst.UID, "createdAt", inst.CreatedAt) + out = append(out, inst) + } + log.V(1).Info("proxmox vms listed", "vms", len(rs), "candidates", candidates, "managed", len(out), "skipped", skipped) + return out, nil +} + +// isCandidate picks the resource entries worth a config read: QEMU VMs, not +// templates, named, and either tagged or named like a proxy. A nameless +// entry is a clone in progress, whose Create is still running. +func isCandidate(r vmResource) bool { + if r.Type != "qemu" || r.Template != 0 || r.Name == "" { + return false + } + return strings.HasPrefix(r.Name, proxyNamePrefix) || hasTag(r.Tags, proxyTag) +} + +// hasTag reports whether tags, as Proxmox lists them (";"-separated, though +// "," and spaces are accepted on input), contains tag. Proxmox folds tags to +// lower case. +func hasTag(tags, tag string) bool { + for _, t := range strings.FieldsFunc(tags, func(r rune) bool { return r == ';' || r == ',' || r == ' ' }) { + if strings.EqualFold(t, tag) { + return true + } + } + return false +} diff --git a/internal/provider/proxmox/list_test.go b/internal/provider/proxmox/list_test.go new file mode 100644 index 0000000..422d979 --- /dev/null +++ b/internal/provider/proxmox/list_test.go @@ -0,0 +1,227 @@ +package proxmox + +import ( + "context" + "errors" + "slices" + "strings" + "testing" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +func listIDs(list []provider.Instance) []string { + ids := make([]string, len(list)) + for i, inst := range list { + ids[i] = inst.ID + } + return ids +} + +func (fx *fixture) list() []provider.Instance { + fx.t.Helper() + list, err := fx.p.ListByTag(fx.ctx) + if err != nil { + fx.t.Fatalf("ListByTag() error = %v", err) + } + return list +} + +func TestListByTag_excludesUnmarkedVMs(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running") + fx.fake.addVM(150100, "proxy-nomark0000000000", "running") + fx.fake.addVM(150101, "proxy-managedfalse0000", "stopped").config["description"] = "proxy-operator-managed=false\nproxy-operator-uid=x" + fx.fake.addVM(150102, "tagged-by-hand", "running").config["tags"] = proxyTag + list := fx.list() + if got := listIDs(list); !slices.Equal(got, []string{fx.id()}) { + t.Fatalf("ListByTag = %v, want only %s", got, fx.id()) + } + if list[0].UID != testUID || !list[0].CreatedAt.Equal(testCreatedAt) || list[0].State != "" || list[0].IP != "" { + t.Errorf("listed = %+v, want uid and createdAt only", list[0]) + } + lines := fx.logs.matching(`"msg"="proxmox vms listed"`) + if len(lines) != 1 || !strings.Contains(lines[0], `"candidates"=4`) || !strings.Contains(lines[0], `"managed"=1`) || !strings.Contains(lines[0], `"skipped"=0`) { + t.Errorf("listed line = %v, want candidates 4, managed 1, skipped 0", lines) + } +} + +func TestListByTag_emptyUIDStillListed(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "stopped").config["description"] = "proxy-operator-managed=true" + list := fx.list() + if len(list) != 1 || list[0].UID != "" || list[0].ID != fx.id() { + t.Errorf("ListByTag = %+v, want the VM listed with an empty UID", list) + } +} + +func TestListByTag_candidatesByTagOrNamePrefix(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + // Named like a proxy, no tag: a candidate. + fx.addOurVM(150001, "proxy-notag00000000000", "running").config["tags"] = "" + // Tagged, renamed by an admin: a candidate. Tags as Proxmox lists them. + fx.addOurVM(150002, "renamed", "running").config["tags"] = "web;proxy_operator" + // Marked but neither tagged nor named like a proxy: never read. + fx.addOurVM(150003, "unrelated", "running").config["tags"] = "web" + got := listIDs(fx.list()) + want := []string{formatProviderID(150001, "proxy-notag00000000000"), formatProviderID(150002, "renamed")} + if !slices.Equal(got, want) { + t.Errorf("ListByTag = %v, want %v", got, want) + } + if n := fx.fake.callCount("VMConfig"); n != 2 { + t.Errorf("VMConfig calls = %d, want 2, one per candidate", n) + } +} + +func TestHasTag_table(t *testing.T) { + t.Parallel() + tests := []struct { + tags string + want bool + }{ + {tags: "proxy_operator", want: true}, + {tags: "a;proxy_operator;b", want: true}, + {tags: "a,proxy_operator", want: true}, + {tags: "a proxy_operator", want: true}, + {tags: "Proxy_Operator", want: true}, + {tags: ""}, + {tags: "proxy_operator2"}, + {tags: "proxy-operator"}, + } + for _, tc := range tests { + if got := hasTag(tc.tags, proxyTag); got != tc.want { + t.Errorf("hasTag(%q) = %v, want %v", tc.tags, got, tc.want) + } + } +} + +func TestListByTag_skipsLXCAndTemplates(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(150001, "proxy-container0000000", "running").kind = "lxc" + fx.addOurVM(150002, "proxy-template00000000", "stopped").template = true + if list := fx.list(); len(list) != 0 { + t.Errorf("ListByTag = %+v, want nothing", list) + } + if n := fx.fake.callCount("VMConfig"); n != 0 { + t.Errorf("VMConfig calls = %d, want 0", n) + } +} + +func TestListByTag_midCloneEntryIsNotACandidate(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.addTemplate(9001, "debian-13-template") + if _, err := fx.fake.Clone(fx.ctx, fakeNode, 9001, cloneParams{NewID: testVMID, Name: testName, Description: formatDescription(testUID, testCreatedAt)}); err != nil { + t.Fatalf("Clone error = %v", err) + } + if list := fx.list(); len(list) != 0 { + t.Errorf("ListByTag = %+v, want nothing while the clone runs", list) + } +} + +func TestListByTag_resourcesFailureFailsWhole(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running") + fx.fake.failWith("ClusterResources", fakeErr(595, "Connection refused")) + list, err := fx.p.ListByTag(fx.ctx) + if list != nil { + t.Errorf("ListByTag = %+v, want nil", list) + } + assertProviderError(t, err, "list", provider.ErrTransient) +} + +func TestListByTag_unreadableVMSkipped(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(150001, "proxy-destroyedmidswee", "running") + fx.addOurVM(testVMID, testName, "running") + // The first candidate is destroyed between the listing and its config + // read. + mutateOn(fx.fake, "VMConfig", func() { delete(fx.fake.vms, 150001) }) + if got := listIDs(fx.list()); !slices.Equal(got, []string{fx.id()}) { + t.Errorf("ListByTag = %v, want only %s", got, fx.id()) + } + lines := fx.logs.matching(`"msg"="skipping unreadable proxmox vm"`) + if len(lines) != 1 || !strings.Contains(lines[0], `"vmid"=150001`) || !strings.Contains(lines[0], "does not exist") { + t.Errorf("skip lines = %v, want one naming 150001 and the error", lines) + } + if lines := fx.logs.matching(`"skipped"=1`); len(lines) != 1 { + t.Errorf("listed line with skipped 1 = %v", lines) + } +} + +func TestListByTag_unreadableVMWithEndedContextFails(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running") + ctx, cancel := context.WithCancel(fx.ctx) + defer cancel() + fx.fake.hookOn("VMConfig", func() error { + cancel() + return context.Canceled + }) + list, err := fx.p.ListByTag(ctx) + if list != nil { + t.Errorf("ListByTag = %+v, want nil", list) + } + assertProviderError(t, err, "list", provider.ErrTransient) + if !errors.Is(err, context.Canceled) { + t.Errorf("error = %v, want it to wrap context.Canceled", err) + } +} + +func TestListByTag_unknownStatusStillAttempted(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").status = "unknown" + if got := listIDs(fx.list()); !slices.Equal(got, []string{fx.id()}) { + t.Errorf("ListByTag = %v, want %s despite status unknown", got, fx.id()) + } +} + +func TestListByTag_doesNotReadStatusOrAgent(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running").ifaces = capturedIfaces(t) + fx.addOurVM(150001, "proxy-second0000000000", "stopped") + fx.list() + want := []string{"ClusterResources", "VMConfig", "VMConfig"} + if got := fx.fake.callLog(); !slices.Equal(got, want) { + t.Errorf("calls = %v, want %v", got, want) + } +} + +func TestListByTag_idsAcceptedByParse(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "running") + fx.addOurVM(999999999, "proxy-topofrange000000", "stopped") + fx.addOurVM(100, "proxy-bottomofrange000", "stopped") + list := fx.list() + if len(list) != 3 { + t.Fatalf("ListByTag = %+v, want 3", list) + } + for _, inst := range list { + vmid, name, err := parseProviderID(inst.ID) + if err != nil || formatProviderID(vmid, name) != inst.ID { + t.Errorf("parseProviderID(%q) = %d, %q, %v; want a round trip", inst.ID, vmid, name, err) + } + } +} + +func TestListByTag_quietAtV0(t *testing.T) { + t.Parallel() + fx := newFixture(t, 0) + fx.addOurVM(testVMID, testName, "running") + fx.addOurVM(150001, "proxy-destroyedmidswee", "running") + mutateOn(fx.fake, "VMConfig", func() { delete(fx.fake.vms, 150001) }) + fx.list() + if lines := fx.logs.all(); len(lines) != 0 { + t.Errorf("V(0) lines = %v, want none", lines) + } +} diff --git a/internal/provider/proxmox/provider_test.go b/internal/provider/proxmox/provider_test.go new file mode 100644 index 0000000..3e9dc4b --- /dev/null +++ b/internal/provider/proxmox/provider_test.go @@ -0,0 +1,103 @@ +package proxmox + +import ( + "context" + "errors" + "testing" + "time" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +const ( + testProvider = "proxmox-home" + testUID = "0b9c6f3e-2d4a-4e1b-9c7f-5a8d3e2f1b60" + testVMID = 150042 +) + +var ( + testName = provider.NameFromUID(testUID) + testCreatedAt = time.Date(2026, 9, 23, 10, 0, 0, 0, time.UTC) +) + +// fixture is a Provider over fakeAPI, logging into a capture, with every +// budget shrunk so waits take milliseconds. +type fixture struct { + t *testing.T + p *Provider + fake *fakeAPI + logs *logCapture + ctx context.Context +} + +func newFixture(t *testing.T, verbosity int) *fixture { + t.Helper() + fx := &fixture{t: t, fake: newFakeAPI(), logs: &logCapture{}} + fx.p = newProvider(testProvider, testConfig(), fx.fake) + fx.p.budgets = testBudgets() + fx.ctx = fx.logs.context(verbosity) + return fx +} + +func testBudgets() budgets { + return budgets{ + clone: time.Second, + upload: time.Second, + resize: time.Second, + start: time.Second, + stop: time.Second, + destroy: time.Second, + seedDelete: time.Second, + running: time.Second, + poll: time.Millisecond, + } +} + +// addOurVM places a VM carrying the mark for testUID and the tag, as +// Create leaves it. +func (fx *fixture) addOurVM(vmid int, name, status string) *fakeVM { + fx.t.Helper() + vm := fx.fake.addVM(vmid, name, status) + vm.config["description"] = formatDescription(testUID, testCreatedAt) + vm.config["tags"] = proxyTag + return vm +} + +// id is the providerID of the VM addOurVM(testVMID, testName, …) places. +func (fx *fixture) id() string { return formatProviderID(testVMID, testName) } + +func (fx *fixture) get(id string) *provider.Instance { + fx.t.Helper() + inst, err := fx.p.Get(fx.ctx, id) + if err != nil { + fx.t.Fatalf("Get(%s) error = %v", id, err) + } + return inst +} + +// assertProviderError checks that err is a *provider.Error from this +// provider for op, in class and in no other class. +func assertProviderError(t *testing.T, err error, op string, class error) { + t.Helper() + if err == nil { + t.Fatalf("error = nil, want op %s class %v", op, class) + } + var pe *provider.Error + if !errors.As(err, &pe) { + t.Fatalf("error = %T (%v), want *provider.Error", err, err) + } + if pe.Provider != testProvider { + t.Errorf("error.Provider = %q, want %q", pe.Provider, testProvider) + } + if pe.Op != op { + t.Errorf("error.Op = %q, want %q", pe.Op, op) + } + if !errors.Is(err, class) { + t.Errorf("error = %v, want class %v", err, class) + } + for _, other := range []error{provider.ErrNotFound, provider.ErrQuotaExceeded, provider.ErrPermanent, provider.ErrTransient} { + if other != class && errors.Is(err, other) { + t.Errorf("error = %v, also matches %v", err, other) + } + } +} diff --git a/internal/provider/proxmox/proxmox.go b/internal/provider/proxmox/proxmox.go index 66fe882..1be34e8 100644 --- a/internal/provider/proxmox/proxmox.go +++ b/internal/provider/proxmox/proxmox.go @@ -173,22 +173,14 @@ func (p *Provider) lockName(ctx context.Context, name string) (func(), error) { return release, nil } -// The four contract methods are stubs until the phases that implement them -// (Get and ListByTag in Phase 5, Create in 6, Delete in 7). Each answers -// Transient, so a Proxy on this provider retries rather than latching. +// Create and Delete are stubs until the phases that implement them (Create +// in 6, Delete in 7). Each answers Transient, so a Proxy on this provider +// retries rather than latching. func (p *Provider) Create(_ context.Context, req provider.CreateRequest) (string, error) { return "", p.wrapErr("create", "", transientf("proxmox Create is not implemented yet (request %s)", req.Name)) } -func (p *Provider) Get(_ context.Context, id string) (*provider.Instance, error) { - return nil, p.wrapErr("get", id, transientf("proxmox Get is not implemented yet")) -} - func (p *Provider) Delete(_ context.Context, id string) error { return p.wrapErr("delete", id, transientf("proxmox Delete is not implemented yet")) } - -func (p *Provider) ListByTag(_ context.Context) ([]provider.Instance, error) { - return nil, p.wrapErr("list", "", transientf("proxmox ListByTag is not implemented yet")) -} diff --git a/internal/provider/proxmox/proxmox_test.go b/internal/provider/proxmox/proxmox_test.go index 2add898..42d540e 100644 --- a/internal/provider/proxmox/proxmox_test.go +++ b/internal/provider/proxmox/proxmox_test.go @@ -216,16 +216,14 @@ func TestWrapErr_carriesProviderNameOpAndID(t *testing.T) { } // TestStubs_answerTransient keeps a Proxy on this provider retrying, not -// latched, until the methods are implemented in later phases. +// latched, until Create and Delete are implemented in later phases. func TestStubs_answerTransient(t *testing.T) { t.Parallel() ctx := context.Background() p := newProvider("proxmox-home", testConfig(), nil) _, createErr := p.Create(ctx, provider.CreateRequest{Name: "proxy-x"}) - _, getErr := p.Get(ctx, "vms/150042/proxy-x") deleteErr := p.Delete(ctx, "vms/150042/proxy-x") - _, listErr := p.ListByTag(ctx) - for op, err := range map[string]error{"create": createErr, "get": getErr, "delete": deleteErr, "list": listErr} { + for op, err := range map[string]error{"create": createErr, "delete": deleteErr} { if !errors.Is(err, provider.ErrTransient) { t.Errorf("%s error = %v, want ErrTransient", op, err) } diff --git a/internal/provider/proxmox/task.go b/internal/provider/proxmox/task.go new file mode 100644 index 0000000..3ebe972 --- /dev/null +++ b/internal/provider/proxmox/task.go @@ -0,0 +1,109 @@ +package proxmox + +import ( + "context" + "fmt" + "regexp" + "time" + + "github.com/go-logr/logr" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// upidPattern is Proxmox's task id: UPID::::: +// ::: with hex pid, pstart and starttime. +var upidPattern = regexp.MustCompile(`^UPID:([a-zA-Z0-9]([a-zA-Z0-9-]*[a-zA-Z0-9])?):[0-9A-Fa-f]{8}:[0-9A-Fa-f]{8,9}:[0-9A-Fa-f]{8}:[^:\s]+:[^:\s]*:[^:\s]+:$`) + +// parseUPID returns the node a task runs on, which is where its status is +// read (spec entry 10). A malformed UPID is Permanent: Proxmox never sends +// one, so retrying cannot help. +func parseUPID(upid string) (string, error) { + m := upidPattern.FindStringSubmatch(upid) + if m == nil { + return "", permanentf("malformed task id %q", upid) + } + return m[1], nil +} + +// taskSucceeded reads a stopped task's exitstatus the way Proxmox does: +// "OK", or "WARNINGS: " for a task that finished but logged warnings. +func taskSucceeded(exit string) bool { + return exit == "OK" || taskWarnings.MatchString(exit) +} + +var taskWarnings = regexp.MustCompile(`^WARNINGS: [0-9]+$`) + +// pollUntil calls check every interval until it reports done or fails. Once +// budget has passed or ctx has ended it gives up with ErrTransient naming +// what it waited for; the backend keeps working, and the next attempt picks +// the result up. +func pollUntil(ctx context.Context, interval, budget time.Duration, what string, check func() (bool, error)) error { + deadline := time.NewTimer(budget) + defer deadline.Stop() + tick := time.NewTicker(interval) + defer tick.Stop() + for { + done, err := check() + if err != nil || done { + return err + } + select { + case <-ctx.Done(): + return fmt.Errorf("%w: waiting for %s: %w", provider.ErrTransient, what, ctx.Err()) + case <-deadline.C: + return transientf("waiting for %s: not done after %v", what, budget) + case <-tick.C: + } + } +} + +// waitTask polls a task on the node its UPID names until it stops. A +// failed task is ErrTransient carrying its exitstatus, and logged at V(1) +// (spec entries 10 and 20). +func (p *Provider) waitTask(ctx context.Context, log logr.Logger, upid string, budget time.Duration) error { + node, err := parseUPID(upid) + if err != nil { + return err + } + var exit string + err = pollUntil(ctx, p.budgets.poll, budget, "task "+upid, func() (bool, error) { + s, err := p.api.TaskStatus(ctx, node, upid) + if err != nil { + return false, fmt.Errorf("reading task %s: %w", upid, err) + } + exit = s.ExitStatus + return s.Status == "stopped", nil + }) + if err != nil { + return err + } + if !taskSucceeded(exit) { + log.V(1).Info("proxmox task failed", "upid", upid, "exitstatus", exit) + return transientf("proxmox task %s failed: %s", upid, exit) + } + if exit != "OK" { + log.V(1).Info("proxmox task finished with warnings", "upid", upid, "exitstatus", exit) + } + return nil +} + +// waitUnlocked polls a VM's config until no task holds its lock and returns +// that config, for Create's ownership check (spec entry 2). A VM that +// disappears meanwhile, as a failed clone does, returns the does-not-exist +// error for the caller to judge. +func (p *Provider) waitUnlocked(ctx context.Context, node string, vmid int, budget time.Duration) (vmConfig, error) { + var cfg vmConfig + err := pollUntil(ctx, p.budgets.poll, budget, fmt.Sprintf("the lock on VM %d", vmid), func() (bool, error) { + c, err := p.api.VMConfig(ctx, node, vmid) + if err != nil { + return false, fmt.Errorf("reading config of VM %d: %w", vmid, err) + } + cfg = c + return c["lock"] == "", nil + }) + if err != nil { + return nil, err + } + return cfg, nil +} diff --git a/internal/provider/proxmox/task_test.go b/internal/provider/proxmox/task_test.go new file mode 100644 index 0000000..f25d401 --- /dev/null +++ b/internal/provider/proxmox/task_test.go @@ -0,0 +1,233 @@ +package proxmox + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// addTask registers a fake task on node that ends with exit. +func addTask(f *fakeAPI, node, exit string) string { + f.mu.Lock() + defer f.mu.Unlock() + return f.newTask(node, "qmstart", "150042", exit, nil) +} + +func TestParseUPID_table(t *testing.T) { + t.Parallel() + tests := []struct { + name string + upid string + wantNode string + }{ + {name: "captured clone", upid: decodeData[string](t, "clone_response"), wantNode: "pv1-1"}, + {name: "captured task status", upid: decodeData[taskStatus](t, "task_status_stopped_ok").UPID, wantNode: "pv1-1"}, + {name: "captured upload", upid: decodeData[string](t, "upload_response_proxy_seeds"), wantNode: "pv1-1"}, + {name: "fake task on another node", upid: addTask(newFakeAPI(), "pv1-2", "OK"), wantNode: "pv1-2"}, + {name: "empty", upid: ""}, + {name: "prefix only", upid: "UPID:"}, + {name: "lower-case prefix", upid: "upid:pv1-1:0010F9DC:02801DA1:6AB37871:qmclone:9001:root@pam:"}, + {name: "no trailing colon", upid: "UPID:pv1-1:0010F9DC:02801DA1:6AB37871:qmclone:9001:root@pam"}, + {name: "slash in node", upid: "UPID:pv1/1:0010F9DC:02801DA1:6AB37871:qmclone:9001:root@pam:"}, + {name: "pid not hex", upid: "UPID:pv1-1:0010F9DZ:02801DA1:6AB37871:qmclone:9001:root@pam:"}, + {name: "missing user", upid: "UPID:pv1-1:0010F9DC:02801DA1:6AB37871:qmclone:9001::"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + node, err := parseUPID(tc.upid) + if tc.wantNode == "" { + if !errors.Is(err, provider.ErrPermanent) { + t.Errorf("parseUPID(%q) = %q, %v; want ErrPermanent", tc.upid, node, err) + } + return + } + if err != nil || node != tc.wantNode { + t.Errorf("parseUPID(%q) = %q, %v; want %q", tc.upid, node, err, tc.wantNode) + } + }) + } +} + +func TestWaitTask_table(t *testing.T) { + t.Parallel() + failedExit := loadTaskExit(t, "task_status_stopped_error") + tests := []struct { + name string + node string + exit string + upid string // overrides the registered task's UPID + polls int + budget time.Duration + cancelled bool + statusErr error + wantClass error // nil for success + wantCalls int + wantInErr string + wantLog string + }{ + {name: "ok on the first poll", exit: "OK", wantCalls: 1}, + {name: "polls until stopped", exit: "OK", polls: 3, wantCalls: 4}, + {name: "polls the node the UPID names", node: "pv1-2", exit: "OK", wantCalls: 1}, + {name: "warnings count as success", exit: "WARNINGS: 2", wantCalls: 1, wantLog: `"msg"="proxmox task finished with warnings"`}, + { + name: "failed exitstatus is transient and logged", exit: failedExit, wantClass: provider.ErrTransient, wantCalls: 1, + wantInErr: failedExit, wantLog: `"msg"="proxmox task failed"`, + }, + {name: "budget exceeded names the UPID", exit: "OK", polls: 1 << 30, budget: 20 * time.Millisecond, wantClass: provider.ErrTransient, wantCalls: -1, wantInErr: "UPID:pv1-1:"}, + {name: "ended ctx is transient", exit: "OK", polls: 1 << 30, cancelled: true, wantClass: provider.ErrTransient, wantCalls: 1, wantInErr: context.Canceled.Error()}, + {name: "malformed UPID makes no call", upid: "UPID:bogus", wantClass: provider.ErrPermanent, wantCalls: 0}, + {name: "status read failing is returned", exit: "OK", statusErr: loadAPIError(t, "err_forbidden"), wantClass: provider.ErrPermanent, wantCalls: 1, wantInErr: "Permission check failed"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.taskPolls = tc.polls + if tc.statusErr != nil { + fx.fake.failWith("TaskStatus", tc.statusErr) + } + node := tc.node + if node == "" { + node = fakeNode + } + upid := addTask(fx.fake, node, tc.exit) + if tc.upid != "" { + upid = tc.upid + } + budget := tc.budget + if budget == 0 { + budget = time.Minute + } + ctx := fx.ctx + if tc.cancelled { + var cancel context.CancelFunc + ctx, cancel = context.WithCancel(ctx) + cancel() + } + start := time.Now() + err := fx.p.waitTask(ctx, fx.p.logger(ctx), upid, budget) + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Errorf("waitTask took %v", elapsed) + } + if tc.wantClass == nil { + if err != nil { + t.Fatalf("waitTask error = %v, want nil", err) + } + } else if classify("create", err) != tc.wantClass { + t.Fatalf("waitTask error = %v, want class %v", err, tc.wantClass) + } + if tc.wantInErr != "" && !strings.Contains(err.Error(), tc.wantInErr) { + t.Errorf("waitTask error = %q, want it to contain %q", err, tc.wantInErr) + } + if got := fx.fake.callCount("TaskStatus"); tc.wantCalls >= 0 && got != tc.wantCalls { + t.Errorf("TaskStatus calls = %d, want %d", got, tc.wantCalls) + } + if tc.wantLog != "" { + lines := fx.logs.matching(tc.wantLog) + if len(lines) != 1 || !strings.Contains(lines[0], `"upid"="`+upid+`"`) || !strings.Contains(lines[0], `"exitstatus"="`+tc.exit+`"`) { + t.Errorf("log lines %q = %v, want one with upid and exitstatus", tc.wantLog, lines) + } + } + }) + } +} + +func TestWaitTask_quietAtV0(t *testing.T) { + t.Parallel() + fx := newFixture(t, 0) + err := fx.p.waitTask(fx.ctx, fx.p.logger(fx.ctx), addTask(fx.fake, fakeNode, "boom"), time.Minute) + if err == nil { + t.Fatal("waitTask on a failed task returned nil") + } + if lines := fx.logs.all(); len(lines) != 0 { + t.Errorf("V(0) lines = %v, want none", lines) + } +} + +func TestWaitUnlocked_table(t *testing.T) { + t.Parallel() + tests := []struct { + name string + lock string + // onCall runs on the n-th VMConfig call (1-based) under f.mu. + onCall func(f *fakeAPI, n int) + budget time.Duration + cancelled bool + wantErr func(error) bool + wantCalls int + }{ + {name: "unlocked answers at once", wantCalls: 1}, + { + name: "waits until the lock clears", lock: "clone", wantCalls: 3, + onCall: func(f *fakeAPI, n int) { + if n == 3 { + f.vms[testVMID].lock = "" + } + }, + }, + { + name: "lock that never clears is transient", lock: "clone", budget: 20 * time.Millisecond, wantCalls: -1, + wantErr: func(err error) bool { + return errors.Is(err, provider.ErrTransient) && strings.Contains(err.Error(), "the lock on VM 150042") + }, + }, + { + name: "VM removed by a failed clone gives does not exist", lock: "clone", wantCalls: 2, + onCall: func(f *fakeAPI, n int) { + if n == 2 { + delete(f.vms, testVMID) + } + }, + wantErr: isNotExist, + }, + { + name: "ended ctx is transient", lock: "clone", cancelled: true, wantCalls: 1, + wantErr: func(err error) bool { + return errors.Is(err, provider.ErrTransient) && errors.Is(err, context.Canceled) + }, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.addOurVM(testVMID, testName, "stopped").lock = tc.lock + var n int + fx.fake.hookOn("VMConfig", func() error { + fx.fake.mu.Lock() + defer fx.fake.mu.Unlock() + n++ + if tc.onCall != nil { + tc.onCall(fx.fake, n) + } + return nil + }) + budget := tc.budget + if budget == 0 { + budget = time.Minute + } + ctx := fx.ctx + if tc.cancelled { + var cancel context.CancelFunc + ctx, cancel = context.WithCancel(ctx) + cancel() + } + cfg, err := fx.p.waitUnlocked(ctx, fakeNode, testVMID, budget) + if tc.wantErr == nil { + if err != nil || cfg["name"] != testName || cfg["lock"] != "" { + t.Errorf("waitUnlocked = %v, %v; want the unlocked config", cfg, err) + } + } else if !tc.wantErr(err) || cfg != nil { + t.Errorf("waitUnlocked = %v, %v; want no config and the expected error", cfg, err) + } + if got := fx.fake.callCount("VMConfig"); tc.wantCalls >= 0 && got != tc.wantCalls { + t.Errorf("VMConfig calls = %d, want %d", got, tc.wantCalls) + } + }) + } +}