Extract aggregate logic to ApplyAggregates

Almost the same logic code was used in the aggregates and the onboarding
controller, with the variation that the onboarding controller was also
creating the zone aggregate, and the aggregates controller kept
aggregates modified outside of the controller untouched.

Let's drop that functionality and expect that all aggregates are somehow
created externally (i.e. by another controller), and then we can
combine that logic in one function, that also can be called by the
offboarding controller.

Cortex wants to rely on the aggregates in the status, so we keeping
extra aggregates is unsupported either way.

Author: fwiesel

internal/controller/aggregates_controller.go

@@ -52,7 +52,6 @@ type AggregatesController struct {
 // +kubebuilder:rbac:groups=kvm.cloud.sap,resources=hypervisors/status,verbs=get;list;watch;create;update;patch;delete
 
 func (ac *AggregatesController) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	log := logger.FromContext(ctx)
 	hv := &kvmv1.Hypervisor{}
 	if err := ac.Get(ctx, req.NamespacedName, hv); err != nil {
 		return ctrl.Result{}, k8sclient.IgnoreNotFound(err)
@@ -69,44 +68,9 @@ func (ac *AggregatesController) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{}, nil
 	}
 
-	aggs, err := openstack.GetAggregatesByName(ctx, ac.computeClient)
+	err := openstack.ApplyAggregates(ctx, ac.computeClient, hv.Name, hv.Spec.Aggregates)
 	if err != nil {
-		err = fmt.Errorf("failed listing aggregates: %w", err)
-		if err2 := ac.setErrorCondition(ctx, hv, err.Error()); err2 != nil {
-			return ctrl.Result{}, errors.Join(err, err2)
-		}
-		return ctrl.Result{}, err
-	}
-
-	toAdd := Difference(hv.Status.Aggregates, hv.Spec.Aggregates)
-	toRemove := Difference(hv.Spec.Aggregates, hv.Status.Aggregates)
-
-	// We need to add first the host to the aggregates, because if we first drop
-	// an aggregate with a filter criterion and then add a new one, we leave the host
-	// open for period of time. Still, this may fail due to a conflict of aggregates
-	// with different availability zones, so we collect all the errors and return them
-	// so it hopefully will converge eventually.
-	var errs []error
-	if len(toAdd) > 0 {
-		log.Info("Adding", "aggregates", toAdd)
-		for item := range slices.Values(toAdd) {
-			if err = openstack.AddToAggregate(ctx, ac.computeClient, aggs, hv.Name, item, ""); err != nil {
-				errs = append(errs, err)
-			}
-		}
-	}
-
-	if len(toRemove) > 0 {
-		log.Info("Removing", "aggregates", toRemove)
-		for item := range slices.Values(toRemove) {
-			if err = openstack.RemoveFromAggregate(ctx, ac.computeClient, aggs, hv.Name, item); err != nil {
-				errs = append(errs, err)
-			}
-		}
-	}
-
-	if errs != nil {
-		err = fmt.Errorf("encountered errors during aggregate update: %w", errors.Join(errs...))
+		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)
 		}

internal/controller/aggregates_controller_test.go

