egress-proxies-operator/internal/provider/kvm2/backend_test.go

package kvm2

import (
	"context"
	"errors"
	"io"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/digitalocean/go-libvirt"
	"github.com/digitalocean/go-libvirt/libvirttest"

	"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)

// blockingFn returns a func that blocks until release is closed, plus a
// counter of how often it was started.
func blockingFn(t *testing.T) (fn func() error, started *atomic.Int32) {
	t.Helper()
	release := make(chan struct{})
	t.Cleanup(func() { close(release) })
	started = &atomic.Int32{}
	return func() error {
		started.Add(1)
		<-release
		return nil
	}, started
}

func TestCall_deadlineClosesConnectionAndReturnsTransient(t *testing.T) {
	t.Parallel()
	var killed atomic.Int32
	l := newLink("beelink", func() { killed.Add(1) })
	fn, started := blockingFn(t)

	start := time.Now()
	err := l.run(context.Background(), "DomainCreate", 50*time.Millisecond, fn)
	elapsed := time.Since(start)
	if !errors.Is(err, provider.ErrTransient) {
		t.Fatalf("run() error = %v, want ErrTransient", err)
	}
	if !errors.Is(err, context.DeadlineExceeded) {
		t.Errorf("run() error = %v, want it to wrap context.DeadlineExceeded", err)
	}
	if !strings.Contains(err.Error(), "DomainCreate") || !strings.Contains(err.Error(), "beelink") {
		t.Errorf("run() error = %q, want the call name and host in it", err)
	}
	if elapsed > time.Second {
		t.Errorf("run() took %v, want about the 50ms budget", elapsed)
	}
	if killed.Load() != 1 {
		t.Errorf("kill hook called %d times, want 1", killed.Load())
	}
	if !l.isDead() {
		t.Error("link not dead after an abandoned call")
	}
	if !strings.Contains(l.deadReason(), "budget") {
		t.Errorf("deadReason() = %q, want the budget mentioned", l.deadReason())
	}

	start = time.Now()
	err = l.run(context.Background(), "DomainGetState", time.Minute, fn)
	if !errors.Is(err, provider.ErrTransient) {
		t.Fatalf("second run() error = %v, want ErrTransient", err)
	}
	if time.Since(start) > 100*time.Millisecond {
		t.Errorf("second run() took %v, want immediate", time.Since(start))
	}
	if started.Load() != 1 {
		t.Errorf("fn started %d times, want 1: a dead link must not touch the socket", started.Load())
	}
	if killed.Load() != 1 {
		t.Errorf("kill hook called %d times, want still 1", killed.Load())
	}
}

func TestCall_cancelUnblocksRPC(t *testing.T) {
	t.Parallel()
	var killed atomic.Int32
	l := newLink("beelink", func() { killed.Add(1) })
	fn, _ := blockingFn(t)
	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		time.Sleep(10 * time.Millisecond)
		cancel()
	}()

	start := time.Now()
	err := l.run(ctx, "ConnectListAllDomains", time.Minute, fn)
	if time.Since(start) > 100*time.Millisecond {
		t.Errorf("run() took %v after cancel, want under 100ms", time.Since(start))
	}
	if !errors.Is(err, provider.ErrTransient) || !errors.Is(err, context.Canceled) {
		t.Errorf("run() error = %v, want ErrTransient wrapping context.Canceled", err)
	}
	if killed.Load() != 1 {
		t.Errorf("kill hook called %d times, want 1", killed.Load())
	}
}

func TestCall_alreadyCancelledDoesNotStart(t *testing.T) {
	t.Parallel()
	l := newLink("beelink", nil)
	fn, started := blockingFn(t)
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	err := l.run(ctx, "DomainGetState", time.Minute, fn)
	if !errors.Is(err, provider.ErrTransient) || !errors.Is(err, context.Canceled) {
		t.Errorf("run() error = %v, want ErrTransient wrapping context.Canceled", err)
	}
	if started.Load() != 0 {
		t.Error("fn started although ctx was already cancelled")
	}
	if l.isDead() {
		t.Error("link marked dead although nothing was abandoned")
	}
}

func TestCall_returnsFnResult(t *testing.T) {
	t.Parallel()
	l := newLink("beelink", nil)
	want := libvirt.Error{Code: uint32(libvirt.ErrNoDomain), Message: "gone"}
	err := l.run(context.Background(), "DomainLookupByName", time.Second, func() error { return want })
	if !hasCode(err, libvirt.ErrNoDomain) {
		t.Errorf("run() error = %v, want the libvirt error through unchanged", err)
	}
	if l.isDead() {
		t.Error("a backend error must not kill the link")
	}
}

