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

provider: extract nocloud seed builder and keyedlock from kvm2

Browse Source 
kvm2's NoCloud seed builder moves to internal/provider/nocloud (exported,
with an ErrEmptyName sentinel instead of permanentf) and its keyed locks
to internal/provider/keyedlock, tests moved under their existing names.
kvm2 behaviour is unchanged. proxmox gains the per-name lock and the pure
seed helpers (seedISOName, seedVolID, buildSeed); unlike kvm2 (review
finding S5.1), a seed staging failure there is Transient, not Permanent.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
kacerr
2026-09-23 14:01:27 +00:00
co-authored by Claude Opus 5.5
parent 6e996ae2c7
commit 0a6c8624f3
18 changed files with 504 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/provider/keyedlock/keyedlock.go
+60
View File
@@ -0,0 +1,60 @@
// Package keyedlock provides one-slot semaphores by key, shared by the
// providers that serialise work per instance name or per cached image
// (kvm2, proxmox).
package keyedlock
import (
"context"
"sync"
)
// Locks is a set of one-slot semaphores by key: a semaphore rather than a
// mutex so a waiter can give up with its ctx. Each entry counts the callers
// holding or waiting for it and is dropped from the map at zero, so the map
// does not grow with every key ever seen while two waiters on one key still
// always share one semaphore. The zero value is not usable; call New.
type Locks struct {
mu sync.Mutex
locks map[string]*lock
}
type lock struct {
sem chan struct{}
refs int
}
func New() *Locks {
return &Locks{locks: make(map[string]*lock)}
}
// Acquire takes the lock for key, or gives up with ctx's error. The
// returned func releases it and must be called exactly once.
func (k *Locks) Acquire(ctx context.Context, key string) (release func(), err error) {
k.mu.Lock()
l, ok := k.locks[key]
if !ok {
l = &lock{sem: make(chan struct{}, 1)}
k.locks[key] = l
}
l.refs++
k.mu.Unlock()
select {
case l.sem <- struct{}{}:
return func() {
<-l.sem
k.unref(key, l)
}, nil
case <-ctx.Done():
k.unref(key, l)
return nil, ctx.Err()
}
}
func (k *Locks) unref(key string, l *lock) {
k.mu.Lock()
defer k.mu.Unlock()
l.refs--
if l.refs == 0 {
delete(k.locks, key)
}
}
internal/provider/keyedlock/keyedlock_test.go
+90
View File
@@ -0,0 +1,90 @@
package keyedlock
import (
"context"
"errors"
"sync"
"sync/atomic"
"testing"
"time"
)
func lockCount(k *Locks) int {
k.mu.Lock()
defer k.mu.Unlock()
return len(k.locks)
}
func TestKeyedLocks_entryRemovedWhenUncontended(t *testing.T) {
t.Parallel()
k := New()
release, err := k.Acquire(context.Background(), "a")
if err != nil {
t.Fatalf("acquire error = %v", err)
}
if lockCount(k) != 1 {
t.Errorf("entries while held = %d, want 1", lockCount(k))
}
release()
if lockCount(k) != 0 {
t.Errorf("entries after release = %d, want 0", lockCount(k))
}
if _, err := k.Acquire(context.Background(), "a"); err != nil {
t.Fatalf("acquire after release error = %v", err)
}
}
func TestKeyedLocks_waitersShareOneLock(t *testing.T) {
t.Parallel()
k := New()
const n = 20
var inside, overlaps atomic.Int32
var wg sync.WaitGroup
for range n {
wg.Go(func() {
release, err := k.Acquire(context.Background(), "shared")
if err != nil {
t.Errorf("acquire error = %v", err)
return
}
if inside.Add(1) > 1 {
overlaps.Add(1)
}
time.Sleep(time.Millisecond)
inside.Add(-1)
release()
})
}
wg.Wait()
if overlaps.Load() != 0 {
t.Errorf("%d holders overlapped on one key", overlaps.Load())
}
if lockCount(k) != 0 {
t.Errorf("entries after every holder released = %d, want 0", lockCount(k))
}
}
func TestKeyedLocks_waiterGivesUpWithContext(t *testing.T) {
t.Parallel()
k := New()
release, err := k.Acquire(context.Background(), "a")
if err != nil {
t.Fatalf("acquire error = %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
start := time.Now()
if _, err := k.Acquire(ctx, "a"); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("waiter error = %v, want DeadlineExceeded", err)
}
if time.Since(start) > time.Second {
t.Errorf("waiter took %v, want to give up with its ctx", time.Since(start))
}
if lockCount(k) != 1 {
t.Errorf("entries with the holder still in = %d, want 1: the waiter's exit must not drop the holder's entry", lockCount(k))
}
release()
if lockCount(k) != 0 {
t.Errorf("entries after the holder released = %d, want 0", lockCount(k))
}
}
internal/provider/kvm2/create.go
+3 -2
View File
@@ -11,6 +11,7 @@ import (
"github.com/go-logr/logr"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/nocloud"
)
// The wait after DomainCreate (spec entry 18): Create returns only once the
@@ -191,12 +192,12 @@ func (p *Provider) writeSeed(ctx context.Context, log logr.Logger, be backend, p
return libvirt.StorageVol{}, err
}
installAgent := p.cfg.InstallGuestAgent == nil || *p.cfg.InstallGuestAgent
vendorData, err := buildVendorData(p.cfg.VendorData, installAgent)
vendorData, err := nocloud.BuildVendorData(p.cfg.VendorData, installAgent)
if err != nil {
return libvirt.StorageVol{}, permanentf("building vendor-data for %s: %v", req.Name, err)
}
var iso bytes.Buffer
if err := buildSeedISO(req.Name, req.CloudInit, vendorData, &iso); err != nil {
if err := nocloud.BuildSeedISO(req.Name, req.CloudInit, vendorData, &iso); err != nil {
return libvirt.StorageVol{}, permanentf("building seed ISO for %s: %v", req.Name, err)
}
xmlDoc, err := buildVolumeXML(volumeSpec{Name: name, CapacityBytes: int64(iso.Len()), Format: provider.KVMFormatRaw})
internal/provider/kvm2/create_test.go
+3 -2
View File
@@ -12,6 +12,7 @@ import (
"github.com/digitalocean/go-libvirt"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/nocloud"
)
func TestCreate_repeatReturnsSameIDAndCreatesNothing(t *testing.T) {
@@ -95,7 +96,7 @@ func TestCreate_resumesAfterEachPartialStep(t *testing.T) {
},
check: func(t *testing.T, fx *fixture) {
got := fx.fake.vol(seed)
if got == nil || len(got.content) <= 3 || !bytes.Contains(got.content, []byte(seedLabel)) {
if got == nil || len(got.content) <= 3 || !bytes.Contains(got.content, []byte(nocloud.SeedLabel)) {
t.Errorf("seed volume not rebuilt")
}
},
@@ -461,7 +462,7 @@ func TestCreate_seedAlwaysRebuilt(t *testing.T) {
if got := fx.fake.callCount("StorageVolUpload"); got != uploads+1 {
t.Errorf("StorageVolUpload calls = %d, want %d", got, uploads+1)
}
if len(second.content) != len(first.content) || !bytes.Contains(second.content, []byte(seedLabel)) {
if len(second.content) != len(first.content) || !bytes.Contains(second.content, []byte(nocloud.SeedLabel)) {
t.Errorf("rebuilt seed is %d bytes, first was %d; want a complete seed of the same size", len(second.content), len(first.content))
}
}
internal/provider/kvm2/image.go
+1 -1
View File
@@ -73,7 +73,7 @@ func resolveImage(entry string, cfg provider.KVMConfig, image string) (imageSour
// returns promptly (contract §1.5) instead of waiting out a 20-minute
// transfer.
func (p *Provider) lockImage(ctx context.Context, hostName, volume string) (func(), error) {
release, err := p.imageLocks.acquire(ctx, hostName+"/"+volume)
release, err := p.imageLocks.Acquire(ctx, hostName+"/"+volume)
if err != nil {
return nil, fmt.Errorf("%w: waiting for image %s on host %s: %w", provider.ErrTransient, volume, hostName, err)
}
internal/provider/kvm2/kvm2.go
+6 -57
View File
@@ -9,13 +9,13 @@ import (
"fmt"
"net/http"
"os"
"sync"
"time"
"github.com/go-logr/logr"
logf "sigs.k8s.io/controller-runtime/pkg/log"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/keyedlock"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
)
@@ -41,12 +41,12 @@ type Provider struct {
runningPoll time.Duration
// imageLocks serialises image fetches per (host, volume), spec entry 19.
imageLocks *keyedLocks
imageLocks *keyedlock.Locks
// nameLocks serialises Create and Delete per instance name, spec entry
// 19, so two Deletes never race on destroy and undefine and a Delete
// cannot interleave with a Create's steps.
nameLocks *keyedLocks
nameLocks *keyedlock.Locks
}
// providerOption adjusts what newProvider builds; tests inject an HTTP
@@ -111,8 +111,8 @@ func newProvider(name string, cfg provider.KVMConfig, d dialer, opts ...provider
http: http.DefaultClient,
runningWait: defaultRunningWait,
runningPoll: defaultRunningPoll,
imageLocks: newKeyedLocks(),
nameLocks: newKeyedLocks(),
imageLocks: keyedlock.New(),
nameLocks: keyedlock.New(),
}
for _, opt := range opts {
opt(p)
@@ -187,60 +187,9 @@ func (p *Provider) wrapErr(op, id string, err error) error {
// so a caller whose ctx ends while another caller's Create is downloading
// an image returns promptly.
func (p *Provider) lockName(ctx context.Context, name string) (func(), error) {
release, err := p.nameLocks.acquire(ctx, name)
release, err := p.nameLocks.Acquire(ctx, name)
if err != nil {
return nil, fmt.Errorf("%w: waiting for the lock on %s: %w", provider.ErrTransient, name, err)
}
return release, nil
}
// keyedLocks is a set of one-slot semaphores by key (spec entry 19): a
// semaphore rather than a mutex so a waiter can give up with its ctx. Each
// entry counts the callers holding or waiting for it and is dropped from
// the map at zero, so the map does not grow with every name ever seen while
// two waiters on one key still always share one semaphore.
type keyedLocks struct {
mu sync.Mutex
locks map[string]*keyedLock
}
type keyedLock struct {
sem chan struct{}
refs int
}
func newKeyedLocks() *keyedLocks {
return &keyedLocks{locks: make(map[string]*keyedLock)}
}
// acquire takes the lock for key, or gives up with ctx's error. The
// returned func releases it and must be called exactly once.
func (k *keyedLocks) acquire(ctx context.Context, key string) (func(), error) {
k.mu.Lock()
l, ok := k.locks[key]
if !ok {
l = &keyedLock{sem: make(chan struct{}, 1)}
k.locks[key] = l
}
l.refs++
k.mu.Unlock()
select {
case l.sem <- struct{}{}:
return func() {
<-l.sem
k.unref(key, l)
}, nil
case <-ctx.Done():
k.unref(key, l)
return nil, ctx.Err()
}
}
func (k *keyedLocks) unref(key string, l *keyedLock) {
k.mu.Lock()
defer k.mu.Unlock()
l.refs--
if l.refs == 0 {
delete(k.locks, key)
}
}
internal/provider/kvm2/kvm2_test.go
-81
View File
@@ -5,7 +5,6 @@ import (
"errors"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
@@ -267,86 +266,6 @@ func TestWithDefaults_doesNotAliasInput(t *testing.T) {
}
}
func lockCount(k *keyedLocks) int {
k.mu.Lock()
defer k.mu.Unlock()
return len(k.locks)
}
func TestKeyedLocks_entryRemovedWhenUncontended(t *testing.T) {
t.Parallel()
k := newKeyedLocks()
release, err := k.acquire(context.Background(), "a")
if err != nil {
t.Fatalf("acquire error = %v", err)
}
if lockCount(k) != 1 {
t.Errorf("entries while held = %d, want 1", lockCount(k))
}
release()
if lockCount(k) != 0 {
t.Errorf("entries after release = %d, want 0", lockCount(k))
}
if _, err := k.acquire(context.Background(), "a"); err != nil {
t.Fatalf("acquire after release error = %v", err)
}
}
func TestKeyedLocks_waitersShareOneLock(t *testing.T) {
t.Parallel()
k := newKeyedLocks()
const n = 20
var inside, overlaps atomic.Int32
var wg sync.WaitGroup
for range n {
wg.Go(func() {
release, err := k.acquire(context.Background(), "shared")
if err != nil {
t.Errorf("acquire error = %v", err)
return
}
if inside.Add(1) > 1 {
overlaps.Add(1)
}
time.Sleep(time.Millisecond)
inside.Add(-1)
release()
})
}
wg.Wait()
if overlaps.Load() != 0 {
t.Errorf("%d holders overlapped on one key", overlaps.Load())
}
if lockCount(k) != 0 {
t.Errorf("entries after every holder released = %d, want 0", lockCount(k))
}
}
func TestKeyedLocks_waiterGivesUpWithContext(t *testing.T) {
t.Parallel()
k := newKeyedLocks()
release, err := k.acquire(context.Background(), "a")
if err != nil {
t.Fatalf("acquire error = %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
start := time.Now()
if _, err := k.acquire(ctx, "a"); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("waiter error = %v, want DeadlineExceeded", err)
}
if time.Since(start) > time.Second {
t.Errorf("waiter took %v, want to give up with its ctx", time.Since(start))
}
if lockCount(k) != 1 {
t.Errorf("entries with the holder still in = %d, want 1: the waiter's exit must not drop the holder's entry", lockCount(k))
}
release()
if lockCount(k) != 0 {
t.Errorf("entries after the holder released = %d, want 0", lockCount(k))
}
}
func TestWrapErr_carriesProviderNameOpAndID(t *testing.T) {
t.Parallel()
p := newProvider("kvm2-home", twoHosts(), &stubDialer{})
internal/provider/kvm2/seed.go → internal/provider/nocloud/seed.go
+21 -15
View File
@@ -1,4 +1,7 @@
package kvm2
// Package nocloud builds cloud-init NoCloud seed ISOs in process, shared by
// the providers that attach a seed as a CD-ROM (kvm2, proxmox). It never
// parses or logs the payload.
package nocloud
import (
"bytes"
@@ -12,19 +15,22 @@ import (
"github.com/kdomanski/iso9660"
)
// seedLabel is the volume label cloud-init's NoCloud datasource looks for.
const seedLabel = "cidata"
// SeedLabel is the volume label cloud-init's NoCloud datasource looks for.
const SeedLabel = "cidata"
// guestAgentScript is the vendor-data payload that gets qemu-guest-agent
// running so the provider can read the VM's address through libvirt. It is
// ErrEmptyName is returned by BuildSeedISO for an empty instance name.
var ErrEmptyName = errors.New("seed ISO: instance name is empty")
// GuestAgentScript is the vendor-data payload that gets qemu-guest-agent
// running so the provider can read the VM's address from the hypervisor. It is
// a shell script and not a cloud-config on purpose: cloud-init merges a
// vendor-data cloud-config under the user-data with mergers that keep the
// user-data's value for every top-level key and never append lists, so a
// Proxy with its own packages: would silently lose the agent. Scripts
// bypass the merge.
const guestAgentScript = `#!/bin/sh
const GuestAgentScript = `#!/bin/sh
# Added by egress-proxies-operator: installs and starts qemu-guest-agent so
# the operator can learn this VM's IP address through libvirt.
# the operator can learn this VM's IP address from the hypervisor.
# To leave this out, set installGuestAgent: false in the provider's config.
set -e
if command -v qemu-ga >/dev/null 2>&1 || [ -x /usr/sbin/qemu-ga ]; then
@@ -53,15 +59,15 @@ else
fi
`
// buildSeedISO writes a NoCloud seed to w: user-data byte for byte,
// BuildSeedISO writes a NoCloud seed to w: user-data byte for byte,
// meta-data naming the instance, and vendor-data (an empty file when there
// is none, so cloud-init does not look elsewhere). iso9660.NewWriter stages
// the files in os.MkdirTemp("", ""), that is under os.TempDir(): $TMPDIR
// when set, else /tmp, the manager's 64Mi emptyDir. The staging dir is
// removed before returning. Nothing about the payload is logged here.
func buildSeedISO(name, userData, vendorData string, w io.Writer) (err error) {
func BuildSeedISO(name, userData, vendorData string, w io.Writer) (err error) {
if name == "" {
return permanentf("seed ISO: instance name is empty")
return ErrEmptyName
}
iw, err := iso9660.NewWriter()
if err != nil {
@@ -83,30 +89,30 @@ func buildSeedISO(name, userData, vendorData string, w io.Writer) (err error) {
return fmt.Errorf("adding %s to seed ISO: %w", f.name, err)
}
}
if err := iw.WriteTo(w, seedLabel); err != nil {
if err := iw.WriteTo(w, SeedLabel); err != nil {
return fmt.Errorf("writing seed ISO: %w", err)
}
return nil
}
// buildVendorData composes the seed's vendor-data from the configured
// BuildVendorData composes the seed's vendor-data from the configured
// document and the guest agent script. With both, they travel as separate
// parts of a MIME multipart, so cloud-init handles each by its own type; the
// MIME-Version header comes first because cloud-init detects multipart by
// finding "mime-version:" within the first 4096 bytes.
func buildVendorData(configured string, installAgent bool) (string, error) {
func BuildVendorData(configured string, installAgent bool) (string, error) {
switch {
case !installAgent:
return configured, nil
case strings.TrimSpace(configured) == "":
return guestAgentScript, nil
return GuestAgentScript, nil
}
var buf bytes.Buffer
mw := multipart.NewWriter(&buf)
fmt.Fprintf(&buf, "MIME-Version: 1.0\r\nContent-Type: multipart/mixed; boundary=%q\r\n\r\n", mw.Boundary())
parts := []struct{ contentType, body string }{
{"text/plain", configured},
{"text/x-shellscript", guestAgentScript},
{"text/x-shellscript", GuestAgentScript},
}
for _, p := range parts {
pw, err := mw.CreatePart(textproto.MIMEHeader{"Content-Type": {p.contentType}})
internal/provider/kvm2/seed_test.go → internal/provider/nocloud/seed_test.go
+15 -14
View File
@@ -1,8 +1,9 @@
package kvm2
package nocloud
import (
"bufio"
"bytes"
"errors"
"io"
"mime"
"mime/multipart"
@@ -18,8 +19,8 @@ import (
func readSeed(t *testing.T, name, userData, vendorData string) (map[string]string, string) {
t.Helper()
var buf bytes.Buffer
if err := buildSeedISO(name, userData, vendorData, &buf); err != nil {
t.Fatalf("buildSeedISO: %v", err)
if err := BuildSeedISO(name, userData, vendorData, &buf); err != nil {
t.Fatalf("BuildSeedISO: %v", err)
}
img, err := iso9660.OpenImage(bytes.NewReader(buf.Bytes()))
if err != nil {
@@ -57,8 +58,8 @@ func TestSeedISO_filesAndLabel(t *testing.T) {
userData := "#cloud-config\npackages: [squid]\n"
vendorData := "#!/bin/sh\necho hi\n"
files, label := readSeed(t, "proxy-abc", userData, vendorData)
if label != seedLabel {
t.Errorf("label = %q, want %q", label, seedLabel)
if label != SeedLabel {
t.Errorf("label = %q, want %q", label, SeedLabel)
}
want := map[string]string{
"user-data": userData,
@@ -117,8 +118,8 @@ func TestSeedISO_tempDirRemoved(t *testing.T) {
t.Fatalf("os.TempDir() = %q, want %q", got, tmp)
}
var buf bytes.Buffer
if err := buildSeedISO("proxy-abc", "#cloud-config\n", guestAgentScript, &buf); err != nil {
t.Fatalf("buildSeedISO: %v", err)
if err := BuildSeedISO("proxy-abc", "#cloud-config\n", GuestAgentScript, &buf); err != nil {
t.Fatalf("BuildSeedISO: %v", err)
}
if buf.Len() == 0 {
t.Fatal("seed ISO is empty")
@@ -151,8 +152,8 @@ func TestSeedISO_emptyVendorDataStillHasFile(t *testing.T) {
func TestSeedISO_rejectsEmptyName(t *testing.T) {
t.Parallel()
var buf bytes.Buffer
if err := buildSeedISO("", "#cloud-config\n", "", &buf); err == nil {
t.Fatal("buildSeedISO with empty name succeeded")
if err := BuildSeedISO("", "#cloud-config\n", "", &buf); !errors.Is(err, ErrEmptyName) {
t.Fatalf("BuildSeedISO with empty name error = %v, want ErrEmptyName", err)
}
if buf.Len() != 0 {
t.Errorf("wrote %d bytes alongside the error", buf.Len())
@@ -161,7 +162,7 @@ func TestSeedISO_rejectsEmptyName(t *testing.T) {
func TestVendorData_isShellScriptNotCloudConfig(t *testing.T) {
t.Parallel()
got, err := buildVendorData("", true)
got, err := BuildVendorData("", true)
if err != nil {
t.Fatal(err)
}
@@ -193,7 +194,7 @@ func bufioReader(s string) *bufio.Reader {
func TestVendorData_multipartWhenExtraProvided(t *testing.T) {
t.Parallel()
configured := "#cloud-config\nntp:\n enabled: true\n"
got, err := buildVendorData(configured, true)
got, err := BuildVendorData(configured, true)
if err != nil {
t.Fatal(err)
}
@@ -229,7 +230,7 @@ func TestVendorData_multipartWhenExtraProvided(t *testing.T) {
}
parts = append(parts, part{p.Header.Get("Content-Type"), string(body)})
}
want := []part{{"text/plain", configured}, {"text/x-shellscript", guestAgentScript}}
want := []part{{"text/plain", configured}, {"text/x-shellscript", GuestAgentScript}}
if len(parts) != len(want) {
t.Fatalf("got %d parts, want %d", len(parts), len(want))
}
@@ -242,7 +243,7 @@ func TestVendorData_multipartWhenExtraProvided(t *testing.T) {
func TestVendorData_offWithoutConfiguredIsEmpty(t *testing.T) {
t.Parallel()
got, err := buildVendorData("", false)
got, err := BuildVendorData("", false)
if err != nil {
t.Fatal(err)
}
@@ -254,7 +255,7 @@ func TestVendorData_offWithoutConfiguredIsEmpty(t *testing.T) {
func TestVendorData_offWithConfiguredPassesThrough(t *testing.T) {
t.Parallel()
configured := "#cloud-config\nntp:\n enabled: true"
got, err := buildVendorData(configured, false)
got, err := BuildVendorData(configured, false)
if err != nil {
t.Fatal(err)
}
internal/provider/proxmox/proxmox.go
+24 -5
View File
@@ -6,6 +6,7 @@ package proxmox
import (
"context"
"fmt"
"os"
"time"
@@ -13,6 +14,7 @@ import (
logf "sigs.k8s.io/controller-runtime/pkg/log"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/keyedlock"
)
const (
@@ -57,6 +59,11 @@ type Provider struct {
api api
now func() time.Time
budgets budgets
// nameLocks serialises Create and Delete per instance name (spec entry
// 19), so the reconciler's Delete and GC's Delete never race on stop and
// destroy, and a Delete cannot interleave with a Create's steps.
nameLocks *keyedlock.Locks
}
var _ provider.Provider = (*Provider)(nil)
@@ -85,11 +92,12 @@ func New(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, er
// newProvider is the constructor tests use, with a in-memory api.
func newProvider(name string, cfg provider.ProxmoxConfig, a api) *Provider {
return &Provider{
name: name,
cfg: withDefaults(cfg),
api: a,
now: time.Now,
budgets: defaultBudgets(),
name: name,
cfg: withDefaults(cfg),
api: a,
now: time.Now,
budgets: defaultBudgets(),
nameLocks: keyedlock.New(),
}
}
@@ -152,6 +160,17 @@ func (p *Provider) wrapErr(op, id string, err error) error {
return provider.Wrap(classify(op, err), op, p.name, id, err)
}
// lockName takes the per-name lock of spec entry 19, or gives up with ctx,
// so a caller whose ctx ends while another caller's Create waits on a clone
// returns promptly.
func (p *Provider) lockName(ctx context.Context, name string) (func(), error) {
release, err := p.nameLocks.Acquire(ctx, name)
if err != nil {
return nil, fmt.Errorf("%w: waiting for the lock on %s: %w", provider.ErrTransient, name, err)
}
return release, nil
}
// The four contract methods are stubs until the phases that implement them
// (Get and ListByTag in Phase 5, Create in 6, Delete in 7). Each answers
// Transient, so a Proxy on this provider retries rather than latching.
internal/provider/proxmox/proxmox_test.go
+21
View File
@@ -206,3 +206,24 @@ func TestStubs_answerTransient(t *testing.T) {
}
}
}
func TestLockName_givesUpWithContextAsTransient(t *testing.T) {
t.Parallel()
p := newProvider("proxmox-home", testConfig(), nil)
release, err := p.lockName(context.Background(), "proxy-abc")
if err != nil {
t.Fatalf("lockName error = %v", err)
}
defer release()
other, err := p.lockName(context.Background(), "proxy-def")
if err != nil {
t.Fatalf("lockName on another name error = %v, want it free", err)
}
other()
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
_, err = p.lockName(ctx, "proxy-abc")
if !errors.Is(err, provider.ErrTransient) || !errors.Is(err, context.DeadlineExceeded) {
t.Errorf("lockName on a held name error = %v, want ErrTransient wrapping DeadlineExceeded", err)
}
}
internal/provider/proxmox/seed.go
+43
View File
@@ -0,0 +1,43 @@
package proxmox
import (
"bytes"
"errors"
"fmt"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/nocloud"
)
// seedISOName is the seed's file name on the ISO storage. Upload refuses a
// name without the .iso suffix (F12), and GC's blind spot is documented by
// this pattern (spec entry 11).
func seedISOName(name string) string {
return name + "-seed.iso"
}
// seedVolID is the seed's volume id, as ide2 references it and as the
// storage content delete takes it.
func seedVolID(storage, name string) string {
return storage + ":iso/" + seedISOName(name)
}
// buildSeed renders the NoCloud seed for req in memory (spec entry 5): a
// seed is tens of KiB. Nothing about its content is logged or put in an
// error. Only an empty name is Permanent; a staging failure (no /tmp, a
// full emptyDir) is left unclassified, so classify makes it Transient and
// the reconciler retries instead of latching the Proxy.
func (p *Provider) buildSeed(req provider.CreateRequest) (*bytes.Buffer, error) {
vendorData, err := nocloud.BuildVendorData(p.cfg.VendorData, *p.cfg.InstallGuestAgent)
if err != nil {
return nil, fmt.Errorf("building vendor-data for %s: %w", req.Name, err)
}
var iso bytes.Buffer
if err := nocloud.BuildSeedISO(req.Name, req.CloudInit, vendorData, &iso); err != nil {
if errors.Is(err, nocloud.ErrEmptyName) {
return nil, permanentf("building seed ISO: %v", err)
}
return nil, fmt.Errorf("building seed ISO for %s: %w", req.Name, err)
}
return &iso, nil
}
internal/provider/proxmox/seed_test.go
+139
View File
@@ -0,0 +1,139 @@
package proxmox
import (
"bytes"
"errors"
"io"
"strings"
"testing"
"github.com/kdomanski/iso9660"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/nocloud"
)
// seedFiles opens a built seed and returns its label and files by name.
func seedFiles(t *testing.T, iso *bytes.Buffer) (string, map[string]string) {
t.Helper()
img, err := iso9660.OpenImage(bytes.NewReader(iso.Bytes()))
if err != nil {
t.Fatalf("OpenImage: %v", err)
}
label, err := img.Label()
if err != nil {
t.Fatalf("Label: %v", err)
}
root, err := img.RootDir()
if err != nil {
t.Fatalf("RootDir: %v", err)
}
children, err := root.GetChildren()
if err != nil {
t.Fatalf("GetChildren: %v", err)
}
files := make(map[string]string, len(children))
for _, c := range children {
data, err := io.ReadAll(c.Reader())
if err != nil {
t.Fatalf("reading %s: %v", c.Name(), err)
}
files[c.Name()] = string(data)
}
return label, files
}
func TestSeedNames_table(t *testing.T) {
t.Parallel()
if got, want := seedISOName("proxy-abc"), "proxy-abc-seed.iso"; got != want {
t.Errorf("seedISOName = %q, want %q", got, want)
}
if got, want := seedVolID("proxy-seeds", "proxy-abc"), "proxy-seeds:iso/proxy-abc-seed.iso"; got != want {
t.Errorf("seedVolID = %q, want %q", got, want)
}
}
func TestBuildSeed_table(t *testing.T) {
t.Parallel()
off := false
userData := "#cloud-config\r\npackages: [squid]\r\nwrite_files:\n - content: \"caf\xc3\xa9\"\n"
configured := "#cloud-config\nntp:\n enabled: true\n"
tests := []struct {
name string
vendorData string
install *bool
want func(t *testing.T, got string)
}{
{name: "agent script by default", want: func(t *testing.T, got string) {
if got != nocloud.GuestAgentScript {
t.Errorf("vendor-data is not the guest agent script: %q", got)
}
}},
{name: "configured joined with the agent script", vendorData: configured, want: func(t *testing.T, got string) {
if !strings.HasPrefix(got, "MIME-Version: 1.0\r\n") || !strings.Contains(got, configured) || !strings.Contains(got, nocloud.GuestAgentScript) {
t.Errorf("vendor-data is not a multipart of both documents: %q", got)
}
}},
{name: "agent off passes configured through", vendorData: configured, install: &off, want: func(t *testing.T, got string) {
if got != configured {
t.Errorf("vendor-data = %q, want the configured document", got)
}
}},
{name: "agent off with nothing configured is empty", install: &off, want: func(t *testing.T, got string) {
if got != "" {
t.Errorf("vendor-data = %q, want empty", got)
}
}},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
cfg := testConfig()
cfg.VendorData = tc.vendorData
cfg.InstallGuestAgent = tc.install
p := newProvider("proxmox-home", cfg, nil)
iso, err := p.buildSeed(provider.CreateRequest{Name: "proxy-abc", CloudInit: userData})
if err != nil {
t.Fatalf("buildSeed error = %v", err)
}
label, files := seedFiles(t, iso)
if label != nocloud.SeedLabel {
t.Errorf("label = %q, want %q", label, nocloud.SeedLabel)
}
if files["user-data"] != userData {
t.Errorf("user-data changed:\n got %q\nwant %q", files["user-data"], userData)
}
if want := "instance-id: proxy-abc\nlocal-hostname: proxy-abc\n"; files["meta-data"] != want {
t.Errorf("meta-data = %q, want %q", files["meta-data"], want)
}
tc.want(t, files["vendor-data"])
})
}
}
func TestBuildSeed_emptyNameIsPermanentWithoutPayload(t *testing.T) {
t.Parallel()
const sentinel = "SENTINEL-cloud-init-payload"
p := newProvider("proxmox-home", testConfig(), nil)
_, err := p.buildSeed(provider.CreateRequest{CloudInit: sentinel})
if !errors.Is(err, provider.ErrPermanent) {
t.Fatalf("buildSeed error = %v, want ErrPermanent", err)
}
if strings.Contains(err.Error(), sentinel) {
t.Errorf("error carries the cloud-init payload: %v", err)
}
}
// Not parallel: it points TMPDIR at a path that cannot exist for the
// process.
func TestBuildSeed_stagingFailureIsTransient(t *testing.T) {
t.Setenv("TMPDIR", t.TempDir()+"/missing")
p := newProvider("proxmox-home", testConfig(), nil)
_, err := p.buildSeed(provider.CreateRequest{Name: "proxy-abc", CloudInit: "#cloud-config\n"})
if err == nil {
t.Fatal("buildSeed with no staging dir succeeded")
}
if got := classify("create", err); got != provider.ErrTransient {
t.Errorf("classify(buildSeed error) = %v, want ErrTransient: %v", got, err)
}
}