egress-proxies-operator/internal/provider/proxmox/observe.go

package proxmox

import (
	"context"
	"errors"
	"io"
	"strconv"
	"time"

	logf "sigs.k8s.io/controller-runtime/pkg/log"
)

// observed is the logging decorator of spec entry 20: one V(1)
// "proxmox api call" line per seam call, with request and response detail
// at V(2). It starts no span: the transport already traces each request
// (spec entry 17), so nothing is counted twice.
type observed struct {
	inner    api
	provider string
}

var _ api = (*observed)(nil)

func observe(inner api, providerName string) api {
	return &observed{inner: inner, provider: providerName}
}

// target is what a call names at V(1); zero fields are left out.
type target struct {
	node    string
	vmid    int
	storage string
	upid    string
	extra   []any
}

// detail returns V(2)-only key/value pairs. It runs only when V(2) is on.
type detail func() []any

// call wraps one seam call. fn returns the V(2) detail and may fill in
// t.upid from the call's answer.
func (o *observed) call(ctx context.Context, name string, t *target, fn func() (detail, error)) error {
	log := logf.FromContext(ctx).WithName("proxmox").WithValues("provider", o.provider)
	start := time.Now()
	more, err := fn()
	kv := make([]any, 0, 16)
	kv = append(kv, "call", name)
	if t.node != "" {
		kv = append(kv, "node", t.node)
	}
	if t.vmid != 0 {
		kv = append(kv, "vmid", t.vmid)
	}
	if t.storage != "" {
		kv = append(kv, "storage", t.storage)
	}
	if t.upid != "" {
		kv = append(kv, "upid", t.upid)
	}
	kv = append(kv, t.extra...)
	kv = append(kv, "duration", time.Since(start))
	if err != nil {
		var ae *apiError
		if errors.As(err, &ae) {
			kv = append(kv, "httpStatus", ae.Status)
		}
		kv = append(kv, "error", err.Error())
	}
	if more != nil && log.V(2).Enabled() {
		kv = append(kv, more()...)
	}
	log.V(1).Info("proxmox api call", kv...)
	return err
}

func (o *observed) Version(ctx context.Context) (v versionInfo, err error) {
	err = o.call(ctx, "Version", &target{}, func() (detail, error) {
		v, err = o.inner.Version(ctx)
		return func() []any { return []any{"version", v.Version} }, err
	})
	return v, err
}

func (o *observed) NodeStatus(ctx context.Context, node string) (s nodeStatus, err error) {
	err = o.call(ctx, "NodeStatus", &target{node: node}, func() (detail, error) {
		s, err = o.inner.NodeStatus(ctx, node)
		return func() []any { return []any{"pveVersion", s.PVEVersion, "kernel", s.CurrentKernel.Release} }, err
	})
	return s, err
}

func (o *observed) StorageStatus(ctx context.Context, node, storage string) (s storageStatus, err error) {
	err = o.call(ctx, "StorageStatus", &target{node: node, storage: storage}, func() (detail, error) {
		s, err = o.inner.StorageStatus(ctx, node, storage)
		return func() []any {
			return []any{"type", s.Type, "content", s.Content, "active", s.Active, "avail", s.Avail}
		}, err
	})
	return s, err
}

func (o *observed) ClusterResources(ctx context.Context) (rs []vmResource, err error) {
	err = o.call(ctx, "ClusterResources", &target{}, func() (detail, error) {
		rs, err = o.inner.ClusterResources(ctx)
		return func() []any { return []any{"count", len(rs), "vms", resourceSummary(rs)} }, err
	})
	return rs, err
}

func (o *observed) VMConfig(ctx context.Context, node string, vmid int) (cfg vmConfig, err error) {
	err = o.call(ctx, "VMConfig", &target{node: node, vmid: vmid}, func() (detail, error) {
		cfg, err = o.inner.VMConfig(ctx, node, vmid)
		return func() []any { return []any{"config", configSummary(cfg)} }, err
	})
	return cfg, err
}

func (o *observed) SetVMConfig(ctx context.Context, node string, vmid int, params map[string]string) error {
	return o.call(ctx, "SetVMConfig", &target{node: node, vmid: vmid}, func() (detail, error) {
		return func() []any { return []any{"params", params} }, o.inner.SetVMConfig(ctx, node, vmid, params)
	})
}

// Clone names the new VM at V(1); the description, which holds the mark,
// only at V(2).
func (o *observed) Clone(ctx context.Context, node string, vmid int, p cloneParams) (upid string, err error) {
	t := &target{node: node, vmid: vmid, extra: []any{"newid", p.NewID, "name", p.Name}}
	err = o.call(ctx, "Clone", t, func() (detail, error) {
		upid, err = o.inner.Clone(ctx, node, vmid, p)
		t.upid = upid
		return func() []any {
			return []any{"description", p.Description, "storage", p.Storage, "target", p.Target}
		}, err
	})
	return upid, err
}