func TestCall_killedByAnotherCallerFailsTransient(t *testing.T) {
	t.Parallel()
	l := newLink("beelink", nil)
	fn, _ := blockingFn(t)
	done := make(chan error, 1)
	go func() { done <- l.run(context.Background(), "DomainGetState", time.Minute, fn) }()
	time.Sleep(10 * time.Millisecond)
	l.markDead("keepalive failed")
	select {
	case err := <-done:
		if !errors.Is(err, provider.ErrTransient) || !strings.Contains(err.Error(), "keepalive failed") {
			t.Errorf("run() error = %v, want ErrTransient with the death reason", err)
		}
	case <-time.After(time.Second):
		t.Fatal("run() did not return after the link was killed")
	}
}

// TestMarkDead_slowKillDoesNotBlockCallers pins the guarantee the pooled
// connection rests on: a kill that takes its time — closing a transport can
// block for as long as the kernel retransmits to a host that went away —
// holds up neither the callers waiting on dead nor the next markDead. The
// call's own fn signals when it is in flight, so the parked caller is
// genuinely parked rather than refused by the dead check on the way in.
func TestMarkDead_slowKillDoesNotBlockCallers(t *testing.T) {
	t.Parallel()
	release := make(chan struct{})
	t.Cleanup(func() { close(release) })
	killing := make(chan struct{})
	var killed atomic.Int32
	l := newLink("beelink", func() {
		killed.Add(1)
		close(killing)
		<-release
	})

	inflight := make(chan struct{})
	waiting := make(chan error, 1)
	go func() {
		waiting <- l.run(context.Background(), "StorageVolUpload", time.Minute, func() error {
			close(inflight)
			<-release
			return nil
		})
	}()
	<-inflight

	go l.markDead("keepalive failed")
	<-killing

	select {
	case <-l.dead():
	case <-time.After(time.Second):
		t.Fatal("link not signalled dead while its kill is still running")
	}
	select {
	case err := <-waiting:
		if !errors.Is(err, provider.ErrTransient) || !strings.Contains(err.Error(), "keepalive failed") {
			t.Errorf("parked call error = %v, want ErrTransient with the death reason", err)
		}
	case <-time.After(time.Second):
		t.Fatal("parked call not released while the kill is still running")
	}

	second := make(chan struct{})
	go func() {
		l.markDead("closed")
		close(second)
	}()
	select {
	case <-second:
	case <-time.After(time.Second):
		t.Fatal("second markDead blocked behind the running kill")
	}

	if killed.Load() != 1 {
		t.Errorf("kill hook called %d times, want 1", killed.Load())
	}
	if !strings.Contains(l.deadReason(), "keepalive failed") {
		t.Errorf("deadReason() = %q, want the first reason", l.deadReason())
	}
}

// TestEOFReader_sourceErrorNeverReachesTheStream pins the one property the
// upload seam needs: go-libvirt's stream sender must never see an error
// other than io.EOF, because it answers one by parking forever on a channel
// nothing reads. The real error has to survive for the caller.
func TestEOFReader_sourceErrorNeverReachesTheStream(t *testing.T) {
	t.Parallel()
	boom := errors.New("connection reset by peer")
	tests := []struct {
		name    string
		reader  io.Reader
		wantN   int
		wantErr error
	}{
		{name: "error after data", reader: io.MultiReader(strings.NewReader("abc"), errReader{boom}), wantN: 3, wantErr: boom},
		{name: "error with no data", reader: errReader{boom}, wantErr: boom},
		{name: "clean eof passes through", reader: strings.NewReader("abc"), wantN: 3},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			e := &eofReader{r: tc.reader}
			n, err := io.Copy(io.Discard, e)
			if err != nil {
				t.Fatalf("io.Copy through eofReader = %v, want nil: the stream sender must only ever see io.EOF", err)
			}
			if n != int64(tc.wantN) {
				t.Errorf("copied %d bytes, want %d", n, tc.wantN)
			}
			if !errors.Is(e.sourceErr(), tc.wantErr) {
				t.Errorf("sourceErr() = %v, want %v", e.sourceErr(), tc.wantErr)
			}
		})
	}
}

