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

package kvm2

import (
	"context"
	"io"
	"time"

	"github.com/digitalocean/go-libvirt"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"
	logf "sigs.k8s.io/controller-runtime/pkg/log"

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

// observed is the span-decorator of spec entries 17 and 20: one child span
// kvm2.<Method> and one V(1) "libvirt call" line per call, with request and
// response detail at V(2). Applied once, in the dialer, over every backend
// that leaves it.
type observed struct {
	inner    backend
	provider string
	host     string
	tracer   trace.Tracer
}

var (
	_ backend  = (*observed)(nil)
	_ liveness = (*observed)(nil)
)

func observe(inner backend, providerName, hostName string, opts ...tracing.Option) backend {
	return &observed{inner: inner, provider: providerName, host: hostName, tracer: tracing.Tracer(opts...)}
}

// dead forwards the inner backend's liveness; a backend without one is
// never dropped by the host table.
func (o *observed) dead() <-chan struct{} {
	if l, ok := o.inner.(liveness); ok {
		return l.dead()
	}
	return nil
}

// id is one identifier a call names: a domain, pool or volume. It becomes
// both the span attribute kvm2.<key> and the log key.
type id struct{ key, val string }

func domainID(name string) id { return id{"domain", name} }
func poolID(name string) id   { return id{"pool", name} }
func volumeID(name string) id { return id{"volume", name} }

// detail returns V(2)-only key/value pairs. It runs only when V(2) is on,
// so building the list of a hundred domain names costs nothing otherwise.
type detail func() []any

// call wraps one backend call: span, timing, status and the log line. fn
// returns the V(2) detail for the line, or nil.
func (o *observed) call(ctx context.Context, rpc string, ids []id, fn func(ctx context.Context) (detail, error)) error {
	attrs := make([]attribute.KeyValue, 0, len(ids)+1)
	attrs = append(attrs, attribute.String("kvm2.host", o.host))
	for _, i := range ids {
		attrs = append(attrs, attribute.String("kvm2."+i.key, i.val))
	}
	ctx, span := tracing.StartSpan(ctx, o.tracer, "kvm2."+rpc, trace.WithSpanKind(trace.SpanKindClient), trace.WithAttributes(attrs...))
	defer span.End()
	log := logf.FromContext(ctx).WithName("kvm2").WithValues("provider", o.provider, "host", o.host)
	start := time.Now()
	more, err := fn(ctx)
	kv := make([]any, 0, 8+2*len(ids))
	kv = append(kv, "rpc", rpc)
	for _, i := range ids {
		kv = append(kv, i.key, i.val)
	}
	kv = append(kv, "duration", time.Since(start))
	if err != nil {
		span.SetStatus(codes.Error, err.Error())
		if code, ok := libvirtCode(err); ok {
			span.SetAttributes(attribute.Int("kvm2.libvirt_code", int(code)))
		}
		kv = append(kv, "error", err.Error())
		kv = append(kv, libvirtErrorFields(err)...)
	}
	if more != nil && log.V(2).Enabled() {
		kv = append(kv, more()...)
	}
	log.V(1).Info("libvirt call", kv...)
	return err
}

func (o *observed) DomainLookupByName(ctx context.Context, name string) (dom libvirt.Domain, err error) {
	err = o.call(ctx, "DomainLookupByName", []id{domainID(name)}, func(ctx context.Context) (detail, error) {
		dom, err = o.inner.DomainLookupByName(ctx, name)
		return nil, err
	})
	return dom, err
}

func (o *observed) DomainDefineXML(ctx context.Context, xml string) (dom libvirt.Domain, err error) {
	err = o.call(ctx, "DomainDefineXML", nil, func(ctx context.Context) (detail, error) {
		dom, err = o.inner.DomainDefineXML(ctx, xml)
		return func() []any { return []any{"domain", dom.Name, "xml", xml} }, err
	})
	return dom, err
}

func (o *observed) DomainCreate(ctx context.Context, dom libvirt.Domain) error {
	return o.call(ctx, "DomainCreate", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) {
		return nil, o.inner.DomainCreate(ctx, dom)
	})
}

