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 3c5e57e..ca16638 100644 --- a/docs/plans-executions/2026-09-18-1502-proxmox-provider.md +++ b/docs/plans-executions/2026-09-18-1502-proxmox-provider.md @@ -8,7 +8,7 @@ Pairs with [docs/plans/2026-09-18-1502-proxmox-provider.md](../plans/2026-09-18- - [x] Phase 1 — The spec (`docs/proxmox-provider-spec.md`) - [x] Phase 2 — Config block, registration, package skeleton - [x] Phase 3 — Extract `nocloud` and `keyedlock` from kvm2 -- [ ] Phase 4 — Seam, HTTP client, decorator, fake +- [x] Phase 4 — Seam, HTTP client, decorator, fake - [ ] Phase 5 — Get, ListByTag, task helpers - [ ] Phase 6 — Create - [ ] Phase 7 — Delete, Checker, CRD comment @@ -447,3 +447,110 @@ Worth noting: test starts its 200 ms release timers before `Check` starts its clock, so under load the measured latency can come in just under 200 ms. The flake predates this phase, and the test is not touched here. + +## Phase 4 — Seam, HTTP client, decorator, fake + +`client.go` holds `httpAPI`, the thin `net/http` adapter for all sixteen +seam methods. It reads the token file on every request, turns any non-2xx +answer into `*apiError` from the body's `message` or the reason phrase +(F1), and wraps its transport in `tracing.RestConfigWrapper`. `observe.go` +is the logging decorator: one V(1) `proxmox api call` line per call, +detail at V(2), and no span, because the transport already traces each +request. `New` now builds `observe(newHTTPAPI(...))` instead of passing a +nil `api`, still without touching the network or the token file. +`fake_test.go` holds `fakeAPI`, which Phases 5 to 7 build on. + +Where this phase departs from the plan's Phase 4 text: + +- **The `http.Client` is built on first use, not in `newHTTPAPI`.** Spec + entry 14 wins over the plan: the CA is loaded lazily, a missing CA file + is Transient and retried until the file lands, one without a certificate + is Permanent, and the pool is cached only once it loads. A transport's + `RootCAs` cannot change after first use, so the whole client waits for + the CA (`TestClient_caFileLoadedLazily`). +- **The upload is not an `io.Pipe` with a goroutine.** The multipart + framing (the `content=iso` field, the `filename` part header and the + closing boundary) is rendered into two small buffers, and the body is + `io.MultiReader(head, io.LimitReader(r, size), tail)` with an exact + `Content-Length`. The seed is still streamed and never held by the + client. There is also no goroutine to leak, and the request carries a + length, as Phase 0's `curl -F` upload did. A chunked upload to pveproxy + was never tried. `TestClient_uploadStreamsMultipart` proves the + streaming with a reader that stalls after its first 256 KiB until the + server has seen file bytes. +- **The fake's `DeleteVolume` tolerates a missing volume**, as spec + correction 5 and F13 say, instead of the plan's error. +- **Added:** secret redaction in `apiError` (contract §7), because + otherwise a server that echoes the header would put the secret into + `Proxy` status; an `errCredentials` sentinel on token-file errors, so + Phase 7's Checker can report `CredentialsUnavailable`; and the V(1) + identifiers `newid` and `name` on a clone, `filename` on an upload and + `volid` on a volume delete. Spec entry 20 now lists them. + +**The volid's `/` travels as `%2F`.** Neither the plan nor the spec says +how `{volid}` is encoded in the content-delete path. The Phase 0 session's +transcript shows every delete went through a helper encoding the whole +volid: + +```bash +enc() { python3 -c "import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1], safe=''))" "$1"; } +curl ... -X DELETE ".../storage/proxy-seeds/content/$(enc 'proxy-seeds:iso/proxy-reconaaaaaaaaaaa-seed.iso')" +``` + +So `apiPath` runs `url.PathEscape` over each segment +(`TestClient_volIDSlashIsEscaped`). + +`net/http` cannot send a custom reason phrase, so the client tests replay +every capture byte for byte through a hijacked connection (`serveCapture`). +`TestClient_apiErrorFromStatusReasonAndBody` globs `testdata/err_*.json` +and skips `err_tag_hyphen.json`, which is a 200 despite its name: hyphenated +tags were accepted (F9). + +Tests beyond the plan's list: `TestClient_methodPathAndParams` (the method, +path, query and form of every seam method, with a reflection check that no +method lacks a row), `TestClient_cloneOmitsEmptyStorageAndTarget`, +`TestClient_volIDSlashIsEscaped`, `TestClient_caFileLoadedLazily`, +`TestClient_x509FailureIsPermanent` (from spec entry 21), +`TestDecodeConfig_scalarsAsStrings`, `TestReadToken_trimsTrailingNewline`, +`TestObserve_startsNoSpan`, `TestObserve_identifiersAtV1`, +`TestNew_wiresObservedHTTPClient`, and six `TestFakeAPI_*` tests. +`TestFakeAPI_errorsMatchCaptures` compares the fake's error texts with the +captures, so every test that goes through the fake exercises the real +wording. Texts that recon never produced start with `fake:`. + +A mutation check confirmed the new tests can fail: dropping `redact`, +buffering the upload with `io.ReadAll`, and logging V(2) detail at V(1) +each failed their test. Removing only the empty `TLSNextProto` map did +**not** fail `TestClient_http1Only`. A transport with a custom +`TLSClientConfig` and `DialContext` never offers h2 unless +`ForceAttemptHTTP2` is set, so the map is a second guard. With both +guards removed the test failed with `client spoke HTTP/2`. Every mutation +was reverted, and the files were compared identical to backups. + +```bash +go test -race -count=2 ./internal/provider/proxmox/ # ok, run three times +grep -n 'ReadAll' internal/provider/proxmox/client.go # only the 64 KiB error-body read +grep -n 'context.Background()' internal/provider/proxmox/*.go | grep -v _test.go # nothing +go build ./... && go vet ./... && go vet -tags=integration ./... +go test -short -count=1 ./... # all ok +go test -count=1 -v ./internal/provider/proxmox/ | grep -c '^=== RUN' # 161 before, 289 after +``` + +Worth noting: + +- `bin/golangci-lint run ./internal/provider/proxmox/...` panics inside + staticcheck's IR builder (`unexpected expr: *ast.KeyValueExpr` in the + Go 1.27 standard library's `poll` package). This is a toolchain problem, + not this code. With `--disable staticcheck` the new files' findings are + goconst on form keys (kvm2 carries 50 of the same) and two unparam hits + on the fake's `addTemplate` and `resourceByID`, which Phases 5 and 6 will + call with other VMIDs. The mapsloop and gocyclo findings in `proxmox.go` + and `api_test.go` date from Phase 2. +- Nothing in this phase talked to the host. The client has only met + `httptest` servers replaying the captures. The first real request is + Phase 10's integration run, which is where the `Content-Length` upload + and the `%2F` volid get proven against pveproxy. +- The fake refuses to destroy a running VM with an uncaptured + `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. diff --git a/docs/proxmox-provider-spec.md b/docs/proxmox-provider-spec.md index 8b2bd2f..9e1dda4 100644 --- a/docs/proxmox-provider-spec.md +++ b/docs/proxmox-provider-spec.md @@ -696,8 +696,9 @@ named `proxmox`, with `provider` on every line and `vmid`, `name` and `proxmox vm fetched` with state, ip and uid, `proxmox vm deleted`, `proxmox vms listed` with counts, `proxmox node checked` with the reason), one `proxmox api call` line per seam call from the decorator (`call`, - `node`, `vmid`, `storage`, `upid`, `duration`, and on failure - `httpStatus` and `error`), a `proxmox task failed` line with the UPID and + `node`, `vmid`, `storage`, `upid`, `duration`, plus `newid` and `name` + on a clone, `filename` on an upload and `volid` on a volume delete, and + on failure `httpStatus` and `error`), a `proxmox task failed` line with the UPID and `exitstatus`, and the no-op lines of Delete and ListByTag (foreign VM left alone, unreadable candidate skipped). - V(2): request parameters (without `description`'s content at V(1) and diff --git a/internal/provider/proxmox/client.go b/internal/provider/proxmox/client.go new file mode 100644 index 0000000..c1df283 --- /dev/null +++ b/internal/provider/proxmox/client.go @@ -0,0 +1,378 @@ +package proxmox + +import ( + "bytes" + "context" + "crypto/tls" + "crypto/x509" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + "mime/multipart" + "net" + "net/http" + "net/url" + "os" + "regexp" + "strconv" + "strings" + "sync" + "time" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" +) + +const ( + dialTimeout = 10 * time.Second + tlsHandshakeTimeout = 10 * time.Second + idleConnTimeout = 90 * time.Second + maxIdleConnsPerHost = 4 + // maxErrorBody caps how much of a failed response is read for its + // message; Proxmox error bodies are a few hundred bytes. + maxErrorBody = 64 << 10 +) + +// errCredentials marks a token file that is missing or unusable, so the +// Checker can report CredentialsUnavailable instead of an API failure. +var errCredentials = errors.New("proxmox token file unusable") + +// tokenPattern is @!= (spec entry 14). +var tokenPattern = regexp.MustCompile(`^[^@\s]+@[^!\s]+![^=\s]+=\S+$`) + +// httpAPI is the production adapter behind the api seam: a thin net/http +// client for /api2/json (spec entry 17). It is safe for concurrent use. +type httpAPI struct { + base *url.URL + tokenFile string + caFile string + insecure bool + // timeout bounds one request, response included. Read when the client + // is built; tests shrink it before the first call. + timeout time.Duration + wrap func(http.RoundTripper) http.RoundTripper + + // mu guards client, built on first use because the CA is loaded lazily + // and only cached once it loads (spec entry 14). + mu sync.Mutex + client *http.Client +} + +var _ api = (*httpAPI)(nil) + +// newHTTPAPI builds the adapter without touching the network or any file. +func newHTTPAPI(cfg provider.ProxmoxConfig, opts ...tracing.Option) (*httpAPI, error) { + u, err := url.Parse(cfg.URL) + if err != nil { + return nil, fmt.Errorf("proxmox.url: %w", err) + } + return &httpAPI{ + base: u.JoinPath("api2", "json"), + tokenFile: cfg.TokenFile, + caFile: cfg.CAFile, + insecure: cfg.InsecureSkipTLSVerify, + timeout: time.Duration(cfg.RequestTimeoutSeconds) * time.Second, + wrap: tracing.RestConfigWrapper(opts...), + }, nil +} + +// httpClient returns the shared client, building it once the CA loads. A +// missing CA file is Transient and retried on the next call; one without a +// certificate is Permanent. +func (a *httpAPI) httpClient() (*http.Client, error) { + a.mu.Lock() + defer a.mu.Unlock() + if a.client != nil { + return a.client, nil + } + tlsCfg := &tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: a.insecure} //nolint:gosec // opt-in, warned about at startup + if a.caFile != "" && !a.insecure { + pem, err := os.ReadFile(a.caFile) + if err != nil { + return nil, fmt.Errorf("%w: reading CA file: %w", provider.ErrTransient, err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(pem) { + return nil, permanentf("CA file %s holds no PEM certificate", a.caFile) + } + tlsCfg.RootCAs = pool + } + transport := &http.Transport{ + Proxy: http.ProxyFromEnvironment, + DialContext: (&net.Dialer{Timeout: dialTimeout, KeepAlive: 30 * time.Second}).DialContext, + TLSClientConfig: tlsCfg, + TLSHandshakeTimeout: tlsHandshakeTimeout, + ResponseHeaderTimeout: a.timeout, + IdleConnTimeout: idleConnTimeout, + MaxIdleConnsPerHost: maxIdleConnsPerHost, + // pveproxy speaks HTTP/1.1 only (F18); a non-nil empty map keeps + // the transport from offering h2. + ForceAttemptHTTP2: false, + TLSNextProto: map[string]func(string, *tls.Conn) http.RoundTripper{}, + } + a.client = &http.Client{Transport: a.wrap(transport), Timeout: a.timeout} + return a.client, nil +} + +// readToken reads the token file on every request, so a rotated Secret +// takes effect at once. Neither error names the file's content. +func readToken(path string) (string, error) { + raw, err := os.ReadFile(path) + if errors.Is(err, fs.ErrNotExist) { + return "", fmt.Errorf("%w: %w: %s does not exist yet", provider.ErrTransient, errCredentials, path) + } + if err != nil { + return "", fmt.Errorf("%w: reading %s: %w", errCredentials, path, err) + } + token := strings.TrimSpace(string(raw)) + if !tokenPattern.MatchString(token) { + return "", fmt.Errorf("%w: %w: %s is not @!=", provider.ErrPermanent, errCredentials, path) + } + return token, nil +} + +// apiPath joins path segments, escaping each, so a volid's "/" travels as +// %2F as Phase 0 sent it. +func apiPath(segments ...string) string { + escaped := make([]string, len(segments)) + for i, s := range segments { + escaped[i] = url.PathEscape(s) + } + return strings.Join(escaped, "/") +} + +func vmPath(node string, vmid int, rest ...string) string { + return apiPath(append([]string{"nodes", node, "qemu", strconv.Itoa(vmid)}, rest...)...) +} + +// do sends one request and decodes the response's data member into out, +// which may be nil. POST and PUT send form as the body; every method sends +// query in the URL. Any non-2xx answer is an *apiError. +func (a *httpAPI) do(ctx context.Context, method, path string, query, form url.Values, out any) error { + var body io.Reader + contentType := "" + if form != nil { + body = strings.NewReader(form.Encode()) + contentType = "application/x-www-form-urlencoded" + } + return a.send(ctx, method, path, query, body, -1, contentType, out) +} + +// send is do with a caller-built body. length is the body's size, or -1 +// when do built it from a string the request can measure itself. +func (a *httpAPI) send(ctx context.Context, method, path string, query url.Values, body io.Reader, length int64, contentType string, out any) error { + token, err := readToken(a.tokenFile) + if err != nil { + return err + } + client, err := a.httpClient() + if err != nil { + return err + } + u := a.base.JoinPath(path) + u.RawQuery = query.Encode() + req, err := http.NewRequestWithContext(ctx, method, u.String(), body) + if err != nil { + return fmt.Errorf("building %s %s: %w", method, path, err) + } + if length >= 0 { + req.ContentLength = length + } + if contentType != "" { + req.Header.Set("Content-Type", contentType) + } + req.Header.Set("Accept", "application/json") + req.Header.Set("Authorization", "PVEAPIToken="+token) + resp, err := client.Do(req) + if err != nil { + return err + } + defer func() { + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxErrorBody)) + _ = resp.Body.Close() + }() + if resp.StatusCode < 200 || resp.StatusCode > 299 { + raw, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBody)) + reason := strings.TrimPrefix(resp.Status, strconv.Itoa(resp.StatusCode)) + return redact(newAPIError(resp.StatusCode, reason, raw), token) + } + var env struct { + Data json.RawMessage `json:"data"` + } + if err := json.NewDecoder(resp.Body).Decode(&env); err != nil { + return fmt.Errorf("decoding %s %s: %w", method, path, err) + } + if out == nil || len(env.Data) == 0 { + return nil + } + if err := json.Unmarshal(env.Data, out); err != nil { + return fmt.Errorf("decoding %s %s data: %w", method, path, err) + } + return nil +} + +// redact removes the token's secret from an error Proxmox echoed back, so +// it cannot reach a log line, a span or a Proxy's status (spec entry 14). +func redact(e *apiError, token string) *apiError { + _, secret, _ := strings.Cut(token, "=") + if secret == "" { + return e + } + e.Message = strings.ReplaceAll(e.Message, secret, "") + for k, v := range e.Errors { + e.Errors[k] = strings.ReplaceAll(v, secret, "") + } + return e +} + +// decodeConfig renders every config value as a string: Proxmox sends some +// scalars as JSON numbers (cores) and others as strings (memory). A null +// value is dropped. +func decodeConfig(raw map[string]json.RawMessage) (vmConfig, error) { + cfg := make(vmConfig, len(raw)) + for k, v := range raw { + switch { + case len(v) == 0 || string(v) == "null": + continue + case v[0] == '"': + var s string + if err := json.Unmarshal(v, &s); err != nil { + return nil, fmt.Errorf("config key %s: %w", k, err) + } + cfg[k] = s + default: + cfg[k] = string(v) + } + } + return cfg, nil +} + +func (a *httpAPI) Version(ctx context.Context) (v versionInfo, err error) { + err = a.do(ctx, http.MethodGet, "version", nil, nil, &v) + return v, err +} + +func (a *httpAPI) NodeStatus(ctx context.Context, node string) (s nodeStatus, err error) { + err = a.do(ctx, http.MethodGet, apiPath("nodes", node, "status"), nil, nil, &s) + return s, err +} + +func (a *httpAPI) StorageStatus(ctx context.Context, node, storage string) (s storageStatus, err error) { + err = a.do(ctx, http.MethodGet, apiPath("nodes", node, "storage", storage, "status"), nil, nil, &s) + return s, err +} + +func (a *httpAPI) ClusterResources(ctx context.Context) (rs []vmResource, err error) { + err = a.do(ctx, http.MethodGet, "cluster/resources", url.Values{"type": {"vm"}}, nil, &rs) + return rs, err +} + +func (a *httpAPI) VMConfig(ctx context.Context, node string, vmid int) (vmConfig, error) { + var raw map[string]json.RawMessage + if err := a.do(ctx, http.MethodGet, vmPath(node, vmid, "config"), nil, nil, &raw); err != nil { + return nil, err + } + return decodeConfig(raw) +} + +func (a *httpAPI) SetVMConfig(ctx context.Context, node string, vmid int, params map[string]string) error { + form := url.Values{} + for k, v := range params { + form.Set(k, v) + } + return a.do(ctx, http.MethodPut, vmPath(node, vmid, "config"), nil, form, nil) +} + +func (a *httpAPI) Clone(ctx context.Context, node string, vmid int, p cloneParams) (upid string, err error) { + form := url.Values{ + "newid": {strconv.Itoa(p.NewID)}, + "name": {p.Name}, + "description": {p.Description}, + "full": {"1"}, + } + if p.Storage != "" { + form.Set("storage", p.Storage) + } + if p.Target != "" { + form.Set("target", p.Target) + } + err = a.do(ctx, http.MethodPost, vmPath(node, vmid, "clone"), nil, form, &upid) + return upid, err +} + +func (a *httpAPI) Resize(ctx context.Context, node string, vmid int, disk, size string) (upid string, err error) { + form := url.Values{"disk": {disk}, "size": {size}} + err = a.do(ctx, http.MethodPut, vmPath(node, vmid, "resize"), nil, form, &upid) + return upid, err +} + +func (a *httpAPI) Start(ctx context.Context, node string, vmid int) (upid string, err error) { + err = a.do(ctx, http.MethodPost, vmPath(node, vmid, "status", "start"), nil, url.Values{}, &upid) + return upid, err +} + +func (a *httpAPI) Stop(ctx context.Context, node string, vmid int) (upid string, err error) { + err = a.do(ctx, http.MethodPost, vmPath(node, vmid, "status", "stop"), nil, url.Values{}, &upid) + return upid, err +} + +func (a *httpAPI) VMStatus(ctx context.Context, node string, vmid int) (s vmStatus, err error) { + err = a.do(ctx, http.MethodGet, vmPath(node, vmid, "status", "current"), nil, nil, &s) + return s, err +} + +// Destroy purges the VM from backup jobs and HA and removes disks no config +// references, so nothing of it is left behind (spec entry 11). +func (a *httpAPI) Destroy(ctx context.Context, node string, vmid int) (upid string, err error) { + query := url.Values{"purge": {"1"}, "destroy-unreferenced-disks": {"1"}} + err = a.do(ctx, http.MethodDelete, vmPath(node, vmid), query, nil, &upid) + return upid, err +} + +func (a *httpAPI) TaskStatus(ctx context.Context, node, upid string) (s taskStatus, err error) { + err = a.do(ctx, http.MethodGet, apiPath("nodes", node, "tasks", upid, "status"), nil, nil, &s) + return s, err +} + +func (a *httpAPI) AgentNetworkInterfaces(ctx context.Context, node string, vmid int) ([]agentInterface, error) { + var out struct { + Result []agentInterface `json:"result"` + } + err := a.do(ctx, http.MethodGet, vmPath(node, vmid, "agent", "network-get-interfaces"), nil, nil, &out) + return out.Result, err +} + +// UploadISO streams r as the file part of a multipart upload (F12). The +// multipart framing is rendered up front, so the request carries an exact +// Content-Length, as Phase 0's curl -F upload did, while r itself is read +// in the transport's chunks and never held in memory. +func (a *httpAPI) UploadISO(ctx context.Context, node, storage, filename string, r io.Reader, size int64) (upid string, err error) { + var frame bytes.Buffer + mw := multipart.NewWriter(&frame) + if err := mw.WriteField("content", "iso"); err != nil { + return "", fmt.Errorf("framing upload: %w", err) + } + if _, err := mw.CreateFormFile("filename", filename); err != nil { + return "", fmt.Errorf("framing upload: %w", err) + } + head := bytes.Clone(frame.Bytes()) + frame.Reset() + if err := mw.Close(); err != nil { + return "", fmt.Errorf("framing upload: %w", err) + } + tail := frame.Bytes() + if r == nil { + r = bytes.NewReader(nil) + } + body := io.MultiReader(bytes.NewReader(head), io.LimitReader(r, size), bytes.NewReader(tail)) + length := int64(len(head)) + size + int64(len(tail)) + err = a.send(ctx, http.MethodPost, apiPath("nodes", node, "storage", storage, "upload"), nil, body, length, mw.FormDataContentType(), &upid) + return upid, err +} + +func (a *httpAPI) DeleteVolume(ctx context.Context, node, storage, volid string) (upid string, err error) { + err = a.do(ctx, http.MethodDelete, apiPath("nodes", node, "storage", storage, "content", volid), nil, nil, &upid) + return upid, err +} diff --git a/internal/provider/proxmox/client_test.go b/internal/provider/proxmox/client_test.go new file mode 100644 index 0000000..3295047 --- /dev/null +++ b/internal/provider/proxmox/client_test.go @@ -0,0 +1,874 @@ +package proxmox + +import ( + "bytes" + "context" + "encoding/json" + "encoding/pem" + "errors" + "fmt" + "io" + stdlog "log" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "reflect" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" +) + +const testToken = "proxy-operator@pve!operator=0b4e1c2a-5d6f-4a7b-8c9d-0e1f2a3b4c5d" + +// clientFixture is a TLS test server, a CA file trusting it, a token file +// and the config pointing at them. +type clientFixture struct { + srv *httptest.Server + cfg provider.ProxmoxConfig + tokenFile string + caFile string +} + +// newClientFixture starts h on a TLS server whose error log is discarded: +// the tests that fail a handshake on purpose would otherwise print it. +func newClientFixture(t *testing.T, h http.HandlerFunc) *clientFixture { + t.Helper() + srv := httptest.NewUnstartedServer(h) + srv.Config.ErrorLog = stdlog.New(io.Discard, "", 0) + srv.StartTLS() + t.Cleanup(srv.Close) + return fixtureFor(t, srv) +} + +func fixtureFor(t *testing.T, srv *httptest.Server) *clientFixture { + t.Helper() + dir := t.TempDir() + fx := &clientFixture{srv: srv, tokenFile: filepath.Join(dir, "token"), caFile: filepath.Join(dir, "ca.pem")} + writeFile(t, fx.tokenFile, testToken+"\n") + writeFile(t, fx.caFile, string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: srv.Certificate().Raw}))) + fx.cfg = provider.ProxmoxConfig{URL: srv.URL, TokenFile: fx.tokenFile, CAFile: fx.caFile, RequestTimeoutSeconds: 30} + return fx +} + +func (fx *clientFixture) api(t *testing.T, opts ...tracing.Option) *httpAPI { + t.Helper() + a, err := newHTTPAPI(fx.cfg, opts...) + if err != nil { + t.Fatalf("newHTTPAPI error = %v", err) + } + return a +} + +func writeFile(t *testing.T, path, content string) { + t.Helper() + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + t.Fatalf("writing %s: %v", path, err) + } +} + +// serveCapture writes a Phase 0 capture back byte for byte, reason phrase +// included, which net/http's ResponseWriter cannot do. +func serveCapture(t *testing.T, w http.ResponseWriter, c capture) { + t.Helper() + conn, buf, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Errorf("hijack: %v", err) + return + } + defer func() { _ = conn.Close() }() + _, _ = fmt.Fprintf(buf, "HTTP/1.1 %d %s\r\nContent-Type: application/json;charset=UTF-8\r\nContent-Length: %d\r\nConnection: close\r\n\r\n%s", + c.Status, c.Reason, len(c.Body), c.Body) + _ = buf.Flush() +} + +func serveJSON(w http.ResponseWriter, body string) { + w.Header().Set("Content-Type", "application/json;charset=UTF-8") + _, _ = io.WriteString(w, body) +} + +func assertClass(t *testing.T, err, class error) { + t.Helper() + if got := classify("get", err); got != class { + t.Errorf("classify(%v) = %v, want %v", err, got, class) + } +} + +func TestClient_sendsTokenHeaderFromFileOnEveryRequest(t *testing.T) { + t.Parallel() + var ( + mu sync.Mutex + headers []string + ) + fx := newClientFixture(t, func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + headers = append(headers, r.Header.Get("Authorization")) + mu.Unlock() + serveJSON(w, readCapture(t, "version").Body) + }) + a := fx.api(t) + const rotated = "proxy-operator@pve!operator=11111111-2222-3333-4444-555555555555" + for i, token := range []string{testToken, rotated} { + writeFile(t, fx.tokenFile, token+"\n") + v, err := a.Version(context.Background()) + if err != nil || v.Version != "9.2.20" { + t.Fatalf("call %d: Version() = %+v, %v", i, v, err) + } + } + want := []string{"PVEAPIToken=" + testToken, "PVEAPIToken=" + rotated} + mu.Lock() + defer mu.Unlock() + if !reflect.DeepEqual(headers, want) { + t.Errorf("Authorization headers = %q, want %q", headers, want) + } +} + +func TestClient_missingTokenFileIsTransient(t *testing.T) { + t.Parallel() + var requests atomic.Int32 + fx := newClientFixture(t, func(http.ResponseWriter, *http.Request) { requests.Add(1) }) + fx.cfg.TokenFile = filepath.Join(t.TempDir(), "absent") + _, err := fx.api(t).Version(context.Background()) + if !errors.Is(err, provider.ErrTransient) || !errors.Is(err, errCredentials) { + t.Errorf("error = %v, want ErrTransient and errCredentials", err) + } + if n := requests.Load(); n != 0 { + t.Errorf("server saw %d requests, want none without a token", n) + } +} + +func TestClient_unparsableTokenFileIsPermanent(t *testing.T) { + t.Parallel() + const sentinel = "SENTINEL-7f3e" + tests := []struct{ name, content string }{ + {name: "empty", content: ""}, + {name: "secret only", content: sentinel}, + {name: "no secret", content: "proxy-operator@pve!operator"}, + {name: "no token id", content: "proxy-operator@pve=" + sentinel}, + {name: "no realm", content: "proxy-operator!operator=" + sentinel}, + {name: "two lines", content: "proxy-operator@pve!operator=" + sentinel + "\nsecond=line"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + var requests atomic.Int32 + fx := newClientFixture(t, func(http.ResponseWriter, *http.Request) { requests.Add(1) }) + writeFile(t, fx.tokenFile, tc.content) + _, err := fx.api(t).Version(context.Background()) + if !errors.Is(err, provider.ErrPermanent) || !errors.Is(err, errCredentials) { + t.Errorf("error = %v, want ErrPermanent and errCredentials", err) + } + if err != nil && strings.Contains(err.Error(), sentinel) { + t.Errorf("error carries the file's content: %v", err) + } + if n := requests.Load(); n != 0 { + t.Errorf("server saw %d requests, want none", n) + } + }) + } +} + +// TestClient_apiErrorFromStatusReasonAndBody serves every captured error +// verbatim and expects the apiError the captures describe (F1): the body's +// message, or the reason phrase when the body is empty. +func TestClient_apiErrorFromStatusReasonAndBody(t *testing.T) { + t.Parallel() + paths, err := filepath.Glob(filepath.Join("testdata", "err_*.json")) + if err != nil || len(paths) == 0 { + t.Fatalf("no error captures: %v", err) + } + for _, path := range paths { + name := strings.TrimSuffix(filepath.Base(path), ".json") + if readCapture(t, name).Status < 400 { + continue // err_tag_hyphen: hyphens turned out to be accepted (F9) + } + t.Run(name, func(t *testing.T) { + t.Parallel() + c := readCapture(t, name) + fx := newClientFixture(t, func(w http.ResponseWriter, _ *http.Request) { serveCapture(t, w, c) }) + _, err := fx.api(t).Version(context.Background()) + var ae *apiError + if !errors.As(err, &ae) { + t.Fatalf("error = %v (%T), want *apiError", err, err) + } + want := loadAPIError(t, name) + if ae.Status != want.Status || ae.Message != want.Message || !reflect.DeepEqual(ae.Errors, want.Errors) { + t.Errorf("apiError = %+v, want %+v", ae, want) + } + if ae.Message == "" || strings.HasSuffix(ae.Message, "\n") { + t.Errorf("message %q is empty or untrimmed", ae.Message) + } + }) + } +} + +func TestClient_decodesDataEnvelope(t *testing.T) { + t.Parallel() + ctx := context.Background() + tests := []struct { + capture string + call func(a *httpAPI) (any, error) + check func(t *testing.T, got any) + }{ + {capture: "version", call: func(a *httpAPI) (any, error) { return a.Version(ctx) }, check: func(t *testing.T, got any) { + if v := got.(versionInfo); v.Version != "9.2.20" { + t.Errorf("version = %+v", v) + } + }}, + {capture: "node_status", call: func(a *httpAPI) (any, error) { return a.NodeStatus(ctx, "pv1-1") }, check: func(t *testing.T, got any) { + if s := got.(nodeStatus); s.CurrentKernel.Release != "7.0.14-17-pve" { + t.Errorf("node status = %+v", s) + } + }}, + {capture: "storage_status", call: func(a *httpAPI) (any, error) { return a.StorageStatus(ctx, "pv1-1", "local") }, check: func(t *testing.T, got any) { + if s := got.(storageStatus); s.Avail <= 0 || !strings.Contains(s.Content, "iso") { + t.Errorf("storage status = %+v", s) + } + }}, + {capture: "cluster_resources_during_clone", call: func(a *httpAPI) (any, error) { return a.ClusterResources(ctx) }, check: func(t *testing.T, got any) { + rs := got.([]vmResource) + for _, r := range rs { + if r.VMID == 199999 && r.Lock == "clone" && r.Name == "" { + return + } + } + t.Errorf("resources = %+v, want the nameless locked clone", rs) + }}, + {capture: "vm_config_clone", call: func(a *httpAPI) (any, error) { return a.VMConfig(ctx, "pv1-1", 199999) }, check: func(t *testing.T, got any) { + cfg := got.(vmConfig) + if cfg["cores"] != "2" || cfg["memory"] != "2048" || cfg["agent"] != "1" || cfg["boot"] != "order=scsi0" { + t.Errorf("config scalars = cores %q memory %q agent %q boot %q", cfg["cores"], cfg["memory"], cfg["agent"], cfg["boot"]) + } + if m := parseDescription(cfg["description"]); !m.Managed || m.UID == "" { + t.Errorf("description %q did not survive decoding", cfg["description"]) + } + }}, + {capture: "status_current_paused", call: func(a *httpAPI) (any, error) { return a.VMStatus(ctx, "pv1-1", 199999) }, check: func(t *testing.T, got any) { + if s := got.(vmStatus); s.Status != "running" || s.QMPStatus != "paused" { + t.Errorf("status = %+v", s) + } + }}, + {capture: "task_status_stopped_error", call: func(a *httpAPI) (any, error) { return a.TaskStatus(ctx, "pv1-1", "UPID:pv1-1:x") }, check: func(t *testing.T, got any) { + if s := got.(taskStatus); s.Status != "stopped" || s.ExitStatus != "shrinking disks is not supported" { + t.Errorf("task = %+v", s) + } + }}, + {capture: "agent_network_get_interfaces", call: func(a *httpAPI) (any, error) { return a.AgentNetworkInterfaces(ctx, "pv1-1", 199999) }, check: func(t *testing.T, got any) { + if ip := firstIPv4(got.([]agentInterface)); ip != "192.168.0.150" { + t.Errorf("first IPv4 = %q", ip) + } + }}, + {capture: "set_config_response", call: func(a *httpAPI) (any, error) { + return nil, a.SetVMConfig(ctx, "pv1-1", 199999, map[string]string{"cores": "1"}) + }, check: func(*testing.T, any) {}}, + {capture: "clone_response", call: func(a *httpAPI) (any, error) { return a.Clone(ctx, "pv1-1", 9001, cloneParams{NewID: 199999}) }, check: wantUPID("qmclone")}, + {capture: "resize_response", call: func(a *httpAPI) (any, error) { return a.Resize(ctx, "pv1-1", 199999, "scsi0", "4G") }, check: wantUPID("resize")}, + {capture: "start_response", call: func(a *httpAPI) (any, error) { return a.Start(ctx, "pv1-1", 199999) }, check: wantUPID("qmstart")}, + {capture: "stop_response", call: func(a *httpAPI) (any, error) { return a.Stop(ctx, "pv1-1", 199999) }, check: wantUPID("qmstop")}, + {capture: "destroy_response", call: func(a *httpAPI) (any, error) { return a.Destroy(ctx, "pv1-1", 199999) }, check: wantUPID("qmdestroy")}, + {capture: "upload_response", call: func(a *httpAPI) (any, error) { + return a.UploadISO(ctx, "pv1-1", "local", "x.iso", strings.NewReader("iso"), 3) + }, check: wantUPID("imgcopy")}, + {capture: "content_delete_response", call: func(a *httpAPI) (any, error) { + return a.DeleteVolume(ctx, "pv1-1", "proxy-seeds", "proxy-seeds:iso/x.iso") + }, check: wantUPID("imgdel")}, + } + for _, tc := range tests { + t.Run(tc.capture, func(t *testing.T) { + t.Parallel() + c := readCapture(t, tc.capture) + fx := newClientFixture(t, func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + serveCapture(t, w, c) + }) + got, err := tc.call(fx.api(t)) + if err != nil { + t.Fatalf("call error = %v", err) + } + tc.check(t, got) + }) + } +} + +func wantUPID(kind string) func(t *testing.T, got any) { + return func(t *testing.T, got any) { + t.Helper() + if upid := got.(string); !strings.HasPrefix(upid, "UPID:pv1-1:") || !strings.Contains(upid, ":"+kind+":") { + t.Errorf("upid = %q, want a %s UPID of pv1-1", upid, kind) + } + } +} + +// seenRequest is what a test server records of one request. +type seenRequest struct { + method, path, escapedPath string + query, form url.Values + contentType string +} + +func recordingServer(t *testing.T, seen chan<- seenRequest) *clientFixture { + t.Helper() + return newClientFixture(t, func(w http.ResponseWriter, r *http.Request) { + req := seenRequest{method: r.Method, path: r.URL.Path, escapedPath: r.URL.EscapedPath(), query: r.URL.Query(), contentType: r.Header.Get("Content-Type")} + if r.Method == http.MethodPost || r.Method == http.MethodPut { + if err := r.ParseForm(); err != nil { + t.Errorf("parsing form: %v", err) + } + req.form = r.PostForm + } + seen <- req + serveJSON(w, `{"data":null}`) + }) +} + +// TestClient_methodPathAndParams pins the HTTP method, path and parameters +// of every seam method to the endpoint map of spec entry 17. +func TestClient_methodPathAndParams(t *testing.T) { + t.Parallel() + ctx := context.Background() + const upid = "UPID:pv1-1:0010FBF1:02804D0D:6AB378EA:qmstart:199999:proxy-operator@pve!operator:" + tests := map[string]struct { + call func(a *httpAPI) error + method string + path string + query url.Values + form url.Values + }{ + "Version": {call: func(a *httpAPI) error { _, err := a.Version(ctx); return err }, method: "GET", path: "/api2/json/version"}, + "NodeStatus": {call: func(a *httpAPI) error { _, err := a.NodeStatus(ctx, "pv1-1"); return err }, + method: "GET", path: "/api2/json/nodes/pv1-1/status"}, + "StorageStatus": {call: func(a *httpAPI) error { _, err := a.StorageStatus(ctx, "pv1-1", "proxy-seeds"); return err }, + method: "GET", path: "/api2/json/nodes/pv1-1/storage/proxy-seeds/status"}, + "ClusterResources": {call: func(a *httpAPI) error { _, err := a.ClusterResources(ctx); return err }, + method: "GET", path: "/api2/json/cluster/resources", query: url.Values{"type": {"vm"}}}, + "VMConfig": {call: func(a *httpAPI) error { _, err := a.VMConfig(ctx, "pv1-1", 150042); return err }, + method: "GET", path: "/api2/json/nodes/pv1-1/qemu/150042/config"}, + "SetVMConfig": {call: func(a *httpAPI) error { + return a.SetVMConfig(ctx, "pv1-1", 150042, map[string]string{"cores": "1", "ide2": "proxy-seeds:iso/proxy-x-seed.iso,media=cdrom"}) + }, method: "PUT", path: "/api2/json/nodes/pv1-1/qemu/150042/config", + form: url.Values{"cores": {"1"}, "ide2": {"proxy-seeds:iso/proxy-x-seed.iso,media=cdrom"}}}, + "Clone": {call: func(a *httpAPI) error { + _, err := a.Clone(ctx, "pv1-1", 9001, cloneParams{NewID: 150042, Name: "proxy-x", Description: "a=1\nb=2", Storage: "local-lvm", Target: "pv1-2"}) + return err + }, method: "POST", path: "/api2/json/nodes/pv1-1/qemu/9001/clone", + form: url.Values{"newid": {"150042"}, "name": {"proxy-x"}, "description": {"a=1\nb=2"}, "full": {"1"}, "storage": {"local-lvm"}, "target": {"pv1-2"}}}, + "Resize": {call: func(a *httpAPI) error { _, err := a.Resize(ctx, "pv1-1", 150042, "scsi0", "8G"); return err }, + method: "PUT", path: "/api2/json/nodes/pv1-1/qemu/150042/resize", form: url.Values{"disk": {"scsi0"}, "size": {"8G"}}}, + "Start": {call: func(a *httpAPI) error { _, err := a.Start(ctx, "pv1-1", 150042); return err }, + method: "POST", path: "/api2/json/nodes/pv1-1/qemu/150042/status/start", form: url.Values{}}, + "Stop": {call: func(a *httpAPI) error { _, err := a.Stop(ctx, "pv1-1", 150042); return err }, + method: "POST", path: "/api2/json/nodes/pv1-1/qemu/150042/status/stop", form: url.Values{}}, + "VMStatus": {call: func(a *httpAPI) error { _, err := a.VMStatus(ctx, "pv1-1", 150042); return err }, + method: "GET", path: "/api2/json/nodes/pv1-1/qemu/150042/status/current"}, + "Destroy": {call: func(a *httpAPI) error { _, err := a.Destroy(ctx, "pv1-1", 150042); return err }, + method: "DELETE", path: "/api2/json/nodes/pv1-1/qemu/150042", query: url.Values{"purge": {"1"}, "destroy-unreferenced-disks": {"1"}}}, + "TaskStatus": {call: func(a *httpAPI) error { _, err := a.TaskStatus(ctx, "pv1-1", upid); return err }, + method: "GET", path: "/api2/json/nodes/pv1-1/tasks/" + upid + "/status"}, + "AgentNetworkInterfaces": {call: func(a *httpAPI) error { _, err := a.AgentNetworkInterfaces(ctx, "pv1-1", 150042); return err }, + method: "GET", path: "/api2/json/nodes/pv1-1/qemu/150042/agent/network-get-interfaces"}, + "UploadISO": {call: func(a *httpAPI) error { + _, err := a.UploadISO(ctx, "pv1-1", "proxy-seeds", "proxy-x-seed.iso", strings.NewReader("iso"), 3) + return err + }, method: "POST", path: "/api2/json/nodes/pv1-1/storage/proxy-seeds/upload"}, // multipart, see TestClient_uploadStreamsMultipart + "DeleteVolume": {call: func(a *httpAPI) error { + _, err := a.DeleteVolume(ctx, "pv1-1", "proxy-seeds", "proxy-seeds:iso/proxy-x-seed.iso") + return err + }, method: "DELETE", path: "/api2/json/nodes/pv1-1/storage/proxy-seeds/content/proxy-seeds:iso/proxy-x-seed.iso"}, + } + for m := range reflect.TypeFor[api]().Methods() { + if _, ok := tests[m.Name]; !ok { + t.Errorf("no row for seam method %s", m.Name) + } + } + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + seen := make(chan seenRequest, 1) + fx := recordingServer(t, seen) + if err := tc.call(fx.api(t)); err != nil { + t.Fatalf("call error = %v", err) + } + got := <-seen + if got.method != tc.method || got.path != tc.path { + t.Errorf("request = %s %s, want %s %s", got.method, got.path, tc.method, tc.path) + } + if len(got.query) != 0 || len(tc.query) != 0 { + if !reflect.DeepEqual(got.query, tc.query) { + t.Errorf("query = %v, want %v", got.query, tc.query) + } + } + if tc.form == nil { + if got.form != nil && len(got.form) != 0 { + t.Errorf("form = %v, want none", got.form) + } + return + } + if got.contentType != "application/x-www-form-urlencoded" { + t.Errorf("content type = %q, want a urlencoded form", got.contentType) + } + for k, want := range tc.form { + if !reflect.DeepEqual(got.form[k], want) { + t.Errorf("form %s = %q, want %q", k, got.form[k], want) + } + } + if len(got.form) != len(tc.form) { + t.Errorf("form = %v, want exactly %v", got.form, tc.form) + } + }) + } +} + +func TestClient_cloneOmitsEmptyStorageAndTarget(t *testing.T) { + t.Parallel() + seen := make(chan seenRequest, 1) + fx := recordingServer(t, seen) + if _, err := fx.api(t).Clone(context.Background(), "pv1-1", 9001, cloneParams{NewID: 150042, Name: "proxy-x"}); err != nil { + t.Fatalf("Clone error = %v", err) + } + got := <-seen + if _, ok := got.form["storage"]; ok { + t.Errorf("form = %v, want no storage", got.form) + } + if _, ok := got.form["target"]; ok { + t.Errorf("form = %v, want no target", got.form) + } +} + +// TestClient_destroyQueryFlags checks that destroy leaves nothing behind: +// the purge and unreferenced-disk flags travel in the query of the DELETE +// (spec entry 11). +func TestClient_destroyQueryFlags(t *testing.T) { + t.Parallel() + seen := make(chan seenRequest, 1) + fx := recordingServer(t, seen) + if _, err := fx.api(t).Destroy(context.Background(), "pv1-1", 150042); err != nil { + t.Fatalf("Destroy error = %v", err) + } + got := <-seen + if got.method != http.MethodDelete || got.query.Get("purge") != "1" || got.query.Get("destroy-unreferenced-disks") != "1" { + t.Errorf("request = %s %s?%s, want DELETE with purge=1 and destroy-unreferenced-disks=1", got.method, got.path, got.query.Encode()) + } +} + +// TestClient_volIDSlashIsEscaped sends the volid's "/" as %2F, the way +// Phase 0 deleted volumes, so the router sees one path segment. +func TestClient_volIDSlashIsEscaped(t *testing.T) { + t.Parallel() + seen := make(chan seenRequest, 1) + fx := recordingServer(t, seen) + if _, err := fx.api(t).DeleteVolume(context.Background(), "pv1-1", "proxy-seeds", "proxy-seeds:iso/proxy-x-seed.iso"); err != nil { + t.Fatalf("DeleteVolume error = %v", err) + } + if got := <-seen; !strings.HasSuffix(got.escapedPath, "/content/proxy-seeds:iso%2Fproxy-x-seed.iso") { + t.Errorf("escaped path = %q, want the volid as one segment with %%2F", got.escapedPath) + } +} + +func TestClient_http1Only(t *testing.T) { + t.Parallel() + protos := make(chan int, 2) + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + protos <- r.ProtoMajor + serveJSON(w, readCapture(t, "version").Body) + })) + srv.EnableHTTP2 = true + srv.StartTLS() + t.Cleanup(srv.Close) + + resp, err := srv.Client().Get(srv.URL) + if err != nil { + t.Fatalf("sanity request error = %v", err) + } + _ = resp.Body.Close() + if p := <-protos; p != 2 { + t.Fatalf("sanity: server spoke HTTP/%d to an h2 client, want 2", p) + } + if _, err := fixtureFor(t, srv).api(t).Version(context.Background()); err != nil { + t.Fatalf("Version error = %v", err) + } + if p := <-protos; p != 1 { + t.Errorf("client spoke HTTP/%d, want HTTP/1.1 (F18)", p) + } +} + +func TestClient_requestTimeoutIsTransient(t *testing.T) { + t.Parallel() + fx := newClientFixture(t, func(_ http.ResponseWriter, r *http.Request) { + select { + case <-r.Context().Done(): + case <-time.After(5 * time.Second): + } + }) + a := fx.api(t) + a.timeout = 100 * time.Millisecond + start := time.Now() + _, err := a.Version(context.Background()) + if elapsed := time.Since(start); elapsed > 2*time.Second { + t.Errorf("Version took %v, want the 100ms request timeout to cut it", elapsed) + } + var ne net.Error + if !errors.As(err, &ne) || !ne.Timeout() { + t.Errorf("error = %v, want a timeout", err) + } + assertClass(t, err, provider.ErrTransient) +} + +// TestClient_cancelledContextReturnsPromptly also checks that the +// cancellation touches only that request: the next call on the same client +// succeeds. +func TestClient_cancelledContextReturnsPromptly(t *testing.T) { + t.Parallel() + fx := newClientFixture(t, func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/status") { + select { + case <-r.Context().Done(): + case <-time.After(5 * time.Second): + } + return + } + serveJSON(w, readCapture(t, "version").Body) + }) + a := fx.api(t) + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(50*time.Millisecond, cancel) + start := time.Now() + _, err := a.NodeStatus(ctx, "pv1-1") + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("NodeStatus took %v after cancel, want prompt", elapsed) + } + if !errors.Is(err, context.Canceled) { + t.Errorf("error = %v, want context.Canceled", err) + } + assertClass(t, err, provider.ErrTransient) + if _, err := a.Version(context.Background()); err != nil { + t.Errorf("next call error = %v, want the client unaffected", err) + } +} + +func tracedFixture(t *testing.T) (*clientFixture, *tracetest.SpanRecorder, *sdktrace.TracerProvider, func() []string) { + t.Helper() + var ( + mu sync.Mutex + traceparent []string + ) + fx := newClientFixture(t, func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + traceparent = append(traceparent, r.Header.Get("Traceparent")) + mu.Unlock() + serveJSON(w, readCapture(t, "version").Body) + }) + sr := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr)) + t.Cleanup(func() { _ = tp.Shutdown(context.Background()) }) + sent := func() []string { + mu.Lock() + defer mu.Unlock() + return append([]string(nil), traceparent...) + } + return fx, sr, tp, sent +} + +func TestClient_childSpanUnderParent(t *testing.T) { + t.Parallel() + fx, sr, tp, traceparent := tracedFixture(t) + a := fx.api(t, tracing.WithTracerProvider(tp)) + ctx, parent := tp.Tracer("test").Start(context.Background(), "provider.get") + if _, err := a.Version(ctx); err != nil { + t.Fatalf("Version error = %v", err) + } + parent.End() + + var child sdktrace.ReadOnlySpan + for _, s := range sr.Ended() { + if s.Name() != "provider.get" { + child = s + } + } + if child == nil { + t.Fatalf("no HTTP span recorded; spans: %d", len(sr.Ended())) + } + if child.Parent().SpanID() != parent.SpanContext().SpanID() || child.SpanContext().TraceID() != parent.SpanContext().TraceID() { + t.Errorf("HTTP span %q is not a child of provider.get", child.Name()) + } + secret := testToken[strings.Index(testToken, "=")+1:] + for _, kv := range child.Attributes() { + if strings.Contains(kv.Value.Emit(), secret) { + t.Errorf("span attribute %s carries the token", kv.Key) + } + } + if sent := traceparent(); len(sent) != 1 || !strings.Contains(sent[0], child.SpanContext().TraceID().String()) { + t.Errorf("traceparent sent = %q, want the trace id", sent) + } +} + +func TestClient_noSpanWithoutParent(t *testing.T) { + t.Parallel() + fx, sr, tp, traceparent := tracedFixture(t) + if _, err := fx.api(t, tracing.WithTracerProvider(tp)).Version(context.Background()); err != nil { + t.Fatalf("Version error = %v", err) + } + if n := len(sr.Ended()); n != 0 { + t.Errorf("got %d spans without a parent, want none", n) + } + if sent := traceparent(); len(sent) != 1 || sent[0] != "" { + t.Errorf("traceparent sent = %q, want none", sent) + } +} + +// TestClient_tokenNeverInErrorsOrLogs has the server echo the secret back +// in every place an error's text comes from; neither the error nor the +// decorator's V(2) lines may carry it. +func TestClient_tokenNeverInErrorsOrLogs(t *testing.T) { + t.Parallel() + const secret = "SENTINEL-SECRET-9d41" + token := "proxy-operator@pve!operator=" + secret + tests := []struct { + name string + resp func(auth string) capture + }{ + {name: "401 body echoes the header", resp: func(auth string) capture { + return capture{Status: 401, Reason: "Authentication failed!", Body: `{"data":null,"message":"bad credentials: ` + auth + `\n"}`} + }}, + {name: "401 reason echoes the header", resp: func(auth string) capture { + return capture{Status: 401, Reason: "bad credentials " + auth} + }}, + {name: "400 errors echo the header", resp: func(auth string) capture { + return capture{Status: 400, Reason: "Parameter verification failed.", Body: `{"data":null,"message":"Parameter verification failed.\n","errors":{"token":"` + auth + `"}}`} + }}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newClientFixture(t, func(w http.ResponseWriter, r *http.Request) { + serveCapture(t, w, tc.resp(r.Header.Get("Authorization"))) + }) + writeFile(t, fx.tokenFile, token) + logs := &logCapture{} + ctx := logs.context(2) + _, err := observe(fx.api(t), "proxmox-home").Version(ctx) + var ae *apiError + if !errors.As(err, &ae) { + t.Fatalf("error = %v, want an apiError", err) + } + if !strings.Contains(err.Error(), "") { + t.Errorf("error = %v, want the echoed secret redacted", err) + } + for _, s := range append([]string{err.Error(), fmt.Sprintf("%+v", err), fmt.Sprintf("%#v", ae)}, logs.all()...) { + if strings.Contains(s, secret) { + t.Errorf("secret leaked: %s", s) + } + } + if len(logs.all()) != 1 { + t.Errorf("got %d log lines, want the one call line", len(logs.all())) + } + }) + } +} + +func TestClient_caFileTrustsPrivateCA(t *testing.T) { + t.Parallel() + fx := newClientFixture(t, func(w http.ResponseWriter, _ *http.Request) { serveJSON(w, readCapture(t, "version").Body) }) + if _, err := fx.api(t).Version(context.Background()); err != nil { + t.Errorf("with the CA file: error = %v, want nil", err) + } + fx.cfg.CAFile = "" + _, err := fx.api(t).Version(context.Background()) + if !isTLSVerification(err) { + t.Errorf("with system roots: error = %v, want a verification failure", err) + } +} + +// TestClient_caFileLoadedLazily covers spec entry 14: a missing CA file is +// Transient and retried until it lands; one without a certificate is +// Permanent. +func TestClient_caFileLoadedLazily(t *testing.T) { + t.Parallel() + fx := newClientFixture(t, func(w http.ResponseWriter, _ *http.Request) { serveJSON(w, readCapture(t, "version").Body) }) + pemBytes, err := os.ReadFile(fx.caFile) + if err != nil { + t.Fatal(err) + } + fx.cfg.CAFile = filepath.Join(t.TempDir(), "ca.pem") + a := fx.api(t) + if _, err := a.Version(context.Background()); !errors.Is(err, provider.ErrTransient) { + t.Errorf("missing CA file: error = %v, want ErrTransient", err) + } + writeFile(t, fx.cfg.CAFile, "not a certificate") + if _, err := a.Version(context.Background()); !errors.Is(err, provider.ErrPermanent) { + t.Errorf("CA file without a certificate: error = %v, want ErrPermanent", err) + } + writeFile(t, fx.cfg.CAFile, string(pemBytes)) + if _, err := a.Version(context.Background()); err != nil { + t.Errorf("CA file landed: error = %v, want nil", err) + } +} + +func TestClient_x509FailureIsPermanent(t *testing.T) { + t.Parallel() + tests := []struct { + name string + mutate func(fx *clientFixture) + }{ + {name: "unknown authority", mutate: func(fx *clientFixture) { fx.cfg.CAFile = "" }}, + {name: "host not in the certificate", mutate: func(fx *clientFixture) { + fx.cfg.URL = strings.Replace(fx.srv.URL, "127.0.0.1", "localhost", 1) + }}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + var requests atomic.Int32 + fx := newClientFixture(t, func(http.ResponseWriter, *http.Request) { requests.Add(1) }) + tc.mutate(fx) + _, err := fx.api(t).Version(context.Background()) + if !isTLSVerification(err) { + t.Errorf("error = %v, want a verification failure", err) + } + assertClass(t, err, provider.ErrPermanent) + if n := requests.Load(); n != 0 { + t.Errorf("server saw %d requests, want none past the handshake", n) + } + }) + } +} + +func TestClient_insecureSkipsVerification(t *testing.T) { + t.Parallel() + fx := newClientFixture(t, func(w http.ResponseWriter, _ *http.Request) { serveJSON(w, readCapture(t, "version").Body) }) + fx.cfg.CAFile = "" + fx.cfg.InsecureSkipTLSVerify = true + if _, err := fx.api(t).Version(context.Background()); err != nil { + t.Errorf("error = %v, want nil with verification off", err) + } +} + +// gatedReader hands out first, then waits for release before handing out +// rest. A client that read the whole body before sending would never get +// past first. +type gatedReader struct { + first, rest []byte + release <-chan struct{} + waited bool +} + +func (g *gatedReader) Read(p []byte) (int, error) { + if len(g.first) > 0 { + n := copy(p, g.first) + g.first = g.first[n:] + return n, nil + } + if !g.waited { + select { + case <-g.release: + case <-time.After(5 * time.Second): + return 0, errors.New("the server saw nothing of the file before the reader was drained: the body is not streamed") + } + g.waited = true + } + if len(g.rest) == 0 { + return 0, io.EOF + } + n := copy(p, g.rest) + g.rest = g.rest[n:] + return n, nil +} + +func TestClient_uploadStreamsMultipart(t *testing.T) { + t.Parallel() + first := bytes.Repeat([]byte("a"), 256<<10) + rest := bytes.Repeat([]byte("b"), 256<<10) + sawFirst := make(chan struct{}) + type upload struct { + contentLength int64 + chunked bool + content, name string + fileName, bytes string + } + got := make(chan upload, 1) + fx := newClientFixture(t, func(w http.ResponseWriter, r *http.Request) { + u := upload{contentLength: r.ContentLength, chunked: len(r.TransferEncoding) > 0} + mr, err := r.MultipartReader() + if err != nil { + t.Errorf("multipart reader: %v", err) + return + } + for { + part, err := mr.NextPart() + if err == io.EOF { + break + } + if err != nil { + t.Errorf("next part: %v", err) + return + } + switch part.FormName() { + case "content": + b, _ := io.ReadAll(part) + u.content = string(b) + case "filename": + u.name, u.fileName = part.FormName(), part.FileName() + var buf bytes.Buffer + if _, err := io.CopyN(&buf, part, 1); err != nil { + t.Errorf("reading the file part: %v", err) + return + } + close(sawFirst) + _, _ = io.Copy(&buf, part) + u.bytes = buf.String() + } + } + got <- u + serveJSON(w, readCapture(t, "upload_response").Body) + }) + body := &gatedReader{first: first, rest: rest, release: sawFirst} + size := int64(len(first) + len(rest)) + upid, err := fx.api(t).UploadISO(context.Background(), "pv1-1", "proxy-seeds", "proxy-x-seed.iso", body, size) + if err != nil { + t.Fatalf("UploadISO error = %v", err) + } + if !strings.Contains(upid, ":imgcopy:") { + t.Errorf("upid = %q, want the imgcopy task", upid) + } + u := <-got + if u.chunked || u.contentLength <= size { + t.Errorf("Content-Length %d (chunked %v), want an exact length above the file's %d bytes", u.contentLength, u.chunked, size) + } + if u.content != "iso" || u.name != "filename" || u.fileName != "proxy-x-seed.iso" { + t.Errorf("fields content=%q part=%q file=%q, want iso, filename, proxy-x-seed.iso (F12)", u.content, u.name, u.fileName) + } + if u.bytes != string(first)+string(rest) { + t.Errorf("server got %d file bytes, want the %d sent", len(u.bytes), size) + } +} + +func TestDecodeConfig_scalarsAsStrings(t *testing.T) { + t.Parallel() + in := map[string]json.RawMessage{"cores": json.RawMessage(`2`), "memory": json.RawMessage(`"2048"`), "template": json.RawMessage(`1`), "balloon": json.RawMessage(`null`), "description": json.RawMessage(`"a\nb"`)} + got, err := decodeConfig(in) + if err != nil { + t.Fatalf("decodeConfig error = %v", err) + } + want := vmConfig{"cores": "2", "memory": "2048", "template": "1", "description": "a\nb"} + if !reflect.DeepEqual(got, want) { + t.Errorf("decodeConfig = %v, want %v", got, want) + } +} + +func TestReadToken_trimsTrailingNewline(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "token") + writeFile(t, path, " "+testToken+"\r\n") + if got, err := readToken(path); err != nil || got != testToken { + t.Errorf("readToken = %q, %v; want the token without whitespace", got, err) + } +} diff --git a/internal/provider/proxmox/fake_test.go b/internal/provider/proxmox/fake_test.go new file mode 100644 index 0000000..62242e6 --- /dev/null +++ b/internal/provider/proxmox/fake_test.go @@ -0,0 +1,893 @@ +package proxmox + +import ( + "context" + "errors" + "fmt" + "io" + "maps" + "slices" + "strconv" + "strings" + "sync" + "testing" +) + +// fakeAPI is the in-memory cluster of spec entry 17: VMs, tasks and ISOs, +// with lock simulation (a clone leaves a nameless locked entry until its +// task completes), task polls, per-method errors and blocking, and a call +// log. Error texts Proxmox was seen to send are reproduced exactly and +// checked against the captures by TestFakeAPI_errorsMatchCaptures; texts +// recon never produced start with "fake:". +type fakeAPI struct { + mu sync.Mutex + vms map[int]*fakeVM + tasks map[string]*fakeTask + isos map[string]map[string][]byte // storage, then file name + nodes map[string]nodeStatus + storages map[string]storageStatus + version versionInfo + taskSeq int + + // hooks returns the error a method fails with before touching state. + hooks map[string]func() error + // block makes a method wait on its channel (or ctx) before running. + block map[string]chan struct{} + calls map[string]int + log []string + + // taskPolls is how many TaskStatus polls answer "running" before a + // task stops. + taskPolls int + // cloneFailsWith and startFailsWith, when set, are the exitstatus of + // every clone or start task. A failed clone removes its VM. + cloneFailsWith, startFailsWith string + // agentNotRunning makes the agent of every running VM with agent=1 + // answer "QEMU guest agent is not running". + agentNotRunning bool + // cloneLeavesLock keeps lock=clone after a clone task ends, as an + // interrupted clone would (F15). + cloneLeavesLock bool +} + +type fakeVM struct { + node, name, status, qmpstatus, lock string + template bool + config map[string]string + ifaces []agentInterface +} + +type fakeTask struct { + node string + polls int + exit string + onDone func() + kind string + id string + done bool +} + +const ( + fakeNode = "pv1-1" + fakeISOStorage = "proxy-seeds" +) + +var _ api = (*fakeAPI)(nil) + +func newFakeAPI() *fakeAPI { + var node nodeStatus + node.PVEVersion = "pve-manager/9.2.20/49318c671b82f31e" + node.CurrentKernel.Release = "7.0.14-17-pve" + return &fakeAPI{ + vms: make(map[int]*fakeVM), + tasks: make(map[string]*fakeTask), + isos: make(map[string]map[string][]byte), + nodes: map[string]nodeStatus{fakeNode: node}, + storages: map[string]storageStatus{fakeISOStorage: {Type: "dir", Content: "iso", Active: 1, Enabled: 1, Total: 100 << 30, Used: 10 << 30, Avail: 90 << 30}}, + version: versionInfo{Version: "9.2.20", Release: "9.2", RepoID: "49318c671b82f31e"}, + hooks: make(map[string]func() error), + block: make(map[string]chan struct{}), + calls: make(map[string]int), + } +} + +// addVM places a stopped or running VM cloned from a template like 9001 on +// fakeNode, without a description; tests add the mark they need. +func (f *fakeAPI) addVM(vmid int, name, status string) *fakeVM { + f.mu.Lock() + defer f.mu.Unlock() + vm := &fakeVM{ + node: fakeNode, + name: name, + status: status, + qmpstatus: status, + config: map[string]string{ + "agent": "1", + "boot": "order=scsi0", + "cores": "2", + "memory": "2048", + "net0": "virtio=BC:24:11:48:11:46,bridge=vmbr0", + "scsi0": fmt.Sprintf("local-lvm:vm-%d-disk-1,size=3G", vmid), + }, + } + f.vms[vmid] = vm + return vm +} + +// addTemplate places a template shaped like 9001 (F19) on fakeNode. +func (f *fakeAPI) addTemplate(vmid int, name string) *fakeVM { + vm := f.addVM(vmid, name, "stopped") + f.mu.Lock() + defer f.mu.Unlock() + vm.template = true + vm.config["scsi0"] = fmt.Sprintf("local-lvm:base-%d-disk-1,size=3G", vmid) + return vm +} + +func (f *fakeAPI) failWith(method string, err error) { + f.mu.Lock() + defer f.mu.Unlock() + f.hooks[method] = func() error { return err } +} + +func (f *fakeAPI) blockOn(method string) chan struct{} { + f.mu.Lock() + defer f.mu.Unlock() + ch := make(chan struct{}) + f.block[method] = ch + return ch +} + +func (f *fakeAPI) callCount(method string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls[method] +} + +func (f *fakeAPI) callLog() []string { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.log) +} + +// iso returns an uploaded file's content. +func (f *fakeAPI) iso(storage, filename string) ([]byte, bool) { + f.mu.Lock() + defer f.mu.Unlock() + b, ok := f.isos[storage][filename] + return b, ok +} + +// vmSnapshot returns a copy of the VM at vmid, for assertions. +func (f *fakeAPI) vmSnapshot(vmid int) (fakeVM, bool) { + f.mu.Lock() + defer f.mu.Unlock() + vm, ok := f.vms[vmid] + if !ok { + return fakeVM{}, false + } + c := *vm + c.config = maps.Clone(vm.config) + c.ifaces = slices.Clone(vm.ifaces) + return c, true +} + +func fakeErr(status int, msg string) *apiError { + return &apiError{Status: status, Message: msg} +} + +// The 500s recon captured (F2, F4, F5, F7). +func errVMMissing(node string, vmid int) *apiError { + return fakeErr(500, fmt.Sprintf("Configuration file 'nodes/%s/qemu-server/%d.conf' does not exist", node, vmid)) +} + +func errLocked(lock string) *apiError { return fakeErr(500, "VM is locked ("+lock+")") } + +func errNewIDInUse(vmid int) *apiError { + return fakeErr(500, fmt.Sprintf("unable to create VM %d: config file already exists", vmid)) +} + +func errVMNotRunning(vmid int) *apiError { + return fakeErr(500, fmt.Sprintf("VM %d is not running", vmid)) +} + +// enter records the call, honours a block and runs the hook. It is called +// without f.mu held; state access happens afterwards under the lock. +func (f *fakeAPI) enter(ctx context.Context, method string) error { + f.mu.Lock() + f.calls[method]++ + f.log = append(f.log, method) + block := f.block[method] + hook := f.hooks[method] + f.mu.Unlock() + if block != nil { + select { + case <-block: + case <-ctx.Done(): + return ctx.Err() + } + } + if hook != nil { + return hook() + } + return nil +} + +// vm looks a VM up the way Proxmox does, by node and VMID: a VM on another +// 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 { + return nil, errVMMissing(node, vmid) + } + return vm, nil +} + +// newTask registers a task that stops after f.taskPolls running answers, +// then runs onDone under f.mu. The UPID has the shape parseUPID reads. +func (f *fakeAPI) newTask(node, kind, id, exit string, onDone func()) string { + f.taskSeq++ + upid := fmt.Sprintf("UPID:%s:%08X:%08X:%08X:%s:%s:proxy-operator@pve!operator:", node, 0x100000+f.taskSeq, 0x2800000+f.taskSeq, 0x6AB37800+f.taskSeq, kind, id) + f.tasks[upid] = &fakeTask{node: node, exit: exit, onDone: onDone, kind: kind, id: id} + return upid +} + +func (f *fakeAPI) Version(ctx context.Context) (versionInfo, error) { + if err := f.enter(ctx, "Version"); err != nil { + return versionInfo{}, err + } + f.mu.Lock() + defer f.mu.Unlock() + return f.version, nil +} + +func (f *fakeAPI) NodeStatus(ctx context.Context, node string) (nodeStatus, error) { + if err := f.enter(ctx, "NodeStatus"); err != nil { + return nodeStatus{}, err + } + f.mu.Lock() + defer f.mu.Unlock() + s, ok := f.nodes[node] + if !ok { + return nodeStatus{}, fakeErr(500, "fake: no such node '"+node+"'") + } + return s, nil +} + +func (f *fakeAPI) StorageStatus(ctx context.Context, node, storage string) (storageStatus, error) { + if err := f.enter(ctx, "StorageStatus"); err != nil { + return storageStatus{}, err + } + f.mu.Lock() + defer f.mu.Unlock() + s, ok := f.storages[storage] + if !ok { + return storageStatus{}, fakeErr(500, "fake: storage '"+storage+"' does not exist") + } + return s, nil +} + +// ClusterResources lists VMs by VMID. A VM being cloned is listed as F8 +// saw it: no name, no tags, lock clone and status unknown. +func (f *fakeAPI) ClusterResources(ctx context.Context) ([]vmResource, error) { + if err := f.enter(ctx, "ClusterResources"); err != nil { + return nil, err + } + f.mu.Lock() + defer f.mu.Unlock() + out := make([]vmResource, 0, len(f.vms)) + 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.template { + r.Template = 1 + } + if vm.lock == "clone" { + r.Name, r.Tags, r.Status = "", "", "unknown" + } + out = append(out, r) + } + return out, nil +} + +func (f *fakeAPI) VMConfig(ctx context.Context, node string, vmid int) (vmConfig, error) { + if err := f.enter(ctx, "VMConfig"); err != nil { + return nil, err + } + f.mu.Lock() + defer f.mu.Unlock() + vm, err := f.vm(node, vmid) + if err != nil { + return nil, err + } + cfg := vmConfig(maps.Clone(vm.config)) + if vm.name != "" { + cfg["name"] = vm.name + } + if vm.lock != "" { + cfg["lock"] = vm.lock + } + if vm.template { + cfg["template"] = "1" + } + return cfg, nil +} + +// SetVMConfig is refused while the VM is locked (F7). "delete" removes a +// comma-separated list of keys; "name" renames. +func (f *fakeAPI) SetVMConfig(ctx context.Context, node string, vmid int, params map[string]string) error { + if err := f.enter(ctx, "SetVMConfig"); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + vm, err := f.vm(node, vmid) + if err != nil { + return err + } + if vm.lock != "" { + return errLocked(vm.lock) + } + for k, v := range params { + switch k { + case "delete": + for key := range strings.SplitSeq(v, ",") { + delete(vm.config, key) + } + case "name": + vm.name = v + default: + vm.config[k] = v + } + } + return nil +} + +// Clone creates the VM at once, locked and nameless in the listing, and a +// task that clears the lock, or removes the VM when cloneFailsWith is set. +// A taken newid fails synchronously (F4). +func (f *fakeAPI) Clone(ctx context.Context, node string, vmid int, p cloneParams) (string, error) { + if err := f.enter(ctx, "Clone"); err != nil { + return "", err + } + f.mu.Lock() + defer f.mu.Unlock() + src, err := f.vm(node, vmid) + if err != nil { + return "", err + } + if _, taken := f.vms[p.NewID]; taken { + return "", errNewIDInUse(p.NewID) + } + target := node + if p.Target != "" { + target = p.Target + } + cfg := maps.Clone(src.config) + for k, v := range cfg { + cfg[k] = strings.ReplaceAll(v, fmt.Sprintf("base-%d-", vmid), fmt.Sprintf("vm-%d-", p.NewID)) + } + if p.Description != "" { + cfg["description"] = p.Description + } + vm := &fakeVM{node: target, name: p.Name, status: "stopped", qmpstatus: "stopped", lock: "clone", config: cfg} + f.vms[p.NewID] = vm + exit := "OK" + if f.cloneFailsWith != "" { + exit = f.cloneFailsWith + } + leaveLock := f.cloneLeavesLock + return f.newTask(node, "qmclone", strconv.Itoa(vmid), exit, func() { + switch { + case exit != "OK": + if f.vms[p.NewID] == vm { + delete(f.vms, p.NewID) + } + case !leaveLock: + vm.lock = "" + } + }), nil +} + +// Resize runs as a task (F20): the same size is a no-op, a smaller one +// fails inside the task with the captured exitstatus. +func (f *fakeAPI) Resize(ctx context.Context, node string, vmid int, disk, size string) (string, error) { + if err := f.enter(ctx, "Resize"); err != nil { + return "", err + } + f.mu.Lock() + defer f.mu.Unlock() + vm, err := f.vm(node, vmid) + if err != nil { + return "", err + } + if vm.lock != "" { + return "", errLocked(vm.lock) + } + cur, ok := vm.config[disk] + if !ok { + return "", fakeErr(500, "fake: disk '"+disk+"' does not exist") + } + curBytes, ok := fakeParseSize(diskSizeOption(cur)) + if !ok { + return "", fakeErr(500, "fake: disk '"+disk+"' has no size") + } + want, ok := fakeParseSize(strings.TrimPrefix(size, "+")) + if !ok { + return "", fakeErr(400, "fake: bad size '"+size+"'") + } + if strings.HasPrefix(size, "+") { + want += curBytes + } + exit := "OK" + if want < curBytes { + exit = "shrinking disks is not supported" + } + return f.newTask(node, "resize", strconv.Itoa(vmid), exit, func() { + if exit == "OK" { + vm.config[disk] = strings.Replace(cur, "size="+diskSizeOption(cur), "size="+fakeFormatSize(want), 1) + } + }), nil +} + +// Start is accepted even while a clone holds the lock (F3). +func (f *fakeAPI) Start(ctx context.Context, node string, vmid int) (string, error) { + if err := f.enter(ctx, "Start"); err != nil { + return "", err + } + f.mu.Lock() + defer f.mu.Unlock() + vm, err := f.vm(node, vmid) + if err != nil { + return "", err + } + exit := "OK" + if f.startFailsWith != "" { + exit = f.startFailsWith + } + return f.newTask(node, "qmstart", strconv.Itoa(vmid), exit, func() { + if exit == "OK" { + vm.status, vm.qmpstatus = "running", "running" + } + }), nil +} + +func (f *fakeAPI) Stop(ctx context.Context, node string, vmid int) (string, error) { + if err := f.enter(ctx, "Stop"); err != nil { + return "", err + } + f.mu.Lock() + defer f.mu.Unlock() + vm, err := f.vm(node, vmid) + if err != nil { + return "", err + } + return f.newTask(node, "qmstop", strconv.Itoa(vmid), "OK", func() { + vm.status, vm.qmpstatus = "stopped", "stopped" + }), nil +} + +func (f *fakeAPI) VMStatus(ctx context.Context, node string, vmid int) (vmStatus, error) { + if err := f.enter(ctx, "VMStatus"); err != nil { + return vmStatus{}, err + } + f.mu.Lock() + defer f.mu.Unlock() + vm, err := f.vm(node, vmid) + if err != nil { + return vmStatus{}, err + } + return vmStatus{Status: vm.status, QMPStatus: vm.qmpstatus, Name: vm.name, Lock: vm.lock}, nil +} + +// Destroy refuses a locked or running VM and removes it when its task +// ends. +func (f *fakeAPI) Destroy(ctx context.Context, node string, vmid int) (string, error) { + if err := f.enter(ctx, "Destroy"); err != nil { + return "", err + } + f.mu.Lock() + defer f.mu.Unlock() + vm, err := f.vm(node, vmid) + if err != nil { + return "", err + } + if vm.lock != "" { + return "", errLocked(vm.lock) + } + if vm.status == "running" { + return "", fakeErr(500, fmt.Sprintf("fake: VM %d is running - destroy failed", vmid)) + } + return f.newTask(node, "qmdestroy", strconv.Itoa(vmid), "OK", func() { + if f.vms[vmid] == vm { + delete(f.vms, vmid) + } + }), nil +} + +// TaskStatus answers running for f.taskPolls polls, then stopped with the +// task's exitstatus, running its effect once. +func (f *fakeAPI) TaskStatus(ctx context.Context, node, upid string) (taskStatus, error) { + if err := f.enter(ctx, "TaskStatus"); err != nil { + return taskStatus{}, err + } + f.mu.Lock() + defer f.mu.Unlock() + t, ok := f.tasks[upid] + if !ok || t.node != node { + return taskStatus{}, fakeErr(500, "fake: no such task '"+upid+"'") + } + t.polls++ + s := taskStatus{UPID: upid, Type: t.kind, ID: t.id, Node: t.node, Status: "running"} + if !t.done && t.polls > f.taskPolls { + t.done = true + if t.onDone != nil { + t.onDone() + } + } + if t.done { + s.Status, s.ExitStatus = "stopped", t.exit + } + return s, nil +} + +// AgentNetworkInterfaces answers with the three messages of F5 in the +// order Proxmox checks them. +func (f *fakeAPI) AgentNetworkInterfaces(ctx context.Context, node string, vmid int) ([]agentInterface, error) { + if err := f.enter(ctx, "AgentNetworkInterfaces"); err != nil { + return nil, err + } + f.mu.Lock() + defer f.mu.Unlock() + vm, err := f.vm(node, vmid) + if err != nil { + return nil, err + } + if agent := vm.config["agent"]; agent != "1" && !strings.HasPrefix(agent, "1,") && !strings.Contains(agent, "enabled=1") { + return nil, fakeErr(500, msgAgentNotConfigured) + } + if vm.status != "running" { + return nil, errVMNotRunning(vmid) + } + if f.agentNotRunning { + return nil, fakeErr(500, msgAgentNotRunning) + } + return slices.Clone(vm.ifaces), nil +} + +// UploadISO records the bytes and overwrites an existing file silently +// (F12). A body that disagrees with size fails, as the real Content-Length +// would. +func (f *fakeAPI) UploadISO(ctx context.Context, node, storage, filename string, r io.Reader, size int64) (string, error) { + if err := f.enter(ctx, "UploadISO"); err != nil { + return "", err + } + var content []byte + if r != nil { + var err error + if content, err = io.ReadAll(r); err != nil { + return "", err + } + } + f.mu.Lock() + defer f.mu.Unlock() + if _, ok := f.storages[storage]; !ok { + return "", fakeErr(500, "fake: storage '"+storage+"' does not exist") + } + if int64(len(content)) != size { + return "", fakeErr(400, fmt.Sprintf("fake: body has %d bytes, size says %d", len(content), size)) + } + if f.isos[storage] == nil { + f.isos[storage] = make(map[string][]byte) + } + f.isos[storage][filename] = content + return f.newTask(node, "imgcopy", "", "OK", nil), nil +} + +// DeleteVolume is a task that ends OK whether or not the volume exists +// (F13). +func (f *fakeAPI) DeleteVolume(ctx context.Context, node, storage, volid string) (string, error) { + if err := f.enter(ctx, "DeleteVolume"); err != nil { + return "", err + } + f.mu.Lock() + defer f.mu.Unlock() + if _, ok := f.storages[storage]; !ok { + return "", fakeErr(500, "fake: storage '"+storage+"' does not exist") + } + filename := strings.TrimPrefix(volid, storage+":iso/") + return f.newTask(node, "imgdel", storage, "OK", func() { + delete(f.isos[storage], filename) + }), nil +} + +// diskSizeOption returns the size= option of a disk value. +func diskSizeOption(disk string) string { + for opt := range strings.SplitSeq(disk, ",") { + if v, ok := strings.CutPrefix(opt, "size="); ok { + return v + } + } + return "" +} + +// fakeParseSize reads Proxmox's [KMGT] sizes into bytes. +func fakeParseSize(s string) (int64, bool) { + if s == "" { + return 0, false + } + shift := map[byte]uint{'K': 10, 'M': 20, 'G': 30, 'T': 40} + unit, ok := shift[s[len(s)-1]] + if ok { + s = s[:len(s)-1] + } + n, err := strconv.ParseInt(s, 10, 64) + if err != nil || n < 0 { + return 0, false + } + return n << unit, true +} + +func fakeFormatSize(b int64) string { + if b%(1<<30) == 0 { + return strconv.FormatInt(b>>30, 10) + "G" + } + return strconv.FormatInt(b>>20, 10) + "M" +} + +// TestFakeAPI_errorsMatchCaptures keeps the fake's error texts equal to +// what Proxmox sent, so the provider's matchers are exercised against real +// wording in every test that goes through the fake. +func TestFakeAPI_errorsMatchCaptures(t *testing.T) { + t.Parallel() + tests := []struct { + capture string + fake *apiError + matcher func(error) bool + }{ + {capture: "err_vm_missing", fake: errVMMissing("pv1-1", 999999), matcher: isNotExist}, + {capture: "err_destroy_missing", fake: errVMMissing("pv1-1", 199999), matcher: isNotExist}, + {capture: "err_config_while_clone_locked", fake: errLocked("clone"), matcher: isLocked}, + {capture: "err_newid_in_use", fake: errNewIDInUse(199999), matcher: isAlreadyExists}, + {capture: "err_agent_vm_stopped_agent_enabled", fake: errVMNotRunning(199999), matcher: isVMNotRunning}, + {capture: "err_agent_not_running", fake: fakeErr(500, msgAgentNotRunning), matcher: isAgentNotRunning}, + {capture: "err_agent_vm_stopped", fake: fakeErr(500, msgAgentNotConfigured), matcher: isAgentNotConfigured}, + } + for _, tc := range tests { + t.Run(tc.capture, func(t *testing.T) { + t.Parallel() + want := loadAPIError(t, tc.capture) + if tc.fake.Status != want.Status || tc.fake.Message != want.Message { + t.Errorf("fake error = %d %q, capture = %d %q", tc.fake.Status, tc.fake.Message, want.Status, want.Message) + } + if !tc.matcher(tc.fake) { + t.Errorf("matcher does not accept the fake's %q", tc.fake.Message) + } + }) + } +} + +// pollTask polls a fake task until it stops and returns its exitstatus. +func pollTask(t *testing.T, f *fakeAPI, upid string) string { + t.Helper() + for range 10 { + s, err := f.TaskStatus(context.Background(), fakeNode, upid) + if err != nil { + t.Fatalf("TaskStatus(%s) error = %v", upid, err) + } + if s.Status == "stopped" { + return s.ExitStatus + } + } + t.Fatalf("task %s still running after 10 polls", upid) + return "" +} + +func resourceByID(t *testing.T, f *fakeAPI, vmid int) (vmResource, bool) { + t.Helper() + rs, err := f.ClusterResources(context.Background()) + if err != nil { + t.Fatalf("ClusterResources error = %v", err) + } + for _, r := range rs { + if r.VMID == vmid { + return r, true + } + } + return vmResource{}, false +} + +func TestFakeAPI_cloneLocksUntilTaskEnds(t *testing.T) { + t.Parallel() + ctx := context.Background() + f := newFakeAPI() + f.addTemplate(9001, "debian-13-template") + f.taskPolls = 1 + upid, err := f.Clone(ctx, fakeNode, 9001, cloneParams{NewID: 150042, Name: "proxy-x", Description: "d"}) + if err != nil { + t.Fatalf("Clone error = %v", err) + } + if r, _ := resourceByID(t, f, 150042); r.Name != "" || r.Lock != "clone" || r.Status != "unknown" { + t.Errorf("mid-clone entry = %+v, want nameless, lock clone, status unknown (F8)", r) + } + if err := f.SetVMConfig(ctx, fakeNode, 150042, map[string]string{"cores": "1"}); !isLocked(err) { + t.Errorf("SetVMConfig mid-clone error = %v, want the lock answer (F7)", err) + } + if _, err := f.Clone(ctx, fakeNode, 9001, cloneParams{NewID: 150042, Name: "proxy-y"}); !isAlreadyExists(err) { + t.Errorf("second Clone onto 150042 error = %v, want config file already exists (F4)", err) + } + if s, _ := f.TaskStatus(ctx, fakeNode, upid); s.Status != "running" { + t.Errorf("first poll = %+v, want running with taskPolls 1", s) + } + if exit := pollTask(t, f, upid); exit != "OK" { + t.Errorf("clone exitstatus = %q, want OK", exit) + } + r, _ := resourceByID(t, f, 150042) + if r.Name != "proxy-x" || r.Lock != "" || r.Status != "stopped" || r.Template != 0 { + t.Errorf("cloned entry = %+v, want named, unlocked, stopped", r) + } + cfg, err := f.VMConfig(ctx, fakeNode, 150042) + if err != nil || cfg["description"] != "d" || cfg["scsi0"] != "local-lvm:vm-150042-disk-1,size=3G" || cfg["template"] != "" { + t.Errorf("clone config = %v, %v", cfg, err) + } +} + +func TestFakeAPI_cloneFailureAndLeftLock(t *testing.T) { + t.Parallel() + ctx := context.Background() + t.Run("failed clone removes the VM", func(t *testing.T) { + t.Parallel() + f := newFakeAPI() + f.addTemplate(9001, "debian-13-template") + f.cloneFailsWith = "clone failed: out of space" + upid, _ := f.Clone(ctx, fakeNode, 9001, cloneParams{NewID: 150042, Name: "proxy-x"}) + if exit := pollTask(t, f, upid); exit != "clone failed: out of space" { + t.Errorf("exitstatus = %q", exit) + } + if _, ok := resourceByID(t, f, 150042); ok { + t.Error("failed clone left its VM") + } + }) + t.Run("interrupted clone keeps the lock", func(t *testing.T) { + t.Parallel() + f := newFakeAPI() + f.addTemplate(9001, "debian-13-template") + f.cloneLeavesLock = true + upid, _ := f.Clone(ctx, fakeNode, 9001, cloneParams{NewID: 150042, Name: "proxy-x"}) + pollTask(t, f, upid) + if r, _ := resourceByID(t, f, 150042); r.Lock != "clone" { + t.Errorf("entry = %+v, want lock clone kept", r) + } + }) +} + +func TestFakeAPI_powerAgentAndDestroy(t *testing.T) { + t.Parallel() + ctx := context.Background() + f := newFakeAPI() + vm := f.addVM(150042, "proxy-x", "stopped") + vm.ifaces = []agentInterface{{Name: "ens18", IPAddresses: []agentIP{{Address: "192.168.4.147", Type: "ipv4", Prefix: 24}}}} + + if _, err := f.AgentNetworkInterfaces(ctx, fakeNode, 150042); !isVMNotRunning(err) { + t.Errorf("agent on a stopped VM error = %v, want VM is not running", err) + } + upid, _ := f.Start(ctx, fakeNode, 150042) + pollTask(t, f, upid) + if s, _ := f.VMStatus(ctx, fakeNode, 150042); s.Status != "running" || s.QMPStatus != "running" { + t.Errorf("status after start = %+v", s) + } + f.agentNotRunning = true + if _, err := f.AgentNetworkInterfaces(ctx, fakeNode, 150042); !isAgentNotRunning(err) { + t.Errorf("agent error = %v, want agent not running", err) + } + f.agentNotRunning = false + if ifaces, err := f.AgentNetworkInterfaces(ctx, fakeNode, 150042); err != nil || firstIPv4(ifaces) != "192.168.4.147" { + t.Errorf("agent = %v, %v", ifaces, err) + } + if _, err := f.Destroy(ctx, fakeNode, 150042); err == nil { + t.Error("Destroy of a running VM succeeded") + } + upid, _ = f.Stop(ctx, fakeNode, 150042) + pollTask(t, f, upid) + upid, err := f.Destroy(ctx, fakeNode, 150042) + if err != nil { + t.Fatalf("Destroy error = %v", err) + } + pollTask(t, f, upid) + if _, err := f.VMConfig(ctx, fakeNode, 150042); !isNotExist(err) { + t.Errorf("VMConfig after destroy error = %v, want does not exist", err) + } + if _, err := f.VMStatus(ctx, "pv1-2", 150042); !isNotExist(err) { + t.Errorf("lookup on another node error = %v, want does not exist", err) + } +} + +func TestFakeAPI_resizeIsATask(t *testing.T) { + t.Parallel() + tests := []struct { + name string + size string + wantExit string + wantDisk string + }{ + {name: "grow absolute", size: "8G", wantExit: "OK", wantDisk: "local-lvm:vm-150042-disk-1,size=8G"}, + {name: "grow relative", size: "+1G", wantExit: "OK", wantDisk: "local-lvm:vm-150042-disk-1,size=4G"}, + {name: "same size", size: "3G", wantExit: "OK", wantDisk: "local-lvm:vm-150042-disk-1,size=3G"}, + {name: "shrink fails in the task", size: "2G", wantExit: loadTaskExit(t, "task_status_stopped_error"), wantDisk: "local-lvm:vm-150042-disk-1,size=3G"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + f := newFakeAPI() + f.addVM(150042, "proxy-x", "stopped") + upid, err := f.Resize(context.Background(), fakeNode, 150042, "scsi0", tc.size) + if err != nil { + t.Fatalf("Resize error = %v", err) + } + if exit := pollTask(t, f, upid); exit != tc.wantExit { + t.Errorf("exitstatus = %q, want %q", exit, tc.wantExit) + } + if vm, _ := f.vmSnapshot(150042); vm.config["scsi0"] != tc.wantDisk { + t.Errorf("scsi0 = %q, want %q", vm.config["scsi0"], tc.wantDisk) + } + }) + } +} + +func loadTaskExit(t *testing.T, name string) string { + t.Helper() + return decodeData[taskStatus](t, name).ExitStatus +} + +func TestFakeAPI_uploadOverwritesAndDeleteIsIdempotent(t *testing.T) { + t.Parallel() + ctx := context.Background() + f := newFakeAPI() + for _, body := range []string{"first", "second"} { + upid, err := f.UploadISO(ctx, fakeNode, fakeISOStorage, "proxy-x-seed.iso", strings.NewReader(body), int64(len(body))) + if err != nil { + t.Fatalf("UploadISO error = %v", err) + } + pollTask(t, f, upid) + } + if b, _ := f.iso(fakeISOStorage, "proxy-x-seed.iso"); string(b) != "second" { + t.Errorf("iso = %q, want the second upload (F12 overwrite)", b) + } + if _, err := f.UploadISO(ctx, fakeNode, fakeISOStorage, "x.iso", strings.NewReader("abc"), 4); err == nil { + t.Error("UploadISO with a wrong size succeeded") + } + for range 2 { + upid, err := f.DeleteVolume(ctx, fakeNode, fakeISOStorage, seedVolID(fakeISOStorage, "proxy-x")) + if err != nil { + t.Fatalf("DeleteVolume error = %v", err) + } + if exit := pollTask(t, f, upid); exit != "OK" { + t.Errorf("delete exitstatus = %q, want OK even when absent (F13)", exit) + } + } + if _, ok := f.iso(fakeISOStorage, "proxy-x-seed.iso"); ok { + t.Error("iso still present after delete") + } +} + +func TestFakeAPI_hooksBlocksAndLog(t *testing.T) { + t.Parallel() + f := newFakeAPI() + boom := errors.New("boom") + f.failWith("Version", boom) + if _, err := f.Version(context.Background()); !errors.Is(err, boom) { + t.Errorf("Version error = %v, want the hook's", err) + } + f.blockOn("ClusterResources") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := f.ClusterResources(ctx); !errors.Is(err, context.Canceled) { + t.Errorf("blocked call with ended ctx error = %v, want context.Canceled", err) + } + if got := f.callLog(); !slices.Equal(got, []string{"Version", "ClusterResources"}) { + t.Errorf("call log = %v", got) + } + if f.callCount("Version") != 1 { + t.Errorf("Version count = %d, want 1", f.callCount("Version")) + } +} diff --git a/internal/provider/proxmox/observe.go b/internal/provider/proxmox/observe.go new file mode 100644 index 0000000..7ef38e2 --- /dev/null +++ b/internal/provider/proxmox/observe.go @@ -0,0 +1,259 @@ +package proxmox + +import ( + "context" + "errors" + "io" + "strconv" + "time" + + logf "sigs.k8s.io/controller-runtime/pkg/log" +) + +// observed is the logging decorator of spec entry 20: one V(1) +// "proxmox api call" line per seam call, with request and response detail +// at V(2). It starts no span: the transport already traces each request +// (spec entry 17), so nothing is counted twice. +type observed struct { + inner api + provider string +} + +var _ api = (*observed)(nil) + +func observe(inner api, providerName string) api { + return &observed{inner: inner, provider: providerName} +} + +// target is what a call names at V(1); zero fields are left out. +type target struct { + node string + vmid int + storage string + upid string + extra []any +} + +// detail returns V(2)-only key/value pairs. It runs only when V(2) is on. +type detail func() []any + +// call wraps one seam call. fn returns the V(2) detail and may fill in +// t.upid from the call's answer. +func (o *observed) call(ctx context.Context, name string, t *target, fn func() (detail, error)) error { + log := logf.FromContext(ctx).WithName("proxmox").WithValues("provider", o.provider) + start := time.Now() + more, err := fn() + kv := make([]any, 0, 16) + kv = append(kv, "call", name) + if t.node != "" { + kv = append(kv, "node", t.node) + } + if t.vmid != 0 { + kv = append(kv, "vmid", t.vmid) + } + if t.storage != "" { + kv = append(kv, "storage", t.storage) + } + if t.upid != "" { + kv = append(kv, "upid", t.upid) + } + kv = append(kv, t.extra...) + kv = append(kv, "duration", time.Since(start)) + if err != nil { + var ae *apiError + if errors.As(err, &ae) { + kv = append(kv, "httpStatus", ae.Status) + } + kv = append(kv, "error", err.Error()) + } + if more != nil && log.V(2).Enabled() { + kv = append(kv, more()...) + } + log.V(1).Info("proxmox api call", kv...) + return err +} + +func (o *observed) Version(ctx context.Context) (v versionInfo, err error) { + err = o.call(ctx, "Version", &target{}, func() (detail, error) { + v, err = o.inner.Version(ctx) + return func() []any { return []any{"version", v.Version} }, err + }) + return v, err +} + +func (o *observed) NodeStatus(ctx context.Context, node string) (s nodeStatus, err error) { + err = o.call(ctx, "NodeStatus", &target{node: node}, func() (detail, error) { + s, err = o.inner.NodeStatus(ctx, node) + return func() []any { return []any{"pveVersion", s.PVEVersion, "kernel", s.CurrentKernel.Release} }, err + }) + return s, err +} + +func (o *observed) StorageStatus(ctx context.Context, node, storage string) (s storageStatus, err error) { + err = o.call(ctx, "StorageStatus", &target{node: node, storage: storage}, func() (detail, error) { + s, err = o.inner.StorageStatus(ctx, node, storage) + return func() []any { + return []any{"type", s.Type, "content", s.Content, "active", s.Active, "avail", s.Avail} + }, err + }) + return s, err +} + +func (o *observed) ClusterResources(ctx context.Context) (rs []vmResource, err error) { + err = o.call(ctx, "ClusterResources", &target{}, func() (detail, error) { + rs, err = o.inner.ClusterResources(ctx) + return func() []any { return []any{"count", len(rs), "vms", resourceSummary(rs)} }, err + }) + return rs, err +} + +func (o *observed) VMConfig(ctx context.Context, node string, vmid int) (cfg vmConfig, err error) { + err = o.call(ctx, "VMConfig", &target{node: node, vmid: vmid}, func() (detail, error) { + cfg, err = o.inner.VMConfig(ctx, node, vmid) + return func() []any { return []any{"config", configSummary(cfg)} }, err + }) + return cfg, err +} + +func (o *observed) SetVMConfig(ctx context.Context, node string, vmid int, params map[string]string) error { + return o.call(ctx, "SetVMConfig", &target{node: node, vmid: vmid}, func() (detail, error) { + return func() []any { return []any{"params", params} }, o.inner.SetVMConfig(ctx, node, vmid, params) + }) +} + +// Clone names the new VM at V(1); the description, which holds the mark, +// only at V(2). +func (o *observed) Clone(ctx context.Context, node string, vmid int, p cloneParams) (upid string, err error) { + t := &target{node: node, vmid: vmid, extra: []any{"newid", p.NewID, "name", p.Name}} + err = o.call(ctx, "Clone", t, func() (detail, error) { + upid, err = o.inner.Clone(ctx, node, vmid, p) + t.upid = upid + return func() []any { + return []any{"description", p.Description, "storage", p.Storage, "target", p.Target} + }, err + }) + return upid, err +} + +func (o *observed) Resize(ctx context.Context, node string, vmid int, disk, size string) (upid string, err error) { + t := &target{node: node, vmid: vmid} + err = o.call(ctx, "Resize", t, func() (detail, error) { + upid, err = o.inner.Resize(ctx, node, vmid, disk, size) + t.upid = upid + return func() []any { return []any{"disk", disk, "size", size} }, err + }) + return upid, err +} + +func (o *observed) Start(ctx context.Context, node string, vmid int) (upid string, err error) { + t := &target{node: node, vmid: vmid} + err = o.call(ctx, "Start", t, func() (detail, error) { + upid, err = o.inner.Start(ctx, node, vmid) + t.upid = upid + return nil, err + }) + return upid, err +} + +func (o *observed) Stop(ctx context.Context, node string, vmid int) (upid string, err error) { + t := &target{node: node, vmid: vmid} + err = o.call(ctx, "Stop", t, func() (detail, error) { + upid, err = o.inner.Stop(ctx, node, vmid) + t.upid = upid + return nil, err + }) + return upid, err +} + +func (o *observed) VMStatus(ctx context.Context, node string, vmid int) (s vmStatus, err error) { + err = o.call(ctx, "VMStatus", &target{node: node, vmid: vmid}, func() (detail, error) { + s, err = o.inner.VMStatus(ctx, node, vmid) + return func() []any { return []any{"status", s.Status, "qmpstatus", s.QMPStatus, "lock", s.Lock} }, err + }) + return s, err +} + +func (o *observed) Destroy(ctx context.Context, node string, vmid int) (upid string, err error) { + t := &target{node: node, vmid: vmid} + err = o.call(ctx, "Destroy", t, func() (detail, error) { + upid, err = o.inner.Destroy(ctx, node, vmid) + t.upid = upid + return nil, err + }) + return upid, err +} + +func (o *observed) TaskStatus(ctx context.Context, node, upid string) (s taskStatus, err error) { + err = o.call(ctx, "TaskStatus", &target{node: node, upid: upid}, func() (detail, error) { + s, err = o.inner.TaskStatus(ctx, node, upid) + return func() []any { return []any{"status", s.Status, "exitstatus", s.ExitStatus} }, err + }) + return s, err +} + +func (o *observed) AgentNetworkInterfaces(ctx context.Context, node string, vmid int) (ifaces []agentInterface, err error) { + err = o.call(ctx, "AgentNetworkInterfaces", &target{node: node, vmid: vmid}, func() (detail, error) { + ifaces, err = o.inner.AgentNetworkInterfaces(ctx, node, vmid) + return func() []any { return []any{"addresses", agentAddresses(ifaces)} }, err + }) + return ifaces, err +} + +// UploadISO logs the file name and size, never the body: a seed carries +// CloudInit. +func (o *observed) UploadISO(ctx context.Context, node, storage, filename string, r io.Reader, size int64) (upid string, err error) { + t := &target{node: node, storage: storage, extra: []any{"filename", filename}} + err = o.call(ctx, "UploadISO", t, func() (detail, error) { + upid, err = o.inner.UploadISO(ctx, node, storage, filename, r, size) + t.upid = upid + return func() []any { return []any{"bytes", size} }, err + }) + return upid, err +} + +func (o *observed) DeleteVolume(ctx context.Context, node, storage, volid string) (upid string, err error) { + t := &target{node: node, storage: storage, extra: []any{"volid", volid}} + err = o.call(ctx, "DeleteVolume", t, func() (detail, error) { + upid, err = o.inner.DeleteVolume(ctx, node, storage, volid) + t.upid = upid + return nil, err + }) + return upid, err +} + +// resourceSummary is one "::" per entry, with the lock +// appended when set. +func resourceSummary(rs []vmResource) []string { + out := make([]string, len(rs)) + for i, r := range rs { + s := strconv.Itoa(r.VMID) + ":" + r.Name + ":" + r.Status + if r.Lock != "" { + s += ":lock=" + r.Lock + } + out[i] = s + } + return out +} + +// summaryKeys are the config keys the provider reads or writes. +var summaryKeys = []string{"name", "lock", "tags", "description", "cores", "memory", "agent", "net0", "ide2", "boot", "bootdisk", "scsi0", "virtio0", "sata0", "ide0", "template"} + +func configSummary(cfg vmConfig) map[string]string { + out := make(map[string]string, len(summaryKeys)) + for _, k := range summaryKeys { + if v, ok := cfg[k]; ok { + out[k] = v + } + } + return out +} + +func agentAddresses(ifaces []agentInterface) []string { + var out []string + for _, iface := range ifaces { + for _, a := range iface.IPAddresses { + out = append(out, iface.Name+"="+a.Address) + } + } + return out +} diff --git a/internal/provider/proxmox/observe_test.go b/internal/provider/proxmox/observe_test.go new file mode 100644 index 0000000..64ff65a --- /dev/null +++ b/internal/provider/proxmox/observe_test.go @@ -0,0 +1,229 @@ +package proxmox + +import ( + "context" + "errors" + "reflect" + "strconv" + "strings" + "sync" + "testing" + + "github.com/go-logr/logr/funcr" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + logf "sigs.k8s.io/controller-runtime/pkg/log" +) + +// logCapture collects funcr output, safe for concurrent use. +type logCapture struct { + mu sync.Mutex + lines []string +} + +// context returns a ctx whose logger writes into c at verbosity. +func (c *logCapture) context(verbosity int) context.Context { + log := funcr.New(func(prefix, args string) { + c.mu.Lock() + defer c.mu.Unlock() + c.lines = append(c.lines, prefix+" "+args) + }, funcr.Options{Verbosity: verbosity}) + return logf.IntoContext(context.Background(), log) +} + +func (c *logCapture) all() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.lines...) +} + +func (c *logCapture) matching(substr string) []string { + var out []string + for _, l := range c.all() { + if strings.Contains(l, substr) { + out = append(out, l) + } + } + return out +} + +const callLine = `"msg"="proxmox api call"` + +func newObserveFixture(verbosity int) (*fakeAPI, api, *logCapture, context.Context) { + fake := newFakeAPI() + logs := &logCapture{} + return fake, observe(fake, "proxmox-home"), logs, logs.context(verbosity) +} + +func TestObserve_everySeamMethodLogsOneLine(t *testing.T) { + t.Parallel() + for m := range reflect.TypeFor[api]().Methods() { + t.Run(m.Name, func(t *testing.T) { + t.Parallel() + fake, a, logs, ctx := newObserveFixture(1) + args := make([]reflect.Value, m.Type.NumIn()) + args[0] = reflect.ValueOf(ctx) + for j := 1; j < m.Type.NumIn(); j++ { + args[j] = reflect.Zero(m.Type.In(j)) + } + reflect.ValueOf(a).MethodByName(m.Name).Call(args) + + lines := logs.matching(callLine) + if len(lines) != 1 || len(logs.all()) != 1 { + t.Fatalf("got %d call lines of %d, want exactly 1:\n%s", len(lines), len(logs.all()), strings.Join(logs.all(), "\n")) + } + if !strings.HasPrefix(lines[0], "proxmox ") { + t.Errorf("line is not from the proxmox logger: %s", lines[0]) + } + for _, want := range []string{`"call"="` + m.Name + `"`, `"provider"="proxmox-home"`, `"duration"=`, `"level"=1`} { + if !strings.Contains(lines[0], want) { + t.Errorf("line lacks %s: %s", want, lines[0]) + } + } + if n := fake.callCount(m.Name); n != 1 { + t.Errorf("inner %s called %d times, want 1", m.Name, n) + } + }) + } +} + +// TestObserve_startsNoSpan keeps the decorator from duplicating the +// transport's HTTP spans (spec entry 17). +func TestObserve_startsNoSpan(t *testing.T) { + t.Parallel() + sr := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr)) + t.Cleanup(func() { _ = tp.Shutdown(context.Background()) }) + fake, a, _, ctx := newObserveFixture(1) + fake.addVM(150042, "proxy-x", "running") + ctx, parent := tp.Tracer("test").Start(ctx, "provider.get") + _, _ = a.ClusterResources(ctx) + _, _ = a.VMConfig(ctx, fakeNode, 150042) + parent.End() + if ended := sr.Ended(); len(ended) != 1 || ended[0].Name() != "provider.get" { + t.Errorf("spans = %d, want only the parent", len(ended)) + } +} + +func TestObserve_identifiersAtV1(t *testing.T) { + t.Parallel() + fake, a, logs, ctx := newObserveFixture(1) + fake.addTemplate(9001, "debian-13-template") + upid, err := a.Clone(ctx, fakeNode, 9001, cloneParams{NewID: 150042, Name: "proxy-x", Description: "proxy-operator-managed=true"}) + if err != nil { + t.Fatalf("Clone error = %v", err) + } + _, _ = a.TaskStatus(ctx, fakeNode, upid) + _, _ = a.DeleteVolume(ctx, fakeNode, fakeISOStorage, seedVolID(fakeISOStorage, "proxy-x")) + lines := logs.matching(callLine) + if len(lines) != 3 { + t.Fatalf("got %d lines, want 3:\n%s", len(lines), strings.Join(logs.all(), "\n")) + } + wants := [][]string{ + {`"call"="Clone" "node"="pv1-1" "vmid"=9001 "upid"="` + upid + `" "newid"=150042 "name"="proxy-x"`}, + {`"call"="TaskStatus" "node"="pv1-1" "upid"="` + upid + `"`}, + {`"call"="DeleteVolume" "node"="pv1-1" "storage"="proxy-seeds"`, `"volid"="proxy-seeds:iso/proxy-x-seed.iso"`, `"upid"="UPID:pv1-1:`}, + } + for i, want := range wants { + for _, w := range want { + if !strings.Contains(lines[i], w) { + t.Errorf("line %d lacks %s:\n%s", i, w, lines[i]) + } + } + } + if strings.Contains(lines[0], "description") { + t.Errorf("the description reached V(1): %s", lines[0]) + } +} + +func TestObserve_detailOnlyAtV2(t *testing.T) { + t.Parallel() + tests := []struct { + name string + verbosity int + want bool + }{ + {name: "V1 has no detail", verbosity: 1, want: false}, + {name: "V2 has detail", verbosity: 2, want: true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fake, a, logs, ctx := newObserveFixture(tc.verbosity) + fake.addTemplate(9001, "debian-13-template") + vm := fake.addVM(150042, "proxy-x", "running") + vm.ifaces = []agentInterface{{Name: "ens18", IPAddresses: []agentIP{{Address: "192.168.4.147", Type: "ipv4"}}}} + _, _ = a.ClusterResources(ctx) + _, _ = a.VMConfig(ctx, fakeNode, 150042) + _ = a.SetVMConfig(ctx, fakeNode, 9001, map[string]string{"cores": "3"}) + _, _ = a.Clone(ctx, fakeNode, 9001, cloneParams{NewID: 150043, Name: "proxy-y", Description: "mark-line"}) + _, _ = a.AgentNetworkInterfaces(ctx, fakeNode, 150042) + all := strings.Join(logs.all(), "\n") + for _, detail := range []string{`"vms"=["9001:debian-13-template:stopped" "150042:proxy-x:running"]`, `"config"={`, `"params"={"cores"="3"}`, `"description"="mark-line"`, `"addresses"=["ens18=192.168.4.147"]`} { + if got := strings.Contains(all, detail); got != tc.want { + t.Errorf("%s logged = %v, want %v:\n%s", detail, got, tc.want, all) + } + } + if got := len(logs.matching(callLine)); got != 5 { + t.Errorf("got %d call lines, want 5", got) + } + }) + } +} + +func TestObserve_uploadBodyNeverLogged(t *testing.T) { + t.Parallel() + const sentinel = "SENTINEL-CLOUDINIT-51c9" + fake, a, logs, ctx := newObserveFixture(2) + body := "#cloud-config\npassword: " + sentinel + "\n" + if _, err := a.UploadISO(ctx, fakeNode, fakeISOStorage, "proxy-x-seed.iso", strings.NewReader(body), int64(len(body))); err != nil { + t.Fatalf("UploadISO error = %v", err) + } + if got, _ := fake.iso(fakeISOStorage, "proxy-x-seed.iso"); string(got) != body { + t.Errorf("fake got %q, want the body passed through", got) + } + all := strings.Join(logs.all(), "\n") + if strings.Contains(all, sentinel) { + t.Errorf("upload body leaked into logs:\n%s", all) + } + lines := logs.matching(`"call"="UploadISO"`) + if len(lines) != 1 || !strings.Contains(lines[0], `"filename"="proxy-x-seed.iso"`) || !strings.Contains(lines[0], `"bytes"=`+strconv.Itoa(len(body))) { + t.Errorf("upload line lacks filename or bytes: %v", lines) + } +} + +func TestObserve_errorFieldsOnFailure(t *testing.T) { + t.Parallel() + fake, a, logs, ctx := newObserveFixture(1) + _, err := a.VMConfig(ctx, fakeNode, 999999) + if !isNotExist(err) { + t.Fatalf("error = %v, want the does-not-exist apiError through the decorator", err) + } + lines := logs.matching(callLine) + if len(lines) != 1 { + t.Fatalf("got %d lines, want 1", len(lines)) + } + for _, want := range []string{`"httpStatus"=500`, `"error"="proxmox API 500: Configuration file 'nodes/pv1-1/qemu-server/999999.conf' does not exist"`} { + if !strings.Contains(lines[0], want) { + t.Errorf("line lacks %s: %s", want, lines[0]) + } + } + + fake.failWith("Version", errors.New("dial tcp: connection refused")) + _, _ = a.Version(ctx) + lines = logs.matching(`"call"="Version"`) + if len(lines) != 1 || strings.Contains(lines[0], "httpStatus") || !strings.Contains(lines[0], `"error"="dial tcp: connection refused"`) { + t.Errorf("non-API failure line = %v, want the error without httpStatus", lines) + } +} + +func TestObserve_v0Silent(t *testing.T) { + t.Parallel() + fake, a, logs, ctx := newObserveFixture(0) + fake.addVM(150042, "proxy-x", "running") + _, _ = a.VMConfig(ctx, fakeNode, 150042) + _, _ = a.VMConfig(ctx, fakeNode, 999999) + if lines := logs.all(); len(lines) != 0 { + t.Errorf("got %d lines at V(0), want none:\n%s", len(lines), strings.Join(lines, "\n")) + } +} diff --git a/internal/provider/proxmox/proxmox.go b/internal/provider/proxmox/proxmox.go index 1b19826..66fe882 100644 --- a/internal/provider/proxmox/proxmox.go +++ b/internal/provider/proxmox/proxmox.go @@ -76,9 +76,11 @@ func New(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, er if pc.Proxmox != nil { cfg = *pc.Proxmox } - // The HTTP client behind the api seam arrives in Phase 4 of the plan; - // until then every method is a stub that makes no request. - p := newProvider(pc.Name, cfg, nil) + client, err := newHTTPAPI(withDefaults(cfg)) + if err != nil { + return nil, fmt.Errorf("proxmox provider %s: %w", pc.Name, err) + } + p := newProvider(pc.Name, cfg, observe(client, pc.Name)) log := p.logger(ctx) if _, err := os.Stat(p.cfg.TokenFile); err != nil { log.Info("WARNING: token file is not readable; every call fails until it is", "path", p.cfg.TokenFile, "reason", err.Error()) @@ -89,7 +91,7 @@ func New(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, er return p, nil } -// newProvider is the constructor tests use, with a in-memory api. +// newProvider is the constructor tests use, with an in-memory api. func newProvider(name string, cfg provider.ProxmoxConfig, a api) *Provider { return &Provider{ name: name, diff --git a/internal/provider/proxmox/proxmox_test.go b/internal/provider/proxmox/proxmox_test.go index 1197571..2add898 100644 --- a/internal/provider/proxmox/proxmox_test.go +++ b/internal/provider/proxmox/proxmox_test.go @@ -95,6 +95,31 @@ func TestNew_nilBlockDoesNotPanic(t *testing.T) { } } +// TestNew_wiresObservedHTTPClient checks that production calls go through +// the logging decorator to the HTTP client, with /api2/json appended and +// the request timeout from the config. +func TestNew_wiresObservedHTTPClient(t *testing.T) { + t.Parallel() + ctx, _ := captureContext() + cfg := testConfig() + cfg.RequestTimeoutSeconds = 45 + p, err := New(ctx, provider.ProviderConfig{Name: "proxmox-home", Type: "proxmox", Proxmox: &cfg}) + if err != nil { + t.Fatalf("New() error = %v", err) + } + o, ok := p.(*Provider).api.(*observed) + if !ok { + t.Fatalf("api = %T, want *observed", p.(*Provider).api) + } + h, ok := o.inner.(*httpAPI) + if !ok { + t.Fatalf("observed.inner = %T, want *httpAPI", o.inner) + } + if o.provider != "proxmox-home" || h.base.String() != "https://192.0.2.8:8006/api2/json" || h.timeout != 45*time.Second || h.tokenFile != cfg.TokenFile { + t.Errorf("wiring = provider %q base %q timeout %v token file %q", o.provider, h.base, h.timeout, h.tokenFile) + } +} + func TestWithDefaults_table(t *testing.T) { t.Parallel() off := false