refactor, optimize, don't run API calls in reconciler

Author: notandy

go.mod

@@ -11,7 +11,6 @@ require (
 	github.com/coreos/go-systemd/v22 v22.5.0
 	github.com/digitalocean/go-libvirt v0.0.0-20240916165608-bff44a349d9d
 	github.com/godbus/dbus/v5 v5.1.0
-	github.com/google/uuid v1.6.0
 	github.com/onsi/ginkgo/v2 v2.19.0
 	github.com/onsi/gomega v1.33.1
 	k8s.io/api v0.31.1
@@ -48,6 +47,7 @@ require (
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/pprof v0.0.0-20240727154555-813a5fbdbec8 // indirect
+	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0 // indirect
 	github.com/imdario/mergo v0.3.16 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect

internal/controller/hypervisor_controller.go

@@ -103,7 +103,7 @@ func (r *HypervisorReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	// libvirt
 	// ====================================================================================================
 
-	// Try (re)connect to libvirt
+	// Try (re)connect to libvirt, update status
 	if err := r.libvirt.Connect(); err != nil {
 		log.Error(err, "unable to connect to libvirt system bus")
 		meta.SetStatusCondition(&hypervisor.Status.Conditions, metav1.Condition{
@@ -121,12 +121,8 @@ func (r *HypervisorReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		})
 
 		// Update hypervisor instances
-		hypervisor.Status.Instances, err = r.libvirt.GetInstances()
-		if err != nil {
-			log.Error(err, "unable to list instances")
-			return ctrl.Result{}, err
-		}
-		hypervisor.Status.NumInstances = len(hypervisor.Status.Instances)
+		hypervisor.Status.NumInstances = r.libvirt.GetNumInstances()
+		hypervisor.Status.Instances, _ = r.libvirt.GetInstances()
 	}
 
 	// ====================================================================================================

internal/controller/hypervisor_controller_test.go

@@ -105,6 +105,9 @@ var _ = Describe("Hypervisor Controller", func() {
 					GetVersionFunc: func() string {
 						return "10.9.0"
 					},
+					GetNumInstancesFunc: func() int {
+						return 1
+					},
 				},
 				systemd: &systemd.InterfaceMock{
 					CloseFunc: func() {},

internal/libvirt/interface.go

@@ -43,4 +43,7 @@ type Interface interface {
 
 	// GetVersion returns the version of the libvirt daemon.
 	GetVersion() string
+
+	// GetNumInstances returns the number of instances.
+	GetNumInstances() int
 }

internal/libvirt/interface_mock.go

@@ -25,7 +25,6 @@ import (
 
 	"github.com/digitalocean/go-libvirt"
 	"github.com/digitalocean/go-libvirt/socket/dialers"
-	"github.com/google/uuid"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
@@ -38,6 +37,7 @@ type LibVirt struct {
 	migrationJobs map[string]context.CancelFunc
 	migrationLock sync.Mutex
 	version       string
+	domains       map[libvirt.ConnectListAllDomainsFlags][]libvirt.Domain
 }
 
 func NewLibVirt(k client.Client) *LibVirt {
@@ -52,6 +52,7 @@ func NewLibVirt(k client.Client) *LibVirt {
 		make(map[string]context.CancelFunc),
 		sync.Mutex{},
 		"N/A",
+		make(map[libvirt.ConnectListAllDomainsFlags][]libvirt.Domain, 2),
 	}
 }
 
@@ -74,6 +75,10 @@ func (l *LibVirt) Connect() error {
 		// Run the migration listener in a goroutine
 		ctx := log.IntoContext(context.Background(), log.Log.WithName("libvirt-migration-listener"))
 		go l.runMigrationListener(ctx)
+
+		// Periodic status thread
+		ctx = log.IntoContext(context.Background(), log.Log.WithName("libvirt-status-thread"))
+		go l.runStatusThread(ctx)
 	}
 
 	return err
@@ -92,12 +97,7 @@ func (l *LibVirt) GetInstances() ([]v1alpha1.Instance, error) {
 
 	flags := []libvirt.ConnectListAllDomainsFlags{libvirt.ConnectListDomainsActive, libvirt.ConnectListDomainsInactive}
 	for _, flag := range flags {
-		domains, _, err := l.virt.ConnectListAllDomains(1, flag)
-		if err != nil {
-			return nil, err
-		}
-
-		for _, domain := range domains {
+		for _, domain := range l.domains[flag] {
 			instances = append(instances, v1alpha1.Instance{
 				ID:     GetOpenstackUUID(domain),
 				Name:   domain.Name,
@@ -109,36 +109,13 @@ func (l *LibVirt) GetInstances() ([]v1alpha1.Instance, error) {
 }
 
 func (l *LibVirt) GetDomainsActive() ([]libvirt.Domain, error) {
-	domains, _, err := l.virt.ConnectListAllDomains(1, libvirt.ConnectListDomainsActive)
-	if err != nil {
-		return nil, err
-	}
-	return domains, nil
+	return l.domains[libvirt.ConnectListDomainsActive], nil
 }
 
 func (l *LibVirt) IsConnected() bool {
 	return l.virt.IsConnected()
 }
 
-func GetOpenstackUUID(domain libvirt.Domain) string {
-	u, err := uuid.FromBytes(domain.UUID[:])
-	if err != nil {
-		return ""
-	}
-
-	return u.String()
-}
-
-func ByteCountIEC(b uint64) string {
-	const unit = 1024
-	if b < unit {
-		return fmt.Sprintf("%d B", b)
-	}
-	div, exp := int64(unit), 0
-	for n := b / unit; n >= unit; n /= unit {
-		div *= unit
-		exp++
-	}
-	return fmt.Sprintf("%.1f %ciB",
-		float64(b)/float64(div), "KMGTPE"[exp])
+func (l *LibVirt) GetNumInstances() int {
+	return len(l.domains[libvirt.ConnectListDomainsActive])
 }

internal/libvirt/libvirt.go

@@ -108,14 +108,29 @@ func (l *LibVirt) runMigrationListener(ctx context.Context) {
 				switch e.Msg.Detail {
 				case int32(libvirt.DomainEventDefinedAdded):
 					log.Info("domain added")
+					// add domain to the list of inactive domains
+					l.domains[libvirt.ConnectListDomainsInactive] = append(l.domains[libvirt.ConnectListDomainsInactive], domain)
 				case int32(libvirt.DomainEventDefinedUpdated):
 					log.Info("domain updated")
 				case int32(libvirt.DomainEventDefinedRenamed):
 					log.Info("domain renamed")
 				case int32(libvirt.DomainEventDefinedFromSnapshot):
 					log.Info("domain defined from snapshot")
 				}
+			case int32(libvirt.DomainEventUndefined):
+				log.Info("domain undefined")
+				// remove domain from the list of inactive domains
+				for i, d := range l.domains[libvirt.ConnectListDomainsInactive] {
+					if d.Name == domain.Name {
+						l.domains[libvirt.ConnectListDomainsInactive] = append(
+							l.domains[libvirt.ConnectListDomainsInactive][:i],
+							l.domains[libvirt.ConnectListDomainsInactive][i+1:]...)
+						break
+					}
+				}
 			case int32(libvirt.DomainEventStarted):
+				// add domain to the list of active domains
+				l.domains[libvirt.ConnectListDomainsActive] = append(l.domains[libvirt.ConnectListDomainsActive], domain)
 				switch e.Msg.Detail {
 				case int32(libvirt.DomainEventStartedBooted):
 					log.Info("domain booted")
@@ -138,6 +153,16 @@ func (l *LibVirt) runMigrationListener(ctx context.Context) {
 				}
 			case int32(libvirt.DomainEventStopped):
 				log.Info("domain stopped")
+
+				// remove domain from the list of active domains
+				for i, d := range l.domains[libvirt.ConnectListDomainsActive] {
+					if d.Name == domain.Name {
+						l.domains[libvirt.ConnectListDomainsActive] = append(
+							l.domains[libvirt.ConnectListDomainsActive][:i],
+							l.domains[libvirt.ConnectListDomainsActive][i+1:]...)
+						break
+					}
+				}
 				l.stopMigrationWatch(ctx, domain)
 			case int32(libvirt.DomainEventShutdown):
 				log.Info("domain shutdown")
@@ -146,7 +171,6 @@ func (l *LibVirt) runMigrationListener(ctx context.Context) {
 				log.Info("domain PM suspended")
 			case int32(libvirt.DomainEventCrashed):
 				log.Info("domain crashed")
-
 			}
 
 		case <-ctx.Done():

internal/libvirt/libvirt_events.go

@@ -0,0 +1,69 @@
+/*
+SPDX-FileCopyrightText: Copyright 2025 SAP SE or an SAP affiliate company and cobaltcore-dev contributors
+SPDX-License-Identifier: Apache-2.0
+
+Licensed under the Apache License, LibVirtVersion 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package libvirt
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/digitalocean/go-libvirt"
+	logger "sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+func (l *LibVirt) updateDomains() error {
+	flags := []libvirt.ConnectListAllDomainsFlags{
+		libvirt.ConnectListDomainsActive,
+		libvirt.ConnectListDomainsInactive,
+	}
+
+	// updates all domains (active / inactive)
+	for _, flag := range flags {
+		domains, _, err := l.virt.ConnectListAllDomains(1, flag)
+		if err != nil {
+			return fmt.Errorf("flag %s: %w", fmt.Sprintf("%T", flag), err)
+		}
+
+		// update the domains
+		l.domains[flag] = domains
+	}
+	return nil
+}
+
+func (l *LibVirt) runStatusThread(ctx context.Context) {
+	log := logger.FromContext(ctx)
+	log.Info("starting status thread")
+
+	// run immediately, and every minute after
+	_ = l.updateDomains()
+
+	for {
+		select {
+		case <-time.After(1 * time.Minute):
+			if err := l.updateDomains(); err != nil {
+				log.Error(err, "failed to update domains")
+			}
+		case <-ctx.Done():
+			log.Info("shutting down status thread")
+			return
+		case <-l.virt.Disconnected():
+			log.Info("libvirt disconnected, shutting down status thread")
+			return
+		}
+	}
+}