func (o *observed) DomainDestroy(ctx context.Context, dom libvirt.Domain) error {
	return o.call(ctx, "DomainDestroy", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) {
		return nil, o.inner.DomainDestroy(ctx, dom)
	})
}

func (o *observed) DomainUndefineFlags(ctx context.Context, dom libvirt.Domain, flags libvirt.DomainUndefineFlagsValues) error {
	return o.call(ctx, "DomainUndefineFlags", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) {
		return nil, o.inner.DomainUndefineFlags(ctx, dom, flags)
	})
}

func (o *observed) DomainGetState(ctx context.Context, dom libvirt.Domain, flags uint32) (state int32, reason int32, err error) {
	err = o.call(ctx, "DomainGetState", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) {
		state, reason, err = o.inner.DomainGetState(ctx, dom, flags)
		return func() []any { return []any{"state", libvirt.DomainState(state).String(), "reason", reason} }, err
	})
	return state, reason, err
}

func (o *observed) DomainGetMetadata(ctx context.Context, dom libvirt.Domain, typ int32, uri libvirt.OptString, flags libvirt.DomainModificationImpact) (md string, err error) {
	err = o.call(ctx, "DomainGetMetadata", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) {
		md, err = o.inner.DomainGetMetadata(ctx, dom, typ, uri, flags)
		return func() []any { return []any{"metadata", md} }, err
	})
	return md, err
}

func (o *observed) DomainInterfaceAddresses(ctx context.Context, dom libvirt.Domain, source uint32, flags uint32) (ifaces []libvirt.DomainInterface, err error) {
	err = o.call(ctx, "DomainInterfaceAddresses", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) {
		ifaces, err = o.inner.DomainInterfaceAddresses(ctx, dom, source, flags)
		return func() []any { return []any{"source", source, "addresses", interfaceAddresses(ifaces)} }, err
	})
	return ifaces, err
}

func (o *observed) ConnectListAllDomains(ctx context.Context, needResults int32, flags libvirt.ConnectListAllDomainsFlags) (doms []libvirt.Domain, n uint32, err error) {
	err = o.call(ctx, "ConnectListAllDomains", nil, func(ctx context.Context) (detail, error) {
		doms, n, err = o.inner.ConnectListAllDomains(ctx, needResults, flags)
		return func() []any { return []any{"count", len(doms), "domains", domainNames(doms)} }, err
	})
	return doms, n, err
}

func (o *observed) StoragePoolLookupByName(ctx context.Context, name string) (pool libvirt.StoragePool, err error) {
	err = o.call(ctx, "StoragePoolLookupByName", []id{poolID(name)}, func(ctx context.Context) (detail, error) {
		pool, err = o.inner.StoragePoolLookupByName(ctx, name)
		return nil, err
	})
	return pool, err
}

func (o *observed) StoragePoolRefresh(ctx context.Context, pool libvirt.StoragePool, flags uint32) error {
	return o.call(ctx, "StoragePoolRefresh", []id{poolID(pool.Name)}, func(ctx context.Context) (detail, error) {
		return nil, o.inner.StoragePoolRefresh(ctx, pool, flags)
	})
}

func (o *observed) StoragePoolGetInfo(ctx context.Context, pool libvirt.StoragePool) (state uint8, capacity, allocation, available uint64, err error) {
	err = o.call(ctx, "StoragePoolGetInfo", []id{poolID(pool.Name)}, func(ctx context.Context) (detail, error) {
		state, capacity, allocation, available, err = o.inner.StoragePoolGetInfo(ctx, pool)
		return func() []any {
			return []any{"state", libvirt.StoragePoolState(state).String(), "capacity", capacity, "allocation", allocation, "available", available}
		}, err
	})
	return state, capacity, allocation, available, err
}

