Provider health checks: ProviderHealth CRD, per-provider checks and gauges #6
@@ -4,7 +4,7 @@ Plan: [docs/plans/2026-09-11-1526-provider-health-checks.md](../plans/2026-09-11
|
||||
- [x] Step 0 — Save the plan
|
||||
- [x] Step 1 — ProviderHealth API and CRD
|
||||
- [x] Step 2 — Contract plus the kubernetes and GCP checks
|
||||
- [ ] Step 3 — KVM check
|
||||
- [x] Step 3 — KVM check
|
||||
- [ ] Step 4 — Monitor
|
||||
- [ ] Step 5 — Metrics and wiring
|
||||
- [ ] Step 6 — Docs
|
||||
@@ -95,3 +95,46 @@ Worth noting:
|
||||
a disabled Compute API is reported as `Forbidden` rather than `Unreachable`.
|
||||
- Once a context has ended, both checks report `Timeout` whatever the error says. The
|
||||
monitor's timeout is the usual cause, and the underlying error only repeats it.
|
||||
|
||||
## Step 3 — KVM check
|
||||
|
||||
`conn.go` now wraps each of the four dial stages in a `dialError`: ssh-material, tcp,
|
||||
ssh-handshake and libvirt-connect. `Error()` returns the inner message unchanged and
|
||||
`Unwrap()` returns the cause, so `classify`/`isClassified` and every existing
|
||||
message-matching test behave as before. `TestDialError_keepsClassification` pins this
|
||||
down.
|
||||
|
||||
The three new RPCs went into `libvirtAPI`, `tracingAPI` and `fakeAPI`. The method-count
|
||||
guard in `TestTracingAPI_logsEveryMethod` fails until a traced row exists for each new
|
||||
method, which is what forced the three new rows.
|
||||
|
||||
`Check` runs one goroutine per host, over the shared connection, through
|
||||
`hostPool.call`. The pool verdict is decided only after `call` returns, because
|
||||
returning it as an error from inside the callback would make `call` drop a perfectly
|
||||
good connection.
|
||||
|
||||
Build, vet (also with the integration tag, since that file changed) and race tests:
|
||||
|
||||
```bash
|
||||
go build ./internal/provider/... && go vet ./internal/provider/kvm/ \
|
||||
&& go vet -tags=integration ./internal/provider/kvm/ \
|
||||
&& go test -race -count=1 ./internal/provider/kvm/
|
||||
```
|
||||
|
||||
Worth noting:
|
||||
- **Pool state wording:** `poolState` uses virsh's words (`running`, `inactive`, …)
|
||||
rather than go-libvirt's constant names, so it reads like `virsh pool-list` output.
|
||||
- **Details:** each host also reports its SSH `user`. That is the first thing to check
|
||||
when the reason is `SSHHandshakeFailed`.
|
||||
- **Tests renamed or merged:**
|
||||
- The planned `TestSSHAuth_dial_stages` is `TestLazySSH_dial_tagsTheFailingStage`. It
|
||||
dials through `lazySSH`, so the ssh-material stage is covered as well as the three
|
||||
stages the in-process SSH server can trigger.
|
||||
- The pool-inactive and pool-missing cases are merged into `TestCheck_libvirtVerdicts`,
|
||||
which also covers a daemon error. Each case runs two checks and asserts a single
|
||||
dial.
|
||||
- **Tests added beyond the plan:**
|
||||
- `TestCheck_timeout`: a daemon that stops answering gives `Timeout`.
|
||||
- `TestDialError_keepsClassification`.
|
||||
- **Not run yet:** `TestKVMCheck_realHost` in the integration file only compiles here.
|
||||
It needs beelink, so it goes on the Step 7 checklist for the user to run.
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
package kvm
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
libvirt "github.com/digitalocean/go-libvirt"
|
||||
|
||||
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
|
||||
)
|
||||
|
||||
var _ provider.Checker = (*Provider)(nil)
|
||||
|
||||
// Reasons only the KVM check can tell apart; the rest are provider-wide.
|
||||
const (
|
||||
reasonSSHMaterialUnavailable = "SSHMaterialUnavailable"
|
||||
reasonSSHHandshakeFailed = "SSHHandshakeFailed"
|
||||
reasonLibvirtConnectFailed = "LibvirtConnectFailed"
|
||||
reasonLibvirtError = "LibvirtError"
|
||||
reasonPoolNotFound = "PoolNotFound"
|
||||
reasonPoolNotActive = "PoolNotActive"
|
||||
)
|
||||
|
||||
// hostFacts is what a successful check learns about a host.
|
||||
type hostFacts struct {
|
||||
version uint64
|
||||
hostname string
|
||||
poolState libvirt.StoragePoolState
|
||||
capacity uint64
|
||||
available uint64
|
||||
}
|
||||
|
||||
// Check verifies every configured host concurrently, in kvm.hosts order.
|
||||
// It goes through the same pooled connection Create and Get use, so the
|
||||
// SSH and libvirt handshakes only happen on the first check or after the
|
||||
// link dropped; then two cheap RPCs prove the daemon answers, and the
|
||||
// storage pool VMs are created in must exist and be running.
|
||||
func (p *Provider) Check(ctx context.Context) []provider.TargetStatus {
|
||||
out := make([]provider.TargetStatus, len(p.cfg.Hosts))
|
||||
var wg sync.WaitGroup
|
||||
for i, h := range p.cfg.Hosts {
|
||||
wg.Go(func() { out[i] = p.checkHost(ctx, h) })
|
||||
}
|
||||
wg.Wait()
|
||||
return out
|
||||
}
|
||||
|
||||
// checkHost returns the RPC errors from fn untouched: call drops the
|
||||
// shared connection on anything that is not a libvirt error, and a pool
|
||||
// that exists but is stopped says nothing about the link. facts is read
|
||||
// only after call returned nil, since an abandoned fn may still write it.
|
||||
func (p *Provider) checkHost(ctx context.Context, host provider.KVMHost) provider.TargetStatus {
|
||||
ts := provider.TargetStatus{
|
||||
Name: host.Name,
|
||||
Address: host.Address,
|
||||
Details: map[string]string{"user": host.User, "pool": host.Pool},
|
||||
}
|
||||
start := time.Now()
|
||||
var facts hostFacts
|
||||
err := p.hosts.call(ctx, p.logger(ctx), host, func(api libvirtAPI) error {
|
||||
version, err := api.ConnectGetLibVersion()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
hostname, err := api.ConnectGetHostname()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
pool, err := api.StoragePoolLookupByName(host.Pool)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
state, capacity, _, available, err := api.StoragePoolGetInfo(pool)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
facts = hostFacts{
|
||||
version: version,
|
||||
hostname: hostname,
|
||||
poolState: libvirt.StoragePoolState(state),
|
||||
capacity: capacity,
|
||||
available: available,
|
||||
}
|
||||
return nil
|
||||
})
|
||||
ts.Latency = time.Since(start)
|
||||
if err != nil {
|
||||
ts.Reason = checkReason(err)
|
||||
ts.Message = err.Error()
|
||||
return ts
|
||||
}
|
||||
ts.Details["hostname"] = facts.hostname
|
||||
ts.Details["libvirtVersion"] = formatLibVersion(facts.version)
|
||||
ts.Details["poolState"] = poolStateName(facts.poolState)
|
||||
ts.Details["poolCapacity"] = formatGiB(facts.capacity)
|
||||
ts.Details["poolAvailable"] = formatGiB(facts.available)
|
||||
if facts.poolState != libvirt.StoragePoolRunning {
|
||||
ts.Reason = reasonPoolNotActive
|
||||
ts.Message = fmt.Sprintf("storage pool %q is %s", host.Pool, poolStateName(facts.poolState))
|
||||
return ts
|
||||
}
|
||||
ts.Ready = true
|
||||
ts.Reason = provider.ReasonReachable
|
||||
return ts
|
||||
}
|
||||
|
||||
// checkReason names where a failed check broke. The dial stage wins over
|
||||
// a timeout: "the SSH handshake did not finish" says more than "the check
|
||||
// ran out of time", and the message still carries the deadline.
|
||||
func checkReason(err error) string {
|
||||
var de *dialError
|
||||
if errors.As(err, &de) {
|
||||
switch de.stage {
|
||||
case stageSSHMaterial:
|
||||
return reasonSSHMaterialUnavailable
|
||||
case stageSSHHandshake:
|
||||
return reasonSSHHandshakeFailed
|
||||
case stageLibvirtConnect:
|
||||
return reasonLibvirtConnectFailed
|
||||
default:
|
||||
return provider.ReasonUnreachable
|
||||
}
|
||||
}
|
||||
switch {
|
||||
case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled):
|
||||
return provider.ReasonTimeout
|
||||
case hasCode(err, libvirt.ErrNoStoragePool):
|
||||
return reasonPoolNotFound
|
||||
case isLibvirtError(err):
|
||||
return reasonLibvirtError
|
||||
default:
|
||||
return provider.ReasonUnreachable
|
||||
}
|
||||
}
|
||||
|
||||
// formatLibVersion decodes libvirt's major*1e6 + minor*1e3 + micro.
|
||||
func formatLibVersion(v uint64) string {
|
||||
return fmt.Sprintf("%d.%d.%d", v/1_000_000, v/1_000%1_000, v%1_000)
|
||||
}
|
||||
|
||||
// poolStateName uses virsh's words for a pool state.
|
||||
func poolStateName(s libvirt.StoragePoolState) string {
|
||||
switch s {
|
||||
case libvirt.StoragePoolInactive:
|
||||
return "inactive"
|
||||
case libvirt.StoragePoolBuilding:
|
||||
return "building"
|
||||
case libvirt.StoragePoolRunning:
|
||||
return "running"
|
||||
case libvirt.StoragePoolDegraded:
|
||||
return "degraded"
|
||||
case libvirt.StoragePoolInaccessible:
|
||||
return "inaccessible"
|
||||
default:
|
||||
return fmt.Sprintf("state %d", s)
|
||||
}
|
||||
}
|
||||
|
||||
func formatGiB(bytes uint64) string {
|
||||
return fmt.Sprintf("%.1fGiB", float64(bytes)/(1<<30))
|
||||
}
|
||||
@@ -0,0 +1,270 @@
|
||||
package kvm
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"maps"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
libvirt "github.com/digitalocean/go-libvirt"
|
||||
|
||||
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
|
||||
)
|
||||
|
||||
// withoutLatency zeroes the one field a test cannot predict.
|
||||
func withoutLatency(in []provider.TargetStatus) []provider.TargetStatus {
|
||||
out := make([]provider.TargetStatus, len(in))
|
||||
for i, ts := range in {
|
||||
ts.Latency = 0
|
||||
out[i] = ts
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestCheck_reachable(t *testing.T) {
|
||||
t.Parallel()
|
||||
api := newFakeAPI("default")
|
||||
api.poolCapacity = 10 << 30
|
||||
api.pools["default"]["disk.qcow2"] = &fakeVol{capacity: 4 << 30}
|
||||
p := newTestProvider(hostFakes{"nas1": api}, "nas1")
|
||||
|
||||
got := withoutLatency(p.Check(context.Background()))
|
||||
|
||||
want := provider.TargetStatus{
|
||||
Name: "nas1",
|
||||
Address: "nas1:22",
|
||||
Ready: true,
|
||||
Reason: provider.ReasonReachable,
|
||||
Details: map[string]string{
|
||||
"user": "virt",
|
||||
"pool": "default",
|
||||
"hostname": "fake-host",
|
||||
"libvirtVersion": "10.2.0",
|
||||
"poolState": "running",
|
||||
"poolCapacity": "10.0GiB",
|
||||
"poolAvailable": "6.0GiB",
|
||||
},
|
||||
}
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("Check() returned %d targets, want 1", len(got))
|
||||
}
|
||||
if got[0].Name != want.Name || got[0].Address != want.Address || got[0].Ready != want.Ready ||
|
||||
got[0].Reason != want.Reason || got[0].Message != "" || !maps.Equal(got[0].Details, want.Details) {
|
||||
t.Errorf("Check() = %+v, want %+v", got[0], want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheck_perHostVerdicts(t *testing.T) {
|
||||
t.Parallel()
|
||||
p := newTestProvider(hostFakes{"nas1": newFakeAPI("default"), "nas3": newFakeAPI("default")}, "nas1", "nas2", "nas3")
|
||||
|
||||
got := p.Check(context.Background())
|
||||
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("Check() returned %d targets, want 3", len(got))
|
||||
}
|
||||
for i, want := range []struct {
|
||||
name string
|
||||
ready bool
|
||||
}{{"nas1", true}, {"nas2", false}, {"nas3", true}} {
|
||||
if got[i].Name != want.name || got[i].Ready != want.ready {
|
||||
t.Errorf("target %d = %s ready=%v, want %s ready=%v", i, got[i].Name, got[i].Ready, want.name, want.ready)
|
||||
}
|
||||
}
|
||||
if got[1].Reason != provider.ReasonUnreachable || !strings.Contains(got[1].Message, errHostOff.Error()) {
|
||||
t.Errorf("unreachable host = %s %q, want Unreachable carrying %q", got[1].Reason, got[1].Message, errHostOff)
|
||||
}
|
||||
if got[1].Details["pool"] != "default" || got[1].Details["hostname"] != "" {
|
||||
t.Errorf("unreachable host details = %v, want the configured pool and nothing learned", got[1].Details)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheck_libvirtVerdicts also checks twice per case: none of these
|
||||
// verdicts is evidence against the link, so the connection must survive.
|
||||
func TestCheck_libvirtVerdicts(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
name string
|
||||
setup func(api *fakeAPI)
|
||||
pools []string
|
||||
wantReason string
|
||||
wantMessage string
|
||||
}{
|
||||
{
|
||||
name: "pool inactive",
|
||||
pools: []string{"default"},
|
||||
setup: func(api *fakeAPI) { api.poolStates["default"] = libvirt.StoragePoolInactive },
|
||||
wantReason: reasonPoolNotActive,
|
||||
wantMessage: `storage pool "default" is inactive`,
|
||||
},
|
||||
{
|
||||
name: "pool missing",
|
||||
pools: []string{"other"},
|
||||
wantReason: reasonPoolNotFound,
|
||||
wantMessage: "no storage pool with matching name 'default'",
|
||||
},
|
||||
{
|
||||
name: "daemon error",
|
||||
pools: []string{"default"},
|
||||
setup: func(api *fakeAPI) { api.fail["ConnectGetHostname"] = lvError(libvirt.ErrInternalError, "boom") },
|
||||
wantReason: reasonLibvirtError,
|
||||
wantMessage: "boom",
|
||||
},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
api := newFakeAPI(tc.pools...)
|
||||
if tc.setup != nil {
|
||||
tc.setup(api)
|
||||
}
|
||||
var dials atomic.Int32
|
||||
dial := func(ctx context.Context, h provider.KVMHost) (*hostClient, error) {
|
||||
dials.Add(1)
|
||||
return hostFakes{"nas1": api}.dial(ctx, h)
|
||||
}
|
||||
pc := testConfig("nas1")
|
||||
p := newWithDialer(pc.Name, *pc.KVM, dial, nil)
|
||||
|
||||
for range 2 {
|
||||
got := p.Check(context.Background())
|
||||
if got[0].Ready || got[0].Reason != tc.wantReason || !strings.Contains(got[0].Message, tc.wantMessage) {
|
||||
t.Errorf("Check() = %+v, want not ready, %s, message containing %q", got[0], tc.wantReason, tc.wantMessage)
|
||||
}
|
||||
}
|
||||
if n := dials.Load(); n != 1 {
|
||||
t.Errorf("dialed %d times over two checks, want 1", n)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheck_hostsCheckedConcurrently(t *testing.T) {
|
||||
t.Parallel()
|
||||
nas2Dialing := make(chan struct{})
|
||||
dial := func(ctx context.Context, h provider.KVMHost) (*hostClient, error) {
|
||||
switch h.Name {
|
||||
case "nas1":
|
||||
select {
|
||||
case <-nas2Dialing:
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
case "nas2":
|
||||
close(nas2Dialing)
|
||||
}
|
||||
return &hostClient{api: newFakeAPI("default"), close: func() {}}, nil
|
||||
}
|
||||
pc := testConfig("nas1", "nas2")
|
||||
p := newWithDialer(pc.Name, *pc.KVM, dial, nil)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
got := p.Check(ctx)
|
||||
|
||||
for _, ts := range got {
|
||||
if !ts.Ready {
|
||||
t.Errorf("%s = %s %q, want ready: a sequential check would time out nas1", ts.Name, ts.Reason, ts.Message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheck_timeout(t *testing.T) {
|
||||
t.Parallel()
|
||||
api := newFakeAPI("default")
|
||||
block := make(chan struct{})
|
||||
t.Cleanup(func() { close(block) })
|
||||
blocked := &blockingAPI{fakeAPI: api, block: block}
|
||||
dial := func(context.Context, provider.KVMHost) (*hostClient, error) {
|
||||
return &hostClient{api: blocked, close: func() {}}, nil
|
||||
}
|
||||
pc := testConfig("nas1")
|
||||
p := newWithDialer(pc.Name, *pc.KVM, dial, nil)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
got := p.Check(ctx)
|
||||
|
||||
if got[0].Ready || got[0].Reason != provider.ReasonTimeout {
|
||||
t.Errorf("Check() = %+v, want Timeout", got[0])
|
||||
}
|
||||
}
|
||||
|
||||
// blockingAPI hangs in ConnectGetLibVersion until block closes, like a
|
||||
// daemon that stopped answering.
|
||||
type blockingAPI struct {
|
||||
*fakeAPI
|
||||
block chan struct{}
|
||||
}
|
||||
|
||||
func (b *blockingAPI) ConnectGetLibVersion() (uint64, error) {
|
||||
<-b.block
|
||||
return 0, io.EOF
|
||||
}
|
||||
|
||||
func TestCheckReason(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
name string
|
||||
err error
|
||||
want string
|
||||
}{
|
||||
{"ssh material", &dialError{stage: stageSSHMaterial, err: fmt.Errorf("%w: parsing key", provider.ErrPermanent)}, reasonSSHMaterialUnavailable},
|
||||
{"tcp", &dialError{stage: stageTCP, err: errors.New("ssh dial: connection refused")}, provider.ReasonUnreachable},
|
||||
{"ssh handshake", &dialError{stage: stageSSHHandshake, err: errors.New("ssh handshake: unable to authenticate")}, reasonSSHHandshakeFailed},
|
||||
{"libvirt connect", &dialError{stage: stageLibvirtConnect, err: errors.New("libvirt connect via /sock: EOF")}, reasonLibvirtConnectFailed},
|
||||
{"stage survives the pool's wrapping", fmt.Errorf("connecting to host nas1 (nas1:22): %w",
|
||||
&dialError{stage: stageSSHHandshake, err: context.DeadlineExceeded}), reasonSSHHandshakeFailed},
|
||||
{"deadline", context.DeadlineExceeded, provider.ReasonTimeout},
|
||||
{"cancelled", context.Canceled, provider.ReasonTimeout},
|
||||
{"no pool", lvError(libvirt.ErrNoStoragePool, "no pool"), reasonPoolNotFound},
|
||||
{"other libvirt error", lvError(libvirt.ErrInternalError, "boom"), reasonLibvirtError},
|
||||
{"link died mid-call", io.EOF, provider.ReasonUnreachable},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
if got := checkReason(tc.err); got != tc.want {
|
||||
t.Errorf("checkReason(%v) = %s, want %s", tc.err, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDialError_keepsClassification(t *testing.T) {
|
||||
t.Parallel()
|
||||
err := fmt.Errorf("connecting to host nas1: %w",
|
||||
&dialError{stage: stageSSHMaterial, err: fmt.Errorf("%w: parsing kvm.sshKeyFile: bad", provider.ErrPermanent)})
|
||||
|
||||
if got := classify(err); !errors.Is(got, provider.ErrPermanent) {
|
||||
t.Errorf("classify() = %v, want ErrPermanent through the dialError", got)
|
||||
}
|
||||
if !isClassified(err) {
|
||||
t.Error("isClassified() = false, want true")
|
||||
}
|
||||
if want := "connecting to host nas1: permanent: parsing kvm.sshKeyFile: bad"; !strings.HasSuffix(err.Error(), "parsing kvm.sshKeyFile: bad") {
|
||||
t.Errorf("message = %q, want it unchanged, like %q", err.Error(), want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFormatLibVersion(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
in uint64
|
||||
want string
|
||||
}{
|
||||
{10002000, "10.2.0"},
|
||||
{11006001, "11.6.1"},
|
||||
{9000000, "9.0.0"},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
if got := formatLibVersion(tc.in); got != tc.want {
|
||||
t.Errorf("formatLibVersion(%d) = %s, want %s", tc.in, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -44,6 +44,9 @@ type libvirtAPI interface {
|
||||
StorageVolUpload(vol libvirt.StorageVol, r io.Reader, offset, length uint64, flags libvirt.StorageVolUploadFlags) error
|
||||
StorageVolDownload(vol libvirt.StorageVol, w io.Writer, offset, length uint64, flags libvirt.StorageVolDownloadFlags) error
|
||||
StorageVolDelete(vol libvirt.StorageVol, flags libvirt.StorageVolDeleteFlags) error
|
||||
ConnectGetLibVersion() (uint64, error)
|
||||
ConnectGetHostname() (string, error)
|
||||
StoragePoolGetInfo(pool libvirt.StoragePool) (state uint8, capacity, allocation, available uint64, err error)
|
||||
}
|
||||
|
||||
// hostClient is one live connection to a hypervisor.
|
||||
@@ -68,6 +71,27 @@ func (c *hostClient) alive() bool {
|
||||
// real one; tests inject their own.
|
||||
type hostDialer func(ctx context.Context, host provider.KVMHost) (*hostClient, error)
|
||||
|
||||
// dialStage names the step of a connection attempt that failed, so the
|
||||
// health check can say why a host is unreachable without parsing messages.
|
||||
type dialStage string
|
||||
|
||||
const (
|
||||
stageSSHMaterial dialStage = "ssh-material"
|
||||
stageTCP dialStage = "tcp"
|
||||
stageSSHHandshake dialStage = "ssh-handshake"
|
||||
stageLibvirtConnect dialStage = "libvirt-connect"
|
||||
)
|
||||
|
||||
// dialError tags a dial failure with its stage. It keeps the message and
|
||||
// unwraps to the cause, so classify sees exactly what it saw before.
|
||||
type dialError struct {
|
||||
stage dialStage
|
||||
err error
|
||||
}
|
||||
|
||||
func (e *dialError) Error() string { return e.err.Error() }
|
||||
func (e *dialError) Unwrap() error { return e.err }
|
||||
|
||||
// hostPool caches one connection per host, dialing lazily and redialing
|
||||
// after the connection drops. Dials for different hosts don't block each
|
||||
// other; concurrent callers for the same host share a single dial.
|
||||
@@ -254,7 +278,7 @@ func (l *lazySSH) load() (*sshAuth, error) {
|
||||
func (l *lazySSH) dial(ctx context.Context, host provider.KVMHost) (*hostClient, error) {
|
||||
a, err := l.load()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return nil, &dialError{stage: stageSSHMaterial, err: err}
|
||||
}
|
||||
return a.dial(ctx, host)
|
||||
}
|
||||
@@ -285,7 +309,7 @@ func (a *sshAuth) dialSSH(ctx context.Context, host provider.KVMHost) (*ssh.Clie
|
||||
var d net.Dialer
|
||||
nc, err := d.DialContext(ctx, "tcp", host.Address)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ssh dial: %w", err)
|
||||
return nil, &dialError{stage: stageTCP, err: fmt.Errorf("ssh dial: %w", err)}
|
||||
}
|
||||
_ = nc.SetDeadline(deadline)
|
||||
conn, chans, reqs, err := ssh.NewClientConn(nc, host.Address, &ssh.ClientConfig{
|
||||
@@ -295,7 +319,7 @@ func (a *sshAuth) dialSSH(ctx context.Context, host provider.KVMHost) (*ssh.Clie
|
||||
})
|
||||
if err != nil {
|
||||
nc.Close()
|
||||
return nil, fmt.Errorf("ssh handshake: %w", err)
|
||||
return nil, &dialError{stage: stageSSHHandshake, err: fmt.Errorf("ssh handshake: %w", err)}
|
||||
}
|
||||
_ = nc.SetDeadline(time.Time{})
|
||||
return ssh.NewClient(conn, chans, reqs), nil
|
||||
@@ -319,7 +343,7 @@ func (u unixForward) Dial() (net.Conn, error) {
|
||||
func connectLibvirt(ctx context.Context, client *ssh.Client, socket string) (*hostClient, error) {
|
||||
lv := libvirt.NewWithDialer(unixForward{client: client, path: socket})
|
||||
if err := awaitOrCancel(ctx, lv.Connect, func() { client.Close() }); err != nil {
|
||||
return nil, fmt.Errorf("libvirt connect via %s: %w", socket, err)
|
||||
return nil, &dialError{stage: stageLibvirtConnect, err: fmt.Errorf("libvirt connect via %s: %w", socket, err)}
|
||||
}
|
||||
stop := make(chan struct{})
|
||||
var once sync.Once
|
||||
|
||||
@@ -614,6 +614,86 @@ func TestNew_missingSSHMaterialStillBuilds(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The health check reports where a connection broke, so every stage of a
|
||||
// real dial must come back tagged.
|
||||
func TestLazySSH_dial_tagsTheFailingStage(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
name string
|
||||
mutate func(t *testing.T, f *sshFixture)
|
||||
want dialStage
|
||||
wantReason string
|
||||
}{
|
||||
{
|
||||
name: "key material missing",
|
||||
mutate: func(t *testing.T, f *sshFixture) { f.cfg.SSHKeyFile = filepath.Join(t.TempDir(), "absent") },
|
||||
want: stageSSHMaterial,
|
||||
wantReason: reasonSSHMaterialUnavailable,
|
||||
},
|
||||
{
|
||||
name: "nothing listening",
|
||||
mutate: func(t *testing.T, f *sshFixture) {
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.host.Address = ln.Addr().String()
|
||||
ln.Close()
|
||||
},
|
||||
want: stageTCP,
|
||||
wantReason: provider.ReasonUnreachable,
|
||||
},
|
||||
{
|
||||
name: "host key not in known_hosts",
|
||||
mutate: func(t *testing.T, f *sshFixture) {
|
||||
impostor, _ := newKeyPair(t)
|
||||
line := knownhosts.Line([]string{f.server.addr}, impostor.PublicKey())
|
||||
if err := os.WriteFile(f.cfg.KnownHostsFile, []byte(line+"\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
},
|
||||
want: stageSSHHandshake,
|
||||
wantReason: reasonSSHHandshakeFailed,
|
||||
},
|
||||
{
|
||||
name: "client key rejected",
|
||||
mutate: func(t *testing.T, f *sshFixture) {
|
||||
_, otherPEM := newKeyPair(t)
|
||||
if err := os.WriteFile(f.keyFile, otherPEM, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
},
|
||||
want: stageSSHHandshake,
|
||||
wantReason: reasonSSHHandshakeFailed,
|
||||
},
|
||||
{
|
||||
name: "libvirt socket missing",
|
||||
mutate: func(_ *testing.T, f *sshFixture) { f.host.Socket = "/nonexistent/libvirt-sock" },
|
||||
want: stageLibvirtConnect,
|
||||
wantReason: reasonLibvirtConnectFailed,
|
||||
},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
f := newSSHFixture(t, true)
|
||||
tc.mutate(t, f)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
_, err := newLazySSH(f.cfg).dial(ctx, f.host)
|
||||
|
||||
var de *dialError
|
||||
if !errors.As(err, &de) || de.stage != tc.want {
|
||||
t.Fatalf("dial() error = %v, want a %s dialError", err, tc.want)
|
||||
}
|
||||
if got := checkReason(err); got != tc.wantReason {
|
||||
t.Errorf("checkReason() = %s, want %s", got, tc.wantReason)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestKeepalive_closesClientWhenServerStalls(t *testing.T) {
|
||||
t.Parallel()
|
||||
f := newSSHFixture(t, false)
|
||||
|
||||
@@ -28,6 +28,12 @@ type fakeAPI struct {
|
||||
// uploadHook, when set, sees every StorageVolUpload before the fake
|
||||
// reads the stream; it can return an error to simulate a broken upload.
|
||||
uploadHook func(vol string, r io.Reader) error
|
||||
// hostname, libVersion, poolCapacity and poolStates answer the health
|
||||
// check's RPCs. A pool missing from poolStates is running.
|
||||
hostname string
|
||||
libVersion uint64
|
||||
poolCapacity uint64
|
||||
poolStates map[string]libvirt.StoragePoolState
|
||||
}
|
||||
|
||||
type fakeVol struct {
|
||||
@@ -53,6 +59,11 @@ func newFakeAPI(pools ...string) *fakeAPI {
|
||||
domains: map[string]*fakeDomain{},
|
||||
fail: map[string]error{},
|
||||
failOnce: map[string]error{},
|
||||
|
||||
hostname: "fake-host",
|
||||
libVersion: 10002000,
|
||||
poolCapacity: 100 << 30,
|
||||
poolStates: map[string]libvirt.StoragePoolState{},
|
||||
}
|
||||
for _, p := range pools {
|
||||
f.pools[p] = map[string]*fakeVol{}
|
||||
@@ -409,3 +420,46 @@ func (f *fakeAPI) StorageVolDelete(vol libvirt.StorageVol, _ libvirt.StorageVolD
|
||||
delete(f.pools[vol.Pool], vol.Name)
|
||||
return nil
|
||||
}
|
||||
|
||||
// --- host ---
|
||||
|
||||
func (f *fakeAPI) ConnectGetLibVersion() (uint64, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if err := f.enter("ConnectGetLibVersion"); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return f.libVersion, nil
|
||||
}
|
||||
|
||||
func (f *fakeAPI) ConnectGetHostname() (string, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if err := f.enter("ConnectGetHostname"); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return f.hostname, nil
|
||||
}
|
||||
|
||||
// StoragePoolGetInfo reports the pool's state and poolCapacity minus the
|
||||
// volumes' capacities as available space.
|
||||
func (f *fakeAPI) StoragePoolGetInfo(pool libvirt.StoragePool) (uint8, uint64, uint64, uint64, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if err := f.enter("StoragePoolGetInfo"); err != nil {
|
||||
return 0, 0, 0, 0, err
|
||||
}
|
||||
vols, ok := f.pools[pool.Name]
|
||||
if !ok {
|
||||
return 0, 0, 0, 0, lvError(libvirt.ErrNoStoragePool, "no pool %s", pool.Name)
|
||||
}
|
||||
var used uint64
|
||||
for _, v := range vols {
|
||||
used += uint64(v.capacity)
|
||||
}
|
||||
state := libvirt.StoragePoolRunning
|
||||
if s, ok := f.poolStates[pool.Name]; ok {
|
||||
state = s
|
||||
}
|
||||
return uint8(state), f.poolCapacity, used, f.poolCapacity - used, nil
|
||||
}
|
||||
|
||||
@@ -347,6 +347,36 @@ func TestKVM_lifecycle(t *testing.T) {
|
||||
env.keep = false
|
||||
}
|
||||
|
||||
// TestKVMCheck_realHost runs the provider health check against the host.
|
||||
// It creates nothing: it must reach the host and find the pool running.
|
||||
func TestKVMCheck_realHost(t *testing.T) {
|
||||
env := loadIntegrationEnv(t)
|
||||
log := integrationLogger(t, env.logLevel)
|
||||
ctx, cancel := context.WithTimeout(logr.NewContext(context.Background(), log), time.Minute)
|
||||
defer cancel()
|
||||
|
||||
prov, err := New(ctx, env.cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
got := prov.(*Provider).Check(ctx)
|
||||
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("Check returned %d targets, want 1", len(got))
|
||||
}
|
||||
ts := got[0]
|
||||
t.Logf("%s (%s): ready=%v reason=%s latency=%s details=%v",
|
||||
ts.Name, ts.Address, ts.Ready, ts.Reason, ts.Latency.Round(time.Millisecond), ts.Details)
|
||||
if !ts.Ready {
|
||||
t.Fatalf("check failed: %s: %s", ts.Reason, ts.Message)
|
||||
}
|
||||
for _, key := range []string{"hostname", "libvirtVersion", "poolState", "poolAvailable"} {
|
||||
if ts.Details[key] == "" {
|
||||
t.Errorf("details[%s] is empty, want what the host reported", key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// waitForRunning polls Get until the instance is Running with an IP or ctx
|
||||
// ends. Transient errors are retried; anything else fails the test.
|
||||
func waitForRunning(ctx context.Context, t *testing.T, p *Provider, id string) *provider.Instance {
|
||||
|
||||
@@ -209,3 +209,24 @@ func (t *tracingAPI) StorageVolDelete(vol libvirt.StorageVol, flags libvirt.Stor
|
||||
c.done(err)
|
||||
return err
|
||||
}
|
||||
|
||||
func (t *tracingAPI) ConnectGetLibVersion() (uint64, error) {
|
||||
c := t.begin("ConnectGetLibVersion")
|
||||
v, err := t.inner.ConnectGetLibVersion()
|
||||
c.done(err, "version", formatLibVersion(v))
|
||||
return v, err
|
||||
}
|
||||
|
||||
func (t *tracingAPI) ConnectGetHostname() (string, error) {
|
||||
c := t.begin("ConnectGetHostname")
|
||||
h, err := t.inner.ConnectGetHostname()
|
||||
c.done(err, "hostname", h)
|
||||
return h, err
|
||||
}
|
||||
|
||||
func (t *tracingAPI) StoragePoolGetInfo(pool libvirt.StoragePool) (uint8, uint64, uint64, uint64, error) {
|
||||
c := t.begin("StoragePoolGetInfo", "pool", pool.Name)
|
||||
state, capacity, allocation, available, err := t.inner.StoragePoolGetInfo(pool)
|
||||
c.done(err, "state", poolStateName(libvirt.StoragePoolState(state)), "capacity", capacity, "available", available)
|
||||
return state, capacity, allocation, available, err
|
||||
}
|
||||
|
||||
@@ -103,6 +103,9 @@ func TestTracingAPI_logsEveryMethod(t *testing.T) {
|
||||
return api.StorageVolDownload(overlay, io.Discard, 0, 0, 0)
|
||||
}},
|
||||
{"StorageVolDelete", func(api libvirtAPI) error { return api.StorageVolDelete(overlay, 0) }},
|
||||
{"ConnectGetLibVersion", func(api libvirtAPI) error { _, err := api.ConnectGetLibVersion(); return err }},
|
||||
{"ConnectGetHostname", func(api libvirtAPI) error { _, err := api.ConnectGetHostname(); return err }},
|
||||
{"StoragePoolGetInfo", func(api libvirtAPI) error { _, _, _, _, err := api.StoragePoolGetInfo(pool); return err }},
|
||||
}
|
||||
if n := reflect.TypeOf((*libvirtAPI)(nil)).Elem().NumMethod(); n != len(cases) {
|
||||
t.Fatalf("libvirtAPI has %d methods, this test covers %d; trace the new one", n, len(cases))
|
||||
|
||||
Reference in New Issue
Block a user