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

//go:build integration

package kvm2

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

	"github.com/digitalocean/go-libvirt"
	"github.com/go-logr/logr"
	"github.com/go-logr/logr/funcr"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	"go.opentelemetry.io/otel/sdk/trace/tracetest"
	"k8s.io/apimachinery/pkg/types"
	logf "sigs.k8s.io/controller-runtime/pkg/log"

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

// The real-hypervisor test of spec entry 21. It runs only with
// -tags=integration and skips itself unless KVM2_TEST_* names a host; see
// docs/testing.md for the variables and what one run asserts.
const (
	integrationProvider = "kvm2-integration"
	integrationHost     = "kvm2-test"
	integrationImage    = "test-image"
	integrationMachine  = "small"
	integrationPort     = 3128

	// integrationCloudInit is the user-data the VM boots with. Nothing
	// asserts on the guest running it; it exists so the sentinel check can
	// prove that no span attribute carries the cloud-init text.
	integrationCloudInit = "#cloud-config\nruncmd:\n  - echo kvm2-integration\n"

	defaultIntegrationTimeout = 20 * time.Minute
	defaultIntegrationLogLvl  = 1
	getPollInterval           = 5 * time.Second
	cleanupBudget             = 5 * time.Minute
)

type integrationEnv struct {
	host, user, sshKey, knownHosts, bridge, imageURL string
	imageSHA256, pool, socket                        string
	timeout                                          time.Duration
	keep                                             bool
	logLevel                                         int
}

// loadIntegrationEnv reads the KVM2_TEST_* variables, skipping the test
// under -short or when a required one is missing.
func loadIntegrationEnv(t *testing.T) integrationEnv {
	t.Helper()
	if testing.Short() {
		t.Skip("real-hypervisor test skipped under -short")
	}
	env := integrationEnv{timeout: defaultIntegrationTimeout, logLevel: defaultIntegrationLogLvl}
	required := []struct {
		name string
		dst  *string
	}{
		{"KVM2_TEST_HOST", &env.host},
		{"KVM2_TEST_USER", &env.user},
		{"KVM2_TEST_SSH_KEY", &env.sshKey},
		{"KVM2_TEST_BRIDGE", &env.bridge},
		{"KVM2_TEST_IMAGE_URL", &env.imageURL},
	}
	for _, r := range required {
		v := os.Getenv(r.name)
		if v == "" {
			t.Skipf("%s is not set; the kvm2 integration test needs a real hypervisor (see docs/testing.md)", r.name)
		}
		*r.dst = v
	}
	env.knownHosts = os.Getenv("KVM2_TEST_KNOWN_HOSTS")
	env.imageSHA256 = os.Getenv("KVM2_TEST_IMAGE_SHA256")
	env.pool = os.Getenv("KVM2_TEST_POOL")
	env.socket = os.Getenv("KVM2_TEST_SOCKET")
	env.keep = os.Getenv("KVM2_TEST_KEEP") != ""
	if v := os.Getenv("KVM2_TEST_TIMEOUT"); v != "" {
		d, err := time.ParseDuration(v)
		if err != nil {
			t.Fatalf("KVM2_TEST_TIMEOUT=%q: %v", v, err)
		}
		env.timeout = d
	}
	if v := os.Getenv("KVM2_TEST_LOG_LEVEL"); v != "" {
		n, err := strconv.Atoi(v)
		if err != nil || n < 0 {
			t.Fatalf("KVM2_TEST_LOG_LEVEL=%q: want a non-negative integer", v)
		}
		env.logLevel = n
	}
	return env
}

// providerConfig is what the operator would load for this host: one host,
// one image, one machine type. Host keys are verified only when
// KVM2_TEST_KNOWN_HOSTS is set.
func (e integrationEnv) providerConfig() provider.ProviderConfig {
	return provider.ProviderConfig{
		Name: integrationProvider,
		Type: "kvm2",
		KVM2: &provider.KVMConfig{
			SSHKeyFile:            e.sshKey,
			KnownHostsFile:        e.knownHosts,
			InsecureIgnoreHostKey: e.knownHosts == "",
			Images: map[string]provider.KVMImage{
				integrationImage: {URL: e.imageURL, SHA256: e.imageSHA256},
			},
			MachineTypes: map[string]provider.KVMMachineType{
				integrationMachine: {VCPUs: 1, MemoryMiB: 1024},
			},
			Hosts: []provider.KVMHost{{
				Name:    integrationHost,
				Address: e.host,
				User:    e.user,
				Socket:  e.socket,
				Pool:    e.pool,
				Bridge:  e.bridge,
			}},
		},
	}
}

