fix: calculate hash based on whole spec (#175)

* fix: calculate hash based on whole spec

* fix: pod template hash issue

Author: Code2Life

internal/controller/tensorfusionworkload_controller.go

@@ -22,6 +22,7 @@ import (
 	"sort"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/record"
@@ -31,8 +32,6 @@ import (
 
 	"slices"
 
-	"reflect"
-
 	tfv1 "github.com/NexusGPU/tensor-fusion/api/v1"
 	"github.com/NexusGPU/tensor-fusion/internal/config"
 	"github.com/NexusGPU/tensor-fusion/internal/constants"
@@ -132,7 +131,7 @@ func (r *TensorFusionWorkloadReconciler) Reconcile(ctx context.Context, req ctrl
 	// Create worker generator
 	workerGenerator := &worker.WorkerGenerator{WorkerConfig: pool.Spec.ComponentConfig.Worker, GpuInfos: r.GpuInfos}
 
-	podTemplateHash, err := workerGenerator.PodTemplateHash(workload.Spec.Resources.Limits)
+	podTemplateHash, err := workerGenerator.PodTemplateHash(workload.Spec)
 	if err != nil {
 		return ctrl.Result{}, fmt.Errorf("get pod template hash: %w", err)
 	}
@@ -188,7 +187,7 @@ func (r *TensorFusionWorkloadReconciler) Reconcile(ctx context.Context, req ctrl
 
 		// Calculate how many pods need to be added
 		podsToAdd := int(desiredReplicas - currentReplicas)
-		result, err := r.scaleUpWorkers(ctx, workerGenerator, workload, podsToAdd)
+		result, err := r.scaleUpWorkers(ctx, workerGenerator, workload, podsToAdd, podTemplateHash)
 		if err != nil {
 			return ctrl.Result{}, fmt.Errorf("scale up workers: %w", err)
 		}
@@ -222,9 +221,10 @@ func (r *TensorFusionWorkloadReconciler) tryStartWorker(
 	workerGenerator *worker.WorkerGenerator,
 	gpu *tfv1.GPU,
 	workload *tfv1.TensorFusionWorkload,
+	hash string,
 ) (*corev1.Pod, error) {
 	port := workerGenerator.AllocPort()
-	pod, hash, err := workerGenerator.GenerateWorkerPod(gpu, fmt.Sprintf("%s-tf-worker-", workload.Name), workload.Namespace, port, workload.Spec.Resources.Limits)
+	pod, hash, err := workerGenerator.GenerateWorkerPod(gpu, fmt.Sprintf("%s-tf-worker-", workload.Name), workload.Namespace, port, workload.Spec.Resources.Limits, hash)
 	if err != nil {
 		return nil, fmt.Errorf("generate worker pod %w", err)
 	}
@@ -334,7 +334,7 @@ func (r *TensorFusionWorkloadReconciler) deletePod(ctx context.Context, pod *cor
 }
 
 // scaleUpWorkers handles the scaling up of worker pods
-func (r *TensorFusionWorkloadReconciler) scaleUpWorkers(ctx context.Context, workerGenerator *worker.WorkerGenerator, workload *tfv1.TensorFusionWorkload, count int) (ctrl.Result, error) {
+func (r *TensorFusionWorkloadReconciler) scaleUpWorkers(ctx context.Context, workerGenerator *worker.WorkerGenerator, workload *tfv1.TensorFusionWorkload, count int, hash string) (ctrl.Result, error) {
 	log := log.FromContext(ctx)
 
 	// Create worker pods
@@ -346,7 +346,7 @@ func (r *TensorFusionWorkloadReconciler) scaleUpWorkers(ctx context.Context, wor
 			return ctrl.Result{RequeueAfter: constants.PendingRequeueDuration}, nil
 		}
 
-		pod, err := r.tryStartWorker(ctx, workerGenerator, gpu, workload)
+		pod, err := r.tryStartWorker(ctx, workerGenerator, gpu, workload, hash)
 		if err != nil {
 			// Try to release the GPU resource if pod creation fails
 			releaseErr := r.Scheduler.Release(ctx, workload.Spec.Resources.Requests, gpu)
@@ -426,7 +426,7 @@ func (r *TensorFusionWorkloadReconciler) updateStatus(
 
 	// Check if we need to update status
 	statusChanged := workload.Status.ReadyReplicas != readyReplicas ||
-		!reflect.DeepEqual(workload.Status.WorkerStatuses, workerStatuses)
+		!equality.Semantic.DeepEqual(workload.Status.WorkerStatuses, workerStatuses)
 
 	if statusChanged {
 		log.Info("Updating workload status", "readyReplicas", readyReplicas, "workerCount", len(workerStatuses))

internal/webhook/v1/pod_webhook.go

@@ -21,10 +21,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
-	"reflect"
 
 	"gomodules.xyz/jsonpatch/v2"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/strategicpatch"
@@ -204,7 +204,7 @@ func (m *TensorFusionPodMutator) createOrUpdateWorkload(ctx context.Context, pod
 	}
 
 	// Compare the entire spec at once
-	if !reflect.DeepEqual(workload.Spec, desiredSpec) {
+	if !equality.Semantic.DeepEqual(workload.Spec, desiredSpec) {
 		workload.Spec = desiredSpec
 		if err := m.Client.Update(ctx, workload); err != nil {
 			return fmt.Errorf("failed to update workload: %w", err)

internal/worker/worker.go

@@ -46,13 +46,13 @@ func (wg *WorkerGenerator) AllocPort() int {
 	return rand.Intn(max-min+1) + min
 }
 
-func (wg *WorkerGenerator) PodTemplateHash(limits tfv1.Resource) (string, error) {
+func (wg *WorkerGenerator) PodTemplateHash(workloadSpec any) (string, error) {
 	podTmpl := &corev1.PodTemplate{}
 	err := json.Unmarshal(wg.WorkerConfig.PodTemplate.Raw, podTmpl)
 	if err != nil {
 		return "", fmt.Errorf("failed to unmarshal pod template: %w", err)
 	}
-	return utils.GetObjectHash(podTmpl, limits), nil
+	return utils.GetObjectHash(podTmpl, workloadSpec), nil
 }
 
 func (wg *WorkerGenerator) GenerateWorkerPod(
@@ -61,13 +61,13 @@ func (wg *WorkerGenerator) GenerateWorkerPod(
 	namespace string,
 	port int,
 	limits tfv1.Resource,
+	podTemplateHash string,
 ) (*corev1.Pod, string, error) {
 	podTmpl := &corev1.PodTemplate{}
 	err := json.Unmarshal(wg.WorkerConfig.PodTemplate.Raw, podTmpl)
 	if err != nil {
 		return nil, "", fmt.Errorf("failed to unmarshal pod template: %w", err)
 	}
-	podTemplateHash := utils.GetObjectHash(podTmpl, limits)
 	spec := podTmpl.Template.Spec
 	if spec.NodeSelector == nil {
 		spec.NodeSelector = make(map[string]string)