diff --git a/docs/kvm2-provider-spec.md b/docs/kvm2-provider-spec.md index 1b5f311..7938333 100644 --- a/docs/kvm2-provider-spec.md +++ b/docs/kvm2-provider-spec.md @@ -163,7 +163,7 @@ element ` 2*time.Second { + t.Errorf("Create took %v, want the shrunk cap to end it", elapsed) + } + }) +} + +func TestCreate_rejectsUnknownPlacement(t *testing.T) { + t.Parallel() + tests := []struct { + name string + mutate func(req *provider.CreateRequest) + wantMsg string + }{ + { + name: "zone", + mutate: func(req *provider.CreateRequest) { req.Placement.Zone = "nas9" }, + wantMsg: `placement.zone "nas9" is not in kvm2.hosts`, + }, + { + name: "machineType", + mutate: func(req *provider.CreateRequest) { req.Placement.MachineType = "xlarge" }, + wantMsg: `placement.machineType "xlarge" is not in kvm2.machineTypes`, + }, + { + name: "image", + mutate: func(req *provider.CreateRequest) { req.Placement.Image = "ubuntu-24" }, + wantMsg: `placement.image "ubuntu-24" is neither a key in kvm2.images nor an http(s) URL`, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + req := fx.req(testName) + tc.mutate(&req) + id, err := fx.p.Create(fx.ctx, req) + if id != "" { + t.Errorf("Create = %q, want no ID", id) + } + assertProviderError(t, err, "create", provider.ErrPermanent) + if !strings.Contains(err.Error(), tc.wantMsg) { + t.Errorf("error = %q, want substring %q", err, tc.wantMsg) + } + if got := fx.fake.callLog(); len(got) != 0 { + t.Errorf("backend calls = %v, want none before validation", got) + } + if got := fx.dialer.dials.Load(); got != 0 { + t.Errorf("dials = %d, want 0", got) + } + }) + } +} + +func TestCreate_adoptionWithForeignUIDIsPermanent(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.addDomain(testName, libvirt.DomainRunning, managedMetadata("other-uid", time.Now())) + id, err := fx.p.Create(fx.ctx, fx.req(testName)) + if id != "" { + t.Errorf("Create = %q, want no ID", id) + } + assertProviderError(t, err, "create", provider.ErrPermanent) + if !strings.Contains(err.Error(), "other-uid") || !strings.Contains(err.Error(), testUID) { + t.Errorf("error = %q, want both UIDs named", err) + } + for _, rpc := range []string{"DomainCreate", "DomainDefineXML", "StorageVolCreateXML", "DomainDestroy"} { + if got := fx.fake.callCount(rpc); got != 0 { + t.Errorf("%s calls = %d, want 0", rpc, got) + } + } +} + +func TestCreate_adoptsMatchingOrEmptyUID(t *testing.T) { + t.Parallel() + tests := []struct { + name string + metadata string + }{ + {name: "matching uid", metadata: managedMetadata(testUID, time.Now())}, + {name: "empty uid", metadata: managedMetadata("", time.Now())}, + {name: "no metadata element", metadata: ""}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.addDomain(testName, libvirt.DomainRunning, tc.metadata) + fx.create(testName) + if got := fx.fake.callCount("DomainDefineXML"); got != 0 { + t.Errorf("DomainDefineXML calls = %d, want 0 (adopted)", got) + } + }) + } +} + +func TestCreate_poolOutOfSpaceIsQuotaExceeded(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.mu.Lock() + fx.fake.pools[fakePoolName].available = 1 << 30 + fx.fake.mu.Unlock() + _, err := fx.p.Create(fx.ctx, fx.req(testName)) + assertProviderError(t, err, "create", provider.ErrQuotaExceeded) + if !strings.Contains(err.Error(), fakePoolName) || !strings.Contains(err.Error(), "needed") { + t.Errorf("error = %q, want the pool and the needed bytes named", err) + } + for _, rpc := range []string{"StorageVolCreateXML", "StorageVolUpload", "DomainDefineXML"} { + if got := fx.fake.callCount(rpc); got != 0 { + t.Errorf("%s calls = %d, want 0", rpc, got) + } + } +} + +func TestCreate_libvirtNotFoundIsTransientNotNotFound(t *testing.T) { + t.Parallel() + tests := []struct { + name string + rpc string + err error + }{ + {name: "volume vanished", rpc: "StorageVolGetPath", err: fakeErr(libvirt.ErrNoStorageVol, "Storage volume not found")}, + {name: "domain vanished", rpc: "DomainGetState", err: fakeErr(libvirt.ErrNoDomain, "Domain not found")}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.failWith(tc.rpc, tc.err) + _, err := fx.p.Create(fx.ctx, fx.req(testName)) + assertProviderError(t, err, "create", provider.ErrTransient) + if errors.Is(err, provider.ErrNotFound) { + t.Errorf("Create error = %v, must never be NotFound", err) + } + }) + } +} + +func TestCreate_cloudInitSentinelNeverLoggedOrInErrors(t *testing.T) { + t.Parallel() + const sentinel = "SENTINEL-s3cr3t-passw0rd-7f3a" + fx := newFixture(t, 2) + fx.dialer.backend = observe(fx.fake, testProvider, testHost) + fx.fake.failWith("StorageVolUpload", fakeErr(libvirt.ErrInternalError, "internal error: upload failed")) + req := fx.req(testName) + req.CloudInit = "#cloud-config\nwrite_files:\n - content: " + sentinel + "\n" + _, err := fx.p.Create(fx.ctx, req) + assertProviderError(t, err, "create", provider.ErrTransient) + if strings.Contains(err.Error(), sentinel) { + t.Errorf("error carries the cloud-init: %v", err) + } + lines := fx.logs.all() + if len(lines) == 0 { + t.Fatal("no log lines captured at V(2)") + } + for _, l := range lines { + if strings.Contains(l, sentinel) { + t.Errorf("log line carries the cloud-init: %s", l) + } + } + if got := fx.logs.matching(`"cloudInitBytes"=` + strconv.Itoa(len(req.CloudInit))); len(got) != 1 { + t.Errorf("cloudInitBytes line = %d, want 1:\n%s", len(got), strings.Join(lines, "\n")) + } +} + +func TestCreate_metadataWrittenInDefineCall(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.p.now = func() time.Time { return time.Date(2026, 9, 16, 12, 0, 0, 0, time.UTC) } + fx.create(testName) + fx.withDomain(testName, func(d *fakeDomain) { + for _, want := range []string{`xmlns="` + metadataURI + `"`, "" + testUID + "", "true", "2026-09-16T12:00:00Z"} { + if !strings.Contains(d.xml, want) { + t.Errorf("defined XML lacks %s:\n%s", want, d.xml) + } + } + meta, err := parseInstanceMetadata(d.metadata) + if err != nil { + t.Fatalf("metadata the define stored does not parse: %v", err) + } + if !meta.managed() || meta.UID != testUID || meta.createdAt().IsZero() { + t.Errorf("metadata = %+v, want managed, uid %s and a createdAt", meta, testUID) + } + }) +} + +func TestCreate_seedAlwaysRebuilt(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.create(testName) + seed := seedVolumeName(testName) + first := fx.fake.vol(seed) + if first == nil { + t.Fatal("no seed volume after Create") + } + uploads := fx.fake.callCount("StorageVolUpload") + + // A crash between the seed upload and the define leaves the volumes + // without a domain; the next attempt must not trust the seed. + fx.fake.mu.Lock() + delete(fx.fake.domains, testName) + fx.fake.mu.Unlock() + fx.create(testName) + second := fx.fake.vol(seed) + if second == nil || second == first { + t.Errorf("seed volume reused across attempts") + } + 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)) { + t.Errorf("rebuilt seed is %d bytes, first was %d; want a complete seed of the same size", len(second.content), len(first.content)) + } +} diff --git a/internal/provider/kvm2/delete.go b/internal/provider/kvm2/delete.go new file mode 100644 index 0000000..e006bf2 --- /dev/null +++ b/internal/provider/kvm2/delete.go @@ -0,0 +1,98 @@ +package kvm2 + +import ( + "context" + "fmt" + + "github.com/digitalocean/go-libvirt" + "github.com/go-logr/logr" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// undefineFlags removes everything libvirt keeps for a domain beside its +// definition; UEFI domains keep NVRAM and undefine refuses without the flag. +const undefineFlags = libvirt.DomainUndefineManagedSave | libvirt.DomainUndefineSnapshotsMetadata | libvirt.DomainUndefineNvram + +// Delete is spec entry 11: under the per-name lock, destroy and undefine +// the domain, then remove its overlay and seed. Everything already gone is +// success, and nothing is left when nil is returned, so Get afterwards is +// NotFound. The base image stays: it is a shared cache. +func (p *Provider) Delete(ctx context.Context, providerID string) error { + hostName, name, err := parseProviderID(providerID) + if err != nil { + return provider.Wrap(provider.ErrPermanent, "delete", p.name, providerID, err) + } + log := p.logger(ctx).WithValues("host", hostName, "name", name) + host, err := p.host(hostName) + if err != nil { + log.Info("host is not configured; delete has nothing to clean up", "id", providerID) + return nil + } + unlock, err := p.lockName(ctx, name) + if err != nil { + return p.wrapErr("delete", providerID, err) + } + defer unlock() + + be, err := p.hosts.get(ctx, host) + if err != nil { + logAPIError(log, "delete", err) + return p.wrapErr("delete", providerID, err) + } + if err := removeDomain(ctx, log, be, name); err != nil { + logAPIError(log, "delete", err) + return p.wrapErr("delete", providerID, err) + } + if err := removeVolumes(ctx, log, be, host.Pool, name); err != nil { + logAPIError(log, "delete", err) + return p.wrapErr("delete", providerID, err) + } + log.V(1).Info("kvm2 domain deleted") + return nil +} + +func removeDomain(ctx context.Context, log logr.Logger, be backend, name string) error { + dom, err := be.DomainLookupByName(ctx, name) + if hasCode(err, libvirt.ErrNoDomain) { + log.V(1).Info("domain already gone") + return nil + } + if err != nil { + return fmt.Errorf("looking up domain %s: %w", name, err) + } + switch err := be.DomainDestroy(ctx, dom); { + case err == nil: + log.V(1).Info("domain destroyed") + case hasCode(err, libvirt.ErrOperationInvalid): + log.V(1).Info("domain not running, destroy skipped") + case hasCode(err, libvirt.ErrNoDomain): + log.V(1).Info("domain already gone") + return nil + default: + return fmt.Errorf("destroying domain %s: %w", name, err) + } + if err := be.DomainUndefineFlags(ctx, dom, undefineFlags); err != nil && !hasCode(err, libvirt.ErrNoDomain) { + return fmt.Errorf("undefining domain %s: %w", name, err) + } + log.V(1).Info("domain undefined") + return nil +} + +func removeVolumes(ctx context.Context, log logr.Logger, be backend, poolName, name string) error { + pool, err := be.StoragePoolLookupByName(ctx, poolName) + if hasCode(err, libvirt.ErrNoStoragePool) { + log.Info("pool not found, no volumes to delete", "pool", poolName) + return nil + } + if err != nil { + return fmt.Errorf("looking up storage pool %s: %w", poolName, err) + } + for _, vol := range []string{overlayVolumeName(name), seedVolumeName(name)} { + if err := deleteVolumeIfPresent(ctx, be, pool, vol); err != nil { + return err + } + } + log.V(1).Info("per-VM volumes deleted") + return nil +} diff --git a/internal/provider/kvm2/delete_test.go b/internal/provider/kvm2/delete_test.go new file mode 100644 index 0000000..84a4015 --- /dev/null +++ b/internal/provider/kvm2/delete_test.go @@ -0,0 +1,222 @@ +package kvm2 + +import ( + "errors" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/digitalocean/go-libvirt" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +func TestDelete_malformedIDIsPermanent(t *testing.T) { + t.Parallel() + for _, id := range []string{"", "proxy-x", "hosts/beelink/proxy-x", "hosts//domains/proxy-x", "hosts/beelink/domains/"} { + t.Run(id, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + err := fx.p.Delete(fx.ctx, id) + assertProviderError(t, err, "delete", provider.ErrPermanent) + var pe *provider.Error + if errors.As(err, &pe) && pe.ID != id { + t.Errorf("error.ID = %q, want %q", pe.ID, id) + } + if got := fx.dialer.dials.Load(); got != 0 { + t.Errorf("dials = %d, want 0", got) + } + }) + } +} + +func TestDelete_twiceSucceeds(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + for i := range 2 { + if err := fx.p.Delete(fx.ctx, id); err != nil { + t.Fatalf("Delete #%d error = %v", i+1, err) + } + } + if fx.hasDomain(testName) { + t.Error("domain still defined after Delete") + } + if got := fx.logs.matching("domain already gone"); len(got) != 1 { + t.Errorf("'already gone' lines = %d, want 1 from the second Delete", len(got)) + } + if got := fx.logs.matching("kvm2 domain deleted"); len(got) != 2 { + t.Errorf("'deleted' lines = %d, want 2", len(got)) + } +} + +func TestDelete_removesOverlayAndSeed(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + if err := fx.p.Delete(fx.ctx, id); err != nil { + t.Fatalf("Delete() error = %v", err) + } + for _, name := range []string{overlayVolumeName(testName), seedVolumeName(testName)} { + if fx.fake.vol(name) != nil { + t.Errorf("volume %s still present after Delete", name) + } + } + for _, name := range []string{"base-debian-13.qcow2", "base-debian-13.qcow2.sha256"} { + if fx.fake.vol(name) == nil { + t.Errorf("shared base volume %s removed by Delete", name) + } + } +} + +func TestDelete_runningDomainDestroyedFirst(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + before := len(fx.fake.callLog()) + if err := fx.p.Delete(fx.ctx, id); err != nil { + t.Fatalf("Delete() error = %v", err) + } + calls := fx.fake.callLog()[before:] + destroy := slices.Index(calls, "DomainDestroy") + undefine := slices.Index(calls, "DomainUndefineFlags") + volDelete := slices.Index(calls, "StorageVolDelete") + if destroy < 0 || undefine < 0 || volDelete < 0 || !(destroy < undefine && undefine < volDelete) { + t.Errorf("call order = %v, want destroy before undefine before volume deletes", calls) + } + if n := strings.Count(strings.Join(calls, " "), "StorageVolDelete"); n != 2 { + t.Errorf("StorageVolDelete calls = %d, want 2 (overlay and seed)", n) + } +} + +func TestDelete_stoppedDomainSkipsDestroy(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + fx.withDomain(testName, func(d *fakeDomain) { d.state = libvirt.DomainShutoff }) + if err := fx.p.Delete(fx.ctx, id); err != nil { + t.Fatalf("Delete() error = %v", err) + } + if fx.hasDomain(testName) { + t.Error("domain still defined after Delete") + } + if got := fx.logs.matching("domain not running, destroy skipped"); len(got) != 1 { + t.Errorf("'destroy skipped' lines = %d, want 1:\n%s", len(got), strings.Join(fx.logs.all(), "\n")) + } + if got := fx.fake.callCount("DomainUndefineFlags"); got != 1 { + t.Errorf("DomainUndefineFlags calls = %d, want 1", got) + } +} + +func TestDelete_toleratesMissingVolumesAndPool(t *testing.T) { + t.Parallel() + t.Run("domain without volumes", func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.addDomain(testName, libvirt.DomainRunning, managedMetadata(testUID, time.Now())) + if err := fx.p.Delete(fx.ctx, formatProviderID(testHost, testName)); err != nil { + t.Fatalf("Delete() error = %v", err) + } + if fx.hasDomain(testName) { + t.Error("domain still defined after Delete") + } + if got := fx.fake.callCount("StorageVolDelete"); got != 0 { + t.Errorf("StorageVolDelete calls = %d, want 0 for absent volumes", got) + } + }) + t.Run("pool gone", func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 0) + fx.p.cfg.Hosts[0].Pool = "retired" + fx.fake.addDomain(testName, libvirt.DomainShutoff, managedMetadata(testUID, time.Now())) + if err := fx.p.Delete(fx.ctx, formatProviderID(testHost, testName)); err != nil { + t.Fatalf("Delete() error = %v", err) + } + if fx.hasDomain(testName) { + t.Error("domain still defined after Delete") + } + if got := fx.logs.matching("pool not found, no volumes to delete"); len(got) != 1 || !strings.Contains(got[0], `"pool"="retired"`) { + t.Errorf("'pool not found' Info line = %v", got) + } + }) + t.Run("nothing at all", func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + if err := fx.p.Delete(fx.ctx, formatProviderID(testHost, testName)); err != nil { + t.Fatalf("Delete() error = %v", err) + } + }) +} + +func TestDelete_unconfiguredHostIsNoOp(t *testing.T) { + t.Parallel() + fx := newFixture(t, 0) + id := formatProviderID("nas9", testName) + if err := fx.p.Delete(fx.ctx, id); err != nil { + t.Fatalf("Delete() error = %v, want nil", err) + } + if got := fx.dialer.dials.Load(); got != 0 { + t.Errorf("dials = %d, want 0", got) + } + if got := fx.fake.callLog(); len(got) != 0 { + t.Errorf("backend calls = %v, want none", got) + } + if got := fx.logs.matching("host is not configured; delete has nothing to clean up"); len(got) != 1 || !strings.Contains(got[0], `"host"="nas9"`) { + t.Errorf("Info line = %v, want one naming the host", got) + } +} + +func TestDelete_concurrentSameNameSafe(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + const n = 10 + errs := make([]error, n) + var wg sync.WaitGroup + for i := range n { + wg.Add(1) + go func() { + defer wg.Done() + errs[i] = fx.p.Delete(fx.ctx, id) + }() + } + wg.Wait() + for i, err := range errs { + if err != nil { + t.Errorf("Delete #%d error = %v, want nil", i, err) + } + } + if fx.hasDomain(testName) { + t.Error("domain still defined after concurrent Deletes") + } + for _, name := range []string{overlayVolumeName(testName), seedVolumeName(testName)} { + if fx.fake.vol(name) != nil { + t.Errorf("volume %s still present", name) + } + } + if got := fx.fake.callCount("DomainUndefineFlags"); got != 1 { + t.Errorf("DomainUndefineFlags calls = %d, want 1 (the lock serialises the rest)", got) + } +} + +func TestDelete_undefineUsesNvramFlag(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + if err := fx.p.Delete(fx.ctx, id); err != nil { + t.Fatalf("Delete() error = %v", err) + } + fx.fake.mu.Lock() + flags, ok := fx.fake.undefineFlags[testName] + fx.fake.mu.Unlock() + if !ok { + t.Fatal("DomainUndefineFlags was not called") + } + for _, want := range []libvirt.DomainUndefineFlagsValues{libvirt.DomainUndefineNvram, libvirt.DomainUndefineManagedSave, libvirt.DomainUndefineSnapshotsMetadata} { + if flags&want == 0 { + t.Errorf("undefine flags = %d, lack %d", flags, want) + } + } +} diff --git a/internal/provider/kvm2/fake_test.go b/internal/provider/kvm2/fake_test.go index 6550f66..2368f8f 100644 --- a/internal/provider/kvm2/fake_test.go +++ b/internal/provider/kvm2/fake_test.go @@ -36,6 +36,13 @@ type fakeBackend struct { // sha256 in fixed chunks instead of holding the bytes, for tests that // push more through the fake than they want in memory. discardUploads bool + // startDelayPolls, when set, makes DomainCreate leave the domain shut + // off; it turns running on the startDelayPolls-th DomainGetState after + // that, as a real host that is still booting QEMU would. + startDelayPolls int + // undefineFlags records the flags of the last DomainUndefineFlags per + // domain name. + undefineFlags map[string]libvirt.DomainUndefineFlagsValues calls map[string]int log []string @@ -51,6 +58,9 @@ type fakeDomain struct { addrs map[uint32][]libvirt.DomainInterface uuid libvirt.UUID id int32 + // pendingStart counts the DomainGetState calls left before a delayed + // start (see startDelayPolls) turns the domain running. + pendingStart int } type fakePool struct { @@ -72,15 +82,16 @@ const fakePoolName = "default" func newFakeBackend() *fakeBackend { return &fakeBackend{ - mu: &sync.Mutex{}, - domains: make(map[string]*fakeDomain), - pools: map[string]*fakePool{fakePoolName: newFakePool()}, - hostname: "fake-host", - version: 11006000, - hooks: make(map[string]func() error), - block: make(map[string]chan struct{}), - calls: make(map[string]int), - deadCh: make(chan struct{}), + mu: &sync.Mutex{}, + domains: make(map[string]*fakeDomain), + pools: map[string]*fakePool{fakePoolName: newFakePool()}, + hostname: "fake-host", + version: 11006000, + hooks: make(map[string]func() error), + block: make(map[string]chan struct{}), + undefineFlags: make(map[string]libvirt.DomainUndefineFlagsValues), + calls: make(map[string]int), + deadCh: make(chan struct{}), } } @@ -97,6 +108,7 @@ func (f *fakeBackend) twin() *fakeBackend { hooks: make(map[string]func() error), block: make(map[string]chan struct{}), discardUploads: f.discardUploads, + undefineFlags: make(map[string]libvirt.DomainUndefineFlagsValues), calls: make(map[string]int), deadCh: make(chan struct{}), } @@ -279,6 +291,10 @@ func (f *fakeBackend) DomainCreate(ctx context.Context, dom libvirt.Domain) erro if isActive(d.state) { return fakeErr(libvirt.ErrOperationInvalid, "Requested operation is not valid: domain is already running") } + if f.startDelayPolls > 0 { + d.pendingStart = f.startDelayPolls + return nil + } d.state = libvirt.DomainRunning d.id = int32(len(f.domains)) return nil @@ -301,7 +317,7 @@ func (f *fakeBackend) DomainDestroy(ctx context.Context, dom libvirt.Domain) err return nil } -func (f *fakeBackend) DomainUndefineFlags(ctx context.Context, dom libvirt.Domain, _ libvirt.DomainUndefineFlagsValues) error { +func (f *fakeBackend) DomainUndefineFlags(ctx context.Context, dom libvirt.Domain, flags libvirt.DomainUndefineFlagsValues) error { if err := f.enter(ctx, "DomainUndefineFlags"); err != nil { return err } @@ -310,6 +326,7 @@ func (f *fakeBackend) DomainUndefineFlags(ctx context.Context, dom libvirt.Domai if _, err := f.domain(dom.Name); err != nil { return err } + f.undefineFlags[dom.Name] = flags delete(f.domains, dom.Name) return nil } @@ -324,6 +341,13 @@ func (f *fakeBackend) DomainGetState(ctx context.Context, dom libvirt.Domain, _ if err != nil { return 0, 0, err } + if d.pendingStart > 0 { + d.pendingStart-- + if d.pendingStart == 0 { + d.state = libvirt.DomainRunning + d.id = int32(len(f.domains)) + } + } return int32(d.state), 0, nil } diff --git a/internal/provider/kvm2/get.go b/internal/provider/kvm2/get.go new file mode 100644 index 0000000..0b19496 --- /dev/null +++ b/internal/provider/kvm2/get.go @@ -0,0 +1,127 @@ +package kvm2 + +import ( + "context" + "errors" + "fmt" + + "github.com/digitalocean/go-libvirt" + "github.com/go-logr/logr" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// Get is spec entries 7 to 9: look the domain up by name, read the mark +// back, then state and address. Only ErrNoDomain becomes NotFound; every +// other failure is classified, never guessed. +func (p *Provider) Get(ctx context.Context, providerID string) (*provider.Instance, error) { + hostName, name, err := parseProviderID(providerID) + if err != nil { + return nil, provider.Wrap(provider.ErrPermanent, "get", p.name, providerID, err) + } + host, err := p.host(hostName) + if err != nil { + return nil, p.wrapErr("get", providerID, err) + } + log := p.logger(ctx).WithValues("host", host.Name, "name", name) + be, err := p.hosts.get(ctx, host) + if err != nil { + logAPIError(log, "get", err) + return nil, p.wrapErr("get", providerID, err) + } + inst, err := get(ctx, log, be, host.Name, name) + if err != nil { + if hasCode(err, libvirt.ErrNoDomain) { + log.V(1).Info("kvm2 domain not found") + } else { + logAPIError(log, "get", err) + } + return nil, p.wrapErr("get", providerID, err) + } + log.V(1).Info("kvm2 domain fetched", "state", inst.State, "ip", inst.IP, "uid", inst.UID) + return inst, nil +} + +func get(ctx context.Context, log logr.Logger, be backend, hostName, name string) (*provider.Instance, error) { + dom, err := be.DomainLookupByName(ctx, name) + if err != nil { + return nil, fmt.Errorf("looking up domain %s: %w", name, err) + } + meta, err := readMetadata(ctx, be, dom) + if err != nil { + return nil, err + } + inst, err := inspect(ctx, log, be, hostName, dom, meta) + if err != nil { + return nil, err + } + return &inst, nil +} + +// readMetadata returns the kvm2 mark of a domain, or the zero value (not +// managed, empty UID) when the domain carries none. A mark that does not +// parse is Permanent, from parseInstanceMetadata. +func readMetadata(ctx context.Context, be backend, dom libvirt.Domain) (instanceMetadata, error) { + raw, err := be.DomainGetMetadata(ctx, dom, int32(libvirt.DomainMetadataElement), libvirt.OptString{metadataURI}, libvirt.DomainAffectCurrent) + if hasCode(err, libvirt.ErrNoDomainMetadata) { + return instanceMetadata{}, nil + } + if err != nil { + return instanceMetadata{}, fmt.Errorf("reading metadata of domain %s: %w", dom.Name, err) + } + return parseInstanceMetadata(raw) +} + +// inspect builds the Instance from the backend's state and, for an active +// domain, its discovered address. +func inspect(ctx context.Context, log logr.Logger, be backend, hostName string, dom libvirt.Domain, meta instanceMetadata) (provider.Instance, error) { + state, _, err := be.DomainGetState(ctx, dom, 0) + if err != nil { + return provider.Instance{}, fmt.Errorf("reading state of domain %s: %w", dom.Name, err) + } + s := libvirt.DomainState(state) + var ip string + if isActive(s) { + if ip, err = address(ctx, log, be, dom); err != nil { + return provider.Instance{}, err + } + } + return provider.Instance{ + ID: formatProviderID(hostName, dom.Name), + IP: ip, + State: mapState(s, ip), + UID: meta.UID, + CreatedAt: meta.createdAt(), + }, nil +} + +// addressSources is the discovery order of spec entry 7: the guest agent +// answers where the host's ARP table never sees guest traffic. +var addressSources = []struct { + name string + src libvirt.DomainInterfaceAddressesSource +}{ + {"agent", libvirt.DomainInterfaceAddressesSrcAgent}, + {"arp", libvirt.DomainInterfaceAddressesSrcArp}, +} + +// address returns the first usable IPv4 from the first source that yields +// one, or "" for "no address yet". A source that fails is logged and +// skipped; only ctx ending is an error, so a cancelled Get does not read +// as Provisioning. +func address(ctx context.Context, log logr.Logger, be backend, dom libvirt.Domain) (string, error) { + for _, s := range addressSources { + ifaces, err := be.DomainInterfaceAddresses(ctx, dom, uint32(s.src), 0) + if err != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return "", fmt.Errorf("reading addresses of domain %s: %w", dom.Name, err) + } + log.V(2).Info("no addresses from source", "source", s.name, "error", err.Error()) + continue + } + if ip := firstIPv4(ifaces); ip != "" { + return ip, nil + } + } + return "", nil +} diff --git a/internal/provider/kvm2/get_test.go b/internal/provider/kvm2/get_test.go new file mode 100644 index 0000000..61188fe --- /dev/null +++ b/internal/provider/kvm2/get_test.go @@ -0,0 +1,243 @@ +package kvm2 + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/digitalocean/go-libvirt" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +func TestGet_afterDeleteReturnsNotFound(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + if err := fx.p.Delete(fx.ctx, id); err != nil { + t.Fatalf("Delete() error = %v", err) + } + inst, err := fx.p.Get(fx.ctx, id) + if inst != nil { + t.Errorf("Get after Delete = %+v, want nil", inst) + } + assertProviderError(t, err, "get", provider.ErrNotFound) +} + +func TestGet_duringDeleteStillReturnsInstance(t *testing.T) { + t.Parallel() + t.Run("destroyed but still defined is stopped", func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + fx.withDomain(testName, func(d *fakeDomain) { d.state = libvirt.DomainShutoff }) + inst := fx.get(id) + if inst.State != provider.StateStopped || inst.UID != testUID { + t.Errorf("Get = %+v, want Stopped with uid %s", inst, testUID) + } + }) + t.Run("undefine in flight still answers", func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + release := fx.fake.blockOn("DomainUndefineFlags") + done := make(chan error, 1) + go func() { done <- fx.p.Delete(fx.ctx, id) }() + waitFor(t, "the undefine to block", func() bool { return fx.fake.callCount("DomainUndefineFlags") == 1 }) + + inst, err := fx.p.Get(fx.ctx, id) + if err != nil { + t.Fatalf("Get during Delete error = %v, want the instance", err) + } + if inst.State != provider.StateStopped || inst.ID != id { + t.Errorf("Get during Delete = %+v, want Stopped %s", inst, id) + } + close(release) + if err := <-done; err != nil { + t.Fatalf("Delete() error = %v", err) + } + _, err = fx.p.Get(fx.ctx, id) + assertProviderError(t, err, "get", provider.ErrNotFound) + }) +} + +func TestGet_backendBusyIsTransientNeverNotFound(t *testing.T) { + t.Parallel() + tests := []struct { + name string + err error + }{ + {name: "operation invalid", err: fakeErr(libvirt.ErrOperationInvalid, "Requested operation is not valid: domain is being migrated")}, + {name: "call timed out", err: fmt.Errorf("%w: libvirt call DomainGetState on beelink exceeded its 30s budget: %w", provider.ErrTransient, context.DeadlineExceeded)}, + {name: "unknown libvirt code", err: fakeErr(libvirt.ErrInternalError, "internal error: qemu monitor unresponsive")}, + {name: "not a libvirt error", err: errors.New("read tcp: connection reset by peer")}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + fx.fake.failWith("DomainGetState", tc.err) + inst, err := fx.p.Get(fx.ctx, id) + if inst != nil { + t.Errorf("Get = %+v, want nil", inst) + } + assertProviderError(t, err, "get", provider.ErrTransient) + if errors.Is(err, provider.ErrNotFound) { + t.Errorf("Get error = %v, must never be NotFound", err) + } + }) + } +} + +func TestGet_runningWithoutIPIsProvisioning(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + inst := fx.get(fx.create(testName)) + if inst.State != provider.StateProvisioning || inst.IP != "" { + t.Errorf("Get = %+v, want Provisioning with no IP", inst) + } +} + +func TestGet_runningWithIPIsRunning(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + fx.setAddrs(testName, libvirt.DomainInterfaceAddressesSrcAgent, "127.0.0.1", "169.254.1.2", "10.0.0.5") + inst := fx.get(id) + if inst.State != provider.StateRunning || inst.IP != "10.0.0.5" { + t.Errorf("Get = %+v, want Running with 10.0.0.5", inst) + } + if got := fx.logs.matching("kvm2 domain fetched"); len(got) != 1 || !strings.Contains(got[0], `"ip"="10.0.0.5"`) || !strings.Contains(got[0], `"state"="Running"`) { + t.Errorf("fetched line = %v, want state and ip", got) + } +} + +func TestGet_agentPreferredOverARP(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + fx.setAddrs(testName, libvirt.DomainInterfaceAddressesSrcAgent, "10.0.0.5") + fx.setAddrs(testName, libvirt.DomainInterfaceAddressesSrcArp, "10.0.0.6") + before := fx.fake.callCount("DomainInterfaceAddresses") + inst := fx.get(id) + if inst.IP != "10.0.0.5" { + t.Errorf("IP = %q, want the agent's 10.0.0.5", inst.IP) + } + if got := fx.fake.callCount("DomainInterfaceAddresses") - before; got != 1 { + t.Errorf("DomainInterfaceAddresses calls = %d, want 1 (ARP not consulted)", got) + } +} + +func TestGet_arpFallbackWhenAgentFails(t *testing.T) { + t.Parallel() + fx := newFixture(t, 2) + id := fx.create(testName) + fx.setAddrs(testName, libvirt.DomainInterfaceAddressesSrcArp, "10.0.0.6") + var calls int + fx.fake.mu.Lock() + fx.fake.hooks["DomainInterfaceAddresses"] = func() error { + calls++ + if calls%2 == 1 { + return fakeErr(libvirt.ErrAgentUnresponsive, "Guest agent is not responding") + } + return nil + } + fx.fake.mu.Unlock() + inst := fx.get(id) + if inst.State != provider.StateRunning || inst.IP != "10.0.0.6" { + t.Errorf("Get = %+v, want Running with ARP's 10.0.0.6", inst) + } + if got := fx.logs.matching("no addresses from source"); len(got) != 1 || !strings.Contains(got[0], `"source"="agent"`) { + t.Errorf("agent failure line = %v", got) + } +} + +func TestGet_addressErrorIsProvisioning(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + fx.fake.failWith("DomainInterfaceAddresses", fakeErr(libvirt.ErrAgentUnresponsive, "Guest agent is not responding")) + inst := fx.get(id) + if inst.State != provider.StateProvisioning || inst.IP != "" { + t.Errorf("Get = %+v, want Provisioning with no IP", inst) + } +} + +func TestGet_readsUIDAndCreatedAtFromBackend(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + created := time.Date(2026, 9, 16, 12, 0, 0, 0, time.UTC) + fx.p.now = func() time.Time { return created } + id := fx.create(testName) + inst := fx.get(id) + if inst.UID != testUID || !inst.CreatedAt.Equal(created) { + t.Errorf("Get = %+v, want uid %s created %s", inst, testUID, created) + } + + later := created.Add(48 * time.Hour) + fx.withDomain(testName, func(d *fakeDomain) { d.metadata = managedMetadata("rewritten-uid", later) }) + inst = fx.get(id) + if inst.UID != "rewritten-uid" || !inst.CreatedAt.Equal(later) { + t.Errorf("Get after metadata change = %+v, want uid rewritten-uid created %s (read back, not remembered)", inst, later) + } +} + +func TestGet_unmanagedDomainHasEmptyUID(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.addDomain("legacy-vm", libvirt.DomainRunning, "") + inst := fx.get(formatProviderID(testHost, "legacy-vm")) + if inst.UID != "" || !inst.CreatedAt.IsZero() || inst.State != provider.StateProvisioning { + t.Errorf("Get = %+v, want empty uid, zero createdAt, Provisioning", inst) + } +} + +func TestGet_malformedIDIsPermanent(t *testing.T) { + t.Parallel() + ids := []string{"", "proxy-x", "beelink/proxy-x", "hosts/beelink/proxy-x", "hosts//domains/proxy-x", "hosts/beelink/domains/", "hosts/beelink/domains/a/b", "nodes/beelink/domains/proxy-x"} + for _, id := range ids { + t.Run(id, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + inst, err := fx.p.Get(fx.ctx, id) + if inst != nil { + t.Errorf("Get(%q) = %+v, want nil", id, inst) + } + assertProviderError(t, err, "get", provider.ErrPermanent) + var pe *provider.Error + if errors.As(err, &pe) && pe.ID != id { + t.Errorf("error.ID = %q, want %q", pe.ID, id) + } + if got := fx.dialer.dials.Load(); got != 0 { + t.Errorf("dials = %d, want 0", got) + } + }) + } +} + +func TestGet_unconfiguredHostIsPermanent(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + _, err := fx.p.Get(fx.ctx, formatProviderID("nas9", testName)) + assertProviderError(t, err, "get", provider.ErrPermanent) + if !strings.Contains(err.Error(), `"nas9"`) { + t.Errorf("error = %v, want the host named", err) + } + if got := fx.dialer.dials.Load(); got != 0 { + t.Errorf("dials = %d, want 0", got) + } +} + +func TestGet_unknownStateIsStopped(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.addDomain(testName, libvirt.DomainState(42), managedMetadata(testUID, time.Now())) + inst := fx.get(formatProviderID(testHost, testName)) + if inst.State != provider.StateStopped { + t.Errorf("State = %s, want Stopped for an unknown backend state", inst.State) + } +} diff --git a/internal/provider/kvm2/kvm2.go b/internal/provider/kvm2/kvm2.go index 50625c0..4c8fb4e 100644 --- a/internal/provider/kvm2/kvm2.go +++ b/internal/provider/kvm2/kvm2.go @@ -6,7 +6,7 @@ package kvm2 import ( "context" - "errors" + "fmt" "net/http" "os" "sync" @@ -36,9 +36,20 @@ type Provider struct { budgets budgets tracing []tracing.Option + // runningWait caps the wait for a started domain to report running, + // polled every runningPoll (spec entry 18). + runningWait time.Duration + runningPoll time.Duration + // imageLocks serialises image fetches per (host, volume), spec entry 19. imageMu sync.Mutex imageLocks map[string]chan struct{} + + // 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. + nameMu sync.Mutex + nameLocks map[string]chan struct{} } // providerOption adjusts what newProvider builds; tests inject an HTTP @@ -57,6 +68,12 @@ func withTracing(opts ...tracing.Option) providerOption { return func(p *Provider) { p.tracing = opts } } +// withRunningWait shrinks the poll interval and the cap of the wait for a +// domain to run. +func withRunningWait(poll, cap time.Duration) providerOption { + return func(p *Provider) { p.runningPoll, p.runningWait = poll, cap } +} + var _ provider.Provider = (*Provider)(nil) // New builds a Provider from pc.KVM2. It never dials a host: the credential @@ -90,13 +107,16 @@ func New(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, er // per-request ctx carries the transfer budget) and the default budgets. func newProvider(name string, cfg provider.KVMConfig, d dialer, opts ...providerOption) *Provider { p := &Provider{ - name: name, - cfg: withDefaults(cfg), - now: time.Now, - hosts: newHostTable(d), - http: http.DefaultClient, - budgets: defaultBudgets, - imageLocks: make(map[string]chan struct{}), + name: name, + cfg: withDefaults(cfg), + now: time.Now, + hosts: newHostTable(d), + http: http.DefaultClient, + budgets: defaultBudgets, + runningWait: defaultRunningWait, + runningPoll: defaultRunningPoll, + imageLocks: make(map[string]chan struct{}), + nameLocks: make(map[string]chan struct{}), } for _, opt := range opts { opt(p) @@ -167,24 +187,21 @@ func (p *Provider) wrapErr(op, id string, err error) error { return provider.Wrap(classify(op, err), op, p.name, id, err) } -var errNotImplemented = errors.New("kvm2: not implemented") - -// Create is not implemented yet. -func (p *Provider) Create(_ context.Context, _ provider.CreateRequest) (string, error) { - return "", provider.Wrap(provider.ErrPermanent, "create", p.name, "", errNotImplemented) -} - -// Get is not implemented yet. -func (p *Provider) Get(_ context.Context, _ string) (*provider.Instance, error) { - return nil, provider.Wrap(provider.ErrPermanent, "get", p.name, "", errNotImplemented) -} - -// Delete is not implemented yet. -func (p *Provider) Delete(_ context.Context, _ string) error { - return provider.Wrap(provider.ErrPermanent, "delete", p.name, "", errNotImplemented) -} - -// ListByTag is not implemented yet. -func (p *Provider) ListByTag(_ context.Context) ([]provider.Instance, error) { - return nil, provider.Wrap(provider.ErrPermanent, "list", p.name, "", errNotImplemented) +// lockName takes the per-name lock of spec entry 19, or gives up with ctx. +// A one-slot channel rather than a mutex, 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) { + p.nameMu.Lock() + sem, ok := p.nameLocks[name] + if !ok { + sem = make(chan struct{}, 1) + p.nameLocks[name] = sem + } + p.nameMu.Unlock() + select { + case sem <- struct{}{}: + return func() { <-sem }, nil + case <-ctx.Done(): + return nil, fmt.Errorf("%w: waiting for the lock on %s: %w", provider.ErrTransient, name, ctx.Err()) + } } diff --git a/internal/provider/kvm2/kvm2_test.go b/internal/provider/kvm2/kvm2_test.go index 70145ce..f663db6 100644 --- a/internal/provider/kvm2/kvm2_test.go +++ b/internal/provider/kvm2/kvm2_test.go @@ -260,26 +260,3 @@ func TestWrapErr_carriesProviderNameOpAndID(t *testing.T) { t.Errorf("wrapErr() = %+v, want provider kvm2-home, op get, the id and ErrPermanent", pe) } } - -func TestStubs_returnPermanentWithProviderName(t *testing.T) { - t.Parallel() - ctx := context.Background() - p := newProvider("kvm2-home", twoHosts(), &stubDialer{}) - _, createErr := p.Create(ctx, provider.CreateRequest{Name: "proxy-x"}) - inst, getErr := p.Get(ctx, "hosts/beelink/domains/proxy-x") - deleteErr := p.Delete(ctx, "hosts/beelink/domains/proxy-x") - list, listErr := p.ListByTag(ctx) - if inst != nil || list != nil { - t.Errorf("Get = %+v, ListByTag = %+v, want nil results from stubs", inst, list) - } - for op, err := range map[string]error{"create": createErr, "get": getErr, "delete": deleteErr, "list": listErr} { - var pe *provider.Error - if !errors.As(err, &pe) { - t.Errorf("%s: error = %T, want *provider.Error", op, err) - continue - } - if pe.Provider != "kvm2-home" || pe.Op != op || !errors.Is(err, provider.ErrPermanent) { - t.Errorf("%s: error = %+v, want provider kvm2-home, op %s, ErrPermanent", op, pe, op) - } - } -} diff --git a/internal/provider/kvm2/list.go b/internal/provider/kvm2/list.go new file mode 100644 index 0000000..9a35ce2 --- /dev/null +++ b/internal/provider/kvm2/list.go @@ -0,0 +1,60 @@ +package kvm2 + +import ( + "context" + "fmt" + + "github.com/digitalocean/go-libvirt" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// ListByTag is spec entry 12: every domain on every configured host, in +// config order, filtered client-side on the kvm2 mark. A host that cannot +// be reached or listed fails the whole call (fail-whole); a single domain +// that cannot be read is skipped, unless ctx ended, which fails the call +// rather than pass off an empty view as the truth. +func (p *Provider) ListByTag(ctx context.Context) ([]provider.Instance, error) { + log := p.logger(ctx) + var out []provider.Instance + for _, host := range p.cfg.Hosts { + hlog := log.WithValues("host", host.Name) + be, err := p.hosts.get(ctx, host) + if err != nil { + logAPIError(hlog, "list", err) + return nil, p.wrapErr("list", "", err) + } + doms, _, err := be.ConnectListAllDomains(ctx, 1, libvirt.ConnectListDomainsActive|libvirt.ConnectListDomainsInactive) + if err != nil { + err = fmt.Errorf("listing domains on host %s: %w", host.Name, err) + logAPIError(hlog, "list", err) + return nil, p.wrapErr("list", "", err) + } + for _, dom := range doms { + dlog := hlog.WithValues("name", dom.Name) + meta, err := readMetadata(ctx, be, dom) + if err != nil { + if ctx.Err() != nil { + return nil, p.wrapErr("list", "", err) + } + dlog.V(1).Info("skipping domain with unreadable metadata", "error", err.Error()) + continue + } + if !meta.managed() { + continue + } + inst, err := inspect(ctx, dlog, be, host.Name, dom, meta) + if err != nil { + if ctx.Err() != nil { + return nil, p.wrapErr("list", "", err) + } + dlog.V(1).Info("skipping domain that could not be inspected", "error", err.Error()) + continue + } + dlog.V(2).Info("kvm2 listed instance", "id", inst.ID, "state", inst.State, "uid", inst.UID, "createdAt", inst.CreatedAt) + out = append(out, inst) + } + } + log.V(1).Info("kvm2 domains listed", "hosts", len(p.cfg.Hosts), "managed", len(out)) + return out, nil +} diff --git a/internal/provider/kvm2/list_test.go b/internal/provider/kvm2/list_test.go new file mode 100644 index 0000000..6f49562 --- /dev/null +++ b/internal/provider/kvm2/list_test.go @@ -0,0 +1,178 @@ +package kvm2 + +import ( + "errors" + "strings" + "testing" + "time" + + "github.com/digitalocean/go-libvirt" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +func listIDs(list []provider.Instance) []string { + ids := make([]string, len(list)) + for i, inst := range list { + ids[i] = inst.ID + } + return ids +} + +func TestListByTag_excludesUnmarkedDomains(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + fx.fake.addDomain("legacy-vm", libvirt.DomainRunning, "") + fx.fake.addDomain("unmarked", libvirt.DomainRunning, `falsex`) + fx.fake.addDomain("unrelated", libvirt.DomainShutoff, `hand made`) + list, err := fx.p.ListByTag(fx.ctx) + if err != nil { + t.Fatalf("ListByTag() error = %v", err) + } + if got := listIDs(list); len(got) != 1 || got[0] != id { + t.Errorf("ListByTag = %v, want only %s", got, id) + } + if list[0].UID != testUID || list[0].CreatedAt.IsZero() { + t.Errorf("listed = %+v, want uid and createdAt", list[0]) + } + if got := fx.logs.matching("kvm2 domains listed"); len(got) != 1 || !strings.Contains(got[0], `"hosts"=1`) || !strings.Contains(got[0], `"managed"=1`) { + t.Errorf("listed line = %v", got) + } +} + +func TestListByTag_emptyUIDStillListed(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.addDomain("proxy-nouid", libvirt.DomainRunning, managedMetadata("", time.Now())) + list, err := fx.p.ListByTag(fx.ctx) + if err != nil { + t.Fatalf("ListByTag() error = %v", err) + } + if len(list) != 1 || list[0].UID != "" || list[0].ID != formatProviderID(testHost, "proxy-nouid") { + t.Errorf("ListByTag = %+v, want the domain listed with an empty UID", list) + } +} + +func TestListByTag_acrossHostsInConfigOrder(t *testing.T) { + t.Parallel() + cfg := twoHosts() + cfg.Hosts[1].Pool = "" + first, second := newFakeBackend(), newFakeBackend() + first.addDomain("proxy-z", libvirt.DomainRunning, managedMetadata("uid-z", time.Now())) + first.addDomain("proxy-a", libvirt.DomainShutoff, managedMetadata("uid-a", time.Now())) + second.addDomain("proxy-b", libvirt.DomainRunning, managedMetadata("uid-b", time.Now())) + p := newProvider(testProvider, cfg, &hostDialer{backends: map[string]backend{"beelink": first, "lab2": second}}) + ctx, _ := captureContext() + list, err := p.ListByTag(ctx) + if err != nil { + t.Fatalf("ListByTag() error = %v", err) + } + want := []string{formatProviderID("beelink", "proxy-a"), formatProviderID("beelink", "proxy-z"), formatProviderID("lab2", "proxy-b")} + if got := listIDs(list); strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("ListByTag = %v, want %v (config order, then name order)", got, want) + } + if list[0].State != provider.StateStopped || list[1].State != provider.StateProvisioning { + t.Errorf("states = %s, %s, want Stopped, Provisioning", list[0].State, list[1].State) + } +} + +func TestListByTag_unreachableHostFailsWhole(t *testing.T) { + t.Parallel() + first := newFakeBackend() + first.addDomain("proxy-a", libvirt.DomainRunning, managedMetadata("uid-a", time.Now())) + d := &hostDialer{ + backends: map[string]backend{"beelink": first}, + errs: map[string]error{"lab2": errors.New("dial tcp 192.0.2.2:22: i/o timeout")}, + } + p := newProvider(testProvider, twoHosts(), d) + ctx, _ := captureContext() + list, err := p.ListByTag(ctx) + if list != nil { + t.Errorf("ListByTag = %v, want nil with the error", list) + } + assertProviderError(t, err, "list", provider.ErrTransient) +} + +func TestListByTag_unreadableDomainSkipped(t *testing.T) { + t.Parallel() + tests := []struct { + name string + setup func(fx *fixture) + line string + }{ + { + name: "corrupt metadata on one domain", + setup: func(fx *fixture) { + fx.fake.addDomain("proxy-bad", libvirt.DomainRunning, "not an instance") + }, + line: "skipping domain with unreadable metadata", + }, + { + name: "metadata rpc fails for the first domain only", + setup: func(fx *fixture) { + fx.fake.addDomain("proxy-bad", libvirt.DomainRunning, managedMetadata("uid-bad", time.Now())) + var calls int + fx.fake.mu.Lock() + fx.fake.hooks["DomainGetMetadata"] = func() error { + calls++ + if calls == 1 { + return fakeErr(libvirt.ErrInternalError, "internal error: metadata read failed") + } + return nil + } + fx.fake.mu.Unlock() + }, + line: "skipping domain with unreadable metadata", + }, + { + name: "state rpc fails for the first domain only", + setup: func(fx *fixture) { + fx.fake.addDomain("proxy-bad", libvirt.DomainRunning, managedMetadata("uid-bad", time.Now())) + var calls int + fx.fake.mu.Lock() + fx.fake.hooks["DomainGetState"] = func() error { + calls++ + if calls == 1 { + return fakeErr(libvirt.ErrNoDomain, "Domain not found") + } + return nil + } + fx.fake.mu.Unlock() + }, + line: "skipping domain that could not be inspected", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.fake.addDomain("proxy-good", libvirt.DomainRunning, managedMetadata("uid-good", time.Now())) + tc.setup(fx) + list, err := fx.p.ListByTag(fx.ctx) + if err != nil { + t.Fatalf("ListByTag() error = %v", err) + } + if got := listIDs(list); len(got) != 1 || got[0] != formatProviderID(testHost, "proxy-good") { + t.Errorf("ListByTag = %v, want only proxy-good", got) + } + if got := fx.logs.matching(tc.line); len(got) != 1 || !strings.Contains(got[0], `"name"="proxy-bad"`) { + t.Errorf("skip line = %v, want one naming proxy-bad", got) + } + }) + } +} + +func TestListByTag_includesInactiveDomains(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + id := fx.create(testName) + fx.withDomain(testName, func(d *fakeDomain) { d.state = libvirt.DomainShutoff }) + list, err := fx.p.ListByTag(fx.ctx) + if err != nil { + t.Fatalf("ListByTag() error = %v", err) + } + if len(list) != 1 || list[0].ID != id || list[0].State != provider.StateStopped { + t.Errorf("ListByTag = %+v, want the shut-off domain as Stopped", list) + } +} diff --git a/internal/provider/kvm2/provider_test.go b/internal/provider/kvm2/provider_test.go new file mode 100644 index 0000000..c08b897 --- /dev/null +++ b/internal/provider/kvm2/provider_test.go @@ -0,0 +1,415 @@ +package kvm2 + +import ( + "context" + "errors" + "fmt" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/digitalocean/go-libvirt" + "github.com/go-logr/logr/funcr" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +const ( + testProvider = "kvm2-home" + testHost = "beelink" + testUID = "0f1e2d3c-uid-1" + testName = "proxy-abcdefghijklmnop" +) + +// fixture is one configured host backed by a fake, an image server the +// catalogue points at, a log capture at a chosen verbosity and a provider +// with a fast running-wait, so a Create runs end to end in memory. +type fixture struct { + t *testing.T + p *Provider + fake *fakeBackend + dialer *stubDialer + logs *logCapture + ctx context.Context + image []byte +} + +func newFixture(t *testing.T, verbosity int, opts ...providerOption) *fixture { + t.Helper() + fx := &fixture{t: t, fake: newFakeBackend(), logs: &logCapture{}, image: []byte("QFI\xfb fake base image bytes")} + srv := httptest.NewServer(serveBytes(fx.image)) + t.Cleanup(srv.Close) + fx.dialer = &stubDialer{backend: fx.fake} + cfg := provider.KVMConfig{ + Images: map[string]provider.KVMImage{imageKey: {URL: srv.URL + "/debian-13-generic-amd64.qcow2"}}, + MachineTypes: map[string]provider.KVMMachineType{"small": {VCPUs: 1, MemoryMiB: 1024}}, + Hosts: []provider.KVMHost{{Name: testHost, Address: "beelink:22", User: "virt", Bridge: "br0"}}, + } + opts = append([]providerOption{withHTTPClient(srv.Client()), withRunningWait(time.Millisecond, 500*time.Millisecond)}, opts...) + fx.p = newProvider(testProvider, cfg, fx.dialer, opts...) + log := funcr.New(func(prefix, args string) { + fx.logs.mu.Lock() + defer fx.logs.mu.Unlock() + fx.logs.lines = append(fx.logs.lines, prefix+" "+args) + }, funcr.Options{Verbosity: verbosity}) + fx.ctx = logf.IntoContext(context.Background(), log) + return fx +} + +func (fx *fixture) req(name string) provider.CreateRequest { + return provider.CreateRequest{ + Name: name, + UID: testUID, + Namespace: "default", + ProxyName: "egress", + Placement: provider.Placement{Zone: testHost, MachineType: "small", Image: imageKey}, + CloudInit: "#cloud-config\npackages: [squid]\n", + Port: 3128, + } +} + +func (fx *fixture) create(name string) string { + fx.t.Helper() + id, err := fx.p.Create(fx.ctx, fx.req(name)) + if err != nil { + fx.t.Fatalf("Create(%s) error = %v", name, err) + } + if want := formatProviderID(testHost, name); id != want { + fx.t.Fatalf("Create(%s) = %q, want %q", name, id, want) + } + return id +} + +func (fx *fixture) get(id string) *provider.Instance { + fx.t.Helper() + inst, err := fx.p.Get(fx.ctx, id) + if err != nil { + fx.t.Fatalf("Get(%s) error = %v", id, err) + } + return inst +} + +// withDomain runs fn on the fake's domain under its lock. +func (fx *fixture) withDomain(name string, fn func(d *fakeDomain)) { + fx.t.Helper() + fx.fake.mu.Lock() + defer fx.fake.mu.Unlock() + d, ok := fx.fake.domains[name] + if !ok { + fx.t.Fatalf("fake has no domain %s", name) + } + fn(d) +} + +func (fx *fixture) hasDomain(name string) bool { + fx.fake.mu.Lock() + defer fx.fake.mu.Unlock() + _, ok := fx.fake.domains[name] + return ok +} + +func (fx *fixture) setAddrs(name string, src libvirt.DomainInterfaceAddressesSource, ips ...string) { + fx.withDomain(name, func(d *fakeDomain) { + var addrs []libvirt.DomainIPAddr + for _, ip := range ips { + addrs = append(addrs, libvirt.DomainIPAddr{Type: int32(libvirt.IPAddrTypeIpv4), Addr: ip, Prefix: 24}) + } + d.addrs[uint32(src)] = []libvirt.DomainInterface{{Name: "enp1s0", Addrs: addrs}} + }) +} + +// putVol places a volume in the fake's pool directly. +func (fx *fixture) putVol(name string, content []byte) *fakeVol { + fx.fake.mu.Lock() + defer fx.fake.mu.Unlock() + v := &fakeVol{content: content, capacity: uint64(len(content))} + fx.fake.pools[fakePoolName].vols[name] = v + return v +} + +// cacheImage places a complete base image and its marker, as a previous +// Create on the host would have. +func (fx *fixture) cacheImage() { + src, err := resolveImage(fx.p.cfg, imageKey) + if err != nil { + fx.t.Fatal(err) + } + fx.putVol(src.volumeName(), fx.image) + fx.putVol(src.markerName(), []byte(digestOf(fx.image))) +} + +func managedMetadata(uid string, createdAt time.Time) string { + s, err := marshalInstanceMetadata(newInstanceMetadata(uid, createdAt)) + if err != nil { + panic(err) + } + return s +} + +// assertProviderError checks err is a *provider.Error from the configured +// provider with the given op and class. +func assertProviderError(t *testing.T, err error, op string, class error) { + t.Helper() + if err == nil { + t.Fatalf("error = nil, want op %s class %v", op, class) + } + var pe *provider.Error + if !errors.As(err, &pe) { + t.Fatalf("error = %T (%v), want *provider.Error", err, err) + } + if pe.Provider != testProvider { + t.Errorf("error.Provider = %q, want %q", pe.Provider, testProvider) + } + if pe.Op != op { + t.Errorf("error.Op = %q, want %q", pe.Op, op) + } + if !errors.Is(err, class) { + t.Errorf("error = %v, want class %v", err, class) + } + for _, other := range []error{provider.ErrNotFound, provider.ErrQuotaExceeded, provider.ErrPermanent, provider.ErrTransient} { + if other != class && errors.Is(err, other) { + t.Errorf("error = %v, also matches %v", err, other) + } + } +} + +// hostDialer hands out one backend per host name and fails hosts in errs. +type hostDialer struct { + backends map[string]backend + errs map[string]error +} + +func (d *hostDialer) dial(_ context.Context, host provider.KVMHost) (backend, error) { + if err := d.errs[host.Name]; err != nil { + return nil, err + } + b, ok := d.backends[host.Name] + if !ok { + return nil, fmt.Errorf("hostDialer: no backend for %s", host.Name) + } + return b, nil +} + +// waitFor polls cond until it holds or the deadline passes. +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(time.Millisecond) + } +} + +func TestProvider_cancelledContextReturnsPromptly(t *testing.T) { + t.Parallel() + tests := []struct { + name string + firstRPC string + op string + call func(ctx context.Context, fx *fixture) error + }{ + { + name: "create", firstRPC: "DomainLookupByName", op: "create", + call: func(ctx context.Context, fx *fixture) error { _, err := fx.p.Create(ctx, fx.req(testName)); return err }, + }, + { + name: "get", firstRPC: "DomainLookupByName", op: "get", + call: func(ctx context.Context, fx *fixture) error { + _, err := fx.p.Get(ctx, formatProviderID(testHost, testName)) + return err + }, + }, + { + name: "delete", firstRPC: "DomainLookupByName", op: "delete", + call: func(ctx context.Context, fx *fixture) error { + return fx.p.Delete(ctx, formatProviderID(testHost, testName)) + }, + }, + { + name: "list", firstRPC: "ConnectListAllDomains", op: "list", + call: func(ctx context.Context, fx *fixture) error { _, err := fx.p.ListByTag(ctx); return err }, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + release := fx.fake.blockOn(tc.firstRPC) + defer close(release) + ctx, cancel := context.WithCancel(fx.ctx) + done := make(chan error, 1) + go func() { done <- tc.call(ctx, fx) }() + waitFor(t, "the first RPC to block", func() bool { return fx.fake.callCount(tc.firstRPC) == 1 }) + cancel() + select { + case err := <-done: + assertProviderError(t, err, tc.op, provider.ErrTransient) + case <-time.After(time.Second): + t.Fatalf("%s did not return within 1s of cancellation", tc.name) + } + }) + } +} + +func TestProvider_everyErrorIsWrappedWithConfiguredName(t *testing.T) { + t.Parallel() + dialErr := errors.New("dial tcp 192.0.2.1:22: connect: connection refused") + tests := []struct { + name string + setup func(fx *fixture) + call func(fx *fixture) error + op string + class error + }{ + { + name: "create unknown zone", + call: func(fx *fixture) error { + req := fx.req(testName) + req.Placement.Zone = "nas9" + _, err := fx.p.Create(fx.ctx, req) + return err + }, + op: "create", class: provider.ErrPermanent, + }, + { + name: "create dial failure", + setup: func(fx *fixture) { fx.dialer.err = dialErr }, + call: func(fx *fixture) error { _, err := fx.p.Create(fx.ctx, fx.req(testName)); return err }, + op: "create", class: provider.ErrTransient, + }, + { + name: "create pool out of space", + setup: func(fx *fixture) { + fx.fake.mu.Lock() + fx.fake.pools[fakePoolName].available = 1 + fx.fake.mu.Unlock() + }, + call: func(fx *fixture) error { _, err := fx.p.Create(fx.ctx, fx.req(testName)); return err }, + op: "create", class: provider.ErrQuotaExceeded, + }, + { + name: "create vanished volume", + setup: func(fx *fixture) { fx.fake.failWith("StorageVolGetPath", fakeErr(libvirt.ErrNoStorageVol, "gone")) }, + call: func(fx *fixture) error { _, err := fx.p.Create(fx.ctx, fx.req(testName)); return err }, + op: "create", class: provider.ErrTransient, + }, + { + name: "get malformed id", + call: func(fx *fixture) error { _, err := fx.p.Get(fx.ctx, "beelink/"+testName); return err }, + op: "get", class: provider.ErrPermanent, + }, + { + name: "get missing domain", + call: func(fx *fixture) error { _, err := fx.p.Get(fx.ctx, formatProviderID(testHost, testName)); return err }, + op: "get", class: provider.ErrNotFound, + }, + { + name: "get busy backend", + setup: func(fx *fixture) { + fx.create(testName) + fx.fake.failWith("DomainGetState", fakeErr(libvirt.ErrOperationInvalid, "busy")) + }, + call: func(fx *fixture) error { _, err := fx.p.Get(fx.ctx, formatProviderID(testHost, testName)); return err }, + op: "get", class: provider.ErrTransient, + }, + { + name: "delete malformed id", + call: func(fx *fixture) error { return fx.p.Delete(fx.ctx, "hosts//domains/"+testName) }, + op: "delete", class: provider.ErrPermanent, + }, + { + name: "delete dial failure", + setup: func(fx *fixture) { fx.dialer.err = dialErr }, + call: func(fx *fixture) error { return fx.p.Delete(fx.ctx, formatProviderID(testHost, testName)) }, + op: "delete", class: provider.ErrTransient, + }, + { + name: "delete access denied", + setup: func(fx *fixture) { fx.fake.failWith("DomainLookupByName", fakeErr(libvirt.ErrAccessDenied, "denied")) }, + call: func(fx *fixture) error { return fx.p.Delete(fx.ctx, formatProviderID(testHost, testName)) }, + op: "delete", class: provider.ErrPermanent, + }, + { + name: "list dial failure", + setup: func(fx *fixture) { fx.dialer.err = dialErr }, + call: func(fx *fixture) error { _, err := fx.p.ListByTag(fx.ctx); return err }, + op: "list", class: provider.ErrTransient, + }, + { + name: "list rpc failure", + setup: func(fx *fixture) { + fx.fake.failWith("ConnectListAllDomains", fakeErr(libvirt.ErrInternalError, "boom")) + }, + call: func(fx *fixture) error { _, err := fx.p.ListByTag(fx.ctx); return err }, + op: "list", class: provider.ErrTransient, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + if tc.setup != nil { + tc.setup(fx) + } + assertProviderError(t, tc.call(fx), tc.op, tc.class) + }) + } +} + +func TestLogging_v0SilentV1OutcomeV2Detail(t *testing.T) { + t.Parallel() + outcomes := []string{"overlay volume created", "seed volume uploaded", "domain defined", "kvm2 domain ensured running"} + t.Run("v0 is silent", func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 0) + fx.create(testName) + if lines := fx.logs.all(); len(lines) != 0 { + t.Errorf("Create logged %d lines at V(0), want none:\n%s", len(lines), strings.Join(lines, "\n")) + } + }) + t.Run("v1 has outcomes without xml", func(t *testing.T) { + t.Parallel() + fx := newFixture(t, 1) + fx.create(testName) + for _, want := range outcomes { + if len(fx.logs.matching(want)) != 1 { + t.Errorf("V(1) lines matching %q = %d, want 1:\n%s", want, len(fx.logs.matching(want)), strings.Join(fx.logs.all(), "\n")) + } + } + if got := fx.logs.matching("