// integrationLog routes the provider's logr output into t.Log through funcr
// at the chosen verbosity, keeps every line for assertions, and strips the
// key/value pairs that are constant for the run (stated once up front) so
// each line stays short. Lines arriving after the test ended are dropped:
// t.Log panics then.
type integrationLog struct {
	t *testing.T

	mu    sync.Mutex
	drop  []string
	lines []string
	done  bool
}

func newIntegrationLog(t *testing.T, level int, drop ...string) (*integrationLog, logr.Logger) {
	t.Helper()
	sink := &integrationLog{t: t, drop: drop}
	t.Cleanup(func() {
		sink.mu.Lock()
		defer sink.mu.Unlock()
		sink.done = true
	})
	log := funcr.New(func(prefix, args string) { sink.write(prefix, args) }, funcr.Options{Verbosity: level})
	return sink, log
}

// dropValue adds a rendered pair (`"key"="value"`) to strip from later lines.
func (l *integrationLog) dropValue(key, value string) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.drop = append(l.drop, fmt.Sprintf("%q=%q", key, value))
}

func (l *integrationLog) write(prefix, args string) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.lines = append(l.lines, args)
	if l.done {
		return
	}
	for _, d := range l.drop {
		args = strings.ReplaceAll(args, " "+d, "")
	}
	l.t.Logf("%s %s %s", time.Now().Format("15:04:05.000"), prefix, args)
}

func (l *integrationLog) matching(substr string) []string {
	l.mu.Lock()
	defer l.mu.Unlock()
	var out []string
	for _, line := range l.lines {
		if strings.Contains(line, substr) {
			out = append(out, line)
		}
	}
	return out
}

// newIntegrationProvider builds the provider the way New does, plus the span
// recorder's tracer on both places the package starts spans: the Provider
// (kvm2.download) and the dialer (kvm2.dial and, through observe, every
// kvm2.<Method>). newLibvirtDialer has no option for it, so the field is set
// directly; that is what makes the kvm2.* spans land in the same recorder
// as provider.WithTracing's provider.* spans.
func newIntegrationProvider(pc provider.ProviderConfig, tp *sdktrace.TracerProvider) *Provider {
	cfg := withDefaults(*pc.KVM2)
	opt := tracing.WithTracerProvider(tp)
	d := newLibvirtDialer(pc.Name, cfg)
	d.tracing = []tracing.Option{opt}
	return newProvider(pc.Name, cfg, d, withTracing(opt))
}

// randomUID is a UUID-shaped string, so the run's domain never collides with
// a previous run's leftovers or with a real Proxy's.
func randomUID(t *testing.T) string {
	t.Helper()
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		t.Fatalf("reading random bytes: %v", err)
	}
	h := hex.EncodeToString(b[:])
	return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32]
}

// stepDone logs one lifecycle step with its duration; the log is the record
// of how long a cold image download or a boot took on this host.
func stepDone(t *testing.T, step string, start time.Time) {
	t.Helper()
	t.Logf("step %s: %s", step, time.Since(start).Round(time.Millisecond))
}

// waitUntil polls cond every interval until it is true or ctx ends.
func waitUntil(ctx context.Context, t *testing.T, interval time.Duration, what string, cond func() bool) {
	t.Helper()
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for !cond() {
		select {
		case <-ticker.C:
		case <-ctx.Done():
			t.Fatalf("gave up waiting for %s: %v", what, ctx.Err())
		}
	}
}

// domainCount lists every domain on the host over a dedicated connection.
func domainCount(ctx context.Context, t *testing.T, p *Provider, host provider.KVMHost) int {
	t.Helper()
	be, err := p.hosts.dialFresh(ctx, host)
	if err != nil {
		t.Fatalf("dialling %s to count domains: %v", host.Name, err)
	}
	defer func() { _ = be.Close() }()
	doms, _, err := be.ConnectListAllDomains(ctx, 1, libvirt.ConnectListDomainsActive|libvirt.ConnectListDomainsInactive)
	if err != nil {
		t.Fatalf("listing domains on %s: %v", host.Name, err)
	}
	return len(doms)
}

