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

//go:build integration

package kvm

// Integration test against a real libvirt/KVM hypervisor. It deviates from
// the repo's testcontainers convention on purpose: libvirt needs /dev/kvm
// and a bridge, neither of which a container provides, so this runs
// against whatever host the KVM_TEST_* variables name. It creates one VM
// from a stock cloud image, waits for it to report an address, deletes it,
// and checks the pool afterwards. The base image stays cached on the host.
//
// Required:
//
//	KVM_TEST_HOST       SSH endpoint, host:port
//	KVM_TEST_USER       SSH user (member of the libvirt group)
//	KVM_TEST_SSH_KEY    path to the private key
//	KVM_TEST_BRIDGE     host bridge for the VM's NIC
//	KVM_TEST_IMAGE_URL  cloud image to fetch (http/https)
//
// Optional:
//
//	KVM_TEST_KNOWN_HOSTS   known_hosts file; without it host keys are NOT verified
//	KVM_TEST_IMAGE_SHA256  expected digest of the image
//	KVM_TEST_POOL          storage pool (default "default")
//	KVM_TEST_SOCKET        libvirt socket path on the host
//	KVM_TEST_TIMEOUT       overall budget incl. the image download (default 20m)
//	KVM_TEST_KEEP          set to anything to leave the VM behind for inspection
//	KVM_TEST_LOG_LEVEL     provider log verbosity shown in -v output (default 1:
//	                       one line per libvirt call; 2 adds XML, metadata and
//	                       addresses; 0 leaves only the t.Log lines)
//
// Provider logs are printed through t.Log in a compact form: time, logger
// name, then the libvirt API function (or the message) and its fields.
// The provider/host/name values that are the same on every line of a run
// are dropped; the first t.Log line of the run states them. A run that
// hits its timeout may log one more line from the RPC the pool abandoned,
// after the test has ended; that is a symptom of the timeout, not a bug
// in the tracer.

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"net"
	"os"
	"strconv"
	"strings"
	"testing"
	"time"

	libvirt "github.com/digitalocean/go-libvirt"
	"github.com/go-logr/logr"
	"k8s.io/apimachinery/pkg/types"
	logf "sigs.k8s.io/controller-runtime/pkg/log"

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

const (
	integrationHostName = "kvm-test"
	integrationImageKey = "test-image"
	integrationPoll     = 5 * time.Second
)

type integrationEnv struct {
	cfg      provider.ProviderConfig
	timeout  time.Duration
	keep     bool
	logLevel int
}

func loadIntegrationEnv(t *testing.T) integrationEnv {
	t.Helper()
	if testing.Short() {
		t.Skip("integration test skipped under -short")
	}
	required := map[string]string{}
	for _, key := range []string{"KVM_TEST_HOST", "KVM_TEST_USER", "KVM_TEST_SSH_KEY", "KVM_TEST_BRIDGE", "KVM_TEST_IMAGE_URL"} {
		v := os.Getenv(key)
		if v == "" {
			t.Skipf("%s not set; skipping KVM integration test", key)
		}
		required[key] = v
	}
	cfg := &provider.KVMConfig{
		SSHKeyFile:     required["KVM_TEST_SSH_KEY"],
		KnownHostsFile: os.Getenv("KVM_TEST_KNOWN_HOSTS"),
		Images: map[string]provider.KVMImage{
			integrationImageKey: {URL: required["KVM_TEST_IMAGE_URL"], SHA256: os.Getenv("KVM_TEST_IMAGE_SHA256")},
		},
		MachineTypes: map[string]provider.KVMMachineType{"small": {VCPUs: 1, MemoryMiB: 1024}},
		Hosts: []provider.KVMHost{{
			Name:    integrationHostName,
			Address: required["KVM_TEST_HOST"],
			User:    required["KVM_TEST_USER"],
			Socket:  os.Getenv("KVM_TEST_SOCKET"),
			Pool:    os.Getenv("KVM_TEST_POOL"),
			Bridge:  required["KVM_TEST_BRIDGE"],
		}},
	}
	if cfg.KnownHostsFile == "" {
		cfg.InsecureIgnoreHostKey = true
		t.Log("KVM_TEST_KNOWN_HOSTS not set: host key verification disabled for this run")
	}
	pc := provider.ProviderConfig{Name: "kvm-integration", Type: "kvm", KVM: cfg}
	timeout := 20 * time.Minute
	if v := os.Getenv("KVM_TEST_TIMEOUT"); v != "" {
		d, err := time.ParseDuration(v)
		if err != nil {
			t.Fatalf("KVM_TEST_TIMEOUT: %v", err)
		}
		timeout = d
	}
	logLevel := 1
	if v := os.Getenv("KVM_TEST_LOG_LEVEL"); v != "" {
		n, err := strconv.Atoi(v)
		if err != nil {
			t.Fatalf("KVM_TEST_LOG_LEVEL: %v", err)
		}
		logLevel = n
	}
	return integrationEnv{cfg: pc, timeout: timeout, keep: os.Getenv("KVM_TEST_KEEP") != "", logLevel: logLevel}
}

