ApplyAggregatesAndUpdateStatus for uniform status handling

Three controllers were calling ApplyAggregates, but only one
was maintaining the status. Unify that, so we always an accurate
reflection on the aggregates there.

Author: fwiesel

internal/controller/aggregates_controller.go

@@ -20,7 +20,6 @@ package controller
 import (
 	"context"
 	"errors"
-	"fmt"
 	"slices"
 
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -68,17 +67,15 @@ func (ac *AggregatesController) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{}, nil
 	}
 
-	err := openstack.ApplyAggregates(ctx, ac.computeClient, hv.Name, hv.Spec.Aggregates)
+	base := hv.DeepCopy()
+	err := ApplyAggregatesAndUpdateStatus(ctx, ac.computeClient, hv, hv.Spec.Aggregates)
 	if err != nil {
-		err = fmt.Errorf("failed to apply aggregates: %w", err)
 		if err2 := ac.setErrorCondition(ctx, hv, err.Error()); err2 != nil {
 			return ctrl.Result{}, errors.Join(err, err2)
 		}
 		return ctrl.Result{}, err
 	}
 
-	base := hv.DeepCopy()
-	hv.Status.Aggregates = hv.Spec.Aggregates
 	meta.SetStatusCondition(&hv.Status.Conditions, metav1.Condition{
 		Type:    kvmv1.ConditionTypeAggregatesUpdated,
 		Status:  metav1.ConditionTrue,

internal/controller/decomission_controller.go

@@ -26,6 +26,7 @@ import (
 	"github.com/gophercloud/gophercloud/v2"
 	"github.com/gophercloud/gophercloud/v2/openstack/compute/v2/services"
 	"github.com/gophercloud/gophercloud/v2/openstack/placement/v1/resourceproviders"
+	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -113,13 +114,19 @@ func (r *NodeDecommissionReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		return r.setDecommissioningCondition(ctx, hv, msg)
 	}
 
-	// Before removing the service, first take the node out of all aggregates,
-	// so when the node comes back, it doesn't end up with the old associations
-	host := hv.Name
-	if err := openstack.ApplyAggregates(ctx, r.computeClient, host, []string{}); err != nil {
+	base := hv.DeepCopy()
+	if err := ApplyAggregatesAndUpdateStatus(ctx, r.computeClient, hv, []string{}); err != nil {
 		return r.setDecommissioningCondition(ctx, hv, fmt.Sprintf("failed to remove host from aggregates: %v", err))
 	}
 
+	if !equality.Semantic.DeepEqual(base.Status, hv.Status) {
+		if err := r.Status().Patch(ctx, hv, k8sclient.MergeFromWithOptions(base,
+			k8sclient.MergeFromWithOptimisticLock{}), k8sclient.FieldOwner(DecommissionControllerName)); err != nil {
+			return r.setDecommissioningCondition(ctx, hv, fmt.Sprintf("failed to update status after removing aggregates: %v", err))
+		}
+		return ctrl.Result{}, nil
+	}
+
 	// Deleting and evicted, so better delete the service
 	err = services.Delete(ctx, r.computeClient, hypervisor.Service.ID).ExtractErr()
 	if err != nil && !gophercloud.ResponseCodeIs(err, http.StatusNotFound) {

internal/controller/decomission_controller_test.go

@@ -249,6 +249,25 @@ var _ = Describe("Decommission Controller", func() {
 						),
 					))
 				})
+
+				It("should clear Status.Aggregates when removing from all aggregates", func(ctx SpecContext) {
+					By("Setting initial aggregates in status")
+					hypervisor := &kvmv1.Hypervisor{}
+					Expect(k8sClient.Get(ctx, resourceName, hypervisor)).To(Succeed())
+					hypervisor.Status.Aggregates = []string{"zone-a", "test-aggregate"}
+					hypervisor.Status.ServiceID = serviceId
+					Expect(k8sClient.Status().Update(ctx, hypervisor)).To(Succeed())
+
+					By("Reconciling to trigger aggregate removal")
+					for range 3 {
+						_, err := decommissionReconciler.Reconcile(ctx, reconcileReq)
+						Expect(err).NotTo(HaveOccurred())
+					}
+
+					By("Verifying Status.Aggregates is empty")
+					Expect(k8sClient.Get(ctx, resourceName, hypervisor)).To(Succeed())
+					Expect(hypervisor.Status.Aggregates).To(BeEmpty(), "Status.Aggregates should be cleared after decommissioning")
+				})
 			})
 		})
 	})