// poolVolumes reports, for each name, whether the host's pool holds a volume
// of that name, over a dedicated connection. Any error other than
// ErrNoStorageVol fails the test.
func poolVolumes(ctx context.Context, t *testing.T, p *Provider, host provider.KVMHost, names ...string) map[string]bool {
	t.Helper()
	be, err := p.hosts.dialFresh(ctx, host)
	if err != nil {
		t.Fatalf("dialling %s to inspect the pool: %v", host.Name, err)
	}
	defer func() { _ = be.Close() }()
	pool, err := be.StoragePoolLookupByName(ctx, host.Pool)
	if err != nil {
		t.Fatalf("looking up pool %s: %v", host.Pool, err)
	}
	out := make(map[string]bool, len(names))
	for _, name := range names {
		_, err := be.StorageVolLookupByName(ctx, pool, name)
		switch {
		case err == nil:
			out[name] = true
		case hasCode(err, libvirt.ErrNoStorageVol):
			out[name] = false
		default:
			t.Fatalf("looking up volume %s in pool %s: %v", name, host.Pool, err)
		}
	}
	return out
}

func TestKVM2_lifecycle(t *testing.T) {
	env := loadIntegrationEnv(t)
	pc := env.providerConfig()
	sink, log := newIntegrationLog(t, env.logLevel,
		fmt.Sprintf("%q=%q", "provider", integrationProvider),
		fmt.Sprintf("%q=%q", "host", integrationHost))
	ctx, cancel := context.WithTimeout(logf.IntoContext(context.Background(), log), env.timeout)
	defer cancel()

	t.Logf("provider=%s host=%s address=%s user=%s bridge=%s image=%s budget=%s log level %d; provider, host and name are left off every provider log line",
		integrationProvider, integrationHost, env.host, env.user, env.bridge, env.imageURL, env.timeout, env.logLevel)
	if env.knownHosts == "" {
		t.Log("KVM2_TEST_KNOWN_HOSTS is not set: host keys are NOT verified for this run")
	}

	// 1. New returns without dialling.
	start := time.Now()
	built, err := New(ctx, pc)
	elapsed := time.Since(start)
	if err != nil {
		t.Fatalf("New() error = %v", err)
	}
	if elapsed > 100*time.Millisecond {
		t.Errorf("New() took %s, want under 100ms: it must not dial", elapsed)
	}
	if lines := sink.matching("libvirt host connected"); len(lines) > 0 {
		t.Errorf("New() dialled the host before any call: %q", lines)
	}
	stepDone(t, "1 New", start)

	// 2. Check reports the host Ready with details.
	checker, ok := built.(provider.Checker)
	if !ok {
		t.Fatalf("New() returned %T, which is not a provider.Checker", built)
	}
	start = time.Now()
	targets := checker.Check(ctx)
	if len(targets) != 1 {
		t.Fatalf("Check() returned %d targets, want 1: %+v", len(targets), targets)
	}
	ts := targets[0]
	if !ts.Ready || ts.Reason != provider.ReasonReachable {
		t.Fatalf("Check() = ready %v reason %q message %q, want Ready/%s (details %v)", ts.Ready, ts.Reason, ts.Message, provider.ReasonReachable, ts.Details)
	}
	for _, key := range []string{"hostname", "libvirtVersion", "poolState", "poolAvailableGiB"} {
		if ts.Details[key] == "" {
			t.Errorf("Check() details lack %q: %v", key, ts.Details)
		}
	}
	if ts.Details["poolState"] != "running" {
		t.Errorf("Check() poolState = %q, want running", ts.Details["poolState"])
	}
	t.Logf("host %s: hostname=%s libvirt=%s pool=%s (%s) available=%s GiB capacity=%s GiB latency=%s",
		ts.Name, ts.Details["hostname"], ts.Details["libvirtVersion"], ts.Details["pool"], ts.Details["poolState"],
		ts.Details["poolAvailableGiB"], ts.Details["poolCapacityGiB"], ts.Latency.Round(time.Millisecond))
	stepDone(t, "2 Check", start)

	// 3. Every later call goes through the tracing decorator, with the
	// package's own spans recorded alongside.
	sr := tracetest.NewSpanRecorder()
	tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
	t.Cleanup(func() { _ = tp.Shutdown(context.Background()) })
	p := newIntegrationProvider(pc, tp)
	traced := provider.WithTracing(pc.Name, p, tracing.WithTracerProvider(tp))
	host := p.cfg.Hosts[0]

	// 4. Create.
	uid := randomUID(t)
	name := provider.NameFromUID(types.UID(uid))
	sink.dropValue("name", name)
	id := formatProviderID(integrationHost, name)
	req := provider.CreateRequest{
		Name:      name,
		UID:       uid,
		Namespace: "integration",
		ProxyName: "kvm2-lifecycle",
		Placement: provider.Placement{Zone: integrationHost, MachineType: integrationMachine, Image: integrationImage},
		CloudInit: integrationCloudInit,
		Port:      integrationPort,
	}
	t.Logf("domain name %s, uid %s, id %s", name, uid, id)

	deleted := false
	if env.keep {
		t.Logf("KVM2_TEST_KEEP is set: the VM stays behind. Clean up on the host with:\n"+
			"  virsh undefine --nvram %s\n  virsh vol-delete --pool %s %s\n  virsh vol-delete --pool %s %s",
			name, host.Pool, overlayVolumeName(name), host.Pool, seedVolumeName(name))
	} else {
		t.Cleanup(func() {
			if deleted {
				return
			}
			cctx, ccancel := context.WithTimeout(logf.IntoContext(context.Background(), log), cleanupBudget)
			defer ccancel()
			if err := p.Delete(cctx, id); err != nil {
				t.Errorf("cleanup Delete(%s) error = %v; remove the domain and its volumes by hand", id, err)
			}
		})
	}

	start = time.Now()
	gotID, err := traced.Create(ctx, req)
	if err != nil {
		t.Fatalf("Create() error = %v", err)
	}
	if gotID != id {
		t.Fatalf("Create() = %q, want %q", gotID, id)
	}
	createReturned := time.Now()
	stepDone(t, "4 Create (includes the image download on a cold pool)", start)

	// 5. Create again: same ID, nothing new on the host.
	start = time.Now()
	before := domainCount(ctx, t, p, host)
	again, err := traced.Create(ctx, req)
	if err != nil {
		t.Fatalf("second Create() error = %v", err)
	}
	if again != id {
		t.Errorf("second Create() = %q, want %q", again, id)
	}
	if after := domainCount(ctx, t, p, host); after != before {
		t.Errorf("second Create() changed the host's domain count from %d to %d", before, after)
	}
	stepDone(t, "5 Create again", start)

	// 6. Get reaches Running with an IPv4.
	start = time.Now()
	var inst *provider.Instance
	waitUntil(ctx, t, getPollInterval, "the domain to be Running with an IPv4", func() bool {
		var err error
		inst, err = traced.Get(ctx, id)
		if err != nil {
			t.Fatalf("Get(%s) error = %v", id, err)
		}
		return inst.State == provider.StateRunning && inst.IP != ""
	})
	if inst.UID != uid {
		t.Errorf("Get() UID = %q, want %q", inst.UID, uid)
	}
	if inst.CreatedAt.IsZero() {
		t.Error("Get() CreatedAt is zero; the metadata mark did not round-trip")
	} else if skew := time.Since(inst.CreatedAt).Abs(); skew > 5*time.Minute {
		t.Errorf("Get() CreatedAt = %s, %s from now; want within a few minutes", inst.CreatedAt, skew)
	}
	t.Logf("domain running at %s, %s after Create returned (createdAt %s)", inst.IP, time.Since(createReturned).Round(time.Millisecond), inst.CreatedAt.Format(time.RFC3339))
	stepDone(t, "6 Get until Running", start)

	// 7. ListByTag contains it.
	start = time.Now()
	list, err := traced.ListByTag(ctx)
	if err != nil {
		t.Fatalf("ListByTag() error = %v", err)
	}
	found := false
	for _, li := range list {
		if li.ID == id {
			found = true
			if li.UID != uid {
				t.Errorf("ListByTag() entry %s has UID %q, want %q", id, li.UID, uid)
			}
		}
	}
	if !found {
		t.Errorf("ListByTag() = %d instances, none with ID %s", len(list), id)
	}
	stepDone(t, "7 ListByTag", start)

	// 8. Reachability, logged only: this machine may not route to the bridge.
	start = time.Now()
	addr := net.JoinHostPort(inst.IP, "22")
	if conn, err := net.DialTimeout("tcp", addr, 3*time.Second); err != nil {
		t.Logf("TCP connect to %s failed (informational; the test host may not route to the bridge): %v", addr, err)
	} else {
		_ = conn.Close()
		t.Logf("TCP connect to %s succeeded", addr)
	}
	stepDone(t, "8 TCP connect", start)

	if env.keep {
		t.Log("KVM2_TEST_KEEP is set: skipping Delete and the pool inspection")
	} else {
		// 9. Delete, Get is NotFound, Delete again.
		start = time.Now()
		if err := traced.Delete(ctx, id); err != nil {
			t.Fatalf("Delete() error = %v", err)
		}
		deleted = true
		if _, err := traced.Get(ctx, id); !errors.Is(err, provider.ErrNotFound) {
			t.Errorf("Get() after Delete error = %v, want ErrNotFound", err)
		}
		if err := traced.Delete(ctx, id); err != nil {
			t.Errorf("second Delete() error = %v, want nil", err)
		}
		stepDone(t, "9 Delete, Get, Delete", start)

		// 10. The pool keeps the shared base image and its marker, and holds
		// no per-VM volume.
		start = time.Now()
		base := baseVolumeName(integrationImage, provider.KVMFormatQcow2)
		marker := markerVolumeName(base)
		overlay, seed := overlayVolumeName(name), seedVolumeName(name)
		present := poolVolumes(ctx, t, p, host, base, marker, overlay, seed)
		for _, want := range []string{base, marker} {
			if !present[want] {
				t.Errorf("pool %s lacks %s after Delete; the base image cache must stay", host.Pool, want)
			}
		}
		for _, gone := range []string{overlay, seed} {
			if present[gone] {
				t.Errorf("pool %s still holds %s after Delete", host.Pool, gone)
			}
		}
		stepDone(t, "10 pool inspection", start)
	}

	// 11. Spans: the package's spans are children of the decorator's.
	assertLifecycleSpans(t, sr.Ended())
}