func (o *observed) StorageVolLookupByName(ctx context.Context, pool libvirt.StoragePool, name string) (vol libvirt.StorageVol, err error) {
	err = o.call(ctx, "StorageVolLookupByName", []id{poolID(pool.Name), volumeID(name)}, func(ctx context.Context) (detail, error) {
		vol, err = o.inner.StorageVolLookupByName(ctx, pool, name)
		return nil, err
	})
	return vol, err
}

func (o *observed) StorageVolGetPath(ctx context.Context, vol libvirt.StorageVol) (path string, err error) {
	err = o.call(ctx, "StorageVolGetPath", []id{poolID(vol.Pool), volumeID(vol.Name)}, func(ctx context.Context) (detail, error) {
		path, err = o.inner.StorageVolGetPath(ctx, vol)
		return func() []any { return []any{"path", path} }, err
	})
	return path, err
}

func (o *observed) StorageVolCreateXML(ctx context.Context, pool libvirt.StoragePool, xml string, flags libvirt.StorageVolCreateFlags) (vol libvirt.StorageVol, err error) {
	err = o.call(ctx, "StorageVolCreateXML", []id{poolID(pool.Name)}, func(ctx context.Context) (detail, error) {
		vol, err = o.inner.StorageVolCreateXML(ctx, pool, xml, flags)
		return func() []any { return []any{"volume", vol.Name, "xml", xml} }, err
	})
	return vol, err
}

// StorageVolUpload and StorageVolDownload log the length only; the bodies
// are a seed ISO carrying CloudInit or a base image.
func (o *observed) StorageVolUpload(ctx context.Context, vol libvirt.StorageVol, r io.Reader, offset, length uint64, flags libvirt.StorageVolUploadFlags) error {
	return o.call(ctx, "StorageVolUpload", []id{poolID(vol.Pool), volumeID(vol.Name)}, func(ctx context.Context) (detail, error) {
		err := o.inner.StorageVolUpload(ctx, vol, r, offset, length, flags)
		return func() []any { return []any{"offset", offset, "length", length} }, err
	})
}

func (o *observed) StorageVolDownload(ctx context.Context, vol libvirt.StorageVol, w io.Writer, offset, length uint64, flags libvirt.StorageVolDownloadFlags) error {
	return o.call(ctx, "StorageVolDownload", []id{poolID(vol.Pool), volumeID(vol.Name)}, func(ctx context.Context) (detail, error) {
		err := o.inner.StorageVolDownload(ctx, vol, w, offset, length, flags)
		return func() []any { return []any{"offset", offset, "length", length} }, err
	})
}

func (o *observed) StorageVolDelete(ctx context.Context, vol libvirt.StorageVol, flags libvirt.StorageVolDeleteFlags) error {
	return o.call(ctx, "StorageVolDelete", []id{poolID(vol.Pool), volumeID(vol.Name)}, func(ctx context.Context) (detail, error) {
		return nil, o.inner.StorageVolDelete(ctx, vol, flags)
	})
}

func (o *observed) ConnectGetLibVersion(ctx context.Context) (v uint64, err error) {
	err = o.call(ctx, "ConnectGetLibVersion", nil, func(ctx context.Context) (detail, error) {
		v, err = o.inner.ConnectGetLibVersion(ctx)
		return func() []any { return []any{"libvirtVersion", v} }, err
	})
	return v, err
}

func (o *observed) ConnectGetHostname(ctx context.Context) (name string, err error) {
	err = o.call(ctx, "ConnectGetHostname", nil, func(ctx context.Context) (detail, error) {
		name, err = o.inner.ConnectGetHostname(ctx)
		return func() []any { return []any{"hostname", name} }, err
	})
	return name, err
}

// Close is passed through: it is not a libvirt call and gets no span.
func (o *observed) Close() error { return o.inner.Close() }

func domainNames(doms []libvirt.Domain) []string {
	names := make([]string, len(doms))
	for i, d := range doms {
		names[i] = d.Name
	}
	return names
}

func interfaceAddresses(ifaces []libvirt.DomainInterface) []string {
	var out []string
	for _, iface := range ifaces {
		for _, a := range iface.Addrs {
			out = append(out, iface.Name+"="+a.Addr)
		}
	}
	return out
}