egress-proxies-operator/internal/provider/kvm/check_test.go

package kvm

import (
	"context"
	"errors"
	"fmt"
	"io"
	"maps"
	"strings"
	"sync"
	"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)
		}
	}
}

// A daemon that stopped answering times the check out, but the connection
// the reconciles share stays up, and the next check does not stack a
// second hung RPC on top of the first.
func TestCheck_timeoutKeepsTheConnection(t *testing.T) {
	t.Parallel()
	api := newFakeAPI("default")
	block := make(chan struct{})
	t.Cleanup(func() { close(block) })
	blocked := &blockingAPI{fakeAPI: api, block: block}
	var closed atomic.Bool
	dial := func(context.Context, provider.KVMHost) (*hostClient, error) {
		return &hostClient{api: blocked, close: func() { closed.Store(true) }}, 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])
	}
	if closed.Load() {
		t.Error("the timed-out check closed the connection the reconciles share")
	}
	again := p.Check(context.Background())
	if again[0].Reason != provider.ReasonTimeout || !strings.Contains(again[0].Message, "still waiting") {
		t.Errorf("second Check() = %+v, want an immediate Timeout while the first still waits", again[0])
	}
}

// A reconcile redialing a dead host holds the host's slot for minutes; the
// check must give up at its own deadline rather than wait that out.
func TestCheck_doesNotWaitBehindAReconcileDial(t *testing.T) {
	t.Parallel()
	release := make(chan struct{})
	dialing := make(chan struct{})
	var once sync.Once
	dial := func(context.Context, provider.KVMHost) (*hostClient, error) {
		once.Do(func() { close(dialing) })
		<-release
		return nil, errHostOff
	}
	pc := testConfig("nas1")
	p := newWithDialer(pc.Name, *pc.KVM, dial, nil)
	reconcile := make(chan error, 1)
	go func() {
		_, err := p.Get(context.Background(), formatProviderID("nas1", testName))
		reconcile <- err
	}()
	<-dialing
	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
	defer cancel()
	start := time.Now()

	got := p.Check(ctx)

	if waited := time.Since(start); waited > time.Second {
		t.Errorf("Check() took %s behind the reconcile's dial, want about 50ms", waited)
	}
	if got[0].Reason != provider.ReasonTimeout {
		t.Errorf("Check() = %+v, want Timeout", got[0])
	}
	close(release)
	<-reconcile
}

// 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},
		{"previous check still waiting", errProbeBusy, 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)
		}
	}
}