// assertLifecycleSpans checks that at least one provider.create span has a
// kvm2.DomainDefineXML and a kvm2.dial child (the first Create dials the
// pooled connection and defines the domain), and that no attribute of any
// span carries the cloud-init text.
func assertLifecycleSpans(t *testing.T, spans []sdktrace.ReadOnlySpan) {
	t.Helper()
	var creates []sdktrace.ReadOnlySpan
	for _, s := range spans {
		if s.Name() == "provider.create" {
			creates = append(creates, s)
		}
	}
	if len(creates) == 0 {
		t.Fatalf("no provider.create span among %d spans: %v", len(spans), spanNames(spans))
	}
	satisfied := false
	for _, create := range creates {
		parent := create.SpanContext().SpanID()
		var children, define, dial int
		for _, s := range spans {
			if s.Parent().SpanID() != parent || !strings.HasPrefix(s.Name(), "kvm2.") {
				continue
			}
			children++
			switch s.Name() {
			case "kvm2.DomainDefineXML":
				define++
			case "kvm2.dial":
				dial++
			}
		}
		t.Logf("provider.create span %s: %d kvm2.* children (%d DomainDefineXML, %d dial)", parent, children, define, dial)
		if define > 0 && dial > 0 {
			satisfied = true
		}
	}
	if !satisfied {
		t.Errorf("no provider.create span has both a kvm2.DomainDefineXML and a kvm2.dial child; spans: %v", spanNames(spans))
	}
	for _, s := range spans {
		for _, a := range s.Attributes() {
			v := a.Value.Emit()
			if strings.Contains(v, integrationCloudInit) || strings.Contains(v, "#cloud-config") {
				t.Errorf("span %s attribute %s carries the cloud-init text", s.Name(), a.Key)
			}
		}
	}
}

