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

package kvm

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

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

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

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

// Reasons only the KVM check can tell apart; the rest are provider-wide.
const (
	reasonSSHMaterialUnavailable = "SSHMaterialUnavailable"
	reasonSSHHandshakeFailed     = "SSHHandshakeFailed"
	reasonLibvirtConnectFailed   = "LibvirtConnectFailed"
	reasonLibvirtError           = "LibvirtError"
	reasonPoolNotFound           = "PoolNotFound"
	reasonPoolNotActive          = "PoolNotActive"
)

// hostFacts is what a successful check learns about a host.
type hostFacts struct {
	version   uint64
	hostname  string
	poolState libvirt.StoragePoolState
	capacity  uint64
	available uint64
}

// Check verifies every configured host concurrently, in kvm.hosts order.
// It goes through the same pooled connection Create and Get use, so the
// SSH and libvirt handshakes only happen on the first check or after the
// link dropped; then two cheap RPCs prove the daemon answers, and the
// storage pool VMs are created in must exist and be running.
func (p *Provider) Check(ctx context.Context) []provider.TargetStatus {
	out := make([]provider.TargetStatus, len(p.cfg.Hosts))
	var wg sync.WaitGroup
	for i, h := range p.cfg.Hosts {
		wg.Go(func() { out[i] = p.checkHost(ctx, h) })
	}
	wg.Wait()
	return out
}

// checkHost goes through probe, so a check that runs out of time abandons
// its RPCs rather than closing the connection the reconciles share. fn
// returns the RPC errors untouched, since a transport error still drops the
// link; a pool that exists but is stopped says nothing about the link, so
// that verdict is reached afterwards. facts is read only after probe
// returned nil, since an abandoned fn may still write it.
func (p *Provider) checkHost(ctx context.Context, host provider.KVMHost) provider.TargetStatus {
	ts := provider.TargetStatus{
		Name:    host.Name,
		Address: host.Address,
		Details: map[string]string{"user": host.User, "pool": host.Pool},
	}
	start := time.Now()
	var facts hostFacts
	err := p.hosts.probe(ctx, p.logger(ctx), host, func(api libvirtAPI) error {
		version, err := api.ConnectGetLibVersion()
		if err != nil {
			return err
		}
		hostname, err := api.ConnectGetHostname()
		if err != nil {
			return err
		}
		pool, err := api.StoragePoolLookupByName(host.Pool)
		if err != nil {
			return err
		}
		state, capacity, _, available, err := api.StoragePoolGetInfo(pool)
		if err != nil {
			return err
		}
		facts = hostFacts{
			version:   version,
			hostname:  hostname,
			poolState: libvirt.StoragePoolState(state),
			capacity:  capacity,
			available: available,
		}
		return nil
	})
	ts.Latency = time.Since(start)
	if err != nil {
		ts.Reason = checkReason(err)
		ts.Message = err.Error()
		return ts
	}
	ts.Details["hostname"] = facts.hostname
	ts.Details["libvirtVersion"] = formatLibVersion(facts.version)
	ts.Details["poolState"] = poolStateName(facts.poolState)
	ts.Details["poolCapacity"] = formatGiB(facts.capacity)
	ts.Details["poolAvailable"] = formatGiB(facts.available)
	if facts.poolState != libvirt.StoragePoolRunning {
		ts.Reason = reasonPoolNotActive
		ts.Message = fmt.Sprintf("storage pool %q is %s", host.Pool, poolStateName(facts.poolState))
		return ts
	}
	ts.Ready = true
	ts.Reason = provider.ReasonReachable
	return ts
}

// checkReason names where a failed check broke. The dial stage wins over
// a timeout: "the SSH handshake did not finish" says more than "the check
// ran out of time", and the message still carries the deadline.
func checkReason(err error) string {
	var de *dialError
	if errors.As(err, &de) {
		switch de.stage {
		case stageSSHMaterial:
			return reasonSSHMaterialUnavailable
		case stageSSHHandshake:
			return reasonSSHHandshakeFailed
		case stageLibvirtConnect:
			return reasonLibvirtConnectFailed
		default:
			return provider.ReasonUnreachable
		}
	}
	switch {
	case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled), errors.Is(err, errProbeBusy):
		return provider.ReasonTimeout
	case hasCode(err, libvirt.ErrNoStoragePool):
		return reasonPoolNotFound
	case isLibvirtError(err):
		return reasonLibvirtError
	default:
		return provider.ReasonUnreachable
	}
}

// formatLibVersion decodes libvirt's major*1e6 + minor*1e3 + micro.
func formatLibVersion(v uint64) string {
	return fmt.Sprintf("%d.%d.%d", v/1_000_000, v/1_000%1_000, v%1_000)
}

// poolStateName uses virsh's words for a pool state.
func poolStateName(s libvirt.StoragePoolState) string {
	switch s {
	case libvirt.StoragePoolInactive:
		return "inactive"
	case libvirt.StoragePoolBuilding:
		return "building"
	case libvirt.StoragePoolRunning:
		return "running"
	case libvirt.StoragePoolDegraded:
		return "degraded"
	case libvirt.StoragePoolInaccessible:
		return "inaccessible"
	default:
		return fmt.Sprintf("state %d", s)
	}
}

func formatGiB(bytes uint64) string {
	return fmt.Sprintf("%.1fGiB", float64(bytes)/(1<<30))
}