Refactor libvirt event handling + reconcile when domains change (#41)

With this change, we now reconcile the hypervisor resource if any libvirt domain lifecycle events occur. This keeps the hypervisor allocation up to date and simplifies how we manage the domain list.

We can only subscribe to libvirt events once. This means, we have to pull out the subscription logic from the `runMigrationListener` function in `libvirt_events.go`. We provide a new `WatchDomainChanges` interface function which can be used to subscribe to libvirt events, and reuse it for the existing migration watching and lifecycle logging. Furthermore, the distribution of libvirt events to the listeners is handled by a new event loop. Then, we also separate the logic unrelated to "migration watching" out, improving the overall structure and clarity of the code.

Finally, we make the hypervisor controller reconcile as soon as we obtain a new domain lifecycle event. This is done with the proper mechanism provided by the controller-runtime client library: raw channel sources. The libvirt event subscription is wired together with the controller-runtime event channel on startup of the manager.

Author: PhilippMatthes

internal/controller/hypervisor_controller.go

@@ -25,12 +25,16 @@ import (
 	"time"
 
 	kvmv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1"
+	golibvirt "github.com/digitalocean/go-libvirt"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	logger "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"github.com/cobaltcore-dev/kvm-node-agent/internal/certificates"
 	"github.com/cobaltcore-dev/kvm-node-agent/internal/evacuation"
@@ -48,6 +52,9 @@ type HypervisorReconciler struct {
 
 	osDescriptor     *systemd.Descriptor
 	evacuateOnReboot bool
+
+	// Channel that can be used to trigger reconcile events.
+	reconcileCh chan event.GenericEvent
 }
 
 const (
@@ -287,6 +294,63 @@ func (r *HypervisorReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	return ctrl.Result{RequeueAfter: 1 * time.Minute}, nil
 }
 
+// Trigger a reconcile event for the managed hypervisor through the
+// event channel which is watched by the controller manager.
+func (r *HypervisorReconciler) triggerReconcile() {
+	r.reconcileCh <- event.GenericEvent{
+		Object: &kvmv1.Hypervisor{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       "Hypervisor",
+				APIVersion: "kvm.cloud.sap/v1",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      sys.Hostname,
+				Namespace: sys.Namespace,
+			},
+		},
+	}
+}
+
+// Start is called when the manager starts. It starts the libvirt
+// event subscription to receive events when the hypervisor needs to be
+// reconciled.
+func (r *HypervisorReconciler) Start(ctx context.Context) error {
+	log := logger.FromContext(ctx, "controller", "hypervisor")
+	log.Info("starting libvirt event subscription")
+
+	// Ensure we're connected to libvirt.
+	if err := r.Libvirt.Connect(); err != nil {
+		log.Error(err, "unable to connect to libvirt")
+		return err
+	}
+
+	// Run a ticker which reconciles the hypervisor resource every minute.
+	// This ensures that we periodically reconcile the hypervisor even
+	// if no events are received from libvirt.
+	go func() {
+		ticker := time.NewTicker(1 * time.Minute)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				r.triggerReconcile()
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	// Domain lifecycle events impact the list of active/inactive domains,
+	// as well as the allocation of resources on the hypervisor.
+	r.Libvirt.WatchDomainChanges(
+		golibvirt.DomainEventIDLifecycle,
+		"reconcile-on-domain-lifecycle",
+		func(_ context.Context, _ any) { r.triggerReconcile() },
+	)
+
+	return nil
+}
+
 // SetupWithManager sets up the controller with the Manager.
 func (r *HypervisorReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	ctx := context.Background()
@@ -296,7 +360,16 @@ func (r *HypervisorReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		return fmt.Errorf("unable to get Systemd hostname describe(): %w", err)
 	}
 
+	// Prepare an event channel that will trigger a reconcile event.
+	r.reconcileCh = make(chan event.GenericEvent)
+	src := source.Channel(r.reconcileCh, &handler.EnqueueRequestForObject{})
+	// Run the Start(ctx context.Context) method when the manager starts.
+	if err := mgr.Add(r); err != nil {
+		return err
+	}
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&kvmv1.Hypervisor{}).
+		WatchesRawSource(src).
 		Complete(r)
 }

internal/controller/hypervisor_controller_test.go

@@ -19,15 +19,20 @@ package controller
 
 import (
 	"context"
+	"errors"
+	"time"
 
 	kvmv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1"
 	"github.com/coreos/go-systemd/v22/dbus"
+	golibvirt "github.com/digitalocean/go-libvirt"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	"k8s.io/apimachinery/pkg/api/errors"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	"github.com/cobaltcore-dev/kvm-node-agent/internal/libvirt"
@@ -36,6 +41,154 @@ import (
 )
 
 var _ = Describe("Hypervisor Controller", func() {
+	Context("When testing Start method", func() {
+		It("should successfully start and subscribe to libvirt events", func() {
+			ctx := context.Background()
+			eventCallbackCalled := false
+
+			controllerReconciler := &HypervisorReconciler{
+				Client: k8sClient,
+				Scheme: k8sClient.Scheme(),
+				Libvirt: &libvirt.InterfaceMock{
+					ConnectFunc: func() error {
+						return nil
+					},
+					WatchDomainChangesFunc: func(eventId golibvirt.DomainEventID, handlerId string, handler func(context.Context, any)) {
+						eventCallbackCalled = true
+						Expect(handlerId).To(Equal("reconcile-on-domain-lifecycle"))
+					},
+				},
+				reconcileCh: make(chan event.GenericEvent, 1),
+			}
+
+			err := controllerReconciler.Start(ctx)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(eventCallbackCalled).To(BeTrue())
+		})
+
+		It("should fail when libvirt connection fails", func() {
+			ctx := context.Background()
+
+			controllerReconciler := &HypervisorReconciler{
+				Client: k8sClient,
+				Scheme: k8sClient.Scheme(),
+				Libvirt: &libvirt.InterfaceMock{
+					ConnectFunc: func() error {
+						return errors.New("connection failed")
+					},
+				},
+				reconcileCh: make(chan event.GenericEvent, 1),
+			}
+
+			err := controllerReconciler.Start(ctx)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("connection failed"))
+		})
+	})
+
+	Context("When testing triggerReconcile method", func() {
+		It("should send an event to reconcile channel", func() {
+			const testHostname = "test-host"
+			const testNamespace = "test-namespace"
+
+			// Override hostname and namespace for this test
+			originalHostname := sys.Hostname
+			originalNamespace := sys.Namespace
+			sys.Hostname = testHostname
+			sys.Namespace = testNamespace
+			defer func() {
+				sys.Hostname = originalHostname
+				sys.Namespace = originalNamespace
+			}()
+
+			controllerReconciler := &HypervisorReconciler{
+				Client:      k8sClient,
+				Scheme:      k8sClient.Scheme(),
+				reconcileCh: make(chan event.GenericEvent, 1),
+			}
+
+			// Trigger reconcile in a goroutine to avoid blocking
+			go controllerReconciler.triggerReconcile()
+
+			// Wait for the event with a timeout
+			select {
+			case evt := <-controllerReconciler.reconcileCh:
+				Expect(evt.Object).NotTo(BeNil())
+				hv, ok := evt.Object.(*kvmv1.Hypervisor)
+				Expect(ok).To(BeTrue())
+				Expect(hv.Name).To(Equal(testHostname))
+				Expect(hv.Namespace).To(Equal(testNamespace))
+				Expect(hv.Kind).To(Equal("Hypervisor"))
+				Expect(hv.APIVersion).To(Equal("kvm.cloud.sap/v1"))
+			case <-time.After(2 * time.Second):
+				Fail("timeout waiting for reconcile event")
+			}
+		})
+	})
+
+	Context("When testing SetupWithManager method", func() {
+		It("should successfully setup controller with manager", func() {
+			// Create a test manager
+			mgr, err := ctrl.NewManager(cfg, ctrl.Options{
+				Scheme: k8sClient.Scheme(),
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			controllerReconciler := &HypervisorReconciler{
+				Client: k8sClient,
+				Scheme: k8sClient.Scheme(),
+				Systemd: &systemd.InterfaceMock{
+					DescribeFunc: func(ctx context.Context) (*systemd.Descriptor, error) {
+						return &systemd.Descriptor{
+							OperatingSystemReleaseData: []string{
+								"PRETTY_NAME=\"Garden Linux 1877.8\"",
+								"GARDENLINUX_VERSION=1877.8",
+							},
+							KernelVersion:   "6.1.0",
+							KernelRelease:   "6.1.0-gardenlinux",
+							KernelName:      "Linux",
+							HardwareVendor:  "Test Vendor",
+							HardwareModel:   "Test Model",
+							HardwareSerial:  "TEST123",
+							FirmwareVersion: "1.0",
+							FirmwareVendor:  "Test BIOS",
+							FirmwareDate:    time.Now().UnixMicro(),
+						}, nil
+					},
+				},
+			}
+
+			err = controllerReconciler.SetupWithManager(mgr)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(controllerReconciler.reconcileCh).NotTo(BeNil())
+			Expect(controllerReconciler.osDescriptor).NotTo(BeNil())
+			Expect(controllerReconciler.osDescriptor.OperatingSystemReleaseData).To(HaveLen(2))
+		})
+
+		It("should fail when systemd Describe returns error", func() {
+			// Create a test manager
+			mgr, err := ctrl.NewManager(cfg, ctrl.Options{
+				Scheme: k8sClient.Scheme(),
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			controllerReconciler := &HypervisorReconciler{
+				Client: k8sClient,
+				Scheme: k8sClient.Scheme(),
+				Systemd: &systemd.InterfaceMock{
+					DescribeFunc: func(ctx context.Context) (*systemd.Descriptor, error) {
+						return nil, errors.New("systemd describe failed")
+					},
+				},
+			}
+
+			err = controllerReconciler.SetupWithManager(mgr)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("unable to get Systemd hostname describe()"))
+			Expect(err.Error()).To(ContainSubstring("systemd describe failed"))
+		})
+	})
+
 	Context("When reconciling a resource", func() {
 		const resourceName = "test-resource"
 
@@ -50,7 +203,7 @@ var _ = Describe("Hypervisor Controller", func() {
 		BeforeEach(func() {
 			By("creating the custom resource for the Kind Hypervisor")
 			err := k8sClient.Get(ctx, typeNamespacedName, hypervisor)
-			if err != nil && errors.IsNotFound(err) {
+			if err != nil && apierrors.IsNotFound(err) {
 				resource := &kvmv1.Hypervisor{
 					ObjectMeta: metav1.ObjectMeta{
 						Name:      resourceName,

internal/libvirt/dominfo/client.go

@@ -27,7 +27,10 @@ import (
 // Client that returns information for all domains on our host.
 type Client interface {
 	// Return information for all domains on our host.
-	Get(virt *libvirt.Libvirt) ([]DomainInfo, error)
+	Get(
+		virt *libvirt.Libvirt,
+		flags ...libvirt.ConnectListAllDomainsFlags,
+	) ([]DomainInfo, error)
 }
 
 // Implementation of the Client interface.
@@ -39,9 +42,16 @@ func NewClient() Client {
 }
 
 // Return information for all domains on our host.
-func (m *client) Get(virt *libvirt.Libvirt) ([]DomainInfo, error) {
-	domains, _, err := virt.ConnectListAllDomains(1,
-		libvirt.ConnectListDomainsActive|libvirt.ConnectListDomainsInactive)
+func (m *client) Get(
+	virt *libvirt.Libvirt,
+	flags ...libvirt.ConnectListAllDomainsFlags,
+) ([]DomainInfo, error) {
+
+	flag := libvirt.ConnectListAllDomainsFlags(0)
+	for _, f := range flags {
+		flag |= f
+	}
+	domains, _, err := virt.ConnectListAllDomains(1, flag)
 	if err != nil {
 		log.Log.Error(err, "failed to list all domains")
 		return nil, err
@@ -72,7 +82,11 @@ func NewClientEmulator() Client {
 }
 
 // Get the domain infos of the host we are mounted on.
-func (c *clientEmulator) Get(virt *libvirt.Libvirt) ([]DomainInfo, error) {
+func (c *clientEmulator) Get(
+	virt *libvirt.Libvirt,
+	flags ...libvirt.ConnectListAllDomainsFlags,
+) ([]DomainInfo, error) {
+
 	var info DomainInfo
 	if err := xml.Unmarshal(exampleXML, &info); err != nil {
 		log.Log.Error(err, "failed to unmarshal example capabilities")

internal/libvirt/dominfo/client_test.go

@@ -347,3 +347,28 @@ func TestClientTypes_AreDistinct(t *testing.T) {
 		t.Error("Expected NewClient() and NewClientEmulator() to return different types")
 	}
 }
+
+func TestClientEmulator_Get_WithTwoFlags(t *testing.T) {
+	client := NewClientEmulator()
+
+	// Test that Get accepts multiple flags without error
+	// The emulator doesn't use libvirt, so we pass nil and arbitrary flags
+	domainInfos, err := client.Get(nil, 1, 2)
+
+	if err != nil {
+		t.Fatalf("Get() with 2 flags returned unexpected error: %v", err)
+	}
+
+	if len(domainInfos) == 0 {
+		t.Fatal("Expected at least one domain info from emulator")
+	}
+
+	// Verify the returned domain info has expected structure
+	if domainInfos[0].Name == "" {
+		t.Error("Expected domain to have a name")
+	}
+
+	if domainInfos[0].UUID == "" {
+		t.Error("Expected domain to have a UUID")
+	}
+}