kacerr/egress-proxies-operator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

proxmox: a Proxmox VE provider built from the provider contract #9

Merged
kacerr merged 13 commits from feat/proxmox-provider into main 2026-09-24 13:12:45 +02:00
Conversation 0 Commits 13 Files Changed 131
+2775 -7
9 changed files with 2775 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit e05d994974 - Show all commits
docs/plans-executions/2026-09-18-1502-proxmox-provider.md
+108 -1
View File
@@ -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 <vmid> is running - destroy failed`, so Phase 7's
stop-before-destroy order is enforced by the fake even though the real
text is unknown.
docs/proxmox-provider-spec.md
+3 -2
View File
@@ -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
internal/provider/proxmox/client.go
+378
View File
@@ -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 <user>@<realm>!<tokenid>=<secret> (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 <user>@<realm>!<tokenid>=<secret>", 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, "<redacted>")
for k, v := range e.Errors {
e.Errors[k] = strings.ReplaceAll(v, secret, "<redacted>")
}
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
}
internal/provider/proxmox/client_test.go
+874
View File
@@ -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(), "<redacted>") {
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)
}
}
internal/provider/proxmox/fake_test.go
+893
View File
@@ -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 <n>[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"))
}
}
internal/provider/proxmox/observe.go
+259
View File
@@ -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 "<vmid>:<name>:<status>" 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
}
internal/provider/proxmox/observe_test.go
+229
View File
@@ -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"))
}
}
internal/provider/proxmox/proxmox.go
+6 -4
View File
@@ -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,
internal/provider/proxmox/proxmox_test.go
+25
View File
@@ -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