// integrationLogger routes provider logs into the test output at the
// requested verbosity, and installs it as controller-runtime's root logger
// so nothing falls back to the unfulfilled delegating sink (which would
// print its "SetLogger was never called" warning and swallow the lines).
func integrationLogger(t *testing.T, verbosity int) logr.Logger {
	t.Helper()
	log := logr.New(&compactSink{t: t, verbosity: verbosity})
	logf.SetLogger(log)
	return log
}

// compactSink is a logr sink for test output: one line per record as
// "HH:MM:SS.mmm name headline k=v ...", where the headline is the rpc
// field when present (the libvirt API function) and the message
// otherwise. Keys that are constant for the whole run are omitted.
type compactSink struct {
	t         *testing.T
	verbosity int
	name      string
	values    []any
}

var compactOmit = map[string]bool{"provider": true, "host": true, "name": true}

func (s *compactSink) Init(logr.RuntimeInfo) {}

func (s *compactSink) Enabled(level int) bool { return level <= s.verbosity }

// GetCallStackHelper lets logr mark its frames as helpers, so t.Log
// attributes each line to the provider code that logged it.
func (s *compactSink) GetCallStackHelper() func() { return s.t.Helper }

func (s *compactSink) Info(_ int, msg string, kv ...any) {
	s.t.Helper()
	s.t.Log(s.format(msg, kv))
}

func (s *compactSink) Error(err error, msg string, kv ...any) {
	s.t.Helper()
	s.t.Log(s.format(msg, append(kv, "error", err)))
}

func (s *compactSink) WithValues(kv ...any) logr.LogSink {
	c := *s
	c.values = append(append([]any(nil), s.values...), kv...)
	return &c
}

func (s *compactSink) WithName(name string) logr.LogSink {
	c := *s
	if c.name != "" {
		name = c.name + "/" + name
	}
	c.name = name
	return &c
}

func (s *compactSink) format(msg string, kv []any) string {
	all := append(append([]any(nil), s.values...), kv...)
	headline := msg
	var b strings.Builder
	for i := 0; i+1 < len(all); i += 2 {
		key := fmt.Sprint(all[i])
		if key == "rpc" {
			headline = fmt.Sprint(all[i+1])
			continue
		}
		if compactOmit[key] {
			continue
		}
		val := fmt.Sprint(all[i+1])
		if strings.ContainsAny(val, " \t\"=") {
			val = strconv.Quote(val)
		}
		b.WriteString(" " + key + "=" + val)
	}
	name := s.name[strings.LastIndex(s.name, "/")+1:]
	return fmt.Sprintf("%s %s %s%s", time.Now().Format("15:04:05.000"), name, headline, b.String())
}

func randomUID(t *testing.T) string {
	t.Helper()
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		t.Fatal(err)
	}
	h := hex.EncodeToString(b[:])
	return fmt.Sprintf("%s-%s-%s-%s-%s", h[0:8], h[8:12], h[12:16], h[16:20], h[20:32])
}

