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

// Package kvm2 implements the provider contract on libvirt/KVM hypervisors
// reached over SSH: the host's libvirt unix socket is forwarded through the
// SSH connection and driven with go-libvirt's RPC client. The design is in
// docs/kvm2-provider-spec.md.
package kvm2

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"sync"
	"time"

	"github.com/go-logr/logr"
	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"
)

const (
	defaultSocket     = "/var/run/libvirt/libvirt-sock"
	defaultPool       = "default"
	defaultDiskSizeGB = 10
)

// Provider is one configured kvm2 entry: a set of hosts sharing one SSH key
// and one image and machine-type catalogue.
type Provider struct {
	name    string
	cfg     provider.KVMConfig
	now     func() time.Time
	hosts   *hostTable
	http    *http.Client
	tracing []tracing.Option

	// runningWait caps the wait for a started domain to report running,
	// polled every runningPoll (spec entry 18).
	runningWait time.Duration
	runningPoll time.Duration

	// imageLocks serialises image fetches per (host, volume), spec entry 19.
	imageLocks *keyedLocks

	// nameLocks serialises Create and Delete per instance name, spec entry
	// 19, so two Deletes never race on destroy and undefine and a Delete
	// cannot interleave with a Create's steps.
	nameLocks *keyedLocks
}

// providerOption adjusts what newProvider builds; tests inject an HTTP
// client bound to a test server, a span recorder and a shorter running
// wait.
type providerOption func(*Provider)

func withHTTPClient(c *http.Client) providerOption {
	return func(p *Provider) { p.http = c }
}

func withTracing(opts ...tracing.Option) providerOption {
	return func(p *Provider) { p.tracing = opts }
}

// withRunningWait shrinks the poll interval and the cap of the wait for a
// domain to run.
func withRunningWait(poll, cap time.Duration) providerOption {
	return func(p *Provider) { p.runningPoll, p.runningWait = poll, cap }
}

var _ provider.Provider = (*Provider)(nil)

// New builds a Provider from pc.KVM2. It never dials a host: the credential
// files are only checked for existence, and a missing one is a warning, not
// an error, because the Secret mount is optional and may land later.
func New(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, error) {
	cfg := provider.KVMConfig{}
	if pc.KVM2 != nil {
		cfg = *pc.KVM2
	}
	cfg = withDefaults(cfg)
	p := newProvider(pc.Name, cfg, newLibvirtDialer(pc.Name, cfg))
	log := p.logger(ctx)
	if _, err := os.Stat(p.cfg.SSHKeyFile); err != nil {
		log.Info("WARNING: ssh key file is not readable; every call fails until it is", "path", p.cfg.SSHKeyFile, "reason", err.Error())
	}
	switch {
	case p.cfg.InsecureIgnoreHostKey:
		log.Info("WARNING: insecureIgnoreHostKey is set; host keys are not verified")
	default:
		if _, err := os.Stat(p.cfg.KnownHostsFile); err != nil {
			log.Info("WARNING: known_hosts file is not readable; every call fails until it is", "path", p.cfg.KnownHostsFile, "reason", err.Error())
		}
	}
	return p, nil
}

// newProvider is the constructor tests use: d hands out the backends, the
// production one being newLibvirtDialer. Without options the image
// downloads use http.DefaultClient (system CA roots, no client timeout: the
// per-request ctx carries the transfer budget).
func newProvider(name string, cfg provider.KVMConfig, d dialer, opts ...providerOption) *Provider {
	p := &Provider{
		name:        name,
		cfg:         withDefaults(cfg),
		now:         time.Now,
		hosts:       newHostTable(d),
		http:        http.DefaultClient,
		runningWait: defaultRunningWait,
		runningPoll: defaultRunningPoll,
		imageLocks:  newKeyedLocks(),
		nameLocks:   newKeyedLocks(),
	}
	for _, opt := range opts {
		opt(p)
	}
	if p.http == nil {
		p.http = http.DefaultClient
	}
	return p
}