func (o *observed) Resize(ctx context.Context, node string, vmid int, disk, size string) (upid string, err error) {
	t := &target{node: node, vmid: vmid}
	err = o.call(ctx, "Resize", t, func() (detail, error) {
		upid, err = o.inner.Resize(ctx, node, vmid, disk, size)
		t.upid = upid
		return func() []any { return []any{"disk", disk, "size", size} }, err
	})
	return upid, err
}

func (o *observed) Start(ctx context.Context, node string, vmid int) (upid string, err error) {
	t := &target{node: node, vmid: vmid}
	err = o.call(ctx, "Start", t, func() (detail, error) {
		upid, err = o.inner.Start(ctx, node, vmid)
		t.upid = upid
		return nil, err
	})
	return upid, err
}

func (o *observed) Stop(ctx context.Context, node string, vmid int) (upid string, err error) {
	t := &target{node: node, vmid: vmid}
	err = o.call(ctx, "Stop", t, func() (detail, error) {
		upid, err = o.inner.Stop(ctx, node, vmid)
		t.upid = upid
		return nil, err
	})
	return upid, err
}

func (o *observed) VMStatus(ctx context.Context, node string, vmid int) (s vmStatus, err error) {
	err = o.call(ctx, "VMStatus", &target{node: node, vmid: vmid}, func() (detail, error) {
		s, err = o.inner.VMStatus(ctx, node, vmid)
		return func() []any { return []any{"status", s.Status, "qmpstatus", s.QMPStatus, "lock", s.Lock} }, err
	})
	return s, err
}

func (o *observed) Destroy(ctx context.Context, node string, vmid int) (upid string, err error) {
	t := &target{node: node, vmid: vmid}
	err = o.call(ctx, "Destroy", t, func() (detail, error) {
		upid, err = o.inner.Destroy(ctx, node, vmid)
		t.upid = upid
		return nil, err
	})
	return upid, err
}

func (o *observed) TaskStatus(ctx context.Context, node, upid string) (s taskStatus, err error) {
	err = o.call(ctx, "TaskStatus", &target{node: node, upid: upid}, func() (detail, error) {
		s, err = o.inner.TaskStatus(ctx, node, upid)
		return func() []any { return []any{"status", s.Status, "exitstatus", s.ExitStatus} }, err
	})
	return s, err
}

func (o *observed) AgentNetworkInterfaces(ctx context.Context, node string, vmid int) (ifaces []agentInterface, err error) {
	err = o.call(ctx, "AgentNetworkInterfaces", &target{node: node, vmid: vmid}, func() (detail, error) {
		ifaces, err = o.inner.AgentNetworkInterfaces(ctx, node, vmid)
		return func() []any { return []any{"addresses", agentAddresses(ifaces)} }, err
	})
	return ifaces, err
}

// UploadISO logs the file name and size, never the body: a seed carries
// CloudInit.
func (o *observed) UploadISO(ctx context.Context, node, storage, filename string, r io.Reader, size int64) (upid string, err error) {
	t := &target{node: node, storage: storage, extra: []any{"filename", filename}}
	err = o.call(ctx, "UploadISO", t, func() (detail, error) {
		upid, err = o.inner.UploadISO(ctx, node, storage, filename, r, size)
		t.upid = upid
		return func() []any { return []any{"bytes", size} }, err
	})
	return upid, err
}

func (o *observed) DeleteVolume(ctx context.Context, node, storage, volid string) (upid string, err error) {
	t := &target{node: node, storage: storage, extra: []any{"volid", volid}}
	err = o.call(ctx, "DeleteVolume", t, func() (detail, error) {
		upid, err = o.inner.DeleteVolume(ctx, node, storage, volid)
		t.upid = upid
		return nil, err
	})
	return upid, err
}

// resourceSummary is one "<vmid>:<name>:<status>" per entry, with the lock
// appended when set.
func resourceSummary(rs []vmResource) []string {
	out := make([]string, len(rs))
	for i, r := range rs {
		s := strconv.Itoa(r.VMID) + ":" + r.Name + ":" + r.Status
		if r.Lock != "" {
			s += ":lock=" + r.Lock
		}
		out[i] = s
	}
	return out
}

// summaryKeys are the config keys the provider reads or writes.
var summaryKeys = []string{"name", "lock", "tags", "description", "cores", "memory", "agent", "net0", "ide2", "boot", "bootdisk", "scsi0", "virtio0", "sata0", "ide0", "template"}

func configSummary(cfg vmConfig) map[string]string {
	out := make(map[string]string, len(summaryKeys))
	for _, k := range summaryKeys {
		if v, ok := cfg[k]; ok {
			out[k] = v
		}
	}
	return out
}

func agentAddresses(ifaces []agentInterface) []string {
	var out []string
	for _, iface := range ifaces {
		for _, a := range iface.IPAddresses {
			out = append(out, iface.Name+"="+a.Address)
		}
	}
	return out
}