fix: subscribe the pool's nodeselector in the nodecontroller (#49)

Author: 0x5457

cmd/insights/main.go

@@ -119,7 +119,6 @@ func (p AlibabaGPUNodeProvider) CreateNode(ctx context.Context, param *types.Nod
 		return nil, fmt.Errorf("instance creation failed: %s", response.String())
 	}
 
-	// TODO, SHOULD use request id to check instance status util created
 	return &types.GPUNodeStatus{
 		InstanceID: response.RequestId,
 		CreatedAt:  time.Now(),

internal/cloudprovider/alibaba/ecs.go

@@ -43,7 +43,6 @@ type GPUReconciler struct {
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
 // move the current state of the cluster closer to the desired state.
 func (r *GPUReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	// TODO: Calculate tflops and update capacity here
 	return ctrl.Result{}, nil
 }
 

internal/controller/gpu_controller.go

@@ -70,7 +70,10 @@ func (r *GPUPoolCompactionReconciler) checkNodeCompaction(ctx context.Context, p
 			// Not is in-using, should not be terminated
 			continue
 		}
-
+		if gpuNode.Status.Phase != tfv1.TensorFusionGPUNodePhaseRunning {
+			// Not running, should not be terminated
+			continue
+		}
 		// Protect new nodes at least 5 minutes to avoid flapping
 		if gpuNode.CreationTimestamp.Time.After(time.Now().Add(newNodeProtectionDuration)) {
 			continue

internal/controller/gpupool_compaction_controller.go

@@ -65,8 +65,9 @@ func (r *GPUPoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		return ctrl.Result{}, err
 	}
 
+	// TODO: if phase is destroying, stop all existing workers and hypervisors, stop time series flow aggregations
 	deleted, err := utils.HandleFinalizer(ctx, pool, r.Client, func(ctx context.Context, pool *tfv1.GPUPool) error {
-		// TODO: stop all existing workers and hypervisors, stop time series flow aggregations
+		// TODO: stop all existing components
 		return nil
 	})
 	if err != nil {
@@ -101,9 +102,6 @@ func (r *GPUPoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		}
 	}
 
-	// if !isProvisioningMode {
-	// TODO: move GPUNode CR creation here, rather than node_controller
-	// }
 	// TODO, when componentConfig changed, it should notify corresponding resource to upgrade
 	// eg. when hypervisor changed, should change all owned GPUNode's status.phase to Updating
 
@@ -172,5 +170,6 @@ func (r *GPUPoolReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&tfv1.GPUPool{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
 		Named("gpupool").
+		Owns(&tfv1.GPUNode{}).
 		Complete(r)
 }

internal/controller/gpupool_controller.go

@@ -20,8 +20,6 @@ import (
 )
 
 // Controller and trigger logic for abstract layer of node provisioning
-// TODO: implement the logic
-
 func (r *GPUPoolReconciler) reconcilePoolCapacityWithProvisioner(ctx context.Context, pool *tfv1.GPUPool) (bool, error) {
 	log := log.FromContext(ctx)
 	// check if min resource constraint is satisfied

internal/controller/gpupool_node_provision.go

@@ -24,6 +24,7 @@ import (
 
 	tfv1 "github.com/NexusGPU/tensor-fusion-operator/api/v1"
 	"github.com/NexusGPU/tensor-fusion-operator/internal/constants"
+	"github.com/NexusGPU/tensor-fusion-operator/internal/utils"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -33,8 +34,11 @@ import (
 	"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/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	schedulingcorev1 "k8s.io/component-helpers/scheduling/corev1"
 )
@@ -95,7 +99,14 @@ func (r *NodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 			return ctrl.Result{}, err
 		}
 		if !matched {
-			log.Info("No matched GPU pool found, skip reconcile the Node", "node", node.Name, "labels", node.Labels)
+			// delete gpunode if no matched pool
+			if err := r.Client.Delete(ctx, &tfv1.GPUNode{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: node.Name,
+				},
+			}); err != nil {
+				return ctrl.Result{}, fmt.Errorf("can not delete gpuNode(%s) : %w", node.Name, err)
+			}
 			return ctrl.Result{}, nil
 		}
 
@@ -169,11 +180,36 @@ func (r *NodeReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	if err != nil {
 		return fmt.Errorf("unable to create predicate: %w", err)
 	}
+
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&corev1.Node{}, builder.WithPredicates(p)).
 		Named("node").
+		Watches(&tfv1.GPUPool{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, obj client.Object) []reconcile.Request {
+			nodelist := &tfv1.GPUNodeList{}
+			if err := mgr.GetClient().List(ctx, nodelist, client.MatchingLabels{
+				selectors[0]: selectors[1],
+			}); err != nil {
+				log.FromContext(ctx).Error(err, "failed to list GPUNode")
+				return []reconcile.Request{}
+			}
+			var requests []reconcile.Request
+			for _, n := range nodelist.Items {
+				requests = append(requests, reconcile.Request{NamespacedName: client.ObjectKey{Name: n.Name}})
+			}
+			return requests
+		}), builder.WithPredicates(predicate.Funcs{
+			UpdateFunc: func(e event.UpdateEvent) bool {
+				oldObj, ok1 := e.ObjectOld.(*tfv1.GPUPool)
+				newObj, ok2 := e.ObjectNew.(*tfv1.GPUPool)
+				if !ok1 || !ok2 {
+					return false
+				}
+				oldNodeSelector := oldObj.Spec.NodeManagerConfig.NodeSelector
+				newNodeSelector := newObj.Spec.NodeManagerConfig.NodeSelector
+				return utils.GetObjectHash(oldNodeSelector) != utils.GetObjectHash(newNodeSelector)
+			},
+		})).
 		Complete(r)
-	// TODO: When Pool changed, all nodes should re-generated, delete not matched ones, this logic should be added into GPUPool controller
 }
 
 func getMatchedPoolName(node *corev1.Node, poolList []tfv1.GPUPool) (*tfv1.GPUPool, bool, error) {