chore: throw errors out of nodeclaim informer loop (#2749)

Author: DerekFrank

pkg/controllers/disruption/consolidation_test.go

@@ -1066,6 +1066,7 @@ var _ = Describe("Consolidation", func() {
 				corev1.LabelTopologyZone:       mostExpSpotOffering.Requirements.Get(corev1.LabelTopologyZone).Any(),
 			})
 
+			ExpectSingletonReconciled(ctx, pricingController)
 			rs := test.ReplicaSet()
 			ExpectApplied(ctx, env.Client, rs)
 			Expect(env.Client.Get(ctx, client.ObjectKeyFromObject(rs), rs)).To(Succeed())
@@ -1180,6 +1181,7 @@ var _ = Describe("Consolidation", func() {
 				v1.CapacityTypeLabelKey:        spotOffering.Requirements.Get(v1.CapacityTypeLabelKey).Any(),
 				corev1.LabelTopologyZone:       spotOffering.Requirements.Get(corev1.LabelTopologyZone).Any(),
 			})
+			ExpectSingletonReconciled(ctx, pricingController)
 
 			rs := test.ReplicaSet()
 			ExpectApplied(ctx, env.Client, rs)
@@ -1714,7 +1716,8 @@ var _ = Describe("Consolidation", func() {
 				nodeClaim = lo.Ternary(spotToSpot, spotNodeClaim, nodeClaim)
 				node = lo.Ternary(spotToSpot, spotNode, node)
 				// Add these spot instance with this special condition to cloud provider instancetypes
-				cloudProvider.InstanceTypes = lo.Ternary(spotToSpot, spotInstances, onDemandInstances)
+				cloudProvider.InstanceTypes = append(cloudProvider.InstanceTypes, lo.Ternary(spotToSpot, spotInstances, onDemandInstances)...)
+				ExpectSingletonReconciled(ctx, pricingController)
 
 				rs := test.ReplicaSet()
 				ExpectApplied(ctx, env.Client, rs)
@@ -2229,6 +2232,7 @@ var _ = Describe("Consolidation", func() {
 				currentInstance,
 				replacementInstance,
 			}
+			ExpectSingletonReconciled(ctx, pricingController)
 
 			// create our RS so we can link a pod to it
 			rs := test.ReplicaSet()
@@ -2317,6 +2321,7 @@ var _ = Describe("Consolidation", func() {
 				currentInstance,
 				replacementInstance,
 			}
+			ExpectSingletonReconciled(ctx, pricingController)
 
 			// create our RS so we can link a pod to it
 			rs := test.ReplicaSet()
@@ -2917,6 +2922,7 @@ var _ = Describe("Consolidation", func() {
 				defaultInstanceType,
 				smallInstanceType,
 			}
+			ExpectSingletonReconciled(ctx, pricingController)
 			// create our RS so we can link a pod to it
 			rs := test.ReplicaSet()
 			ExpectApplied(ctx, env.Client, rs)
@@ -4468,6 +4474,7 @@ var _ = Describe("Consolidation", func() {
 					v1alpha1.LabelReservationID: mostExpensiveReservationID,
 				}),
 			})
+			ExpectSingletonReconciled(ctx, pricingController)
 			reservedNodeClaim, reservedNode = test.NodeClaimAndNode(v1.NodeClaim{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: map[string]string{

pkg/controllers/disruption/emptiness_test.go

@@ -65,8 +65,8 @@ var _ = Describe("Emptiness", func() {
 				Labels: map[string]string{
 					v1.NodePoolLabelKey:            nodePool.Name,
 					corev1.LabelInstanceTypeStable: leastExpensiveSpotInstance.Name,
-					v1.CapacityTypeLabelKey:        leastExpensiveOffering.Requirements.Get(v1.CapacityTypeLabelKey).Any(),
-					corev1.LabelTopologyZone:       leastExpensiveOffering.Requirements.Get(corev1.LabelTopologyZone).Any(),
+					v1.CapacityTypeLabelKey:        leastExpensiveSpotOffering.Requirements.Get(v1.CapacityTypeLabelKey).Any(),
+					corev1.LabelTopologyZone:       leastExpensiveSpotOffering.Requirements.Get(corev1.LabelTopologyZone).Any(),
 				},
 			},
 			Status: v1.NodeClaimStatus{

pkg/controllers/disruption/suite_test.go

@@ -67,6 +67,7 @@ var env *test.Environment
 var clusterCost *cost.ClusterCost
 var cluster *state.Cluster
 var disruptionController *disruption.Controller
+var pricingController *informer.PricingController
 var prov *provisioning.Provisioner
 var cloudProvider *fake.CloudProvider
 var nodeStateController *informer.NodeController
@@ -95,6 +96,7 @@ var _ = BeforeSuite(func() {
 	cloudProvider = fake.NewCloudProvider()
 	fakeClock = clock.NewFakeClock(time.Now())
 	clusterCost = cost.NewClusterCost(ctx, cloudProvider, env.Client)
+	pricingController = informer.NewPricingController(env.Client, cloudProvider, clusterCost)
 	cluster = state.NewCluster(fakeClock, env.Client, cloudProvider)
 	nodeStateController = informer.NewNodeController(env.Client, cluster)
 	nodeClaimStateController = informer.NewNodeClaimController(env.Client, cloudProvider, cluster, clusterCost)
@@ -110,6 +112,8 @@ var _ = AfterSuite(func() {
 var _ = BeforeEach(func() {
 	cloudProvider.Reset()
 	cloudProvider.InstanceTypes = fake.InstanceTypesAssorted()
+	clusterCost.Reset()
+	ExpectSingletonReconciled(ctx, pricingController)
 
 	recorder.Reset() // Reset the events that we captured during the run
 
@@ -580,6 +584,7 @@ var _ = Describe("Disruption Taints", func() {
 			currentInstance,
 			replacementInstance,
 		}
+		ExpectSingletonReconciled(ctx, pricingController)
 		nodePool.Spec.Disruption.ConsolidateAfter.Duration = lo.ToPtr(time.Duration(0))
 		nodeClaim.StatusConditions().SetTrue(v1.ConditionTypeConsolidatable)
 		ExpectApplied(ctx, env.Client, nodeClaim, nodePool)
@@ -1778,8 +1783,7 @@ var _ = Describe("Candidate Filtering", func() {
 				},
 			},
 		})
-		// Don't apply the NodePool
-		ExpectApplied(ctx, env.Client, nodeClaim, node)
+		ExpectApplied(ctx, env.Client, nodePool, nodeClaim, node)
 		ExpectMakeNodesAndNodeClaimsInitializedAndStateUpdated(ctx, env.Client, nodeStateController, nodeClaimStateController, []*corev1.Node{node}, []*v1.NodeClaim{nodeClaim})
 
 		// Mock the NodePool not existing by removing it from the nodePool and nodePoolInstanceTypes maps

pkg/controllers/state/informer/nodeclaim.go

@@ -24,7 +24,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
-	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
@@ -68,7 +67,7 @@ func (c *NodeClaimController) Reconcile(ctx context.Context, req reconcile.Reque
 			// notify cluster state of the node deletion
 			c.cluster.DeleteNodeClaim(req.Name)
 			if deleteErr := c.clusterCost.DeleteNodeClaim(ctx, nodeClaim); deleteErr != nil {
-				log.FromContext(ctx).Error(deleteErr, "failed to remove nodeclaim from cost tracking")
+				return reconcile.Result{}, deleteErr
 			}
 		}
 		return reconcile.Result{}, client.IgnoreNotFound(err)
@@ -78,7 +77,7 @@ func (c *NodeClaimController) Reconcile(ctx context.Context, req reconcile.Reque
 	}
 	c.cluster.UpdateNodeClaim(nodeClaim)
 	if err := c.clusterCost.UpdateNodeClaim(ctx, nodeClaim); err != nil {
-		log.FromContext(ctx).Error(err, "failed to process nodeclaim for cost tracking")
+		return reconcile.Result{}, err
 	}
 	// ensure it's aware of any nodes we discover, this is a no-op if the node is already known to our cluster state
 	return reconcile.Result{RequeueAfter: stateRetryPeriod}, nil

pkg/controllers/state/informer/pricing.go

@@ -26,6 +26,8 @@ import (
 	"github.com/samber/lo"
 	"go.uber.org/multierr"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	controllerruntime "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -61,6 +63,9 @@ func (c *PricingController) Reconcile(ctx context.Context) (reconciler.Result, e
 		return reconciler.Result{}, err
 	}
 
+	// Add a fake wide open nodepool to capture manually created nodeclaims
+	npl.Items = append(npl.Items, v1.NodePool{ObjectMeta: metav1.ObjectMeta{UID: types.UID("manual")}})
+
 	newNpItMap := make(map[string]map[string]*cloudprovider.InstanceType)
 	var errs error
 	for _, np := range npl.Items {
@@ -83,7 +88,7 @@ func (c *PricingController) Reconcile(ctx context.Context) (reconciler.Result, e
 		})
 	}
 	if errs != nil {
-		return reconciler.Result{}, fmt.Errorf("refreshing pricing info, %w", err)
+		return reconciler.Result{}, fmt.Errorf("refreshing pricing info, %w", errs)
 	}
 	c.npItMap = newNpItMap
 

pkg/state/cost/cost.go

@@ -22,12 +22,11 @@ import (
 	"sync"
 
 	opmetrics "github.com/awslabs/operatorpkg/metrics"
-	"github.com/awslabs/operatorpkg/object"
 	"github.com/awslabs/operatorpkg/serrors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/samber/lo"
 	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
@@ -216,19 +215,18 @@ func (cc *ClusterCost) UpdateNodeClaim(ctx context.Context, nodeClaim *v1.NodeCl
 	}
 
 	np := &v1.NodePool{}
-	if err := cc.client.Get(ctx, types.NamespacedName{Name: nodeClaim.Labels[v1.NodePoolLabelKey]}, np); err != nil {
+	err := cc.client.Get(ctx, types.NamespacedName{Name: nodeClaim.Labels[v1.NodePoolLabelKey]}, np)
+	if err != nil && !errors.IsNotFound(err) {
 		failed = true
 		return serrors.Wrap(err, "nodepool", nodeClaim.Labels[v1.NodePoolLabelKey], "nodeclaim", klog.KObj(nodeClaim))
 	}
-	if _, found := lo.Find(nodeClaim.GetOwnerReferences(), func(o metav1.OwnerReference) bool {
-		return o.Kind == object.GVK(np).Kind && o.UID == np.UID
-	}); !found {
-		failed = true
-		return serrors.Wrap(fmt.Errorf("nodepool not found for nodeclaim"), "nodepool", nodeClaim.Labels[v1.NodePoolLabelKey], "nodeclaim", klog.KObj(nodeClaim))
+	if errors.IsNotFound(err) {
+		np.UID = types.UID("manual")
 	}
+
 	cc.Lock()
 	defer cc.Unlock()
-	err := cc.internalAddOffering(ctx, np, nodeClaim.Labels[corev1.LabelInstanceTypeStable], nodeClaim.Labels[v1.CapacityTypeLabelKey], nodeClaim.Labels[corev1.LabelTopologyZone], true)
+	err = cc.internalAddOffering(ctx, np, nodeClaim.Labels[corev1.LabelInstanceTypeStable], nodeClaim.Labels[v1.CapacityTypeLabelKey], nodeClaim.Labels[corev1.LabelTopologyZone], true)
 	if err != nil {
 		failed = true
 		return serrors.Wrap(err, "nodeclaim", klog.KObj(nodeClaim), "nodepool", klog.KObj(np))
@@ -267,15 +265,13 @@ func (cc *ClusterCost) DeleteNodeClaim(ctx context.Context, nodeClaim *v1.NodeCl
 	nodePoolName := nodeClaim.Labels[v1.NodePoolLabelKey]
 	np := &v1.NodePool{}
 	err := cc.client.Get(ctx, client.ObjectKey{Name: nodePoolName}, np)
-	if err != nil {
+	if err != nil && !errors.IsNotFound(err) {
 		failed = true
 		return serrors.Wrap(err, "nodepool", nodePoolName, "nodeclaim", klog.KObj(nodeClaim))
 	}
-	if _, found := lo.Find(nodeClaim.GetOwnerReferences(), func(o metav1.OwnerReference) bool {
-		return o.Kind == object.GVK(np).Kind && o.UID == np.UID
-	}); !found {
-		failed = true
-		return serrors.Wrap(fmt.Errorf("nodepool not found for nodeclaim"), "nodepool", nodeClaim.Labels[v1.NodePoolLabelKey], "nodeclaim", klog.KObj(nodeClaim))
+	if errors.IsNotFound(err) {
+		// Technically not an error, as users can create NodeClaims manually without a NodePool.
+		np.UID = types.UID("manual")
 	}
 	cc.Lock()
 	defer cc.Unlock()
@@ -363,6 +359,13 @@ func (cc *ClusterCost) internalRemoveOffering(np *v1.NodePool, instanceName, cap
 	return nil
 }
 
+func (cc *ClusterCost) Reset() {
+	cc.Lock()
+	defer cc.Unlock()
+	cc.npCostMap = make(map[types.UID]*NodePoolCost)
+	cc.nodeClaimSet = make(sets.Set[string])
+}
+
 // GetClusterCost returns the total cost of all compute resources across
 // all NodePools in the cluster.
 func (cc *ClusterCost) GetClusterCost() float64 {