// TestKVM2Check_realHost runs the health check against the host and creates
// nothing.
func TestKVM2Check_realHost(t *testing.T) {
	env := loadIntegrationEnv(t)
	pc := env.providerConfig()
	_, log := newIntegrationLog(t, env.logLevel,
		fmt.Sprintf("%q=%q", "provider", integrationProvider),
		fmt.Sprintf("%q=%q", "host", integrationHost))
	ctx, cancel := context.WithTimeout(logf.IntoContext(context.Background(), log), env.timeout)
	defer cancel()
	if env.knownHosts == "" {
		t.Log("KVM2_TEST_KNOWN_HOSTS is not set: host keys are NOT verified for this run")
	}

	built, err := New(ctx, pc)
	if err != nil {
		t.Fatalf("New() error = %v", err)
	}
	checker, ok := built.(provider.Checker)
	if !ok {
		t.Fatalf("New() returned %T, which is not a provider.Checker", built)
	}
	targets := checker.Check(ctx)
	if len(targets) != 1 {
		t.Fatalf("Check() returned %d targets, want 1: %+v", len(targets), targets)
	}
	ts := targets[0]
	if !ts.Ready || ts.Reason != provider.ReasonReachable {
		t.Fatalf("Check() = ready %v reason %q message %q, want Ready/%s (details %v)", ts.Ready, ts.Reason, ts.Message, provider.ReasonReachable, ts.Details)
	}
	if ts.Details["poolState"] != "running" {
		t.Errorf("Check() poolState = %q, want running", ts.Details["poolState"])
	}
	t.Logf("host %s (%s): hostname=%s libvirt=%s pool=%s (%s) available=%s GiB latency=%s",
		ts.Name, ts.Address, ts.Details["hostname"], ts.Details["libvirtVersion"], ts.Details["pool"], ts.Details["poolState"],
		ts.Details["poolAvailableGiB"], ts.Latency.Round(time.Millisecond))
}