internal/controller/onboarding_controller.go

@@ -126,7 +126,7 @@ func (r *OnboardingController) Reconcile(ctx context.Context, req ctrl.Request)
 
 	switch status.Reason {
 	case kvmv1.ConditionReasonInitial:
-		return ctrl.Result{}, r.initialOnboarding(ctx, hv, computeHost)
+		return ctrl.Result{}, r.initialOnboarding(ctx, hv)
 	case kvmv1.ConditionReasonTesting:
 		if hv.Spec.SkipTests {
 			return r.completeOnboarding(ctx, computeHost, hv)
@@ -190,16 +190,17 @@ func (r *OnboardingController) abortOnboarding(ctx context.Context, hv *kvmv1.Hy
 	return r.patchStatus(ctx, hv, base)
 }
 
-func (r *OnboardingController) initialOnboarding(ctx context.Context, hv *kvmv1.Hypervisor, host string) error {
+func (r *OnboardingController) initialOnboarding(ctx context.Context, hv *kvmv1.Hypervisor) error {
 	zone, found := hv.Labels[corev1.LabelTopologyZone]
 	if !found || zone == "" {
 		return fmt.Errorf("cannot find availability-zone label %v on node", corev1.LabelTopologyZone)
 	}
 
 	// Apply the desired aggregate state (zone and test aggregate only)
 	desiredAggregates := []string{zone, testAggregateName}
-	if err := openstack.ApplyAggregates(ctx, r.computeClient, host, desiredAggregates); err != nil {
-		return fmt.Errorf("failed to apply aggregates: %w", err)
+	base := hv.DeepCopy()
+	if err := ApplyAggregatesAndUpdateStatus(ctx, r.computeClient, hv, desiredAggregates); err != nil {
+		return err
 	}
 
 	// The service may be forced down previously due to an HA event,
@@ -213,7 +214,6 @@ func (r *OnboardingController) initialOnboarding(ctx context.Context, hv *kvmv1.
 		return result.Err
 	}
 
-	base := hv.DeepCopy()
 	meta.SetStatusCondition(&hv.Status.Conditions, metav1.Condition{
 		Type:    kvmv1.ConditionTypeOnboarding,
 		Status:  metav1.ConditionTrue,
@@ -327,8 +327,9 @@ func (r *OnboardingController) completeOnboarding(ctx context.Context, host stri
 
 	// Remove host from test aggregate by only keeping the zone aggregate
 	desiredAggregates := []string{zone}
-	if err := openstack.ApplyAggregates(ctx, r.computeClient, host, desiredAggregates); err != nil {
-		return ctrl.Result{}, fmt.Errorf("failed to apply aggregates: %w", err)
+	base := hv.DeepCopy()
+	if err := ApplyAggregatesAndUpdateStatus(ctx, r.computeClient, hv, desiredAggregates); err != nil {
+		return ctrl.Result{}, err
 	}
 	log.Info("removed from test-aggregate", "name", testAggregateName)
 
@@ -338,7 +339,6 @@ func (r *OnboardingController) completeOnboarding(ctx context.Context, host stri
 		return ctrl.Result{}, err
 	}
 
-	base := hv.DeepCopy()
 	// Set hypervisor ready condition
 	meta.SetStatusCondition(&hv.Status.Conditions, metav1.Condition{
 		Type:    kvmv1.ConditionTypeReady,

internal/controller/utils.go

@@ -19,6 +19,7 @@ package controller
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -28,11 +29,15 @@ import (
 	"strings"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	v1ac "k8s.io/client-go/applyconfigurations/meta/v1"
 
+	"github.com/gophercloud/gophercloud/v2"
+
 	kvmv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1"
+	"github.com/cobaltcore-dev/openstack-hypervisor-operator/internal/openstack"
 )
 
 func InstanceHaUrl(region, zone, hostname string) string {
@@ -146,3 +151,26 @@ func HasKubectlManagedFields(object *metav1.ObjectMeta) bool {
 	}
 	return false
 }
+
+// ApplyAggregatesAndUpdateStatus applies the desired aggregates to the host in OpenStack
+// and updates hv.Status.Aggregates in memory. It removes the ConditionTypeAggregatesUpdated
+// condition before the OpenStack call, ensuring the status reflects an unknown state if the
+// call fails. The aggregates controller will set the condition fresh on its next reconcile.
+// The caller is responsible for patching the status afterwards.
+func ApplyAggregatesAndUpdateStatus(
+	ctx context.Context,
+	computeClient *gophercloud.ServiceClient,
+	hv *kvmv1.Hypervisor,
+	desiredAggregates []string,
+) error {
+	// Remove condition before OpenStack call to reflect unknown state on failure
+	meta.RemoveStatusCondition(&hv.Status.Conditions, kvmv1.ConditionTypeAggregatesUpdated)
+
+	if err := openstack.ApplyAggregates(ctx, computeClient, hv.Name, desiredAggregates); err != nil {
+		return fmt.Errorf("failed to apply aggregates: %w", err)
+	}
+
+	hv.Status.Aggregates = desiredAggregates
+
+	return nil
+}

internal/controller/utils_test.go

@@ -18,8 +18,18 @@ limitations under the License.
 package controller
 
 import (
+	"fmt"
+	"net/http"
+
+	"github.com/gophercloud/gophercloud/v2/testhelper"
+	"github.com/gophercloud/gophercloud/v2/testhelper/client"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+
+	kvmv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1"
 )
 
 var _ = Describe("Utils", func() {
@@ -35,4 +45,157 @@ var _ = Describe("Utils", func() {
 			Expect(difference).To(Equal(expectedDifference))
 		})
 	})
+
+	Context("When using ApplyAggregatesAndUpdateStatus", func() {
+		const (
+			hypervisorName = "test-hypervisor"
+			serviceId      = "service-1234"
+			EOF            = "EOF"
+		)
+
+		var (
+			fakeServer          testhelper.FakeServer
+			resourceName        = types.NamespacedName{Name: hypervisorName}
+			AggregateListWithHv = `
+{
+    "aggregates": [
+		{
+			"name": "test-aggregate2",
+			"availability_zone": "",
+			"deleted": false,
+			"id": 100001,
+			"hosts": ["test-hypervisor"]
+		}
+    ]
+}
+`
+			AggregateRemoveHostBody = `
+{
+        "aggregate": {
+			"name": "test-aggregate2",
+			"availability_zone": "",
+			"deleted": false,
+			"id": 100001
+		}
+}`
+		)
+
+		BeforeEach(func(ctx SpecContext) {
+			By("Setting up the OpenStack http mock server")
+			fakeServer = testhelper.SetupHTTP()
+			DeferCleanup(fakeServer.Teardown)
+
+			// Install default handler to fail unhandled requests
+			fakeServer.Mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+				Fail("Unhandled request to fake server: " + r.Method + " " + r.URL.Path)
+			})
+
+			By("Creating the hypervisor resource")
+			hypervisor := &kvmv1.Hypervisor{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: resourceName.Name,
+				},
+				Spec: kvmv1.HypervisorSpec{
+					LifecycleEnabled: true,
+				},
+			}
+			Expect(k8sClient.Create(ctx, hypervisor)).To(Succeed())
+			DeferCleanup(func(ctx SpecContext) {
+				Expect(k8sClient.Delete(ctx, hypervisor)).To(Succeed())
+			})
+		})
+
+		Context("when removing from aggregates successfully", func() {
+			BeforeEach(func() {
+				By("Mocking OpenStack API endpoints for successful aggregate removal")
+				fakeServer.Mux.HandleFunc("GET /os-aggregates", func(w http.ResponseWriter, r *http.Request) {
+					w.Header().Add("Content-Type", "application/json")
+					w.WriteHeader(http.StatusOK)
+					_, err := fmt.Fprint(w, AggregateListWithHv)
+					Expect(err).NotTo(HaveOccurred())
+				})
+
+				fakeServer.Mux.HandleFunc("POST /os-aggregates/100001/action", func(w http.ResponseWriter, r *http.Request) {
+					Expect(r.Header.Get("Content-Type")).To(Equal("application/json"))
+					expectedBody := `{"remove_host":{"host":"test-hypervisor"}}`
+					body := make([]byte, r.ContentLength)
+					_, err := r.Body.Read(body)
+					Expect(err == nil || err.Error() == EOF).To(BeTrue())
+					Expect(string(body)).To(MatchJSON(expectedBody))
+
+					w.Header().Add("Content-Type", "application/json")
+					w.WriteHeader(http.StatusOK)
+					_, err = fmt.Fprint(w, AggregateRemoveHostBody)
+					Expect(err).NotTo(HaveOccurred())
+				})
+			})
+
+			It("should remove AggregatesUpdated condition and update Status.Aggregates", func(ctx SpecContext) {
+				By("Setting initial aggregates and AggregatesUpdated condition in status")
+				hypervisor := &kvmv1.Hypervisor{}
+				Expect(k8sClient.Get(ctx, resourceName, hypervisor)).To(Succeed())
+				hypervisor.Status.Aggregates = []string{"zone-a", "test-aggregate"}
+				hypervisor.Status.ServiceID = serviceId
+				meta.SetStatusCondition(&hypervisor.Status.Conditions, metav1.Condition{
+					Type:    kvmv1.ConditionTypeAggregatesUpdated,
+					Status:  metav1.ConditionTrue,
+					Reason:  "TestReason",
+					Message: "Test message",
+				})
+				Expect(k8sClient.Status().Update(ctx, hypervisor)).To(Succeed())
+
+				By("Verifying the condition exists before applying aggregates")
+				Expect(k8sClient.Get(ctx, resourceName, hypervisor)).To(Succeed())
+				Expect(meta.FindStatusCondition(hypervisor.Status.Conditions, kvmv1.ConditionTypeAggregatesUpdated)).NotTo(BeNil(), "AggregatesUpdated condition should exist before applying aggregates")
+
+				By("Calling ApplyAggregatesAndUpdateStatus to remove from all aggregates")
+				computeClient := client.ServiceClient(fakeServer)
+				err := ApplyAggregatesAndUpdateStatus(ctx, computeClient, hypervisor, []string{})
+				Expect(err).NotTo(HaveOccurred())
+
+				By("Verifying Status.Aggregates is empty")
+				Expect(hypervisor.Status.Aggregates).To(BeEmpty(), "Status.Aggregates should be cleared")
+
+				By("Verifying the AggregatesUpdated condition was removed")
+				Expect(meta.FindStatusCondition(hypervisor.Status.Conditions, kvmv1.ConditionTypeAggregatesUpdated)).To(BeNil(), "AggregatesUpdated condition should be removed")
+			})
+		})
+
+		Context("when ApplyAggregates fails", func() {
+			BeforeEach(func() {
+				By("Mocking OpenStack API to fail")
+				fakeServer.Mux.HandleFunc("GET /os-aggregates", func(w http.ResponseWriter, r *http.Request) {
+					w.WriteHeader(http.StatusInternalServerError)
+					fmt.Fprint(w, `{"error": "Internal Server Error"}`)
+				})
+			})
+
+			It("should remove AggregatesUpdated condition even when the API call fails", func(ctx SpecContext) {
+				By("Setting initial aggregates and AggregatesUpdated condition in status")
+				hypervisor := &kvmv1.Hypervisor{}
+				Expect(k8sClient.Get(ctx, resourceName, hypervisor)).To(Succeed())
+				hypervisor.Status.Aggregates = []string{"zone-a", "test-aggregate"}
+				hypervisor.Status.ServiceID = serviceId
+				meta.SetStatusCondition(&hypervisor.Status.Conditions, metav1.Condition{
+					Type:    kvmv1.ConditionTypeAggregatesUpdated,
+					Status:  metav1.ConditionTrue,
+					Reason:  "TestReason",
+					Message: "Test message",
+				})
+				Expect(k8sClient.Status().Update(ctx, hypervisor)).To(Succeed())
+
+				By("Verifying the condition exists before applying aggregates")
+				Expect(k8sClient.Get(ctx, resourceName, hypervisor)).To(Succeed())
+				Expect(meta.FindStatusCondition(hypervisor.Status.Conditions, kvmv1.ConditionTypeAggregatesUpdated)).NotTo(BeNil(), "AggregatesUpdated condition should exist before applying aggregates")
+
+				By("Calling ApplyAggregatesAndUpdateStatus which will fail")
+				computeClient := client.ServiceClient(fakeServer)
+				err := ApplyAggregatesAndUpdateStatus(ctx, computeClient, hypervisor, []string{})
+				Expect(err).To(HaveOccurred(), "ApplyAggregatesAndUpdateStatus should return an error")
+
+				By("Verifying the AggregatesUpdated condition was still removed despite the error")
+				Expect(meta.FindStatusCondition(hypervisor.Status.Conditions, kvmv1.ConditionTypeAggregatesUpdated)).To(BeNil(), "AggregatesUpdated condition should be removed even on failure, leaving it in unknown state")
+			})
+		})
+	})
 })