Merge pull request #58 from cobaltcore-dev/wait-for-connection

notandy · web-flow · commit 51cbffdb4145 · 2026-02-25T11:26:53.000-05:00
Wait for libvirt connection in hypervisor controller
diff --git a/internal/controller/hypervisor_controller.go b/internal/controller/hypervisor_controller.go
@@ -58,6 +58,11 @@ type HypervisorReconciler struct {
 
 	// Channel that can be used to trigger reconcile events.
 	reconcileCh chan event.GenericEvent
+
+	// An interval that determines how long to wait between connection attempts
+	// to libvirt. This is used in the Start method when trying to connect to
+	// libvirt, and can be set to a lower value for testing purposes.
+	libvirtConnectInterval time.Duration
 }
 
 const (
@@ -195,6 +200,9 @@ func (r *HypervisorReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			Message: fmt.Sprintf("unable to connect to libvirtd: %v", err),
 			Reason:  "ConnectFailed",
 		})
+		// TODO: When libvirt is down, we should also set the overall Ready
+		// condition to false, because without libvirt connection, we won't be
+		// able to detect capacity and other scheduling-relevant details.
 	} else {
 		// We're connected.
 		meta.SetStatusCondition(&hypervisor.Status.Conditions, metav1.Condition{
@@ -326,10 +334,45 @@ 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 {
+	// Get the hypervisor we will reconcile.
+	var hypervisor kvmv1.Hypervisor
+	key := client.ObjectKey{Name: sys.Hostname} // Cluster-scoped
+	if err := r.Get(ctx, key, &hypervisor); err != nil {
+		return fmt.Errorf("unable to get hypervisor: %w", err)
+	}
+
+	// Block until we're connected to libvirt.
+	for {
+		// Exit if the context is done, e.g. when the manager is shutting down.
+		if ctx.Err() != nil {
+			return fmt.Errorf("context done while trying to connect to libvirt: %w", ctx.Err())
+		}
+		err := r.Libvirt.Connect()
+		if err == nil {
+			log.Info("connected to libvirt")
+			break // Connected successfully
+		}
 		log.Error(err, "unable to connect to libvirt")
-		return err
+		// Set the hypervisor's LibVirtType condition to false with the
+		// error message, so that it's visible in the status.
+		base := hypervisor.DeepCopy()
+		meta.SetStatusCondition(&hypervisor.Status.Conditions, metav1.Condition{
+			Type:    LibVirtType, // TODO: This should be a kvmv1 condition.
+			Status:  metav1.ConditionFalse,
+			Message: fmt.Sprintf("unable to connect to libvirt: %v", err),
+			Reason:  "ConnectFailed",
+		})
+		patch := client.MergeFromWithOptions(base, client.MergeFromWithOptimisticLock{})
+		if err := r.Status().Patch(ctx, &hypervisor, patch); err != nil {
+			log.Error(err, "unable to update hypervisor status after failed libvirt connection")
+		}
+		log.Info("updated hypervisor status after failed libvirt connection")
+		timeToSleep := r.libvirtConnectInterval
+		if timeToSleep == 0 {
+			timeToSleep = 5 * time.Second // default value
+		}
+		log.Info("retrying libvirt connection after sleeping", "duration", timeToSleep)
+		time.Sleep(timeToSleep)
 	}
 
 	// Run a ticker which reconciles the hypervisor resource every minute.
diff --git a/internal/controller/hypervisor_controller_test.go b/internal/controller/hypervisor_controller_test.go
@@ -20,6 +20,7 @@ package controller
 import (
 	"context"
 	"errors"
+	"strings"
 	"time"
 
 	kvmv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1"
@@ -45,6 +46,26 @@ var _ = Describe("Hypervisor Controller", func() {
 	Context("When testing Start method", func() {
 		It("should successfully start and subscribe to libvirt events", func() {
 			ctx := context.Background()
+
+			// Create a hypervisor resource for this test
+			hypervisorName := "start-success-test-hypervisor"
+			originalHostname := sys.Hostname
+			sys.Hostname = hypervisorName
+			defer func() {
+				sys.Hostname = originalHostname
+			}()
+
+			hypervisor := &kvmv1.Hypervisor{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: hypervisorName,
+				},
+			}
+			Expect(k8sClient.Create(ctx, hypervisor)).To(Succeed())
+			defer func() {
+				err := k8sClient.Delete(ctx, hypervisor)
+				Expect(err).NotTo(HaveOccurred())
+			}()
+
 			eventCallbackCalled := false
 
 			controllerReconciler := &HypervisorReconciler{
@@ -68,7 +89,27 @@ var _ = Describe("Hypervisor Controller", func() {
 		})
 
 		It("should fail when libvirt connection fails", func() {
-			ctx := context.Background()
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			// Create a hypervisor resource for this test
+			hypervisorName := "start-fail-test-hypervisor"
+			originalHostname := sys.Hostname
+			sys.Hostname = hypervisorName
+			defer func() {
+				sys.Hostname = originalHostname
+			}()
+
+			hypervisor := &kvmv1.Hypervisor{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: hypervisorName,
+				},
+			}
+			Expect(k8sClient.Create(context.Background(), hypervisor)).To(Succeed())
+			defer func() {
+				err := k8sClient.Delete(context.Background(), hypervisor)
+				Expect(err).NotTo(HaveOccurred())
+			}()
 
 			controllerReconciler := &HypervisorReconciler{
 				Client: k8sClient,
@@ -78,12 +119,124 @@ var _ = Describe("Hypervisor Controller", func() {
 						return errors.New("connection failed")
 					},
 				},
+				reconcileCh:            make(chan event.GenericEvent, 1),
+				libvirtConnectInterval: 10 * time.Millisecond,
+			}
+
+			// Start runs in a goroutine so we can cancel the context
+			done := make(chan error, 1)
+			go func() {
+				done <- controllerReconciler.Start(ctx)
+			}()
+
+			// Wait for the hypervisor status to reflect the failed libvirt connection
+			// This must happen BEFORE we cancel the context to ensure the Start method
+			// had time to attempt connection and update the status
+			var updatedHypervisor kvmv1.Hypervisor
+			Eventually(func() bool {
+				err := k8sClient.Get(context.Background(), types.NamespacedName{Name: hypervisorName}, &updatedHypervisor)
+				if err != nil {
+					return false
+				}
+				for _, condition := range updatedHypervisor.Status.Conditions {
+					if condition.Type == "LibVirtConnection" {
+						return condition.Status == metav1.ConditionFalse &&
+							condition.Reason == "ConnectFailed" &&
+							strings.Contains(condition.Message, "connection failed")
+					}
+				}
+				return false
+			}, 5*time.Second, 100*time.Millisecond).Should(BeTrue(), "hypervisor status should reflect failed libvirt connection")
+
+			// Cancel the context to stop the Start method
+			cancel()
+
+			// Wait for Start to return with context cancellation error
+			select {
+			case err := <-done:
+				Expect(err).To(HaveOccurred())
+				Expect(err.Error()).To(ContainSubstring("context done while trying to connect to libvirt"))
+			case <-time.After(2 * time.Second):
+				Fail("timeout waiting for Start to return after context cancellation")
+			}
+		})
+
+		It("should fail when hypervisor resource does not exist", func() {
+			ctx := context.Background()
+
+			// Set hostname to a non-existent hypervisor
+			originalHostname := sys.Hostname
+			sys.Hostname = "non-existent-hypervisor"
+			defer func() {
+				sys.Hostname = originalHostname
+			}()
+
+			controllerReconciler := &HypervisorReconciler{
+				Client: k8sClient,
+				Scheme: k8sClient.Scheme(),
+				Libvirt: &libvirt.InterfaceMock{
+					ConnectFunc: func() error {
+						return nil
+					},
+				},
 				reconcileCh: make(chan event.GenericEvent, 1),
 			}
 
 			err := controllerReconciler.Start(ctx)
 			Expect(err).To(HaveOccurred())
-			Expect(err.Error()).To(ContainSubstring("connection failed"))
+			Expect(err.Error()).To(ContainSubstring("unable to get hypervisor"))
+		})
+
+		It("should retry libvirt connection and succeed after initial failures", func() {
+			ctx := context.Background()
+
+			// Create a hypervisor resource for this test
+			hypervisorName := "start-retry-test-hypervisor"
+			originalHostname := sys.Hostname
+			sys.Hostname = hypervisorName
+			defer func() {
+				sys.Hostname = originalHostname
+			}()
+
+			hypervisor := &kvmv1.Hypervisor{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: hypervisorName,
+				},
+			}
+			Expect(k8sClient.Create(ctx, hypervisor)).To(Succeed())
+			defer func() {
+				err := k8sClient.Delete(ctx, hypervisor)
+				Expect(err).NotTo(HaveOccurred())
+			}()
+
+			// Track connection attempts
+			connectAttempts := 0
+			eventCallbackCalled := false
+
+			controllerReconciler := &HypervisorReconciler{
+				Client: k8sClient,
+				Scheme: k8sClient.Scheme(),
+				Libvirt: &libvirt.InterfaceMock{
+					ConnectFunc: func() error {
+						connectAttempts++
+						// Fail first 2 attempts, succeed on 3rd
+						if connectAttempts < 3 {
+							return errors.New("connection failed")
+						}
+						return nil
+					},
+					WatchDomainChangesFunc: func(eventId golibvirt.DomainEventID, handlerId string, handler func(context.Context, any)) {
+						eventCallbackCalled = true
+					},
+				},
+				reconcileCh:            make(chan event.GenericEvent, 1),
+				libvirtConnectInterval: 10 * time.Millisecond, // Use short interval for fast test
+			}
+
+			err := controllerReconciler.Start(ctx)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(connectAttempts).To(Equal(3))
+			Expect(eventCallbackCalled).To(BeTrue())
 		})
 	})
 
