From ce62fe57f6f0f1652a577363554c9a15ff1c0aa8 Mon Sep 17 00:00:00 2001 From: kacerr Date: Wed, 16 Sep 2026 19:26:51 +0000 Subject: [PATCH] kvm2: health checker on dedicated connections; kvm2 in the ProviderHealth type comment Co-Authored-By: Claude Fable 5.1 --- api/v1alpha1/providerhealth_types.go | 2 +- .../crawl.example.com_providerhealths.yaml | 2 +- .../2026-09-16-1333-kvm2-provider.md | 33 +- internal/provider/kvm2/check.go | 182 ++++++ internal/provider/kvm2/check_test.go | 572 ++++++++++++++++++ internal/provider/kvm2/kvm2_test.go | 6 + 6 files changed, 794 insertions(+), 3 deletions(-) create mode 100644 internal/provider/kvm2/check.go create mode 100644 internal/provider/kvm2/check_test.go diff --git a/api/v1alpha1/providerhealth_types.go b/api/v1alpha1/providerhealth_types.go index 329463f..c12e058 100644 --- a/api/v1alpha1/providerhealth_types.go +++ b/api/v1alpha1/providerhealth_types.go @@ -68,7 +68,7 @@ type ProviderTarget struct { // ProviderHealthStatus is the result of the latest check of one provider. type ProviderHealthStatus struct { // type is the provider's type from the providers config: kubernetes, - // gcp or kvm. + // gcp, kvm or kvm2. // +optional Type string `json:"type,omitempty"` diff --git a/config/crd/bases/crawl.example.com_providerhealths.yaml b/config/crd/bases/crawl.example.com_providerhealths.yaml index 838d3f8..2f207cf 100644 --- a/config/crd/bases/crawl.example.com_providerhealths.yaml +++ b/config/crd/bases/crawl.example.com_providerhealths.yaml @@ -183,7 +183,7 @@ spec: type: description: |- type is the provider's type from the providers config: kubernetes, - gcp or kvm. + gcp, kvm or kvm2. type: string type: object type: object diff --git a/docs/plans-executions/2026-09-16-1333-kvm2-provider.md b/docs/plans-executions/2026-09-16-1333-kvm2-provider.md index d2a559d..0d84110 100644 --- a/docs/plans-executions/2026-09-16-1333-kvm2-provider.md +++ b/docs/plans-executions/2026-09-16-1333-kvm2-provider.md @@ -9,7 +9,7 @@ Plan: [docs/plans/2026-09-16-1333-kvm2-provider.md](../plans/2026-09-16-1333-kvm - [x] Step 4 — Transport, seam, decorator, fakes - [x] Step 5 — Image cache and pool-space precheck - [x] Step 6 — The four contract methods -- [ ] Step 7 — Checker and the CRD comment +- [x] Step 7 — Checker and the CRD comment - [ ] Step 8 — Deployment files, Makefile, integration test - [ ] Step 9 — Documentation - [ ] Step 10 — Milestone review @@ -250,3 +250,34 @@ grep -n 'provider.Wrap(' internal/provider/kvm2/*.go | grep -v _test.go | grep - ``` All `ok`. Attestation: "files read outside the allowed set: none". + +## Step 7 — Checker and the CRD comment + +2026-09-16 18:52. Sixth subagent step: `check.go` and `check_test.go` +(ten tests, the reasons table has 19 rows). Every host is checked +concurrently on a fresh dedicated connection that is closed when the probe +ends; the pooled table is never consulted, which the test proves with a +counting dialer. Details fill in as facts arrive, so a `PoolNotFound` +verdict still says which libvirt version answered. The `status.type` comment +in `api/v1alpha1/providerhealth_types.go` gained `kvm2` and `make manifests` +changed only the CRD's description line: + +```sh +make manifests && git diff --stat config/ api/ +# api/v1alpha1/providerhealth_types.go | 2 +- +# config/crd/bases/crawl.example.com_providerhealths.yaml | 2 +- +``` + +Decisions the agent made, accepted: a dial that runs out of time is +`Unreachable` (the host did not answer), and `Timeout` is reserved for ctx or +budget expiry on an established connection; `Close` runs synchronously in +the probe so the close-count assertion is deterministic, at the cost of up +to five seconds past ctx if a disconnect hangs on a live link. + +The agent also reported that the `captureContext` helper in `kvm2_test.go` +appends to an unlocked slice, which the race detector flagged once `Check` +logged from several goroutines. The helper was outside its allowed file set, +so it built its own locked capture; the orchestrator fixed `captureContext` +with a mutex in this commit and re-ran the package at `-race -count=3`. + +Attestation: "files read outside the allowed set: none". diff --git a/internal/provider/kvm2/check.go b/internal/provider/kvm2/check.go new file mode 100644 index 0000000..ebce097 --- /dev/null +++ b/internal/provider/kvm2/check.go @@ -0,0 +1,182 @@ +package kvm2 + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "time" + + "github.com/digitalocean/go-libvirt" + "golang.org/x/crypto/ssh/knownhosts" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +var _ provider.Checker = (*Provider)(nil) + +// Reasons only this provider can tell apart (spec entry 16). Each names a +// failure an admin fixes somewhere other than where the shared reasons +// point. +const ( + // reasonCredentialsUnavailable: the key or known_hosts file is missing + // or unparsable; fix the Secret in the cluster, not the host. + reasonCredentialsUnavailable = "CredentialsUnavailable" + // reasonPoolNotFound: the configured pool is not defined on the host; + // compare with `virsh pool-list --all`. + reasonPoolNotFound = "PoolNotFound" + // reasonPoolNotActive: the pool exists but is not running; `virsh + // pool-start ` (and autostart) brings it back. + reasonPoolNotActive = "PoolNotActive" +) + +// Check verifies every configured host concurrently, each on a fresh +// dedicated connection: dial, libvirt version, hostname, pool lookup, pool +// info, close. The pooled connection is never touched, so a check that runs +// out of time closes only its own connection and cannot fail a reconcile's +// call in flight. Every target gets a verdict, in config order. +func (p *Provider) Check(ctx context.Context) []provider.TargetStatus { + out := make([]provider.TargetStatus, len(p.cfg.Hosts)) + var wg sync.WaitGroup + for i, host := range p.cfg.Hosts { + wg.Go(func() { out[i] = p.checkHost(ctx, host) }) + } + wg.Wait() + return out +} + +func (p *Provider) checkHost(ctx context.Context, host provider.KVMHost) provider.TargetStatus { + start := time.Now() + log := p.logger(ctx).WithValues("host", host.Name) + ts := provider.TargetStatus{ + Name: host.Name, + Address: host.Address, + Details: map[string]string{"user": host.User, "pool": host.Pool}, + } + err := p.probeHost(ctx, host, ts.Details) + ts.Latency = time.Since(start) + switch { + case err == nil: + ts.Ready = true + ts.Reason = provider.ReasonReachable + default: + var inactive *poolNotActiveError + if errors.As(err, &inactive) { + ts.Reason = reasonPoolNotActive + } else { + ts.Reason = checkReason(err) + } + ts.Message = err.Error() + } + log.V(1).Info("kvm2 host checked", "ready", ts.Ready, "reason", ts.Reason, "latency", ts.Latency) + return ts +} + +// probeHost runs the check sequence on its own connection and fills details +// as facts arrive, so a failure late in the sequence still reports what the +// host said before it. +func (p *Provider) probeHost(ctx context.Context, host provider.KVMHost, details map[string]string) error { + b, err := p.hosts.dialFresh(ctx, host) + if err != nil { + return err + } + defer func() { _ = b.Close() }() + + version, err := b.ConnectGetLibVersion(ctx) + if err != nil { + return err + } + details["libvirtVersion"] = formatLibVersion(version) + hostname, err := b.ConnectGetHostname(ctx) + if err != nil { + return err + } + details["hostname"] = hostname + pool, err := b.StoragePoolLookupByName(ctx, host.Pool) + if err != nil { + return err + } + state, capacity, _, available, err := b.StoragePoolGetInfo(ctx, pool) + if err != nil { + return err + } + details["poolState"] = poolStateName(state) + details["poolCapacityGiB"] = formatGiB(capacity) + details["poolAvailableGiB"] = formatGiB(available) + if libvirt.StoragePoolState(state) != libvirt.StoragePoolRunning { + return &poolNotActiveError{pool: host.Pool, state: poolStateName(state)} + } + return nil +} + +// poolNotActiveError is the one verdict that is not an error from the +// backend: the pool answered, but is not usable. +type poolNotActiveError struct { + pool string + state string +} + +func (e *poolNotActiveError) Error() string { + return fmt.Sprintf("storage pool %q is %s", e.pool, e.state) +} + +// checkReason maps a failed probe to a reason. A dial failure is judged by +// its stage, so a TCP connect or an SSH handshake that ran out of time is +// Unreachable, not Timeout: the host did not answer. Timeout is for ctx +// ending or a budget expiring once the connection was up. +func checkReason(err error) string { + var de *dialError + if errors.As(err, &de) { + switch de.stage { + case stageCredentials: + return reasonCredentialsUnavailable + case stageSSH: + var keyErr *knownhosts.KeyError + if errors.As(de.err, &keyErr) || strings.Contains(de.err.Error(), sshAuthFailed) { + return provider.ReasonUnauthorized + } + } + return provider.ReasonUnreachable + } + switch { + case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled): + return provider.ReasonTimeout + case hasCode(err, libvirt.ErrAccessDenied), hasCode(err, libvirt.ErrAuthFailed): + return provider.ReasonForbidden + case hasCode(err, libvirt.ErrNoStoragePool): + return reasonPoolNotFound + } + if _, ok := libvirtCode(err); ok { + return provider.ReasonAPIError + } + return provider.ReasonUnreachable +} + +// formatLibVersion decodes ConnectGetLibVersion's major*1_000_000 + +// minor*1_000 + 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(state uint8) string { + switch libvirt.StoragePoolState(state) { + 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", state) + } +} + +func formatGiB(bytes uint64) string { + return fmt.Sprintf("%.1f", float64(bytes)/(1<<30)) +} diff --git a/internal/provider/kvm2/check_test.go b/internal/provider/kvm2/check_test.go new file mode 100644 index 0000000..7356cd0 --- /dev/null +++ b/internal/provider/kvm2/check_test.go @@ -0,0 +1,572 @@ +package kvm2 + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "slices" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/digitalocean/go-libvirt" + "github.com/go-logr/logr/funcr" + "golang.org/x/crypto/ssh/knownhosts" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// checkContext is captureContext with a sink that tolerates the concurrent +// per-host log lines of Check. +func checkContext() (context.Context, *logCapture) { + logs := &logCapture{} + log := funcr.New(func(prefix, args string) { + logs.mu.Lock() + defer logs.mu.Unlock() + logs.lines = append(logs.lines, prefix+" "+args) + }, funcr.Options{Verbosity: 2}) + return logf.IntoContext(context.Background(), log), logs +} + +// checkDialer hands out a distinct fake per host name, fails the hosts in +// errs, counts every dial, and, with block set, waits for ctx to end. +type checkDialer struct { + backends map[string]*fakeBackend + errs map[string]error + block bool + + dials atomic.Int32 +} + +func (d *checkDialer) dial(ctx context.Context, host provider.KVMHost) (backend, error) { + d.dials.Add(1) + if d.block { + <-ctx.Done() + return nil, fmt.Errorf("%w: dialing %s: %w", provider.ErrTransient, host.Name, ctx.Err()) + } + if err := d.errs[host.Name]; err != nil { + return nil, err + } + b, ok := d.backends[host.Name] + if !ok { + return nil, fmt.Errorf("checkDialer: no backend for %s", host.Name) + } + return b, nil +} + +func threeHosts() provider.KVMConfig { + cfg := twoHosts() + cfg.Hosts = append(cfg.Hosts, provider.KVMHost{Name: "lab3", Address: "lab3:22", User: "ops", Bridge: "br0", Pool: "ssd"}) + return cfg +} + +func fakesFor(cfg provider.KVMConfig) map[string]*fakeBackend { + out := make(map[string]*fakeBackend, len(cfg.Hosts)) + for _, h := range cfg.Hosts { + f := newFakeBackend() + f.hostname = h.Name + ".home.lan" + if h.Pool != "" { + f.pools[h.Pool] = newFakePool() + } + out[h.Name] = f + } + return out +} + +func detailKeys(d map[string]string) []string { + keys := make([]string, 0, len(d)) + for k := range d { + keys = append(keys, k) + } + slices.Sort(keys) + return keys +} + +func TestCheck_oneEntryPerHostInConfigOrder(t *testing.T) { + t.Parallel() + cfg := threeHosts() + d := &checkDialer{backends: fakesFor(cfg), errs: map[string]error{"lab2": errors.New("dial tcp 192.0.2.2:22: connect: no route to host")}} + p := newProvider(testProvider, cfg, d) + ctx, _ := checkContext() + got := p.Check(ctx) + if len(got) != len(cfg.Hosts) { + t.Fatalf("Check() returned %d targets, want %d", len(got), len(cfg.Hosts)) + } + for i, h := range cfg.Hosts { + ts := got[i] + if ts.Name != h.Name || ts.Address != h.Address { + t.Errorf("target %d = %s (%s), want %s (%s)", i, ts.Name, ts.Address, h.Name, h.Address) + } + if ts.Name == "" || ts.Address == "" { + t.Errorf("target %d has an empty name or address: %+v", i, ts) + } + wantReady := h.Name != "lab2" + if ts.Ready != wantReady { + t.Errorf("target %s ready = %v, want %v (%s: %s)", ts.Name, ts.Ready, wantReady, ts.Reason, ts.Message) + } + } + if got[1].Reason != provider.ReasonUnreachable || got[0].Reason != provider.ReasonReachable { + t.Errorf("reasons = %s, %s, %s", got[0].Reason, got[1].Reason, got[2].Reason) + } +} + +func TestCheck_hostsCheckedConcurrently(t *testing.T) { + t.Parallel() + cfg := twoHosts() + fakes := fakesFor(cfg) + for _, f := range fakes { + release := f.blockOn("ConnectGetLibVersion") + time.AfterFunc(200*time.Millisecond, func() { close(release) }) + } + p := newProvider(testProvider, cfg, &checkDialer{backends: fakes}) + ctx, _ := checkContext() + start := time.Now() + got := p.Check(ctx) + elapsed := time.Since(start) + if elapsed > 350*time.Millisecond { + t.Errorf("Check() took %v, want under 350ms for two 200ms hosts", elapsed) + } + for _, ts := range got { + if !ts.Ready { + t.Errorf("target %s not ready: %s: %s", ts.Name, ts.Reason, ts.Message) + } + if ts.Latency < 200*time.Millisecond { + t.Errorf("target %s latency = %v, want at least the 200ms block", ts.Name, ts.Latency) + } + } +} + +func TestCheck_expiredContextReturnsPromptly(t *testing.T) { + t.Parallel() + cfg := twoHosts() + p := newProvider(testProvider, cfg, &checkDialer{block: true}) + ctx, _ := checkContext() + ctx, cancel := context.WithCancel(ctx) + cancel() + start := time.Now() + got := p.Check(ctx) + if elapsed := time.Since(start); elapsed > time.Second { + t.Errorf("Check() took %v with an expired ctx, want under 1s", elapsed) + } + if len(got) != len(cfg.Hosts) { + t.Fatalf("Check() returned %d targets, want %d", len(got), len(cfg.Hosts)) + } + for _, ts := range got { + if ts.Ready || ts.Reason != provider.ReasonTimeout { + t.Errorf("target %s = ready %v reason %s, want not ready, Timeout", ts.Name, ts.Ready, ts.Reason) + } + if ts.Message == "" { + t.Errorf("target %s has no message", ts.Name) + } + } +} + +// keyMaterial is the sentinel in the unusable key file of the credentials +// row; it must never reach a message. +const keyMaterial = "SENTINEL-PRIVATE-KEY-BYTES-9f3a" + +func TestCheck_reasonsTable(t *testing.T) { + t.Parallel() + sshHandshake := func(cause error) error { + return &dialError{stage: stageSSH, host: testHost, err: fmt.Errorf("ssh: handshake failed: %w", cause)} + } + tests := []struct { + name string + // dialErr builds the dial failure; nil means the dial succeeds. + dialErr func(t *testing.T) error + prep func(f *fakeBackend) + timeout time.Duration + cancel bool + + wantReady bool + wantReason string + wantMessage string + }{ + { + name: "credentials unusable", + dialErr: func(t *testing.T) error { + path := filepath.Join(t.TempDir(), "id_ed25519") + if err := os.WriteFile(path, []byte(keyMaterial), 0o600); err != nil { + t.Fatal(err) + } + _, err := loadCredentials(provider.KVMConfig{SSHKeyFile: path, KnownHostsFile: "/nonexistent/known_hosts"}) + if err == nil { + t.Fatal("loadCredentials accepted the sentinel file") + } + return &dialError{stage: stageCredentials, host: testHost, err: err} + }, + wantReason: reasonCredentialsUnavailable, + }, + { + name: "credentials missing", + dialErr: func(*testing.T) error { + _, err := loadCredentials(provider.KVMConfig{SSHKeyFile: "/nonexistent/kvm2/id_ed25519"}) + return &dialError{stage: stageCredentials, host: testHost, err: err} + }, + wantReason: reasonCredentialsUnavailable, + }, + { + name: "tcp refused", + dialErr: func(*testing.T) error { + return &dialError{stage: stageTCP, host: testHost, err: errors.New("dial tcp 192.0.2.1:22: connect: connection refused")} + }, + wantReason: provider.ReasonUnreachable, + }, + { + name: "tcp timed out", + dialErr: func(*testing.T) error { + return &dialError{stage: stageTCP, host: testHost, err: fmt.Errorf("dial tcp 192.0.2.1:22: %w", context.DeadlineExceeded)} + }, + wantReason: provider.ReasonUnreachable, + }, + { + name: "ssh host key unknown", + dialErr: func(*testing.T) error { return sshHandshake(&knownhosts.KeyError{}) }, + wantReason: provider.ReasonUnauthorized, + }, + { + name: "ssh host key mismatch", + dialErr: func(*testing.T) error { + return sshHandshake(&knownhosts.KeyError{Want: []knownhosts.KnownKey{{Filename: "known_hosts", Line: 3}}}) + }, + wantReason: provider.ReasonUnauthorized, + }, + { + name: "ssh authentication refused", + dialErr: func(*testing.T) error { + return sshHandshake(errors.New("ssh: unable to authenticate, attempted methods [none publickey], no supported methods remain")) + }, + wantReason: provider.ReasonUnauthorized, + }, + { + name: "ssh handshake reset", + dialErr: func(*testing.T) error { return sshHandshake(errors.New("read tcp: connection reset by peer")) }, + wantReason: provider.ReasonUnreachable, + }, + { + name: "ssh handshake timed out", + dialErr: func(*testing.T) error { return sshHandshake(context.DeadlineExceeded) }, + wantReason: provider.ReasonUnreachable, + }, + { + name: "socket refused", + dialErr: func(*testing.T) error { + return &dialError{stage: stageSocket, host: testHost, err: errors.New("opening /var/run/libvirt/libvirt-sock: ssh: rejected: connect failed (No such file or directory)")} + }, + wantReason: provider.ReasonUnreachable, + wantMessage: "/var/run/libvirt/libvirt-sock", + }, + { + name: "libvirt access denied", + prep: func(f *fakeBackend) { + f.failWith("ConnectGetLibVersion", fakeErr(libvirt.ErrAccessDenied, "access denied")) + }, + wantReason: provider.ReasonForbidden, + }, + { + name: "libvirt auth failed", + prep: func(f *fakeBackend) { + f.failWith("ConnectGetHostname", fakeErr(libvirt.ErrAuthFailed, "authentication failed")) + }, + wantReason: provider.ReasonForbidden, + }, + { + name: "pool not found", + prep: func(f *fakeBackend) { + f.mu.Lock() + delete(f.pools, fakePoolName) + f.mu.Unlock() + }, + wantReason: reasonPoolNotFound, + wantMessage: fakePoolName, + }, + { + name: "other libvirt error", + prep: func(f *fakeBackend) { + f.failWith("StoragePoolGetInfo", fakeErr(libvirt.ErrInternalError, "internal error: pool info")) + }, + wantReason: provider.ReasonAPIError, + }, + { + name: "rpc outlives the deadline", + prep: func(f *fakeBackend) { f.blockOn("ConnectGetLibVersion") }, + timeout: 50 * time.Millisecond, + wantReason: provider.ReasonTimeout, + }, + { + name: "ctx cancelled mid check", + prep: func(f *fakeBackend) { f.blockOn("StoragePoolLookupByName") }, + cancel: true, + wantReason: provider.ReasonTimeout, + }, + { + name: "link killed mid check", + prep: func(f *fakeBackend) { + f.failWith("StoragePoolLookupByName", transientf("libvirt call StoragePoolLookupByName on beelink: connection closed while waiting: keepalive: EOF")) + }, + wantReason: provider.ReasonUnreachable, + }, + { + name: "pool inactive", + prep: func(f *fakeBackend) { f.pools[fakePoolName].state = libvirt.StoragePoolInactive }, + wantReason: reasonPoolNotActive, + wantMessage: `storage pool "default" is inactive`, + }, + { + name: "ok", + wantReady: true, + wantReason: provider.ReasonReachable, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + cfg := twoHosts() + cfg.Hosts = cfg.Hosts[:1] + f := newFakeBackend() + d := &checkDialer{backends: map[string]*fakeBackend{testHost: f}} + if tc.dialErr != nil { + d.errs = map[string]error{testHost: tc.dialErr(t)} + } + if tc.prep != nil { + tc.prep(f) + } + p := newProvider(testProvider, cfg, d) + ctx, _ := checkContext() + switch { + case tc.timeout > 0: + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, tc.timeout) + defer cancel() + case tc.cancel: + var cancel context.CancelFunc + ctx, cancel = context.WithCancel(ctx) + time.AfterFunc(50*time.Millisecond, cancel) + } + got := p.Check(ctx) + if len(got) != 1 { + t.Fatalf("Check() returned %d targets, want 1", len(got)) + } + ts := got[0] + if ts.Ready != tc.wantReady || ts.Reason != tc.wantReason { + t.Errorf("verdict = ready %v reason %s (%s), want ready %v reason %s", ts.Ready, ts.Reason, ts.Message, tc.wantReady, tc.wantReason) + } + if !tc.wantReady && ts.Message == "" { + t.Error("failed check has no message") + } + if tc.wantReady && ts.Message != "" { + t.Errorf("successful check has a message: %q", ts.Message) + } + if strings.Contains(ts.Message, keyMaterial) { + t.Errorf("message leaks the key file content: %q", ts.Message) + } + if tc.wantMessage != "" && !strings.Contains(ts.Message, tc.wantMessage) { + t.Errorf("message = %q, want it to contain %q", ts.Message, tc.wantMessage) + } + if ts.Details["user"] != "virt" || ts.Details["pool"] != "default" { + t.Errorf("details = %v, want user and pool", ts.Details) + } + if tc.wantReady { + want := map[string]string{ + "user": "virt", "pool": "default", "hostname": "fake-host", "libvirtVersion": "11.6.0", + "poolState": "running", "poolCapacityGiB": "100.0", "poolAvailableGiB": "50.0", + } + for k, v := range want { + if ts.Details[k] != v { + t.Errorf("details[%s] = %q, want %q", k, ts.Details[k], v) + } + } + } + }) + } +} + +func TestCheck_neverUsesPooledConnection(t *testing.T) { + t.Parallel() + cfg := threeHosts() + fakes := fakesFor(cfg) + d := &checkDialer{backends: fakes} + p := newProvider(testProvider, cfg, d) + ctx, _ := checkContext() + got := p.Check(ctx) + for _, ts := range got { + if !ts.Ready { + t.Errorf("target %s not ready: %s: %s", ts.Name, ts.Reason, ts.Message) + } + } + if n := d.dials.Load(); int(n) != len(cfg.Hosts) { + t.Errorf("dials = %d, want one fresh dial per host (%d)", n, len(cfg.Hosts)) + } + p.hosts.mu.Lock() + pooled := len(p.hosts.hosts) + p.hosts.mu.Unlock() + if pooled != 0 { + t.Errorf("pooled table has %d entries after Check, want none", pooled) + } + for name, f := range fakes { + if f.closeCount() != 1 { + t.Errorf("fresh backend for %s closed %d times, want 1", name, f.closeCount()) + } + } +} + +func TestCheck_detailsKeys(t *testing.T) { + t.Parallel() + full := []string{"hostname", "libvirtVersion", "pool", "poolAvailableGiB", "poolCapacityGiB", "poolState", "user"} + tests := []struct { + name string + dialErr error + prep func(f *fakeBackend) + want []string + }{ + {name: "success", want: full}, + {name: "pool not active", prep: func(f *fakeBackend) { f.pools[fakePoolName].state = libvirt.StoragePoolDegraded }, want: full}, + { + name: "dial failure", + dialErr: &dialError{stage: stageTCP, host: testHost, err: errors.New("dial tcp: connection refused")}, + want: []string{"pool", "user"}, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + cfg := twoHosts() + cfg.Hosts = cfg.Hosts[:1] + f := newFakeBackend() + if tc.prep != nil { + tc.prep(f) + } + d := &checkDialer{backends: map[string]*fakeBackend{testHost: f}} + if tc.dialErr != nil { + d.errs = map[string]error{testHost: tc.dialErr} + } + p := newProvider(testProvider, cfg, d) + ctx, _ := checkContext() + got := p.Check(ctx)[0] + if keys := detailKeys(got.Details); !slices.Equal(keys, tc.want) { + t.Errorf("details keys = %v, want %v", keys, tc.want) + } + for k, v := range got.Details { + if v == "" { + t.Errorf("details[%s] is empty", k) + } + } + }) + } +} + +func TestCheck_logsVerdictPerHost(t *testing.T) { + t.Parallel() + cfg := twoHosts() + d := &checkDialer{backends: fakesFor(cfg), errs: map[string]error{"lab2": &dialError{stage: stageTCP, host: "lab2", err: errors.New("dial tcp: i/o timeout")}}} + p := newProvider(testProvider, cfg, d) + ctx, logs := checkContext() + p.Check(ctx) + checked := logs.matching("kvm2 host checked") + if len(checked) != 2 { + t.Fatalf("got %d verdict lines, want 2: %v", len(checked), checked) + } + for _, l := range checked { + switch { + case strings.Contains(l, `"host"="beelink"`): + if !strings.Contains(l, `"ready"=true`) || !strings.Contains(l, `"reason"="Reachable"`) || !strings.Contains(l, `"latency"=`) { + t.Errorf("beelink verdict line = %s", l) + } + case strings.Contains(l, `"host"="lab2"`): + if !strings.Contains(l, `"ready"=false`) || !strings.Contains(l, `"reason"="Unreachable"`) { + t.Errorf("lab2 verdict line = %s", l) + } + default: + t.Errorf("verdict line without a host: %s", l) + } + } +} + +func TestCheckReason_table(t *testing.T) { + t.Parallel() + dial := func(stage dialStage, cause error) error { + return fmt.Errorf("wrapped: %w", &dialError{stage: stage, host: testHost, err: cause}) + } + tests := []struct { + name string + err error + want string + }{ + {name: "credentials stage", err: dial(stageCredentials, errors.New("reading ssh key file /k: no such file")), want: reasonCredentialsUnavailable}, + {name: "credentials stage even when permanent", err: dial(stageCredentials, permanentf("parsing ssh key file /k: ssh: no key found")), want: reasonCredentialsUnavailable}, + {name: "tcp stage", err: dial(stageTCP, errors.New("connect: connection refused")), want: provider.ReasonUnreachable}, + {name: "tcp stage with a deadline", err: dial(stageTCP, context.DeadlineExceeded), want: provider.ReasonUnreachable}, + {name: "ssh stage unknown host key", err: dial(stageSSH, &knownhosts.KeyError{}), want: provider.ReasonUnauthorized}, + {name: "ssh stage wrapped key error", err: dial(stageSSH, fmt.Errorf("ssh: handshake failed: %w", &knownhosts.KeyError{Want: []knownhosts.KnownKey{{}}})), want: provider.ReasonUnauthorized}, + {name: "ssh stage unable to authenticate", err: dial(stageSSH, errors.New("ssh: handshake failed: ssh: unable to authenticate, attempted methods [none publickey], no supported methods remain")), want: provider.ReasonUnauthorized}, + {name: "ssh stage reset", err: dial(stageSSH, errors.New("read tcp: connection reset by peer")), want: provider.ReasonUnreachable}, + {name: "ssh stage cancelled", err: dial(stageSSH, context.Canceled), want: provider.ReasonUnreachable}, + {name: "socket stage", err: dial(stageSocket, errors.New("opening /run/libvirt/libvirt-sock: ssh: rejected")), want: provider.ReasonUnreachable}, + {name: "deadline exceeded", err: fmt.Errorf("%w: libvirt call ConnectGetHostname on beelink exceeded its 30s budget: %w", provider.ErrTransient, context.DeadlineExceeded), want: provider.ReasonTimeout}, + {name: "cancelled", err: fmt.Errorf("%w: libvirt call ConnectGetHostname on beelink abandoned: %w", provider.ErrTransient, context.Canceled), want: provider.ReasonTimeout}, + {name: "access denied", err: fakeErr(libvirt.ErrAccessDenied, "access denied"), want: provider.ReasonForbidden}, + {name: "auth failed wrapped", err: fmt.Errorf("connecting to libvirt on host beelink: %w", fakeErr(libvirt.ErrAuthFailed, "authentication failed")), want: provider.ReasonForbidden}, + {name: "no storage pool", err: fakeErr(libvirt.ErrNoStoragePool, "Storage pool not found"), want: reasonPoolNotFound}, + {name: "other libvirt error", err: fakeErr(libvirt.ErrInternalError, "internal error"), want: provider.ReasonAPIError}, + {name: "link killed", err: transientf("libvirt call X on beelink: connection closed while waiting: keepalive: EOF"), want: provider.ReasonUnreachable}, + {name: "plain error", err: errors.New("unexpected EOF"), want: provider.ReasonUnreachable}, + {name: "nil", err: nil, want: 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 TestFormatLibVersion(t *testing.T) { + t.Parallel() + tests := []struct { + in uint64 + want string + }{ + {0, "0.0.0"}, + {11006000, "11.6.0"}, + {10000003, "10.0.3"}, + {9010012, "9.10.12"}, + {100999999, "100.999.999"}, + } + for _, tc := range tests { + t.Run(tc.want, func(t *testing.T) { + t.Parallel() + if got := formatLibVersion(tc.in); got != tc.want { + t.Errorf("formatLibVersion(%d) = %s, want %s", tc.in, got, tc.want) + } + }) + } +} + +func TestPoolStateName(t *testing.T) { + t.Parallel() + tests := []struct { + in uint8 + want string + }{ + {uint8(libvirt.StoragePoolInactive), "inactive"}, + {uint8(libvirt.StoragePoolBuilding), "building"}, + {uint8(libvirt.StoragePoolRunning), "running"}, + {uint8(libvirt.StoragePoolDegraded), "degraded"}, + {uint8(libvirt.StoragePoolInaccessible), "inaccessible"}, + {42, "state 42"}, + } + for _, tc := range tests { + t.Run(tc.want, func(t *testing.T) { + t.Parallel() + if got := poolStateName(tc.in); got != tc.want { + t.Errorf("poolStateName(%d) = %s, want %s", tc.in, got, tc.want) + } + }) + } +} diff --git a/internal/provider/kvm2/kvm2_test.go b/internal/provider/kvm2/kvm2_test.go index f663db6..5b31548 100644 --- a/internal/provider/kvm2/kvm2_test.go +++ b/internal/provider/kvm2/kvm2_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "strings" + "sync" "testing" "time" @@ -13,9 +14,14 @@ import ( "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" ) +// captureContext collects funcr output at V(2). The sink is locked because +// Check logs from one goroutine per host. func captureContext() (context.Context, *[]string) { lines := &[]string{} + var mu sync.Mutex log := funcr.New(func(prefix, args string) { + mu.Lock() + defer mu.Unlock() *lines = append(*lines, prefix+" "+args) }, funcr.Options{Verbosity: 2}) return logf.IntoContext(context.Background(), log), lines