KVM/libvirt provider: proxies as VMs on your own hypervisors #5
@@ -270,16 +270,14 @@ func (r *ProxyReconciler) reconcileDelete(ctx context.Context, p *crawlv1alpha1.
|
||||
if p.Status.ProviderID == "" {
|
||||
// Nothing was ever recorded as created; orphan GC reaps any stray
|
||||
// instance a crashed create might have left behind.
|
||||
controllerutil.RemoveFinalizer(p, crawlv1alpha1.FinalizerName)
|
||||
return ctrl.Result{}, r.Update(ctx, p)
|
||||
return r.releaseFinalizer(ctx, p)
|
||||
}
|
||||
if p.Annotations[crawlv1alpha1.AnnotationSkipCleanup] == "true" {
|
||||
// Checked before the provider lookup so it also releases a Proxy whose
|
||||
// provider is gone from the config. The instance is left running.
|
||||
logf.FromContext(ctx).Info("skip-cleanup annotation set; abandoning the instance",
|
||||
"providerID", p.Status.ProviderID, "provider", p.Spec.Provider)
|
||||
controllerutil.RemoveFinalizer(p, crawlv1alpha1.FinalizerName)
|
||||
return ctrl.Result{}, r.Update(ctx, p)
|
||||
return r.releaseFinalizer(ctx, p)
|
||||
}
|
||||
prov, ok := r.Providers[p.Spec.Provider]
|
||||
if !ok {
|
||||
@@ -290,8 +288,7 @@ func (r *ProxyReconciler) reconcileDelete(ctx context.Context, p *crawlv1alpha1.
|
||||
switch {
|
||||
case err == nil:
|
||||
case provider.Class(err) == provider.ErrNotFound:
|
||||
controllerutil.RemoveFinalizer(p, crawlv1alpha1.FinalizerName)
|
||||
return ctrl.Result{}, r.Update(ctx, p)
|
||||
return r.releaseFinalizer(ctx, p)
|
||||
case provider.Class(err) == provider.ErrPermanent:
|
||||
// Get will never answer for this instance — a host retired from the
|
||||
// config, a providerID the provider can no longer resolve. Delete is
|
||||
@@ -302,8 +299,7 @@ func (r *ProxyReconciler) reconcileDelete(ctx context.Context, p *crawlv1alpha1.
|
||||
}
|
||||
logf.FromContext(ctx).Info("instance unreachable but cleanup succeeded; removing the finalizer",
|
||||
"providerID", p.Status.ProviderID, "error", err.Error())
|
||||
controllerutil.RemoveFinalizer(p, crawlv1alpha1.FinalizerName)
|
||||
return ctrl.Result{}, r.Update(ctx, p)
|
||||
return r.releaseFinalizer(ctx, p)
|
||||
default:
|
||||
return r.deletionFailure(err)
|
||||
}
|
||||
@@ -315,6 +311,11 @@ func (r *ProxyReconciler) reconcileDelete(ctx context.Context, p *crawlv1alpha1.
|
||||
return ctrl.Result{RequeueAfter: r.DeletionPoll}, nil
|
||||
}
|
||||
|
||||
func (r *ProxyReconciler) releaseFinalizer(ctx context.Context, p *crawlv1alpha1.Proxy) (ctrl.Result, error) {
|
||||
controllerutil.RemoveFinalizer(p, crawlv1alpha1.FinalizerName)
|
||||
return ctrl.Result{}, r.Update(ctx, p)
|
||||
}
|
||||
|
||||
func (r *ProxyReconciler) reconcileExternal(_ context.Context, p *crawlv1alpha1.Proxy) (ctrl.Result, error) {
|
||||
if p.Spec.Endpoint == nil {
|
||||
// CEL guarantees an endpoint on any object that went through the API
|
||||
|
||||
@@ -88,6 +88,21 @@ type KVMHost struct {
|
||||
Bridge string `json:"bridge"`
|
||||
}
|
||||
|
||||
// Accepted KVMImage.Format and KVMImage.Firmware values.
|
||||
const (
|
||||
KVMFormatQcow2 = "qcow2"
|
||||
KVMFormatRaw = "raw"
|
||||
KVMFirmwareUEFI = "uefi"
|
||||
KVMFirmwareBIOS = "bios"
|
||||
)
|
||||
|
||||
// IsHTTPURL reports whether s is an absolute http(s) URL with a host, the
|
||||
// only kind of image source the kvm provider can fetch.
|
||||
func IsHTTPURL(s string) bool {
|
||||
u, err := url.Parse(s)
|
||||
return err == nil && (u.Scheme == "http" || u.Scheme == "https") && u.Host != ""
|
||||
}
|
||||
|
||||
var sha256Hex = regexp.MustCompile(`^[0-9a-fA-F]{64}$`)
|
||||
|
||||
// kvmIdentifier constrains the config names that are pasted into
|
||||
@@ -117,18 +132,17 @@ func validateKVM(prefix string, c *KVMConfig) error {
|
||||
return fmt.Errorf("%s: key must be letters, digits, dots, dashes or underscores"+
|
||||
" — it becomes the cached volume name base-<key>.<format>", p)
|
||||
}
|
||||
u, err := url.Parse(img.URL)
|
||||
if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
|
||||
if !IsHTTPURL(img.URL) {
|
||||
return fmt.Errorf("%s.url must be an http(s) URL (got %q)", p, img.URL)
|
||||
}
|
||||
if img.SHA256 != "" && !sha256Hex.MatchString(img.SHA256) {
|
||||
return fmt.Errorf("%s.sha256 must be 64 hex characters", p)
|
||||
}
|
||||
if img.Format != "" && img.Format != "qcow2" && img.Format != "raw" {
|
||||
return fmt.Errorf("%s.format must be qcow2 or raw (got %q)", p, img.Format)
|
||||
if img.Format != "" && img.Format != KVMFormatQcow2 && img.Format != KVMFormatRaw {
|
||||
return fmt.Errorf("%s.format must be %s or %s (got %q)", p, KVMFormatQcow2, KVMFormatRaw, img.Format)
|
||||
}
|
||||
if img.Firmware != "" && img.Firmware != "uefi" && img.Firmware != "bios" {
|
||||
return fmt.Errorf("%s.firmware must be uefi or bios (got %q)", p, img.Firmware)
|
||||
if img.Firmware != "" && img.Firmware != KVMFirmwareUEFI && img.Firmware != KVMFirmwareBIOS {
|
||||
return fmt.Errorf("%s.firmware must be %s or %s (got %q)", p, KVMFirmwareUEFI, KVMFirmwareBIOS, img.Firmware)
|
||||
}
|
||||
}
|
||||
if len(c.MachineTypes) == 0 {
|
||||
|
||||
@@ -182,7 +182,7 @@ func awaitOrCancel(ctx context.Context, fn func() error, abort func()) error {
|
||||
|
||||
// sshAuth holds the pieces of the SSH client config shared by every host:
|
||||
// the private key and the host-key policy. Both are loaded on the first
|
||||
// connection rather than at startup — see lazyDialer.
|
||||
// connection rather than at startup — see lazySSH.
|
||||
type sshAuth struct {
|
||||
signer ssh.Signer
|
||||
hostKey ssh.HostKeyCallback
|
||||
@@ -204,7 +204,7 @@ func newSSHAuth(cfg provider.KVMConfig) (*sshAuth, error) {
|
||||
hostKey := ssh.InsecureIgnoreHostKey() //nolint:gosec // opt-in via config, documented as insecure
|
||||
if !cfg.InsecureIgnoreHostKey {
|
||||
hostKey, err = knownhosts.New(cfg.KnownHostsFile)
|
||||
if err != nil && errors.Is(err, fs.ErrNotExist) {
|
||||
if errors.Is(err, fs.ErrNotExist) {
|
||||
return nil, fmt.Errorf("loading kvm.knownHostsFile: %w", err)
|
||||
}
|
||||
if err != nil {
|
||||
@@ -214,47 +214,54 @@ func newSSHAuth(cfg provider.KVMConfig) (*sshAuth, error) {
|
||||
return &sshAuth{signer: signer, hostKey: hostKey}, nil
|
||||
}
|
||||
|
||||
// lazyAuth defers loading the SSH material to the first connection and
|
||||
// lazySSH defers loading the SSH material until it is first needed and
|
||||
// caches it once it works. The kvm-ssh Secret is an optional mount: the
|
||||
// manager has to start without it and keep every other provider running,
|
||||
// and a Secret created afterwards has to heal the KVM proxies on the next
|
||||
// reconcile rather than needing a restart.
|
||||
func lazyAuth(cfg provider.KVMConfig) func() (*sshAuth, error) {
|
||||
var (
|
||||
type lazySSH struct {
|
||||
cfg provider.KVMConfig
|
||||
|
||||
mu sync.Mutex
|
||||
cached *sshAuth
|
||||
)
|
||||
return func() (*sshAuth, error) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if cached != nil {
|
||||
return cached, nil
|
||||
auth *sshAuth
|
||||
}
|
||||
auth, err := newSSHAuth(cfg)
|
||||
|
||||
// newLazySSH keeps only the SSH fields of cfg, since the loader lives as
|
||||
// long as the Provider.
|
||||
func newLazySSH(cfg provider.KVMConfig) *lazySSH {
|
||||
return &lazySSH{cfg: provider.KVMConfig{
|
||||
SSHKeyFile: cfg.SSHKeyFile,
|
||||
KnownHostsFile: cfg.KnownHostsFile,
|
||||
InsecureIgnoreHostKey: cfg.InsecureIgnoreHostKey,
|
||||
}}
|
||||
}
|
||||
|
||||
func (l *lazySSH) load() (*sshAuth, error) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
if l.auth != nil {
|
||||
return l.auth, nil
|
||||
}
|
||||
auth, err := newSSHAuth(l.cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cached = auth
|
||||
l.auth = auth
|
||||
return auth, nil
|
||||
}
|
||||
}
|
||||
|
||||
func lazyDialer(cfg provider.KVMConfig) hostDialer {
|
||||
auth := lazyAuth(cfg)
|
||||
return func(ctx context.Context, host provider.KVMHost) (*hostClient, error) {
|
||||
a, err := auth()
|
||||
// dial is the production hostDialer.
|
||||
func (l *lazySSH) dial(ctx context.Context, host provider.KVMHost) (*hostClient, error) {
|
||||
a, err := l.load()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a.dialer()(ctx, host)
|
||||
}
|
||||
return a.dial(ctx, host)
|
||||
}
|
||||
|
||||
// dialer returns the production hostDialer: SSH to the host, forward its
|
||||
// libvirt unix socket through the connection, and open a libvirt session
|
||||
// over that.
|
||||
func (a *sshAuth) dialer() hostDialer {
|
||||
return func(ctx context.Context, host provider.KVMHost) (*hostClient, error) {
|
||||
// dial SSHes to the host, forwards its libvirt unix socket through the
|
||||
// connection, and opens a libvirt session over that.
|
||||
func (a *sshAuth) dial(ctx context.Context, host provider.KVMHost) (*hostClient, error) {
|
||||
client, err := a.dialSSH(ctx, host)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -266,7 +273,6 @@ func (a *sshAuth) dialer() hostDialer {
|
||||
}
|
||||
return hc, nil
|
||||
}
|
||||
}
|
||||
|
||||
// dialSSH establishes the SSH connection. The handshake is bounded by ctx's
|
||||
// deadline, or defaultHandshakeTimeout when ctx has none, since a
|
||||
|
||||
@@ -110,41 +110,48 @@ func TestHostPool_call_redialsAfterDisconnect(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostPool_call_transportErrorDropsConnection(t *testing.T) {
|
||||
// Only a transport error is evidence about the link. A daemon-reported
|
||||
// libvirt error, or a verdict fn reached locally, must not cost the
|
||||
// connection.
|
||||
func TestHostPool_call_dropsConnectionOnlyOnTransportError(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
name string
|
||||
err error
|
||||
wantDropped bool
|
||||
}{
|
||||
{"transport error", io.ErrUnexpectedEOF, true},
|
||||
{"libvirt error", lvErr(libvirt.ErrNoDomain), false},
|
||||
{"locally classified error", fmt.Errorf("%w: creating iso writer: read-only file system", provider.ErrPermanent), false},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
d := &fakeDialer{}
|
||||
p := newHostPool(d.dial)
|
||||
err := p.call(context.Background(), discard, testHost, func(libvirtAPI) error { return io.ErrUnexpectedEOF })
|
||||
if !errors.Is(err, io.ErrUnexpectedEOF) {
|
||||
t.Fatalf("call() error = %v, want the transport error passed through", err)
|
||||
log, snapshot := captureLogger(1)
|
||||
err := p.call(context.Background(), log, testHost, func(libvirtAPI) error { return tc.err })
|
||||
if !errors.Is(err, tc.err) {
|
||||
t.Fatalf("call() error = %v, want %v passed through", err, tc.err)
|
||||
}
|
||||
if !d.clients[0].closed.Load() {
|
||||
t.Error("connection not closed after a transport error")
|
||||
if got := d.clients[0].closed.Load(); got != tc.wantDropped {
|
||||
t.Errorf("connection closed = %v, want %v", got, tc.wantDropped)
|
||||
}
|
||||
joined := strings.Join(snapshot(), "\n")
|
||||
if logged := strings.Contains(joined, `"msg"="libvirt connection dropped"`) && strings.Contains(joined, `"reason"="transport error"`); logged != tc.wantDropped {
|
||||
t.Errorf("transport drop logged = %v, want %v:\n%s", logged, tc.wantDropped, joined)
|
||||
}
|
||||
if err := p.call(context.Background(), discard, testHost, noop); err != nil {
|
||||
t.Fatalf("call() after transport error = %v", err)
|
||||
t.Fatalf("next call() error = %v", err)
|
||||
}
|
||||
if d.dialCount() != 2 {
|
||||
t.Errorf("dials = %d, want 2", d.dialCount())
|
||||
wantDials := 1
|
||||
if tc.wantDropped {
|
||||
wantDials = 2
|
||||
}
|
||||
if d.dialCount() != wantDials {
|
||||
t.Errorf("dials = %d, want %d", d.dialCount(), wantDials)
|
||||
}
|
||||
|
||||
func TestHostPool_call_libvirtErrorKeepsConnection(t *testing.T) {
|
||||
t.Parallel()
|
||||
d := &fakeDialer{}
|
||||
p := newHostPool(d.dial)
|
||||
err := p.call(context.Background(), discard, testHost, func(libvirtAPI) error { return lvErr(libvirt.ErrNoDomain) })
|
||||
if !hasCode(err, libvirt.ErrNoDomain) {
|
||||
t.Fatalf("call() error = %v, want the libvirt error passed through", err)
|
||||
}
|
||||
if err := p.call(context.Background(), discard, testHost, noop); err != nil {
|
||||
t.Fatalf("call() error = %v", err)
|
||||
}
|
||||
if d.dialCount() != 1 {
|
||||
t.Errorf("dials = %d, want 1: a daemon-reported error must not cost the connection", d.dialCount())
|
||||
}
|
||||
if d.clients[0].closed.Load() {
|
||||
t.Error("connection closed after a libvirt error")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -388,13 +395,19 @@ func newSSHFixture(t *testing.T, answerGlobalRequests bool) *sshFixture {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSSHAuth_dialSSH_forwardsUnixSocket(t *testing.T) {
|
||||
t.Parallel()
|
||||
f := newSSHFixture(t, true)
|
||||
auth, err := newSSHAuth(f.cfg)
|
||||
func mustSSHAuth(t *testing.T, cfg provider.KVMConfig) *sshAuth {
|
||||
t.Helper()
|
||||
auth, err := newSSHAuth(cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("newSSHAuth() error = %v", err)
|
||||
}
|
||||
return auth
|
||||
}
|
||||
|
||||
func TestSSHAuth_dialSSH_forwardsUnixSocket(t *testing.T) {
|
||||
t.Parallel()
|
||||
f := newSSHFixture(t, true)
|
||||
auth := mustSSHAuth(t, f.cfg)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
client, err := auth.dialSSH(ctx, f.host)
|
||||
@@ -428,10 +441,7 @@ func TestSSHAuth_dialSSH_rejectsUnknownHostKey(t *testing.T) {
|
||||
if err := os.WriteFile(f.cfg.KnownHostsFile, []byte(line+"\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
auth, err := newSSHAuth(f.cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("newSSHAuth() error = %v", err)
|
||||
}
|
||||
auth := mustSSHAuth(t, f.cfg)
|
||||
client, err := auth.dialSSH(context.Background(), f.host)
|
||||
if err == nil {
|
||||
client.Close()
|
||||
@@ -448,11 +458,7 @@ func TestSSHAuth_dialSSH_insecureIgnoreHostKey(t *testing.T) {
|
||||
f := newSSHFixture(t, true)
|
||||
f.cfg.KnownHostsFile = ""
|
||||
f.cfg.InsecureIgnoreHostKey = true
|
||||
auth, err := newSSHAuth(f.cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("newSSHAuth() error = %v", err)
|
||||
}
|
||||
client, err := auth.dialSSH(context.Background(), f.host)
|
||||
client, err := mustSSHAuth(t, f.cfg).dialSSH(context.Background(), f.host)
|
||||
if err != nil {
|
||||
t.Fatalf("dialSSH() error = %v", err)
|
||||
}
|
||||
@@ -466,11 +472,7 @@ func TestSSHAuth_dialSSH_rejectsWrongClientKey(t *testing.T) {
|
||||
if err := os.WriteFile(f.keyFile, otherPEM, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
auth, err := newSSHAuth(f.cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("newSSHAuth() error = %v", err)
|
||||
}
|
||||
client, err := auth.dialSSH(context.Background(), f.host)
|
||||
client, err := mustSSHAuth(t, f.cfg).dialSSH(context.Background(), f.host)
|
||||
if err == nil {
|
||||
client.Close()
|
||||
t.Fatal("dialSSH() succeeded with a key the server does not know")
|
||||
@@ -483,13 +485,9 @@ func TestSSHAuth_dialSSH_rejectsWrongClientKey(t *testing.T) {
|
||||
func TestSSHAuth_dialSSH_cancelledContext(t *testing.T) {
|
||||
t.Parallel()
|
||||
f := newSSHFixture(t, true)
|
||||
auth, err := newSSHAuth(f.cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("newSSHAuth() error = %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
if client, err := auth.dialSSH(ctx, f.host); err == nil {
|
||||
if client, err := mustSSHAuth(t, f.cfg).dialSSH(ctx, f.host); err == nil {
|
||||
client.Close()
|
||||
t.Fatal("dialSSH() succeeded with an already-cancelled context")
|
||||
}
|
||||
@@ -528,7 +526,7 @@ func TestNewSSHAuth_missingFiles(t *testing.T) {
|
||||
// The kvm-ssh Secret is an optional mount, so the provider has to be
|
||||
// constructible without it and has to start working once it appears — the
|
||||
// alternative was a constructor error that crash-looped the whole manager.
|
||||
func TestLazyAuth_healsWhenTheKeyAppears(t *testing.T) {
|
||||
func TestLazySSH_load_healsWhenTheKeyAppears(t *testing.T) {
|
||||
t.Parallel()
|
||||
f := newSSHFixture(t, true)
|
||||
keyPEM, err := os.ReadFile(f.keyFile)
|
||||
@@ -538,7 +536,7 @@ func TestLazyAuth_healsWhenTheKeyAppears(t *testing.T) {
|
||||
if err := os.Remove(f.keyFile); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
auth := lazyAuth(f.cfg)
|
||||
auth := newLazySSH(f.cfg).load
|
||||
|
||||
if _, err := auth(); err == nil {
|
||||
t.Fatal("loading with no key file succeeded, want an error")
|
||||
@@ -562,13 +560,13 @@ func TestLazyAuth_healsWhenTheKeyAppears(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestLazyAuth_unparseableKeyIsPermanent(t *testing.T) {
|
||||
func TestLazySSH_load_unparseableKeyIsPermanent(t *testing.T) {
|
||||
t.Parallel()
|
||||
f := newSSHFixture(t, true)
|
||||
if err := os.WriteFile(f.keyFile, []byte("not a key"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err := lazyAuth(f.cfg)()
|
||||
_, err := newLazySSH(f.cfg).load()
|
||||
if classify(err) != provider.ErrPermanent {
|
||||
t.Errorf("class = %v, want ErrPermanent: retrying cannot fix a key that does not parse", classify(err))
|
||||
}
|
||||
@@ -578,13 +576,13 @@ func TestLazyAuth_unparseableKeyIsPermanent(t *testing.T) {
|
||||
// not a stale or empty one. The echo socket is not a libvirt server, so the
|
||||
// call gets as far as the libvirt handshake and no further; what matters is
|
||||
// that it is past SSH by then.
|
||||
func TestLazyDialer_authenticatesOverSSH(t *testing.T) {
|
||||
func TestLazySSH_dial_authenticatesOverSSH(t *testing.T) {
|
||||
t.Parallel()
|
||||
f := newSSHFixture(t, true)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
_, err := lazyDialer(f.cfg)(ctx, f.host)
|
||||
_, err := newLazySSH(f.cfg).dial(ctx, f.host)
|
||||
if err == nil {
|
||||
t.Fatal("dial to an echo socket succeeded, want a libvirt handshake error")
|
||||
}
|
||||
@@ -619,11 +617,7 @@ func TestNew_missingSSHMaterialStillBuilds(t *testing.T) {
|
||||
func TestKeepalive_closesClientWhenServerStalls(t *testing.T) {
|
||||
t.Parallel()
|
||||
f := newSSHFixture(t, false)
|
||||
auth, err := newSSHAuth(f.cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("newSSHAuth() error = %v", err)
|
||||
}
|
||||
client, err := auth.dialSSH(context.Background(), f.host)
|
||||
client, err := mustSSHAuth(t, f.cfg).dialSSH(context.Background(), f.host)
|
||||
if err != nil {
|
||||
t.Fatalf("dialSSH() error = %v", err)
|
||||
}
|
||||
@@ -644,11 +638,7 @@ func TestKeepalive_closesClientWhenServerStalls(t *testing.T) {
|
||||
func TestKeepalive_keepsHealthyClientOpen(t *testing.T) {
|
||||
t.Parallel()
|
||||
f := newSSHFixture(t, true)
|
||||
auth, err := newSSHAuth(f.cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("newSSHAuth() error = %v", err)
|
||||
}
|
||||
client, err := auth.dialSSH(context.Background(), f.host)
|
||||
client, err := mustSSHAuth(t, f.cfg).dialSSH(context.Background(), f.host)
|
||||
if err != nil {
|
||||
t.Fatalf("dialSSH() error = %v", err)
|
||||
}
|
||||
@@ -705,20 +695,7 @@ func TestHostPool_call_noTracingBelowV1(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostPool_call_logsDropReason(t *testing.T) {
|
||||
t.Parallel()
|
||||
t.Run("transport error", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
d := &fakeDialer{}
|
||||
p := newHostPool(d.dial)
|
||||
log, snapshot := captureLogger(1)
|
||||
_ = p.call(context.Background(), log, testHost, func(libvirtAPI) error { return io.ErrUnexpectedEOF })
|
||||
joined := strings.Join(snapshot(), "\n")
|
||||
if !strings.Contains(joined, `"msg"="libvirt connection dropped"`) || !strings.Contains(joined, `"reason"="transport error"`) {
|
||||
t.Errorf("output missing the drop line:\n%s", joined)
|
||||
}
|
||||
})
|
||||
t.Run("context cancelled", func(t *testing.T) {
|
||||
func TestHostPool_call_logsCancelDropReason(t *testing.T) {
|
||||
t.Parallel()
|
||||
d := &fakeDialer{}
|
||||
p := newHostPool(d.dial)
|
||||
@@ -742,27 +719,4 @@ func TestHostPool_call_logsDropReason(t *testing.T) {
|
||||
if strings.Contains(joined, `"reason"="transport error"`) {
|
||||
t.Errorf("cancel was also reported as a transport error:\n%s", joined)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestHostPool_call_keepsConnectionOnClassifiedError(t *testing.T) {
|
||||
t.Parallel()
|
||||
d := &fakeDialer{}
|
||||
p := newHostPool(d.dial)
|
||||
log, snapshot := captureLogger(1)
|
||||
err := p.call(context.Background(), log, testHost, func(libvirtAPI) error {
|
||||
return fmt.Errorf("%w: creating iso writer: read-only file system", provider.ErrPermanent)
|
||||
})
|
||||
if !errors.Is(err, provider.ErrPermanent) {
|
||||
t.Fatalf("call() error = %v, want the classified error passed through", err)
|
||||
}
|
||||
if d.clients[0].closed.Load() {
|
||||
t.Error("connection closed after a locally classified error; only transport errors should drop it")
|
||||
}
|
||||
if joined := strings.Join(snapshot(), "\n"); strings.Contains(joined, "connection dropped") {
|
||||
t.Errorf("drop logged for a local error:\n%s", joined)
|
||||
}
|
||||
if err := p.call(context.Background(), log, testHost, noop); err != nil || d.dialCount() != 1 {
|
||||
t.Errorf("next call redialled (dials=%d, err=%v), want the cached connection reused", d.dialCount(), err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,10 +19,8 @@ import (
|
||||
// function to recognise. libvirt exposes no quota signal, so nothing maps
|
||||
// to QuotaExceeded.
|
||||
func classify(err error) error {
|
||||
for _, class := range []error{provider.ErrNotFound, provider.ErrQuotaExceeded, provider.ErrPermanent, provider.ErrTransient} {
|
||||
if errors.Is(err, class) {
|
||||
return class
|
||||
}
|
||||
if isClassified(err) {
|
||||
return provider.Class(err)
|
||||
}
|
||||
var lerr libvirt.Error
|
||||
if !errors.As(err, &lerr) {
|
||||
|
||||
@@ -31,7 +31,6 @@ type fakeAPI struct {
|
||||
}
|
||||
|
||||
type fakeVol struct {
|
||||
name string
|
||||
format string
|
||||
capacity int64
|
||||
backing string
|
||||
@@ -94,7 +93,7 @@ func (f *fakeAPI) vol(pool, name string) *fakeVol {
|
||||
func (f *fakeAPI) putVol(pool, name, format string, data []byte) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.pools[pool][name] = &fakeVol{name: name, format: format, capacity: int64(len(data)), data: data}
|
||||
f.pools[pool][name] = &fakeVol{format: format, capacity: int64(len(data)), data: data}
|
||||
}
|
||||
|
||||
func (f *fakeAPI) volNames(pool string) []string {
|
||||
@@ -348,7 +347,7 @@ func (f *fakeAPI) StorageVolCreateXML(pool libvirt.StoragePool, raw string, _ li
|
||||
if _, exists := vols[v.Name]; exists {
|
||||
return libvirt.StorageVol{}, lvError(libvirt.ErrStorageVolExist, "storage volume '%s' exists already", v.Name)
|
||||
}
|
||||
fv := &fakeVol{name: v.Name, format: v.Target.Format.Type, capacity: v.Capacity.Value}
|
||||
fv := &fakeVol{format: v.Target.Format.Type, capacity: v.Capacity.Value}
|
||||
if v.BackingStore != nil {
|
||||
fv.backing = v.BackingStore.Path
|
||||
}
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -23,10 +22,7 @@ import (
|
||||
// provider's images map (Key set) or a URL given directly.
|
||||
type imageSource struct {
|
||||
Key string
|
||||
URL string
|
||||
SHA256 string
|
||||
Format string
|
||||
Firmware string
|
||||
provider.KVMImage
|
||||
}
|
||||
|
||||
// resolveImage maps placement.image to a source. Anything that is neither a
|
||||
@@ -34,11 +30,10 @@ type imageSource struct {
|
||||
// surface, not retry.
|
||||
func resolveImage(cfg provider.KVMConfig, image string) (imageSource, error) {
|
||||
if img, ok := cfg.Images[image]; ok {
|
||||
return imageSource{Key: image, URL: img.URL, SHA256: img.SHA256, Format: img.Format, Firmware: img.Firmware}, nil
|
||||
return imageSource{Key: image, KVMImage: img}, nil
|
||||
}
|
||||
u, err := url.Parse(image)
|
||||
if err == nil && (u.Scheme == "http" || u.Scheme == "https") && u.Host != "" {
|
||||
return imageSource{URL: image, Format: formatQcow2, Firmware: firmwareUEFI}, nil
|
||||
if provider.IsHTTPURL(image) {
|
||||
return imageSource{KVMImage: imageDefaults(provider.KVMImage{URL: image})}, nil
|
||||
}
|
||||
return imageSource{}, fmt.Errorf("%w: placement.image %q is neither a key in kvm.images nor an http(s) URL",
|
||||
provider.ErrPermanent, image)
|
||||
|
||||
@@ -37,12 +37,11 @@ type imageServer struct {
|
||||
status int
|
||||
noLength bool
|
||||
delay time.Duration
|
||||
body []byte
|
||||
}
|
||||
|
||||
func newImageServer(t *testing.T) *imageServer {
|
||||
t.Helper()
|
||||
s := &imageServer{status: http.StatusOK, body: testImage}
|
||||
s := &imageServer{status: http.StatusOK}
|
||||
s.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
s.hits.Add(1)
|
||||
time.Sleep(s.delay)
|
||||
@@ -55,16 +54,16 @@ func newImageServer(t *testing.T) *imageServer {
|
||||
if s.noLength {
|
||||
w.(http.Flusher).Flush()
|
||||
} else {
|
||||
w.Header().Set("Content-Length", strconv.Itoa(len(s.body)))
|
||||
w.Header().Set("Content-Length", strconv.Itoa(len(testImage)))
|
||||
}
|
||||
_, _ = w.Write(s.body)
|
||||
_, _ = w.Write(testImage)
|
||||
}))
|
||||
t.Cleanup(s.Close)
|
||||
return s
|
||||
}
|
||||
|
||||
func testSource(srv *imageServer, sha string) imageSource {
|
||||
return imageSource{Key: "debian-13", URL: srv.URL + "/debian.qcow2", SHA256: sha, Format: formatQcow2}
|
||||
return imageSource{Key: "debian-13", KVMImage: provider.KVMImage{URL: srv.URL + "/debian.qcow2", SHA256: sha, Format: formatQcow2}}
|
||||
}
|
||||
|
||||
func ensureTest(t *testing.T, api *fakeAPI, src imageSource) (string, error) {
|
||||
@@ -88,9 +87,9 @@ func TestResolveImage(t *testing.T) {
|
||||
want imageSource
|
||||
wantErr bool
|
||||
}{
|
||||
{"configured key", "debian-13", imageSource{Key: "debian-13", URL: "https://cloud.debian.org/d.qcow2", SHA256: "abc", Format: "qcow2", Firmware: "uefi"}, false},
|
||||
{"direct https url", "https://x.example/img.qcow2", imageSource{URL: "https://x.example/img.qcow2", Format: "qcow2", Firmware: "uefi"}, false},
|
||||
{"direct http url", "http://mirror/img.qcow2", imageSource{URL: "http://mirror/img.qcow2", Format: "qcow2", Firmware: "uefi"}, false},
|
||||
{"configured key", "debian-13", imageSource{Key: "debian-13", KVMImage: provider.KVMImage{URL: "https://cloud.debian.org/d.qcow2", SHA256: "abc", Format: "qcow2", Firmware: "uefi"}}, false},
|
||||
{"direct https url", "https://x.example/img.qcow2", imageSource{KVMImage: provider.KVMImage{URL: "https://x.example/img.qcow2", Format: "qcow2", Firmware: "uefi"}}, false},
|
||||
{"direct http url", "http://mirror/img.qcow2", imageSource{KVMImage: provider.KVMImage{URL: "http://mirror/img.qcow2", Format: "qcow2", Firmware: "uefi"}}, false},
|
||||
{"unknown key", "ubuntu", imageSource{}, true},
|
||||
{"file url", "file:///tmp/x.qcow2", imageSource{}, true},
|
||||
{"empty", "", imageSource{}, true},
|
||||
@@ -100,9 +99,7 @@ func TestResolveImage(t *testing.T) {
|
||||
t.Parallel()
|
||||
got, err := resolveImage(cfg, tc.image)
|
||||
if tc.wantErr {
|
||||
if !errors.Is(err, provider.ErrPermanent) {
|
||||
t.Fatalf("resolveImage(%q) error = %v, want ErrPermanent", tc.image, err)
|
||||
}
|
||||
wantClass(t, err, provider.ErrPermanent)
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
@@ -117,16 +114,16 @@ func TestResolveImage(t *testing.T) {
|
||||
|
||||
func TestImageSource_volumeNames(t *testing.T) {
|
||||
t.Parallel()
|
||||
keyed := imageSource{Key: "debian-13", URL: "https://a/b.qcow2", Format: "qcow2"}
|
||||
keyed := imageSource{Key: "debian-13", KVMImage: provider.KVMImage{URL: "https://a/b.qcow2", Format: "qcow2"}}
|
||||
if keyed.volumeName() != "base-debian-13.qcow2" || keyed.markerName() != "base-debian-13.qcow2.sha256" {
|
||||
t.Errorf("keyed names = %s / %s", keyed.volumeName(), keyed.markerName())
|
||||
}
|
||||
direct := imageSource{URL: "https://a/b.qcow2", Format: "qcow2"}
|
||||
direct := imageSource{KVMImage: provider.KVMImage{URL: "https://a/b.qcow2", Format: "qcow2"}}
|
||||
name := direct.volumeName()
|
||||
if !strings.HasPrefix(name, "base-") || !strings.HasSuffix(name, ".qcow2") || len(name) != len("base-")+16+len(".qcow2") {
|
||||
t.Errorf("direct-url name = %q, want base-<16 hex>.qcow2", name)
|
||||
}
|
||||
other := imageSource{URL: "https://a/c.qcow2", Format: "qcow2"}
|
||||
other := imageSource{KVMImage: provider.KVMImage{URL: "https://a/c.qcow2", Format: "qcow2"}}
|
||||
if other.volumeName() == name {
|
||||
t.Error("different URLs produced the same volume name")
|
||||
}
|
||||
@@ -181,12 +178,27 @@ func TestImageCache_ensure_noChecksumConfigured(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestImageCache_ensure_alreadyCached(t *testing.T) {
|
||||
func TestImageCache_ensure_existingVolumes(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
name string
|
||||
base []byte
|
||||
marker []byte
|
||||
wantHits int32
|
||||
}{
|
||||
{"already cached", testImage, []byte(testImageSHA()), 0},
|
||||
{"marker missing is a partial upload", []byte("half an image"), nil, 1},
|
||||
{"configured checksum changed", []byte("an older release"), []byte(strings.Repeat("0", 64)), 1},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv := newImageServer(t)
|
||||
api := newFakeAPI("default")
|
||||
api.putVol("default", "base-debian-13.qcow2", "qcow2", testImage)
|
||||
api.putVol("default", "base-debian-13.qcow2.sha256", "raw", []byte(testImageSHA()))
|
||||
api.putVol("default", "base-debian-13.qcow2", "qcow2", tc.base)
|
||||
if tc.marker != nil {
|
||||
api.putVol("default", "base-debian-13.qcow2.sha256", "raw", tc.marker)
|
||||
}
|
||||
path, err := ensureTest(t, api, testSource(srv, testImageSHA()))
|
||||
if err != nil {
|
||||
t.Fatalf("ensure() error = %v", err)
|
||||
@@ -194,44 +206,19 @@ func TestImageCache_ensure_alreadyCached(t *testing.T) {
|
||||
if path != "/var/lib/libvirt/images/base-debian-13.qcow2" {
|
||||
t.Errorf("path = %q", path)
|
||||
}
|
||||
if srv.hits.Load() != 0 {
|
||||
t.Errorf("http hits = %d, want 0 for a cached image", srv.hits.Load())
|
||||
if srv.hits.Load() != tc.wantHits {
|
||||
t.Errorf("http hits = %d, want %d", srv.hits.Load(), tc.wantHits)
|
||||
}
|
||||
if api.hasCall("StorageVolCreateXML") || api.hasCall("StorageVolDelete") {
|
||||
t.Errorf("cached image was touched: calls = %v", api.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestImageCache_ensure_markerMissingRedownloads(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv := newImageServer(t)
|
||||
api := newFakeAPI("default")
|
||||
api.putVol("default", "base-debian-13.qcow2", "qcow2", []byte("half an image"))
|
||||
if _, err := ensureTest(t, api, testSource(srv, testImageSHA())); err != nil {
|
||||
t.Fatalf("ensure() error = %v", err)
|
||||
}
|
||||
if srv.hits.Load() != 1 {
|
||||
t.Errorf("http hits = %d, want 1: a base without its marker is a partial upload", srv.hits.Load())
|
||||
if touched := api.hasCall("StorageVolCreateXML") || api.hasCall("StorageVolDelete"); touched != (tc.wantHits > 0) {
|
||||
t.Errorf("volumes touched = %v, want %v: calls = %v", touched, tc.wantHits > 0, api.calls)
|
||||
}
|
||||
if base := api.vol("default", "base-debian-13.qcow2"); base == nil || !bytes.Equal(base.data, testImage) {
|
||||
t.Error("partial base volume was not replaced")
|
||||
}
|
||||
}
|
||||
|
||||
func TestImageCache_ensure_checksumChangedRedownloads(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv := newImageServer(t)
|
||||
api := newFakeAPI("default")
|
||||
api.putVol("default", "base-debian-13.qcow2", "qcow2", []byte("an older release"))
|
||||
api.putVol("default", "base-debian-13.qcow2.sha256", "raw", []byte(strings.Repeat("0", 64)))
|
||||
if _, err := ensureTest(t, api, testSource(srv, testImageSHA())); err != nil {
|
||||
t.Fatalf("ensure() error = %v", err)
|
||||
}
|
||||
if srv.hits.Load() != 1 {
|
||||
t.Errorf("http hits = %d, want 1 after the configured checksum changed", srv.hits.Load())
|
||||
t.Error("base volume does not hold the verified image")
|
||||
}
|
||||
if marker := api.vol("default", "base-debian-13.qcow2.sha256"); marker == nil || string(marker.data) != testImageSHA() {
|
||||
t.Errorf("marker not updated: %+v", marker)
|
||||
t.Errorf("marker = %+v, want the verified digest", marker)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -240,9 +227,7 @@ func TestImageCache_ensure_checksumMismatchIsPermanent(t *testing.T) {
|
||||
srv := newImageServer(t)
|
||||
api := newFakeAPI("default")
|
||||
_, err := ensureTest(t, api, testSource(srv, strings.Repeat("a", 64)))
|
||||
if !errors.Is(err, provider.ErrPermanent) {
|
||||
t.Fatalf("ensure() error = %v, want ErrPermanent", err)
|
||||
}
|
||||
wantClass(t, err, provider.ErrPermanent)
|
||||
if names := api.volNames("default"); len(names) != 0 {
|
||||
t.Errorf("volumes left behind after a checksum mismatch: %v", names)
|
||||
}
|
||||
@@ -283,8 +268,9 @@ func TestImageCache_ensure_noContentLengthIsPermanent(t *testing.T) {
|
||||
srv.noLength = true
|
||||
api := newFakeAPI("default")
|
||||
_, err := ensureTest(t, api, testSource(srv, ""))
|
||||
if !errors.Is(err, provider.ErrPermanent) || !strings.Contains(err.Error(), "Content-Length") {
|
||||
t.Fatalf("ensure() error = %v, want ErrPermanent about Content-Length", err)
|
||||
wantClass(t, err, provider.ErrPermanent)
|
||||
if !strings.Contains(err.Error(), "Content-Length") {
|
||||
t.Errorf("ensure() error = %v, want one about Content-Length", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -26,10 +26,10 @@ const (
|
||||
defaultDiskSizeGB = 10
|
||||
defaultSocket = "/var/run/libvirt/libvirt-sock"
|
||||
defaultPool = "default"
|
||||
formatQcow2 = "qcow2"
|
||||
formatRaw = "raw"
|
||||
firmwareUEFI = "uefi"
|
||||
firmwareBIOS = "bios"
|
||||
formatQcow2 = provider.KVMFormatQcow2
|
||||
formatRaw = provider.KVMFormatRaw
|
||||
firmwareUEFI = provider.KVMFirmwareUEFI
|
||||
firmwareBIOS = provider.KVMFirmwareBIOS
|
||||
|
||||
// undefineFlags removes everything libvirt may keep alongside a domain
|
||||
// definition, so an undefine never fails on leftover state.
|
||||
@@ -44,13 +44,7 @@ func withDefaults(cfg provider.KVMConfig) provider.KVMConfig {
|
||||
}
|
||||
images := make(map[string]provider.KVMImage, len(cfg.Images))
|
||||
for k, img := range cfg.Images {
|
||||
if img.Format == "" {
|
||||
img.Format = formatQcow2
|
||||
}
|
||||
if img.Firmware == "" {
|
||||
img.Firmware = firmwareUEFI
|
||||
}
|
||||
images[k] = img
|
||||
images[k] = imageDefaults(img)
|
||||
}
|
||||
cfg.Images = images
|
||||
hosts := make([]provider.KVMHost, len(cfg.Hosts))
|
||||
@@ -76,34 +70,41 @@ type Provider struct {
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
func imageDefaults(img provider.KVMImage) provider.KVMImage {
|
||||
if img.Format == "" {
|
||||
img.Format = formatQcow2
|
||||
}
|
||||
if img.Firmware == "" {
|
||||
img.Firmware = firmwareUEFI
|
||||
}
|
||||
return img
|
||||
}
|
||||
|
||||
// New builds a Provider that reaches its hosts over SSH. The key and
|
||||
// known_hosts files are read on the first connection, not here: they come
|
||||
// from an optional Secret mount, and a constructor that failed without it
|
||||
// would abort the whole provider set and crash-loop the manager, taking
|
||||
// the kubernetes and gcp providers down with it. Unreadable files are
|
||||
// still reported at startup, as a warning. Satisfies registry.Constructor.
|
||||
// known_hosts files are tried once here and kept when usable, but a failure
|
||||
// is only a warning, retried on the first connection: they come from an
|
||||
// optional Secret mount, and a constructor that failed without it would
|
||||
// abort the whole provider set and crash-loop the manager, taking the
|
||||
// kubernetes and gcp providers down with it. Satisfies registry.Constructor.
|
||||
func New(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, error) {
|
||||
cfg := provider.KVMConfig{}
|
||||
if pc.KVM != nil {
|
||||
cfg = *pc.KVM
|
||||
}
|
||||
if _, err := newSSHAuth(cfg); err != nil {
|
||||
loader := newLazySSH(cfg)
|
||||
if _, err := loader.load(); err != nil {
|
||||
logf.FromContext(ctx).WithName("kvm").WithValues("provider", pc.Name).Info(
|
||||
"WARNING: kvm ssh material is not usable; proxies on this provider will fail until it is",
|
||||
"error", err.Error())
|
||||
}
|
||||
return newWithDialer(pc, lazyDialer(cfg), nil), nil
|
||||
return newWithDialer(pc.Name, cfg, loader.dial, nil), nil
|
||||
}
|
||||
|
||||
// newWithDialer is the test seam: the dialer decides what libvirtAPI each
|
||||
// host resolves to, and the HTTP client is what image downloads go through.
|
||||
func newWithDialer(pc provider.ProviderConfig, dial hostDialer, httpClient *http.Client) *Provider {
|
||||
cfg := provider.KVMConfig{}
|
||||
if pc.KVM != nil {
|
||||
cfg = *pc.KVM
|
||||
}
|
||||
func newWithDialer(name string, cfg provider.KVMConfig, dial hostDialer, httpClient *http.Client) *Provider {
|
||||
return &Provider{
|
||||
name: pc.Name,
|
||||
name: name,
|
||||
cfg: withDefaults(cfg),
|
||||
hosts: newHostPool(dial),
|
||||
images: newImageCache(httpClient),
|
||||
|
||||
@@ -95,7 +95,8 @@ func testConfig(hosts ...string) provider.ProviderConfig {
|
||||
}
|
||||
|
||||
func newTestProvider(fakes hostFakes, hosts ...string) *Provider {
|
||||
p := newWithDialer(testConfig(hosts...), fakes.dial, nil)
|
||||
pc := testConfig(hosts...)
|
||||
p := newWithDialer(pc.Name, *pc.KVM, fakes.dial, nil)
|
||||
p.now = func() time.Time { return testNow }
|
||||
return p
|
||||
}
|
||||
@@ -103,7 +104,7 @@ func newTestProvider(fakes hostFakes, hosts ...string) *Provider {
|
||||
// cacheImage pre-populates the pool with the configured image and its
|
||||
// marker so Create never touches the network.
|
||||
func cacheImage(api *fakeAPI) {
|
||||
src := imageSource{Key: "debian-13", URL: testImageURL, Format: formatQcow2}
|
||||
src := imageSource{Key: "debian-13", KVMImage: provider.KVMImage{URL: testImageURL, Format: formatQcow2}}
|
||||
api.putVol("default", src.volumeName(), formatQcow2, []byte("qcow2"))
|
||||
api.putVol("default", src.markerName(), formatRaw, []byte(testImageSHA()))
|
||||
}
|
||||
@@ -241,7 +242,7 @@ func TestCreate_fetchesImageOnFirstUse(t *testing.T) {
|
||||
api := newFakeAPI("default")
|
||||
cfg := testConfig("nas1")
|
||||
cfg.KVM.Images["debian-13"] = provider.KVMImage{URL: srv.URL + "/debian.qcow2", SHA256: testImageSHA()}
|
||||
p := newWithDialer(cfg, hostFakes{"nas1": api}.dial, nil)
|
||||
p := newWithDialer(cfg.Name, *cfg.KVM, hostFakes{"nas1": api}.dial, nil)
|
||||
|
||||
if _, err := p.Create(context.Background(), testRequest()); err != nil {
|
||||
t.Fatalf("Create: %v", err)
|
||||
@@ -479,28 +480,25 @@ func TestCreate_hostFailures(t *testing.T) {
|
||||
t.Parallel()
|
||||
cases := []struct {
|
||||
name string
|
||||
setup func() (hostFakes, string)
|
||||
setup func() hostFakes
|
||||
class error
|
||||
}{
|
||||
{
|
||||
name: "host unreachable",
|
||||
setup: func() (hostFakes, string) { return hostFakes{}, "nas1" },
|
||||
setup: func() hostFakes { return hostFakes{} },
|
||||
class: provider.ErrTransient,
|
||||
},
|
||||
{
|
||||
name: "pool missing",
|
||||
setup: func() (hostFakes, string) {
|
||||
api := newFakeAPI("other")
|
||||
return hostFakes{"nas1": api}, "nas1"
|
||||
},
|
||||
setup: func() hostFakes { return hostFakes{"nas1": newFakeAPI("other")} },
|
||||
class: provider.ErrPermanent,
|
||||
},
|
||||
{
|
||||
name: "domain xml rejected",
|
||||
setup: func() (hostFakes, string) {
|
||||
setup: func() hostFakes {
|
||||
api := readyHost()
|
||||
api.fail["DomainDefineXML"] = lvError(libvirt.ErrConfigUnsupported, "no q35 here")
|
||||
return hostFakes{"nas1": api}, "nas1"
|
||||
return hostFakes{"nas1": api}
|
||||
},
|
||||
class: provider.ErrPermanent,
|
||||
},
|
||||
@@ -508,8 +506,7 @@ func TestCreate_hostFailures(t *testing.T) {
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
fakes, host := tc.setup()
|
||||
p := newTestProvider(fakes, host)
|
||||
p := newTestProvider(tc.setup(), "nas1")
|
||||
_, err := p.Create(context.Background(), testRequest())
|
||||
wantClass(t, err, tc.class)
|
||||
})
|
||||
@@ -747,9 +744,8 @@ func TestDelete_unconfiguredHostIsNotAnError(t *testing.T) {
|
||||
}
|
||||
// Get still refuses, so the live path cannot mistake a config typo for a
|
||||
// missing VM and recreate it somewhere else.
|
||||
if _, err := p.Get(context.Background(), "hosts/nas9/domains/"+testName); provider.Class(err) != provider.ErrPermanent {
|
||||
t.Errorf("Get class = %v, want ErrPermanent", provider.Class(err))
|
||||
}
|
||||
_, err := p.Get(context.Background(), "hosts/nas9/domains/"+testName)
|
||||
wantClass(t, err, provider.ErrPermanent)
|
||||
}
|
||||
|
||||
func TestDelete_errors(t *testing.T) {
|
||||
@@ -909,23 +905,6 @@ func TestMapState(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// A bad key path no longer fails the constructor: registry.Build aborts the
|
||||
// whole provider set on a constructor error, so that took the manager down
|
||||
// with every other provider. It degrades per-proxy instead — see
|
||||
// TestNew_missingSSHMaterialStillBuilds and TestLazyAuth_healsWhenTheKeyAppears.
|
||||
func TestNew_badKeyPathDefersToFirstUse(t *testing.T) {
|
||||
t.Parallel()
|
||||
cfg := testConfig("nas1")
|
||||
cfg.KVM.SSHKeyFile = t.TempDir() + "/missing"
|
||||
p, err := New(context.Background(), cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("New() error = %v, want a provider that fails only when used", err)
|
||||
}
|
||||
if err := p.Delete(context.Background(), "hosts/nas1/domains/proxy-x"); provider.Class(err) != provider.ErrTransient {
|
||||
t.Errorf("Delete class = %v, want ErrTransient", provider.Class(err))
|
||||
}
|
||||
}
|
||||
|
||||
// runAllOps drives one full lifecycle over a ready fake host so logging
|
||||
// tests see every provider method and every libvirt call.
|
||||
func runAllOps(t *testing.T, ctx context.Context, req provider.CreateRequest) {
|
||||
|
||||
@@ -38,14 +38,12 @@ func newInstanceMetadata(uid string, createdAt time.Time) instanceMetadata {
|
||||
func parseInstanceMetadata(s string) (instanceMetadata, error) {
|
||||
var raw struct {
|
||||
XMLName xml.Name `xml:"instance"`
|
||||
Managed string `xml:"managed"`
|
||||
UID string `xml:"uid"`
|
||||
CreatedAt string `xml:"createdAt"`
|
||||
instanceMetadata
|
||||
}
|
||||
if err := xml.Unmarshal([]byte(s), &raw); err != nil {
|
||||
return instanceMetadata{}, fmt.Errorf("parsing domain metadata: %w", err)
|
||||
}
|
||||
return instanceMetadata{Managed: raw.Managed, UID: raw.UID, CreatedAt: raw.CreatedAt}, nil
|
||||
return raw.instanceMetadata, nil
|
||||
}
|
||||
|
||||
func (m instanceMetadata) managed() bool {
|
||||
|
||||
@@ -29,47 +29,37 @@ func TestInstanceMetadata_roundTrip(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseInstanceMetadata_libvirtPrefixedForm(t *testing.T) {
|
||||
t.Parallel()
|
||||
// libvirt re-serialises the stored element with whatever prefix it
|
||||
// chooses; only the namespace URI is contractual.
|
||||
in := `<op:instance xmlns:op="` + metadataURI + `">` +
|
||||
`<op:managed>true</op:managed><op:uid>uid-2</op:uid>` +
|
||||
`<op:createdAt>2026-09-04T10:00:00Z</op:createdAt></op:instance>`
|
||||
m, err := parseInstanceMetadata(in)
|
||||
if err != nil {
|
||||
t.Fatalf("parseInstanceMetadata() error = %v", err)
|
||||
}
|
||||
if !m.managed() || m.UID != "uid-2" || m.createdAt().Hour() != 10 {
|
||||
t.Errorf("parsed = %+v", m)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseInstanceMetadata_libvirtBareForm(t *testing.T) {
|
||||
t.Parallel()
|
||||
// What libvirt 11.6 actually returns from DomainGetMetadata: the element
|
||||
// selected by our URI, serialised without any namespace declaration.
|
||||
in := `<instance><managed>true</managed><uid>uid-3</uid>` +
|
||||
`<createdAt>2026-09-10T07:30:00Z</createdAt></instance>`
|
||||
m, err := parseInstanceMetadata(in)
|
||||
if err != nil {
|
||||
t.Fatalf("parseInstanceMetadata() error = %v", err)
|
||||
}
|
||||
if !m.managed() || m.UID != "uid-3" || m.createdAt().Minute() != 30 {
|
||||
t.Errorf("parsed = %+v", m)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseInstanceMetadata_notManaged(t *testing.T) {
|
||||
func TestParseInstanceMetadata(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
name string
|
||||
in string
|
||||
wantErr bool
|
||||
wantManaged bool
|
||||
wantUID string
|
||||
wantCreated time.Time
|
||||
}{
|
||||
{"managed false", `<instance xmlns="` + metadataURI + `"><managed>false</managed><uid>u</uid></instance>`, false},
|
||||
{"foreign element", `<foo xmlns="urn:other"><managed>true</managed></foo>`, true},
|
||||
{"garbage", `not xml`, true},
|
||||
{
|
||||
// libvirt re-serialises the stored element with whatever prefix it
|
||||
// chooses; only the namespace URI is contractual.
|
||||
name: "libvirt prefixed form",
|
||||
in: `<op:instance xmlns:op="` + metadataURI + `">` +
|
||||
`<op:managed>true</op:managed><op:uid>uid-2</op:uid>` +
|
||||
`<op:createdAt>2026-09-04T10:00:00Z</op:createdAt></op:instance>`,
|
||||
wantManaged: true, wantUID: "uid-2", wantCreated: time.Date(2026, 9, 4, 10, 0, 0, 0, time.UTC),
|
||||
},
|
||||
{
|
||||
// What libvirt 11.6 actually returns from DomainGetMetadata: the
|
||||
// element selected by our URI, serialised without any namespace
|
||||
// declaration.
|
||||
name: "libvirt bare form",
|
||||
in: `<instance><managed>true</managed><uid>uid-3</uid>` +
|
||||
`<createdAt>2026-09-10T07:30:00Z</createdAt></instance>`,
|
||||
wantManaged: true, wantUID: "uid-3", wantCreated: time.Date(2026, 9, 10, 7, 30, 0, 0, time.UTC),
|
||||
},
|
||||
{name: "managed false", in: `<instance xmlns="` + metadataURI + `"><managed>false</managed><uid>u</uid></instance>`, wantUID: "u"},
|
||||
{name: "foreign element", in: `<foo xmlns="urn:other"><managed>true</managed></foo>`, wantErr: true},
|
||||
{name: "garbage", in: `not xml`, wantErr: true},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
@@ -78,8 +68,8 @@ func TestParseInstanceMetadata_notManaged(t *testing.T) {
|
||||
if (err != nil) != tc.wantErr {
|
||||
t.Fatalf("error = %v, wantErr %v", err, tc.wantErr)
|
||||
}
|
||||
if m.managed() {
|
||||
t.Errorf("managed() = true for %q", tc.in)
|
||||
if m.managed() != tc.wantManaged || m.UID != tc.wantUID || !m.createdAt().Equal(tc.wantCreated) {
|
||||
t.Errorf("parsed = %+v, want managed=%v uid=%q createdAt=%v", m, tc.wantManaged, tc.wantUID, tc.wantCreated)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -123,7 +123,10 @@ func (t *tracingAPI) DomainGetMetadata(d libvirt.Domain, typ int32, uri libvirt.
|
||||
func (t *tracingAPI) DomainInterfaceAddresses(d libvirt.Domain, source uint32, flags uint32) ([]libvirt.DomainInterface, error) {
|
||||
c := t.begin("DomainInterfaceAddresses", "domain", d.Name, "source", libvirt.DomainInterfaceAddressesSource(source))
|
||||
ifaces, err := t.inner.DomainInterfaceAddresses(d, source, flags)
|
||||
c.detail("addresses", formatAddresses(ifaces)).done(err, "interfaces", len(ifaces))
|
||||
if t.log.V(2).Enabled() {
|
||||
c.detail("addresses", formatAddresses(ifaces))
|
||||
}
|
||||
c.done(err, "interfaces", len(ifaces))
|
||||
return ifaces, err
|
||||
}
|
||||
|
||||
@@ -140,11 +143,14 @@ func formatAddresses(ifaces []libvirt.DomainInterface) []string {
|
||||
func (t *tracingAPI) ConnectListAllDomains(needResults int32, flags libvirt.ConnectListAllDomainsFlags) ([]libvirt.Domain, uint32, error) {
|
||||
c := t.begin("ConnectListAllDomains", flagKV(int64(flags))...)
|
||||
doms, n, err := t.inner.ConnectListAllDomains(needResults, flags)
|
||||
if t.log.V(2).Enabled() {
|
||||
names := make([]string, 0, len(doms))
|
||||
for _, d := range doms {
|
||||
names = append(names, d.Name)
|
||||
}
|
||||
c.detail("names", names).done(err, "count", len(doms))
|
||||
c.detail("names", names)
|
||||
}
|
||||
c.done(err, "count", len(doms))
|
||||
return doms, n, err
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user