fix: delete GPUNode when Kubernetes node no longer exists (#81)

* fix: delete GPUNode when Kubernetes node no longer exists

* fix: test

Author: 0x5457

api/v1/gpupool_types.go

@@ -363,7 +363,7 @@ type GPUPoolStatus struct {
 
 	TotalNodes    int32 `json:"totalNodes,omitempty"`
 	TotalGPUs     int32 `json:"totalGPUs,omitempty"`
-	ReadyNodes    int32 `json:"readyNodes,omitempty"`
+	ReadyNodes    int32 `json:"readyNodes"`
 	NotReadyNodes int32 `json:"notReadyNodes"`
 
 	TotalTFlops resource.Quantity `json:"totalTFlops"`

charts/tensor-fusion/crds/tensor-fusion.ai_gpupools.yaml

@@ -942,6 +942,7 @@ spec:
             - componentStatus
             - notReadyNodes
             - phase
+            - readyNodes
             - totalTFlops
             - totalVRAM
             - virtualTFlops

config/crd/bases/tensor-fusion.ai_gpupools.yaml

@@ -942,6 +942,7 @@ spec:
             - componentStatus
             - notReadyNodes
             - phase
+            - readyNodes
             - totalTFlops
             - totalVRAM
             - virtualTFlops

internal/constants/constants.go

@@ -50,7 +50,7 @@ const (
 	WorkerCudaMemLimitEnv = "TENSOR_FUSION_CUDA_MEM_LIMIT"
 	WorkerPodNameEnv      = "POD_NAME"
 	NamespaceEnv          = "OPERATOR_NAMESPACE"
-	NamespaceDefaultVal   = "tensor-fusion"
+	NamespaceDefaultVal   = "tensor-fusion-sys"
 )
 
 const (

internal/controller/gpunode_controller.go

@@ -38,11 +38,9 @@ import (
 	"k8s.io/client-go/util/retry"
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/log"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
 // GPUNodeReconciler reconciles a GPUNode object
@@ -74,7 +72,6 @@ func (r *GPUNodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	}
 
 	deleted, err := utils.HandleFinalizer(ctx, node, r.Client, func(ctx context.Context, node *tfv1.GPUNode) (bool, error) {
-
 		if node.Status.Phase != tfv1.TensorFusionGPUNodePhaseDestroying {
 			node.Status.Phase = tfv1.TensorFusionGPUNodePhaseDestroying
 			if err := r.Status().Update(ctx, node); err != nil {
@@ -136,13 +133,39 @@ func (r *GPUNodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 			break
 		}
 	}
+	if poolName == "" {
+		log.Error(nil, "failed to get pool name", "node", node.Name)
+		return ctrl.Result{}, nil
+	}
 
 	poolObj := &tfv1.GPUPool{}
 	err = r.Client.Get(ctx, client.ObjectKey{Name: poolName}, poolObj)
 	if err != nil {
 		return ctrl.Result{}, fmt.Errorf("failed to get tensor-fusion pool, can not create node discovery job, pool: %s", poolName)
 	}
 
+	if node.Spec.ManageMode != tfv1.GPUNodeManageModeProvisioned {
+		// Check if the Kubernetes node exists; if not, the GPUNode should delete itself.
+		if node.Status.KubernetesNodeName != "" {
+			// Try to get the Kubernetes node
+			coreNode := &corev1.Node{}
+			err := r.Get(ctx, client.ObjectKey{Name: node.Status.KubernetesNodeName}, coreNode)
+			if err != nil {
+				if errors.IsNotFound(err) {
+					// The Kubernetes node does not exist, delete the GPUNode
+					log.Info("Kubernetes node does not exist, deleting GPUNode",
+						"kubernetesNodeName", node.Status.KubernetesNodeName)
+					if err := r.Delete(ctx, node); err != nil {
+						return ctrl.Result{}, fmt.Errorf("failed to delete GPUNode after Kubernetes node was deleted: %w", err)
+					}
+					// Return early since we've deleted the resource
+					return ctrl.Result{}, nil
+				}
+				return ctrl.Result{}, fmt.Errorf("failed to get Kubernetes node %s: %w",
+					node.Status.KubernetesNodeName, err)
+			}
+		}
+	}
 	if err := r.reconcileCloudVendorNode(ctx, node, poolObj); err != nil {
 		return ctrl.Result{}, err
 	}
@@ -151,10 +174,6 @@ func (r *GPUNodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	if node.Status.KubernetesNodeName == "" {
 		return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
 	}
-	if poolName == "" {
-		log.Error(nil, "failed to get pool name", "node", node.Name)
-		return ctrl.Result{}, nil
-	}
 
 	if err := r.reconcileNodeDiscoveryJob(ctx, node, poolObj); err != nil {
 		return ctrl.Result{}, err
@@ -415,7 +434,6 @@ func (r *GPUNodeReconciler) reconcileHypervisorPod(ctx context.Context, node *tf
 }
 
 func (r *GPUNodeReconciler) reconcileCloudVendorNode(ctx context.Context, node *tfv1.GPUNode, pool *tfv1.GPUPool) error {
-
 	// Avoid creating duplicated cloud vendor nodes, if not working, keep pending status
 	if node.Status.NodeInfo.InstanceID != "" {
 		// node already created, check status
@@ -532,7 +550,7 @@ func (r *GPUNodeReconciler) CalculateVirtualCapacity(node *tfv1.GPUNode, pool *t
 // SetupWithManager sets up the controller with the Manager.
 func (r *GPUNodeReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
-		For(&tfv1.GPUNode{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
+		For(&tfv1.GPUNode{}).
 		Named("gpunode").
 		Owns(&corev1.Node{}).
 		Owns(&batchv1.Job{}).

internal/controller/gpunode_controller_test.go

@@ -63,6 +63,14 @@ var _ = Describe("GPUNode Controller", func() {
 				resource.Status.Phase = tfv1.TensorFusionGPUNodePhaseRunning
 				Expect(k8sClient.Status().Update(ctx, resource)).To(Succeed())
 			}
+			By("creating the core node")
+			coreNode := &corev1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: resourceName,
+				},
+				Spec: corev1.NodeSpec{},
+			}
+			Expect(k8sClient.Create(ctx, coreNode)).To(Succeed())
 		})
 
 		AfterEach(func() {
@@ -72,6 +80,11 @@ var _ = Describe("GPUNode Controller", func() {
 
 			By("Cleanup the specific resource instance GPUNode")
 			Expect(k8sClient.Delete(ctx, resource)).To(Succeed())
+			By("Cleanup the core node")
+			coreNode := &corev1.Node{ObjectMeta: metav1.ObjectMeta{
+				Name: resourceName,
+			}}
+			Expect(k8sClient.Delete(ctx, coreNode)).To(Succeed())
 		})
 
 		It("should successfully reconcile the resource", func() {

internal/controller/gpupool_controller.go

@@ -55,6 +55,7 @@ func (r *GPUPoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 
 	runNow, alreadyQueued, waitTime := utils.DebouncedReconcileCheck(ctx, &r.LastProcessedItems, req.NamespacedName)
 	if alreadyQueued {
+		log.Info("GPUPool already queued for reconcile", "name", req.NamespacedName.Name)
 		return ctrl.Result{}, nil
 	}
 	if !runNow {

internal/controller/node_controller.go

@@ -105,7 +105,6 @@ func (r *NodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 			return ctrl.Result{}, err
 		}
 		if !matched {
-
 			// delete gpunode if no matched pool
 			if err := r.Client.Delete(ctx, &tfv1.GPUNode{
 				ObjectMeta: metav1.ObjectMeta{
@@ -122,10 +121,6 @@ func (r *NodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 		if err := r.Client.Get(ctx, client.ObjectKey{Name: node.Name}, gpuNode); err != nil {
 			if errors.IsNotFound(err) {
 				gpuNode = r.generateGPUNode(node, pool)
-				// Set owner reference to cascade delete after GPU node created
-				if err := controllerutil.SetControllerReference(node, gpuNode, r.Scheme); err != nil {
-					return ctrl.Result{}, fmt.Errorf("failed to set controller reference: %w", err)
-				}
 				_, e := controllerutil.CreateOrUpdate(ctx, r.Client, gpuNode, func() error { return nil })
 				if e != nil {
 					return ctrl.Result{}, fmt.Errorf("failed to create or patch GPUNode: %w", e)
@@ -159,6 +154,7 @@ func (r *NodeReconciler) generateGPUNode(node *corev1.Node, pool *tfv1.GPUPool)
 		ObjectMeta: metav1.ObjectMeta{
 			Name: node.Name,
 			Labels: map[string]string{
+				constants.LabelKeyOwner: pool.Name,
 				fmt.Sprintf(constants.GPUNodePoolIdentifierLabelFormat, pool.Name): "true",
 			},
 		},
@@ -167,7 +163,7 @@ func (r *NodeReconciler) generateGPUNode(node *corev1.Node, pool *tfv1.GPUPool)
 		},
 	}
 
-	_ = controllerutil.SetOwnerReference(pool, gpuNode, r.Scheme)
+	_ = controllerutil.SetControllerReference(pool, gpuNode, r.Scheme)
 	return gpuNode
 }