@@ -302,15 +302,15 @@ var _ = Describe("AggregatesController", func() {
 			Expect(cond.Message).To(ContainSubstring(expectedMessage))
 		}
 
-		Context("when GetAggregates fails", func() {
+		Context("when ApplyAggregates fails", func() {
 			BeforeEach(func(ctx SpecContext) {
 				By("Setting a missing aggregate")
 				hypervisor := &kvmv1.Hypervisor{}
 				Expect(k8sClient.Get(ctx, hypervisorName, hypervisor)).To(Succeed())
 				hypervisor.Spec.Aggregates = []string{"test-aggregate1"}
 				Expect(k8sClient.Update(ctx, hypervisor)).To(Succeed())
 
-				By("Mocking GetAggregates to fail")
+				By("Mocking GET /os-aggregates to fail (first API call in ApplyAggregates)")
 				fakeServer.Mux.HandleFunc("GET /os-aggregates", func(w http.ResponseWriter, r *http.Request) {
 					w.Header().Add("Content-Type", "application/json")
 					w.WriteHeader(http.StatusInternalServerError)
@@ -322,79 +322,7 @@ var _ = Describe("AggregatesController", func() {
 			It("should set error condition", func(ctx SpecContext) {
 				_, err := aggregatesController.Reconcile(ctx, reconcileRequest)
 				Expect(err).To(HaveOccurred())
-				sharedErrorConditionChecks(ctx, "failed listing aggregates")
-			})
-		})
-
-		Context("when AddToAggregate fails", func() {
-			BeforeEach(func(ctx SpecContext) {
-				By("Setting a missing aggregate")
-				hypervisor := &kvmv1.Hypervisor{}
-				Expect(k8sClient.Get(ctx, hypervisorName, hypervisor)).To(Succeed())
-				hypervisor.Spec.Aggregates = []string{"test-aggregate1"}
-				Expect(k8sClient.Update(ctx, hypervisor)).To(Succeed())
-
-				By("Mocking GetAggregates")
-				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, AggregateListBodyEmpty)
-					Expect(err).NotTo(HaveOccurred())
-				})
-
-				By("Mocking CreateAggregate")
-				fakeServer.Mux.HandleFunc("POST /os-aggregates", func(w http.ResponseWriter, r *http.Request) {
-					w.Header().Add("Content-Type", "application/json")
-					w.WriteHeader(http.StatusOK)
-					_, err := fmt.Fprint(w, AggregatesPostBody)
-					Expect(err).NotTo(HaveOccurred())
-				})
-
-				By("Mocking AddHost to fail")
-				fakeServer.Mux.HandleFunc("POST /os-aggregates/42/action", func(w http.ResponseWriter, r *http.Request) {
-					w.Header().Add("Content-Type", "application/json")
-					w.WriteHeader(http.StatusConflict)
-					_, err := fmt.Fprint(w, `{"conflictingRequest": {"message": "Cannot add host to aggregate", "code": 409}}`)
-					Expect(err).NotTo(HaveOccurred())
-				})
-			})
-
-			It("should set error condition", func(ctx SpecContext) {
-				_, err := aggregatesController.Reconcile(ctx, reconcileRequest)
-				Expect(err).To(HaveOccurred())
-				sharedErrorConditionChecks(ctx, "encountered errors during aggregate update")
-			})
-		})
-
-		Context("when RemoveFromAggregate fails", func() {
-			BeforeEach(func(ctx SpecContext) {
-				By("Setting existing aggregate in status")
-				hypervisor := &kvmv1.Hypervisor{}
-				Expect(k8sClient.Get(ctx, hypervisorName, hypervisor)).To(Succeed())
-				hypervisor.Status.Aggregates = []string{"test-aggregate2"}
-				Expect(k8sClient.Status().Update(ctx, hypervisor)).To(Succeed())
-
-				By("Mocking GetAggregates")
-				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, AggregateListBodyFull)
-					Expect(err).NotTo(HaveOccurred())
-				})
-
-				By("Mocking RemoveHost to fail")
-				fakeServer.Mux.HandleFunc("POST /os-aggregates/100001/action", func(w http.ResponseWriter, r *http.Request) {
-					w.Header().Add("Content-Type", "application/json")
-					w.WriteHeader(http.StatusConflict)
-					_, err := fmt.Fprint(w, `{"conflictingRequest": {"message": "Cannot remove host from aggregate", "code": 409}}`)
-					Expect(err).NotTo(HaveOccurred())
-				})
-			})
-
-			It("should set error condition", func(ctx SpecContext) {
-				_, err := aggregatesController.Reconcile(ctx, reconcileRequest)
-				Expect(err).To(HaveOccurred())
-				sharedErrorConditionChecks(ctx, "encountered errors during aggregate update")
+				sharedErrorConditionChecks(ctx, "failed to get aggregates")
 			})
 		})
 

internal/controller/decomission_controller.go

@@ -22,10 +22,8 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
-	"slices"
 
 	"github.com/gophercloud/gophercloud/v2"
-	"github.com/gophercloud/gophercloud/v2/openstack/compute/v2/aggregates"
 	"github.com/gophercloud/gophercloud/v2/openstack/compute/v2/services"
 	"github.com/gophercloud/gophercloud/v2/openstack/placement/v1/resourceproviders"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -115,22 +113,11 @@ 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 the aggregates,
-	// so when the node comes back, it doesn't up with the old associations
-	aggs, err := openstack.GetAggregatesByName(ctx, r.computeClient)
-	if err != nil {
-		return r.setDecommissioningCondition(ctx, hv, fmt.Sprintf("cannot list aggregates due to %v", err))
-	}
-
+	// 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
-	for name, aggregate := range aggs {
-		if slices.Contains(aggregate.Hosts, host) {
-			opts := aggregates.RemoveHostOpts{Host: host}
-			if err = aggregates.RemoveHost(ctx, r.computeClient, aggregate.ID, opts).Err; err != nil {
-				msg := fmt.Sprintf("failed to remove host %v from aggregate %v due to %v", name, host, err)
-				return r.setDecommissioningCondition(ctx, hv, msg)
-			}
-		}
+	if err := openstack.ApplyAggregates(ctx, r.computeClient, host, []string{}); err != nil {
+		return r.setDecommissioningCondition(ctx, hv, fmt.Sprintf("failed to remove host from aggregates: %v", err))
 	}
 
 	// Deleting and evicted, so better delete the service

internal/controller/decomission_controller_test.go

@@ -404,7 +404,7 @@ var _ = Describe("Decommission Controller", func() {
 			})
 		})
 
-		Context("When removing host from aggregate fails", func() {
+		Context("When ApplyAggregates fails", func() {
 			BeforeEach(func() {
 				fakeServer.Mux.HandleFunc("GET /os-hypervisors/detail", func(w http.ResponseWriter, r *http.Request) {
 					w.Header().Add("Content-Type", "application/json")
@@ -425,37 +425,11 @@ var _ = Describe("Decommission Controller", func() {
 					}`)
 				})
 
+				// Mock only the first API call in ApplyAggregates (GET /os-aggregates) to fail
 				fakeServer.Mux.HandleFunc("GET /os-aggregates", func(w http.ResponseWriter, r *http.Request) {
-					w.Header().Add("Content-Type", "application/json")
-					w.WriteHeader(http.StatusOK)
-					fmt.Fprint(w, AggregateListWithHv)
-				})
-
-				fakeServer.Mux.HandleFunc("POST /os-aggregates/100001/action", func(w http.ResponseWriter, r *http.Request) {
 					w.WriteHeader(http.StatusInternalServerError)
 					fmt.Fprint(w, `{"error": "Internal Server Error"}`)
 				})
-
-				// Add handlers for subsequent steps even though we expect failure earlier
-				fakeServer.Mux.HandleFunc("DELETE /os-services/service-1234", func(w http.ResponseWriter, r *http.Request) {
-					w.WriteHeader(http.StatusNoContent)
-				})
-
-				fakeServer.Mux.HandleFunc("GET /resource_providers/c48f6247-abe4-4a24-824e-ea39e108874f", func(w http.ResponseWriter, r *http.Request) {
-					w.Header().Add("Content-Type", "application/json")
-					w.WriteHeader(http.StatusOK)
-					fmt.Fprint(w, `{"uuid": "rp-uuid", "name": "hv-test"}`)
-				})
-
-				fakeServer.Mux.HandleFunc("GET /resource_providers/rp-uuid/allocations", func(w http.ResponseWriter, r *http.Request) {
-					w.Header().Add("Content-Type", "application/json")
-					w.WriteHeader(http.StatusOK)
-					fmt.Fprint(w, `{"allocations": {}}`)
-				})
-
-				fakeServer.Mux.HandleFunc("DELETE /resource_providers/rp-uuid", func(w http.ResponseWriter, r *http.Request) {
-					w.WriteHeader(http.StatusAccepted)
-				})
 			})
 
 			It("should set decommissioning condition with error", func(ctx SpecContext) {

internal/controller/onboarding_controller.go

@@ -22,7 +22,6 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
-	"slices"
 	"strings"
 	"time"
 
@@ -197,34 +196,10 @@ func (r *OnboardingController) initialOnboarding(ctx context.Context, hv *kvmv1.
 		return fmt.Errorf("cannot find availability-zone label %v on node", corev1.LabelTopologyZone)
 	}
 
-	aggs, err := openstack.GetAggregatesByName(ctx, r.computeClient)
-	if err != nil {
-		return fmt.Errorf("cannot list aggregates %w", err)
-	}
-
-	if err = openstack.AddToAggregate(ctx, r.computeClient, aggs, host, zone, zone); err != nil {
-		return fmt.Errorf("failed to agg to availability-zone aggregate %w", err)
-	}
-
-	err = openstack.AddToAggregate(ctx, r.computeClient, aggs, host, testAggregateName, "")
-	if err != nil {
-		return fmt.Errorf("failed to agg to test aggregate %w", err)
-	}
-
-	var errs []error
-	for aggregateName, aggregate := range aggs {
-		if aggregateName == testAggregateName || aggregateName == zone {
-			continue
-		}
-		if slices.Contains(aggregate.Hosts, host) {
-			if err := openstack.RemoveFromAggregate(ctx, r.computeClient, aggs, host, aggregateName); err != nil {
-				errs = append(errs, err)
-			}
-		}
-	}
-
-	if len(errs) > 0 {
-		return fmt.Errorf("failed to remove host %v from aggregates due to %w", host, errors.Join(errs...))
+	// 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)
 	}
 
 	// The service may be forced down previously due to an HA event,
@@ -345,14 +320,15 @@ func (r *OnboardingController) completeOnboarding(ctx context.Context, host stri
 		return ctrl.Result{}, err
 	}
 
-	aggs, err := openstack.GetAggregatesByName(ctx, r.computeClient)
-	if err != nil {
-		return ctrl.Result{}, fmt.Errorf("failed to get aggregates %w", err)
+	zone, found := hv.Labels[corev1.LabelTopologyZone]
+	if !found || zone == "" {
+		return ctrl.Result{}, fmt.Errorf("cannot find availability-zone label %v on node", corev1.LabelTopologyZone)
 	}
 
-	err = openstack.RemoveFromAggregate(ctx, r.computeClient, aggs, host, testAggregateName)
-	if err != nil {
-		return ctrl.Result{}, fmt.Errorf("failed to remove from test aggregate %w", err)
+	// 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)
 	}
 	log.Info("removed from test-aggregate", "name", testAggregateName)
 

internal/controller/onboarding_controller_test.go

@@ -455,6 +455,8 @@ var _ = Describe("Onboarding Controller", func() {
 			})
 			Expect(k8sClient.Status().Update(ctx, hv)).To(Succeed())
 
+			// Mock for ApplyAggregates during completeOnboarding
+			// Returns aggregates with host in both test-az and tenant_filter_tests
 			fakeServer.Mux.HandleFunc("GET /os-aggregates", func(w http.ResponseWriter, r *http.Request) {
 				w.Header().Add("Content-Type", "application/json")
 				w.WriteHeader(http.StatusOK)
@@ -490,10 +492,20 @@ var _ = Describe("Onboarding Controller", func() {
 				Expect(err).NotTo(HaveOccurred())
 			})
 
+			// Mock for removing host from test aggregate (ApplyAggregates will call this)
 			fakeServer.Mux.HandleFunc("POST /os-aggregates/99/action", func(w http.ResponseWriter, r *http.Request) {
 				w.Header().Add("Content-Type", "application/json")
 				w.WriteHeader(http.StatusOK)
-				_, err := fmt.Fprint(w, addedHostToTestBody)
+				// Return aggregate without the host after removal
+				_, err := fmt.Fprint(w, `{
+					"aggregate": {
+						"name": "tenant_filter_tests",
+						"availability_zone": "",
+						"deleted": false,
+						"hosts": [],
+						"id": 99
+					}
+				}`)
 				Expect(err).NotTo(HaveOccurred())
 			})