KVM/libvirt provider: proxies as VMs on your own hypervisors #5
@@ -154,6 +154,11 @@ spec:
|
||||
- name: kvm-ssh
|
||||
mountPath: /etc/proxy-operator/kvm
|
||||
readOnly: true
|
||||
# Scratch space: the root filesystem is read-only, and the kvm
|
||||
# provider builds each VM's cloud-init seed ISO through a temp
|
||||
# directory (a few hundred KB, deleted right after upload).
|
||||
- name: tmp
|
||||
mountPath: /tmp
|
||||
volumes:
|
||||
- name: providers-config
|
||||
configMap:
|
||||
@@ -162,5 +167,8 @@ spec:
|
||||
secret:
|
||||
secretName: kvm-ssh
|
||||
optional: true
|
||||
- name: tmp
|
||||
emptyDir:
|
||||
sizeLimit: 64Mi
|
||||
serviceAccountName: controller-manager
|
||||
terminationGracePeriodSeconds: 10
|
||||
|
||||
@@ -135,10 +135,13 @@ func (p *hostPool) drop(host string, c *hostClient) {
|
||||
// call runs fn against host's connection. go-libvirt RPCs take no context
|
||||
// and block until the daemon answers, so when ctx ends first the connection
|
||||
// is closed — which unblocks fn with an error — and ctx.Err() is returned.
|
||||
// A transport error (anything that is not a libvirt.Error) also drops the
|
||||
// cached connection so the next call redials instead of reusing a dead
|
||||
// link. When log has V(1) enabled, every RPC fn makes is traced through it
|
||||
// (see tracingAPI), as are connection events.
|
||||
// A transport error also drops the cached connection so the next call
|
||||
// redials instead of reusing a dead link. "Transport error" means anything
|
||||
// that is neither a libvirt.Error nor an error fn already classified with a
|
||||
// provider sentinel (a failed seed build, a bad placement): those are local
|
||||
// verdicts, not evidence about the link. When log has V(1) enabled, every
|
||||
// RPC fn makes is traced through it (see tracingAPI), as are connection
|
||||
// events.
|
||||
func (p *hostPool) call(ctx context.Context, log logr.Logger, host provider.KVMHost, fn func(api libvirtAPI) error) error {
|
||||
log = log.WithName("libvirt")
|
||||
c, err := p.get(ctx, log, host)
|
||||
@@ -153,7 +156,7 @@ func (p *hostPool) call(ctx context.Context, log logr.Logger, host provider.KVMH
|
||||
log.V(1).Info("libvirt connection dropped", "reason", "context cancelled", "error", ctx.Err().Error())
|
||||
p.drop(host.Name, c)
|
||||
})
|
||||
if err != nil && !isLibvirtError(err) && !errors.Is(err, ctx.Err()) {
|
||||
if err != nil && !isLibvirtError(err) && !isClassified(err) && !errors.Is(err, ctx.Err()) {
|
||||
log.V(1).Info("libvirt connection dropped", "reason", "transport error", "error", err.Error())
|
||||
p.drop(host.Name, c)
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"crypto/rand"
|
||||
"encoding/pem"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
@@ -652,3 +653,25 @@ func TestHostPool_call_logsDropReason(t *testing.T) {
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -51,6 +51,18 @@ func isLibvirtError(err error) bool {
|
||||
return errors.As(err, &lerr)
|
||||
}
|
||||
|
||||
// isClassified reports whether err already carries one of the provider's
|
||||
// sentinels, i.e. the package reached a verdict itself rather than hitting
|
||||
// a transport failure.
|
||||
func isClassified(err error) bool {
|
||||
for _, class := range []error{provider.ErrNotFound, provider.ErrQuotaExceeded, provider.ErrPermanent, provider.ErrTransient} {
|
||||
if errors.Is(err, class) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func hasCode(err error, code libvirt.ErrorNumber) bool {
|
||||
var lerr libvirt.Error
|
||||
return errors.As(err, &lerr) && libvirt.ErrorNumber(lerr.Code) == code
|
||||
|
||||
Reference in New Issue
Block a user