// TestKVM_lifecycle is one sequential story, not parallel subtests: each
// stage depends on the state the previous one left on the hypervisor.
func TestKVM_lifecycle(t *testing.T) {
	env := loadIntegrationEnv(t)
	log := integrationLogger(t, env.logLevel)
	ctx, cancel := context.WithTimeout(logr.NewContext(context.Background(), log), env.timeout)
	defer cancel()

	prov, err := New(ctx, env.cfg)
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	p := prov.(*Provider)
	host := p.cfg.Hosts[0]

	uid := randomUID(t)
	req := provider.CreateRequest{
		Name:      provider.NameFromUID(types.UID(uid)),
		UID:       uid,
		Namespace: "integration",
		ProxyName: "kvm-integration",
		Placement: provider.Placement{Zone: integrationHostName, MachineType: "small", Image: integrationImageKey},
		CloudInit: "#cloud-config\n",
		Port:      3128,
	}
	t.Logf("host %s (%s), pool %q, bridge %s, domain %s", host.Name, host.Address, host.Pool, host.Bridge, req.Name)

	// Cleanup runs even when a stage fails, with its own budget: the test
	// ctx may already be spent by then.
	t.Cleanup(func() {
		if env.keep {
			t.Logf("KVM_TEST_KEEP set: leaving domain %s and its volumes on %s", req.Name, host.Address)
			return
		}
		cctx, ccancel := context.WithTimeout(logr.NewContext(context.Background(), log), 2*time.Minute)
		defer ccancel()
		if err := p.Delete(cctx, formatProviderID(host.Name, req.Name)); err != nil {
			t.Errorf("cleanup Delete: %v", err)
		}
	})

	// --- create (includes the image fetch on a cold host) ---
	start := time.Now()
	id, err := p.Create(ctx, req)
	if err != nil {
		t.Fatalf("Create: %v", err)
	}
	t.Logf("Create returned %s after %s", id, time.Since(start).Round(time.Second))
	if want := formatProviderID(host.Name, req.Name); id != want {
		t.Fatalf("providerID = %s, want %s", id, want)
	}

	// --- create again: must find what it made, not duplicate it ---
	id2, err := p.Create(ctx, req)
	if err != nil {
		t.Fatalf("second Create: %v", err)
	}
	if id2 != id {
		t.Fatalf("second Create returned %s, want %s", id2, id)
	}

	// --- wait for the guest to come up with an address ---
	inst := waitForRunning(ctx, t, p, id)
	t.Logf("domain Running with IP %s (state reported after %s)", inst.IP, time.Since(start).Round(time.Second))
	if inst.UID != uid {
		t.Errorf("UID read back = %q, want %q (metadata round-trip through libvirt)", inst.UID, uid)
	}
	if age := time.Since(inst.CreatedAt); inst.CreatedAt.IsZero() || age < 0 || age > time.Hour {
		t.Errorf("CreatedAt = %s (age %s), want a recent timestamp", inst.CreatedAt, age)
	}
	if conn, err := net.DialTimeout("tcp", net.JoinHostPort(inst.IP, "22"), 5*time.Second); err != nil {
		t.Logf("note: tcp %s:22 not reachable from this machine (%v); fine if the bridge network is not routed here", inst.IP, err)
	} else {
		conn.Close()
		t.Logf("tcp %s:22 reachable from this machine", inst.IP)
	}

	// --- the GC view must include it ---
	listed, err := p.ListByTag(ctx)
	if err != nil {
		t.Fatalf("ListByTag: %v", err)
	}
	if !containsID(listed, id) {
		t.Errorf("ListByTag = %+v, want it to include %s", listed, id)
	}

	// --- delete, then verify nothing of the VM remains but the base image does ---
	if err := p.Delete(ctx, id); err != nil {
		t.Fatalf("Delete: %v", err)
	}
	if _, err := p.Get(ctx, id); !errors.Is(err, provider.ErrNotFound) {
		t.Errorf("Get after Delete = %v, want ErrNotFound", err)
	}
	if err := p.Delete(ctx, id); err != nil {
		t.Errorf("second Delete: %v, want nil (idempotent)", err)
	}
	listed, err = p.ListByTag(ctx)
	if err != nil {
		t.Fatalf("ListByTag after Delete: %v", err)
	}
	if containsID(listed, id) {
		t.Errorf("ListByTag still includes %s after Delete", id)
	}

	src, err := resolveImage(p.cfg, integrationImageKey)
	if err != nil {
		t.Fatal(err)
	}
	err = p.hosts.call(ctx, p.logger(ctx).WithValues("host", host.Name), host, func(api libvirtAPI) error {
		pool, err := api.StoragePoolLookupByName(host.Pool)
		if err != nil {
			return err
		}
		for _, gone := range []string{overlayVolumeName(req.Name), seedVolumeName(req.Name)} {
			if _, err := api.StorageVolLookupByName(pool, gone); !hasCode(err, libvirt.ErrNoStorageVol) {
				t.Errorf("volume %s after Delete: lookup err = %v, want ErrNoStorageVol", gone, err)
			}
		}
		for _, kept := range []string{src.volumeName(), src.markerName()} {
			if _, err := api.StorageVolLookupByName(pool, kept); err != nil {
				t.Errorf("cached image volume %s: %v, want it kept after Delete", kept, err)
			}
		}
		return nil
	})
	if err != nil {
		t.Fatalf("inspecting pool after Delete: %v", err)
	}
	// KVM_TEST_KEEP is for inspecting a failed run; a passed run has
	// nothing left to keep, so let cleanup's Delete confirm that.
	env.keep = false
}

// 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 {
	t.Helper()
	var last provider.InstanceState
	ticker := time.NewTicker(integrationPoll)
	defer ticker.Stop()
	for {
		inst, err := p.Get(ctx, id)
		switch {
		case err == nil:
			if inst.State != last {
				t.Logf("state %s ip %q", inst.State, inst.IP)
				last = inst.State
			}
			if inst.State == provider.StateRunning {
				return inst
			}
			if inst.State == provider.StateStopped {
				t.Fatalf("domain reported %s while waiting for Running; check `virsh console %s` on the host", inst.State, idName(id))
			}
		case errors.Is(err, provider.ErrTransient):
			t.Logf("Get: transient error, retrying: %v", err)
		default:
			t.Fatalf("Get: %v", err)
		}
		select {
		case <-ctx.Done():
			t.Fatalf("timed out waiting for Running (last state %s): %v", last, ctx.Err())
		case <-ticker.C:
		}
	}
}

func idName(id string) string {
	_, name, _ := parseProviderID(id)
	return name
}

func containsID(list []provider.Instance, id string) bool {
	for _, inst := range list {
		if inst.ID == id {
			return true
		}
	}
	return false
}