feat: add worker version in the connection URL (#64)

Author: 0x5457

api/v1/tensorfusionworkload_types.go

@@ -49,6 +49,8 @@ type WorkerStatus struct {
 	WorkerIp string `json:"workerIp,omitempty"`
 	// +optional
 	WorkerPort int `json:"workerPort,omitempty"`
+	// +optional
+	ResourceVersion string `json:"resourceVersion,omitempty"`
 }
 
 // TensorFusionWorkloadStatus defines the observed state of TensorFusionWorkload.

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

@@ -120,6 +120,8 @@ spec:
                       additionalProperties:
                         type: string
                       type: object
+                    resourceVersion:
+                      type: string
                     workerIp:
                       type: string
                     workerName:

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

@@ -120,6 +120,8 @@ spec:
                       additionalProperties:
                         type: string
                       type: object
+                    resourceVersion:
+                      type: string
                     workerIp:
                       type: string
                     workerName:

internal/controller/tensorfusionconnection_controller.go

@@ -92,7 +92,12 @@ func (r *TensorFusionConnectionReconciler) Reconcile(ctx context.Context, req ct
 
 	connection.Status.Phase = workerStatus.WorkerPhase
 	connection.Status.WorkerName = workerStatus.WorkerName
-	connection.Status.ConnectionURL = fmt.Sprintf("native+%s+%d", workerStatus.WorkerIp, workerStatus.WorkerPort)
+	resourceVersion := workerStatus.ResourceVersion
+	if resourceVersion == "" {
+		resourceVersion = "0"
+	}
+
+	connection.Status.ConnectionURL = fmt.Sprintf("native+%s+%d+%s-%s", workerStatus.WorkerIp, workerStatus.WorkerPort, workerStatus.WorkerName, resourceVersion)
 	if err := r.Status().Update(ctx, connection); err != nil {
 		return ctrl.Result{}, fmt.Errorf("update connection status: %w", err)
 	}

internal/controller/tensorfusionconnection_controller_test.go

@@ -172,7 +172,7 @@ var _ = Describe("TensorFusionConnection Controller", func() {
 			// Verify specific values
 			Expect(connection.Status.WorkerName).To(Equal("test-worker-1"))
 			Expect(connection.Status.Phase).To(Equal(tfv1.WorkerRunning))
-			Expect(connection.Status.ConnectionURL).To(Equal("native+192.168.1.1+8080"))
+			Expect(connection.Status.ConnectionURL).To(Equal("native+192.168.1.1+8080+test-worker-1-0"))
 		})
 
 		It("should handle missing workload label", func() {
@@ -268,7 +268,7 @@ var _ = Describe("TensorFusionConnection Controller", func() {
 					return false
 				}
 				return connection.Status.WorkerName == "test-worker-2" &&
-					connection.Status.ConnectionURL == "native+192.168.1.2+8081"
+					connection.Status.ConnectionURL == "native+192.168.1.2+8081+test-worker-2-0"
 			}, time.Second*5, time.Millisecond*100).Should(BeTrue())
 		})
 	})

internal/controller/tensorfusionworkload_controller.go

@@ -349,11 +349,12 @@ func (r *TensorFusionWorkloadReconciler) updateStatus(
 
 		// Create and append worker status
 		workerStatus := tfv1.WorkerStatus{
-			WorkerPhase:  workerPhase,
-			WorkerName:   pod.Name,
-			WorkerIp:     ip,
-			WorkerPort:   port,
-			NodeSelector: pod.Spec.NodeSelector,
+			WorkerPhase:     workerPhase,
+			WorkerName:      pod.Name,
+			WorkerIp:        ip,
+			WorkerPort:      port,
+			NodeSelector:    pod.Spec.NodeSelector,
+			ResourceVersion: pod.ResourceVersion,
 		}
 
 		workerStatuses = append(workerStatuses, workerStatus)