fix: reconcile bug (#41)

Author: Code2Life

charts/tensor-fusion/Chart.yaml

@@ -15,7 +15,7 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 1.1.3
+version: 1.1.4
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to

charts/tensor-fusion/values.yaml

@@ -20,7 +20,7 @@ serviceAccount:
   annotations: {}
 
 hypervisor:
-  greptimedbEndpoint: greptimedb-standalone.greptimedb.svc.cluster.local:4001
+  greptimedbEndpoint: greptimedb-standalone.tensor-fusion.svc.cluster.local:4001
   image:
     repository: tensorfusion/tensor-fusion-hypervisor
     # Overrides the image tag whose default is the chart appVersion.
@@ -51,7 +51,7 @@ controller:
   tolerations: []
   affinity: {}
 
-  greptimedbEndpoint: http://greptimedb-standalone.greptimedb.svc.cluster.local:4000/v1/prometheus/write?db=public
+  greptimedbEndpoint: http://greptimedb-standalone.tensor-fusion.svc.cluster.local:4000/v1/prometheus/write?db=public
   admissionWebhooks:
     failurePolicy: Fail
     secretName: tensor-fusion-webhook-secret

internal/cloudprovider/common/utils.go

@@ -128,7 +128,7 @@ func CalculateLeastCostGPUNodes(ctx context.Context, provider types.GPUNodeProvi
 	for i := int64(0); i < bestNumInstances; i++ {
 		// Zone and region should ideally be determined from nodeClass's subnet selectors
 		nodes = append(nodes, types.NodeCreationParam{
-			NodeName:     fmt.Sprintf("%s-%s-%d", pool.Name, generateRandomString(6), i+1),
+			NodeName:     fmt.Sprintf("%s-%s", pool.Name, generateRandomString(8)),
 			InstanceType: bestInstance.InstanceType,
 			NodeClass:    nodeClass,
 			Region:       region,

internal/controller/gpunode_controller.go

@@ -116,6 +116,11 @@ func (r *GPUNodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		}
 	}
 
+	// Only reconcile if the node has a kubernetes node name, otherwise the DaemonSet like workloads can not be scheduled
+	if node.Status.KubernetesNodeName == "" {
+		return ctrl.Result{}, nil
+	}
+
 	if err := r.reconcileNodeDiscoveryJob(ctx, node, poolName); err != nil {
 		return ctrl.Result{}, err
 	}
@@ -173,7 +178,7 @@ func (r *GPUNodeReconciler) reconcileNodeDiscoveryJob(
 				{
 					Key:      "metadata.name",
 					Operator: corev1.NodeSelectorOpIn,
-					Values:   []string{gpunode.Name},
+					Values:   []string{gpunode.Status.KubernetesNodeName},
 				},
 			},
 		})
@@ -227,10 +232,9 @@ func (r *GPUNodeReconciler) reconcileHypervisorPod(ctx context.Context, node *tf
 	if poolName == "" {
 		return nil
 	}
-
 	namespace := utils.CurrentNamespace()
 	log := log.FromContext(ctx)
-	hypervisorPodName := fmt.Sprintf("%s-%s-hypervisor", poolName, node.Name)
+	hypervisorPodName := fmt.Sprintf("%s-hypervisor", node.Name)
 	pool := r.GpuPoolState.Get(poolName)
 	if pool == nil {
 		return fmt.Errorf("failed to get tensor-fusion pool, can not create hypervisor pod, pool: %s", poolName)
@@ -245,7 +249,7 @@ func (r *GPUNodeReconciler) reconcileHypervisorPod(ctx context.Context, node *tf
 	if spec.NodeSelector == nil {
 		spec.NodeSelector = make(map[string]string)
 	}
-	spec.NodeSelector["kubernetes.io/hostname"] = node.Name
+	spec.NodeSelector["kubernetes.io/hostname"] = node.Status.KubernetesNodeName
 	newPod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      hypervisorPodName,

internal/controller/gpupool_controller.go

@@ -51,6 +51,7 @@ type GPUPoolReconciler struct {
 func (r *GPUPoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	log := log.FromContext(ctx)
 
+	// TODO, debounce the reconcile when lots of GPUNodes created/updated, GPUPool should not updated too frequently (it owns Job)
 	log.Info("Reconciling GPUPool", "name", req.NamespacedName.Name)
 	defer func() {
 		log.Info("Finished reconciling GPUPool", "name", req.NamespacedName.Name)
@@ -77,6 +78,11 @@ func (r *GPUPoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		return ctrl.Result{}, nil
 	}
 
+	// Update pool capacity at first
+	if err := r.reconcilePoolCurrentCapacityAndReadiness(ctx, pool); err != nil {
+		return ctrl.Result{}, err
+	}
+
 	// For provisioning mode, check if need to scale up GPUNodes upon AvailableCapacity changed
 	isProvisioningMode := pool.Spec.NodeManagerConfig.ProvisioningMode == tfv1.ProvisioningModeProvisioned
 
@@ -91,11 +97,8 @@ func (r *GPUPoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		}
 		// Set phase to updating and let GPUNode event trigger the check and update capacity loop, util all nodes are ready
 		if newNodeCreated {
+			// Refresh the capacity again since new node has been created
 			pool.Status.Phase = tfv1.TensorFusionPoolPhaseUpdating
-			err = r.Client.Status().Update(ctx, pool)
-			if err != nil {
-				return ctrl.Result{}, err
-			}
 			if err := r.reconcilePoolCurrentCapacityAndReadiness(ctx, pool); err != nil {
 				return ctrl.Result{}, err
 			}
@@ -109,10 +112,6 @@ func (r *GPUPoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	// 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
 
-	if err := r.reconcilePoolCurrentCapacityAndReadiness(ctx, pool); err != nil {
-		return ctrl.Result{}, err
-	}
-
 	return ctrl.Result{}, nil
 }
 

internal/controller/gpupool_node_provision.go

@@ -140,11 +140,6 @@ func (r *GPUPoolReconciler) reconcilePoolCapacityWithProvisioner(ctx context.Con
 					ManageMode:  tfv1.GPUNodeManageModeProvisioned,
 					CostPerHour: strconv.FormatFloat(costPerHour, 'f', 6, 64),
 				},
-				Status: tfv1.GPUNodeStatus{
-					TotalTFlops: node.TFlopsOffered,
-					TotalVRAM:   node.VRAMOffered,
-					TotalGPUs:   node.GPUDeviceOffered,
-				},
 			}
 			_ = controllerutil.SetControllerReference(pool, gpuNodeRes, r.Scheme)
 			err := r.Client.Create(ctx, gpuNodeRes)
@@ -153,27 +148,35 @@ func (r *GPUPoolReconciler) reconcilePoolCapacityWithProvisioner(ctx context.Con
 				return
 			}
 			// Update GPUNode status to set the resource quantity
-			// gpuNodeRes.Status =
-			// if err := r.Client.Status().Update(ctx, gpuNodeRes); err != nil {
-			// 	errList = append(errList, err)
-			// 	return
-			// }
-
-			// // Create node on cloud provider
-			// status, err := provider.CreateNode(ctx, &node)
-			// if err != nil {
-			// 	errList = append(errList, err)
-			// 	return
-			// }
-
-			// // Update GPUNode status about the cloud vendor info
-			// // To match GPUNode - K8S node, the --node-label in Kubelet is MUST-have, like Karpenter, it force set userdata to add a provisionerId label, k8s node controller then can set its ownerReference to the GPUNode
-			// gpuNodeRes.Status.NodeInfo.IP = status.PrivateIP
-			// gpuNodeRes.Status.NodeInfo.InstanceID = status.InstanceID
-			// gpuNodeRes.Status.NodeInfo.Region = node.Region
-			// r.Client.Status().Update(ctx, gpuNodeRes)
-
-			// r.Recorder.Eventf(pool, corev1.EventTypeNormal, "GPUNodeCreated", "Created node: %s, IP: %s", status.InstanceID, status.PrivateIP)
+			gpuNodeRes.Status = tfv1.GPUNodeStatus{
+				Phase:               tfv1.TensorFusionGPUNodePhasePending,
+				TotalTFlops:         node.TFlopsOffered,
+				TotalVRAM:           node.VRAMOffered,
+				TotalGPUs:           node.GPUDeviceOffered,
+				AllocationDetails:   []tfv1.GPUNodeAllocationDetails{},
+				LoadedModels:        []string{},
+				ManagedGPUDeviceIDs: []string{},
+			}
+			if err := r.Client.Status().Patch(ctx, gpuNodeRes, client.Merge); err != nil {
+				errList = append(errList, err)
+				return
+			}
+
+			// Create node on cloud provider
+			status, err := provider.CreateNode(ctx, &node)
+			if err != nil {
+				errList = append(errList, err)
+				return
+			}
+
+			// Update GPUNode status about the cloud vendor info
+			// To match GPUNode - K8S node, the --node-label in Kubelet is MUST-have, like Karpenter, it force set userdata to add a provisionerId label, k8s node controller then can set its ownerReference to the GPUNode
+			gpuNodeRes.Status.NodeInfo.IP = status.PrivateIP
+			gpuNodeRes.Status.NodeInfo.InstanceID = status.InstanceID
+			gpuNodeRes.Status.NodeInfo.Region = node.Region
+			r.Client.Status().Update(ctx, gpuNodeRes)
+
+			r.Recorder.Eventf(pool, corev1.EventTypeNormal, "GPUNodeCreated", "Created node: %s, IP: %s", status.InstanceID, status.PrivateIP)
 		}(node)
 	}
 

internal/controller/node_controller.go

@@ -96,6 +96,11 @@ func (r *NodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 		if e != nil {
 			return ctrl.Result{}, fmt.Errorf("failed to create or patch GPUNode: %w", e)
 		}
+
+		gpuNode.Status.KubernetesNodeName = node.Name
+		if err := r.Client.Status().Patch(ctx, gpuNode, client.Merge); err != nil {
+			return ctrl.Result{}, fmt.Errorf("can not add Kubernetes Node info into gpuNode(%s) status : %w", gpuNode.Name, err)
+		}
 		log.Info("Created GPUNode due to selector matched", "name", gpuNode.Name)
 	}
 
@@ -118,10 +123,6 @@ func (r *NodeReconciler) generateGPUNode(ctx context.Context, node *corev1.Node,
 		Spec: tfv1.GPUNodeSpec{
 			ManageMode: tfv1.GPUNodeManageModeAutoSelect,
 		},
-		Status: tfv1.GPUNodeStatus{
-			KubernetesNodeName: node.Name,
-			ObservedGeneration: node.Generation,
-		},
 	}
 	return gpuNode
 }

internal/controller/tensorfusioncluster_controller.go

@@ -153,6 +153,7 @@ func (r *TensorFusionClusterReconciler) Reconcile(ctx context.Context, req ctrl.
 			return ctrl.Result{RequeueAfter: delay}, nil
 		} else {
 			// all components are ready, set cluster as ready
+			tfc.Status.RetryCount = 0
 			tfc.SetAsReady(conditions...)
 			gpupools, err := r.mustListOwnedGPUPools(ctx, tfc)
 			if err != nil {
@@ -349,10 +350,6 @@ func (r *TensorFusionClusterReconciler) checkTFClusterComponentsReady(ctx contex
 			Type:   constants.ConditionStatusTypeTimeSeriesDatabase,
 			Status: metav1.ConditionTrue,
 		},
-		{
-			Type:   constants.ConditionStatusTypeCloudVendorConnection,
-			Status: metav1.ConditionTrue,
-		},
 	}
 
 	// check if all conditions are true, any not ready component will make allPass false
@@ -387,7 +384,8 @@ func (r *TensorFusionClusterReconciler) checkTFClusterComponentsReady(ctx contex
 func (r *TensorFusionClusterReconciler) updateTFClusterStatus(ctx context.Context, tfc *tfv1.TensorFusionCluster, prevStatus *tfv1.TensorFusionClusterStatus) error {
 	// Update the cluster status, ignore retryCount, keep other fields to compare
 	prevStatus.RetryCount = tfc.Status.RetryCount
-	if equality.Semantic.DeepEqual(tfc.Status, prevStatus) {
+
+	if equality.Semantic.DeepEqual(tfc.Status, *prevStatus) {
 		return nil
 	}
 	if err := r.Status().Update(ctx, tfc); err != nil {

internal/utils/reconcile.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"errors"
-	"fmt"
 	"math"
 	"math/rand/v2"
 	"os"
@@ -91,6 +91,11 @@ func CurrentNamespace() string {
 
 func GetObjectHash(obj any) string {
 	hasher := sha256.New()
-	hasher.Write([]byte(fmt.Sprintf("%v", obj)))
+	jsonBytes, err := json.Marshal(obj)
+	if err != nil {
+		panic(err)
+	}
+	str := string(jsonBytes)
+	hasher.Write([]byte(str))
 	return hex.EncodeToString(hasher.Sum(nil))
 }