func TestEOFReader_keepsTheFirstError(t *testing.T) {
	t.Parallel()
	first, second := errors.New("first"), errors.New("second")
	e := &eofReader{r: io.MultiReader(errReader{first}, errReader{second})}
	if _, err := io.Copy(io.Discard, e); err != nil {
		t.Fatalf("io.Copy through eofReader = %v, want nil", err)
	}
	if !errors.Is(e.sourceErr(), first) {
		t.Errorf("sourceErr() = %v, want the first error", e.sourceErr())
	}
}

type errReader struct{ err error }

func (r errReader) Read([]byte) (int, error) { return 0, r.err }

func TestBudgets_upload(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name   string
		length uint64
		want   time.Duration
	}{
		{name: "seed sized", length: 400 << 10, want: defaultBudgets.seedUpload},
		{name: "exactly the limit", length: smallUploadBytes, want: defaultBudgets.seedUpload},
		{name: "one over the limit", length: smallUploadBytes + 1, want: defaultBudgets.transfer},
		{name: "base image", length: 435 << 20, want: defaultBudgets.transfer},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			t.Parallel()
			if got := defaultBudgets.upload(tc.length); got != tc.want {
				t.Errorf("upload(%d) = %v, want %v", tc.length, got, tc.want)
			}
		})
	}
}

// mockAdapter wires a libvirtBackend to go-libvirt's own mock server over a
// net.Pipe, so the adapter is exercised with the real RPC client.
func mockAdapter(t *testing.T, b budgets) *libvirtBackend {
	t.Helper()
	m := libvirttest.New()
	conn, err := m.Dial()
	if err != nil {
		t.Fatalf("mock dial: %v", err)
	}
	ad := newLibvirtBackend(libvirt.New(conn), "mock", conn, b)
	if err := ad.connect(context.Background()); err != nil {
		t.Fatalf("connect: %v", err)
	}
	t.Cleanup(func() { _ = ad.Close() })
	return ad
}

func TestLibvirtBackend_roundTripThroughMock(t *testing.T) {
	t.Parallel()
	ad := mockAdapter(t, defaultBudgets)
	v, err := ad.ConnectGetLibVersion(context.Background())
	if err != nil {
		t.Fatalf("ConnectGetLibVersion() error = %v", err)
	}
	if v == 0 {
		t.Error("ConnectGetLibVersion() = 0, want the mock's version")
	}
	pool, err := ad.StoragePoolLookupByName(context.Background(), "default")
	if err != nil {
		t.Fatalf("StoragePoolLookupByName() error = %v", err)
	}
	if pool.Name == "" {
		t.Error("StoragePoolLookupByName() returned an empty pool")
	}
	if isDead(ad) {
		t.Error("adapter dead after successful calls")
	}
}

func TestLibvirtBackend_unansweredCallIsAbandoned(t *testing.T) {
	t.Parallel()
	b := defaultBudgets
	b.rpc = 100 * time.Millisecond
	ad := mockAdapter(t, b)
	// The mock does not implement ConnectGetHostname, so it never replies.
	_, err := ad.ConnectGetHostname(context.Background())
	if !errors.Is(err, provider.ErrTransient) || !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("ConnectGetHostname() error = %v, want ErrTransient wrapping DeadlineExceeded", err)
	}
	if !isDead(ad) {
		t.Error("adapter not dead after an abandoned call")
	}
	select {
	case <-ad.lv.Disconnected():
	case <-time.After(time.Second):
		t.Error("go-libvirt did not notice the closed socket")
	}
	if _, err := ad.ConnectGetLibVersion(context.Background()); !errors.Is(err, provider.ErrTransient) {
		t.Errorf("call on dead adapter error = %v, want ErrTransient", err)
	}
	if err := ad.Close(); err != nil {
		t.Errorf("Close() on dead adapter error = %v, want nil", err)
	}
}

func TestLibvirtBackend_closeIsIdempotent(t *testing.T) {
	t.Parallel()
	ad := mockAdapter(t, defaultBudgets)
	first := ad.Close()
	second := ad.Close()
	if first != nil || second != nil {
		t.Errorf("Close() errors = %v, %v, want nil, nil", first, second)
	}
	if !isDead(ad) {
		t.Error("adapter not dead after Close")
	}
	if _, err := ad.ConnectGetLibVersion(context.Background()); !errors.Is(err, provider.ErrTransient) {
		t.Errorf("call after Close error = %v, want ErrTransient", err)
	}
}