// withDefaults returns cfg with every optional field filled in, so the rest
// of the package never re-checks for an empty value. The hosts and images
// are copied, not shared with the caller.
func withDefaults(cfg provider.KVMConfig) provider.KVMConfig {
	if cfg.DiskSizeGB == 0 {
		cfg.DiskSizeGB = defaultDiskSizeGB
	}
	hosts := make([]provider.KVMHost, len(cfg.Hosts))
	for i, h := range cfg.Hosts {
		if h.Socket == "" {
			h.Socket = defaultSocket
		}
		if h.Pool == "" {
			h.Pool = defaultPool
		}
		hosts[i] = h
	}
	cfg.Hosts = hosts
	images := make(map[string]provider.KVMImage, len(cfg.Images))
	for k, img := range cfg.Images {
		if img.Format == "" {
			img.Format = provider.KVMFormatQcow2
		}
		if img.Firmware == "" {
			img.Firmware = provider.KVMFirmwareUEFI
		}
		images[k] = img
	}
	cfg.Images = images
	return cfg
}

func (p *Provider) logger(ctx context.Context) logr.Logger {
	return logf.FromContext(ctx).WithName("kvm2").WithValues("provider", p.name)
}

// host resolves a placement.zone or a providerID's host segment to a
// configured host. An empty zone is allowed only when exactly one host is
// configured.
func (p *Provider) host(zone string) (provider.KVMHost, error) {
	if zone == "" {
		if len(p.cfg.Hosts) == 1 {
			return p.cfg.Hosts[0], nil
		}
		return provider.KVMHost{}, permanentf("placement.zone is required when kvm2.hosts lists more than one host")
	}
	for _, h := range p.cfg.Hosts {
		if h.Name == zone {
			return h, nil
		}
	}
	return provider.KVMHost{}, permanentf("host %q is not in kvm2.hosts", zone)
}

// wrapErr is the single exit for errors: it classifies err and wraps it with
// the configured provider name, op and id.
func (p *Provider) wrapErr(op, id string, err error) error {
	return provider.Wrap(classify(op, err), op, p.name, id, err)
}

// lockName takes the per-name lock of spec entry 19, or gives up with ctx,
// so a caller whose ctx ends while another caller's Create is downloading
// an image returns promptly.
func (p *Provider) lockName(ctx context.Context, name string) (func(), error) {
	release, err := p.nameLocks.acquire(ctx, name)
	if err != nil {
		return nil, fmt.Errorf("%w: waiting for the lock on %s: %w", provider.ErrTransient, name, err)
	}
	return release, nil
}

// keyedLocks is a set of one-slot semaphores by key (spec entry 19): a
// semaphore rather than a mutex so a waiter can give up with its ctx. Each
// entry counts the callers holding or waiting for it and is dropped from
// the map at zero, so the map does not grow with every name ever seen while
// two waiters on one key still always share one semaphore.
type keyedLocks struct {
	mu    sync.Mutex
	locks map[string]*keyedLock
}

type keyedLock struct {
	sem  chan struct{}
	refs int
}

func newKeyedLocks() *keyedLocks {
	return &keyedLocks{locks: make(map[string]*keyedLock)}
}

// acquire takes the lock for key, or gives up with ctx's error. The
// returned func releases it and must be called exactly once.
func (k *keyedLocks) acquire(ctx context.Context, key string) (func(), error) {
	k.mu.Lock()
	l, ok := k.locks[key]
	if !ok {
		l = &keyedLock{sem: make(chan struct{}, 1)}
		k.locks[key] = l
	}
	l.refs++
	k.mu.Unlock()
	select {
	case l.sem <- struct{}{}:
		return func() {
			<-l.sem
			k.unref(key, l)
		}, nil
	case <-ctx.Done():
		k.unref(key, l)
		return nil, ctx.Err()
	}
}

func (k *keyedLocks) unref(key string, l *keyedLock) {
	k.mu.Lock()
	defer k.mu.Unlock()
	l.refs--
	if l.refs == 0 {
		delete(k.locks, key)
	}
}