diff --git a/docs/plans-executions/2026-09-11-1526-provider-health-checks.md b/docs/plans-executions/2026-09-11-1526-provider-health-checks.md index 9ba5a38..8566d90 100644 --- a/docs/plans-executions/2026-09-11-1526-provider-health-checks.md +++ b/docs/plans-executions/2026-09-11-1526-provider-health-checks.md @@ -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. diff --git a/internal/provider/kvm/check.go b/internal/provider/kvm/check.go new file mode 100644 index 0000000..a2d7f51 --- /dev/null +++ b/internal/provider/kvm/check.go @@ -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)) +} diff --git a/internal/provider/kvm/check_test.go b/internal/provider/kvm/check_test.go new file mode 100644 index 0000000..ca99639 --- /dev/null +++ b/internal/provider/kvm/check_test.go @@ -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) + } + } +} diff --git a/internal/provider/kvm/conn.go b/internal/provider/kvm/conn.go index 8f83751..da7ff1a 100644 --- a/internal/provider/kvm/conn.go +++ b/internal/provider/kvm/conn.go @@ -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 diff --git a/internal/provider/kvm/conn_test.go b/internal/provider/kvm/conn_test.go index 1340a7b..722c948 100644 --- a/internal/provider/kvm/conn_test.go +++ b/internal/provider/kvm/conn_test.go @@ -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) diff --git a/internal/provider/kvm/fake_test.go b/internal/provider/kvm/fake_test.go index 9599bb5..a0a74b5 100644 --- a/internal/provider/kvm/fake_test.go +++ b/internal/provider/kvm/fake_test.go @@ -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 +} diff --git a/internal/provider/kvm/kvm_integration_test.go b/internal/provider/kvm/kvm_integration_test.go index 6e9e157..2f4070c 100644 --- a/internal/provider/kvm/kvm_integration_test.go +++ b/internal/provider/kvm/kvm_integration_test.go @@ -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 { diff --git a/internal/provider/kvm/trace.go b/internal/provider/kvm/trace.go index a19a38d..6700509 100644 --- a/internal/provider/kvm/trace.go +++ b/internal/provider/kvm/trace.go @@ -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 +} diff --git a/internal/provider/kvm/trace_test.go b/internal/provider/kvm/trace_test.go index 0a11a30..d81883e 100644 --- a/internal/provider/kvm/trace_test.go +++ b/internal/provider/kvm/trace_test.go @@ -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))