Add fqdn

.buildkite/values-kuberay-operator-override.yaml

@@ -24,3 +24,5 @@ featureGates:
     enabled: true
   - name: RayClusterMTLS
     enabled: true
+  - name: RayClusterWorkerFQDN
+    enabled: true

ray-operator/config/overlays/test-overrides/deployment-override.yaml

@@ -9,4 +9,4 @@ spec:
       containers:
       - name: kuberay-operator
         args:
-        - --feature-gates=RayClusterStatusConditions=true,RayJobDeletionPolicy=true,RayMultiHostIndexing=true,RayCronJob=true,RayServiceIncrementalUpgrade=true,SidecarSubmitterRestart=true,GCSFaultToleranceEmbeddedStorage=true,RayClusterMTLS=true
+        - --feature-gates=RayClusterStatusConditions=true,RayJobDeletionPolicy=true,RayMultiHostIndexing=true,RayCronJob=true,RayServiceIncrementalUpgrade=true,SidecarSubmitterRestart=true,GCSFaultToleranceEmbeddedStorage=true,RayClusterMTLS=true,RayClusterWorkerFQDN=true

ray-operator/controllers/ray/common/pod.go

@@ -591,6 +591,19 @@ func getEnableProbesInjection() bool {
 	return true
 }
 
+// generateWorkerHostname returns a stable DNS label for a worker Pod based on
+// its group name and replica/host indices. For multi-host groups the host index
+// is included so each host in a replica gets a unique name.
+func generateWorkerHostname(groupName string, replicaIndex int, hostIndex int, numOfHosts int32) string {
+	var hostname string
+	if numOfHosts > 1 {
+		hostname = fmt.Sprintf("%s-%d-%d", groupName, replicaIndex, hostIndex)
+	} else {
+		hostname = fmt.Sprintf("%s-%d", groupName, replicaIndex)
+	}
+	return utils.CheckLabel(hostname)
+}
+
 // DefaultWorkerPodTemplate sets the config values
 func DefaultWorkerPodTemplate(ctx context.Context, instance rayv1.RayCluster, workerSpec rayv1.WorkerGroupSpec, podName string, fqdnRayIP string, headPort string, replicaGrpName string, replicaIndex int, numHostIndex int) corev1.PodTemplateSpec {
 	podTemplate := workerSpec.Template
@@ -670,16 +683,25 @@ func DefaultWorkerPodTemplate(ctx context.Context, instance rayv1.RayCluster, wo
 	mergedLabels := mergeLabels(workerSpec.Template.ObjectMeta.Labels, workerSpec.Labels)
 	podTemplate.Labels = labelPod(rayv1.WorkerNode, instance.Name, workerSpec.GroupName, mergedLabels)
 
-	// Add additional labels when RayMultihostIndexing is enabled.
-	if features.Enabled(features.RayMultiHostIndexing) {
-		// The ordered replica index can be used for the single-host, multi-slice case.
+	// Assign stable replica indices for RayMultiHostIndexing and/or RayClusterWorkerFQDN.
+	if features.NeedsWorkerIndices() {
+		// The ordered replica index can be used for the single-host, multi-slice case
+		// and for stable FQDN hostnames.
 		podTemplate.Labels[utils.RayWorkerReplicaIndexKey] = strconv.Itoa(replicaIndex)
+		// Slice labels (replica name + host index) when this group has multiple hosts.
 		if workerSpec.NumOfHosts > 1 {
-			// These labels are specific to multi-host group setup and reconciliation.
 			podTemplate.Labels[utils.RayWorkerReplicaNameKey] = replicaGrpName
 			podTemplate.Labels[utils.RayHostIndexKey] = strconv.Itoa(numHostIndex)
 		}
 	}
+
+	// Stable per-pod FQDN via hostname + subdomain matching the headless Service.
+	// See: https://kubernetes.io/docs/concepts/services-networking/dns-pod-service/#pod-hostname-and-subdomain-field
+	if features.Enabled(features.RayClusterWorkerFQDN) {
+		podTemplate.Spec.Hostname = generateWorkerHostname(workerSpec.GroupName, replicaIndex, numHostIndex, workerSpec.NumOfHosts)
+		podTemplate.Spec.Subdomain = instance.Name + utils.DashSymbol + utils.HeadlessServiceSuffix
+	}
+
 	workerSpec.RayStartParams = setMissingRayStartParams(ctx, workerSpec.RayStartParams, rayv1.WorkerNode, headPort, fqdnRayIP)
 
 	initTemplateAnnotations(instance, &podTemplate)

ray-operator/controllers/ray/common/pod_test.go

@@ -1544,7 +1544,9 @@ func TestDeafultWorkerPodTemplateWithReplicaGrpAndIndex(t *testing.T) {
 	fqdnRayIP := utils.GenerateFQDNServiceName(ctx, *cluster, cluster.Namespace)
 	worker := cluster.Spec.WorkerGroupSpecs[0]
 
+	// Production default: indexing on, FQDN off. Slice labels are stamped; hostname/subdomain are not.
 	features.SetFeatureGateDuringTest(t, features.RayMultiHostIndexing, true)
+	features.SetFeatureGateDuringTest(t, features.RayClusterWorkerFQDN, false)
 
 	worker.Template.ObjectMeta.Name = "ray-worker-test"
 	worker.NumOfHosts = 4
@@ -1557,6 +1559,99 @@ func TestDeafultWorkerPodTemplateWithReplicaGrpAndIndex(t *testing.T) {
 	assert.Equal(t, podTemplateSpec.Labels[utils.RayWorkerReplicaNameKey], groupReplicaName)
 	assert.Equal(t, "0", podTemplateSpec.Labels[utils.RayWorkerReplicaIndexKey])
 	assert.Equal(t, "2", podTemplateSpec.Labels[utils.RayHostIndexKey])
+	assert.Empty(t, podTemplateSpec.Spec.Hostname)
+	assert.Empty(t, podTemplateSpec.Spec.Subdomain)
+}
+
+// TestDefaultWorkerPodTemplateStableFQDN covers hostname/subdomain stamping vs the two
+// feature gates. FQDN-off cases assert that indexing labels can still be present while
+// per-pod DNS names stay unset (the production default).
+func TestDefaultWorkerPodTemplateStableFQDN(t *testing.T) {
+	ctx := context.Background()
+
+	cluster := instance.DeepCopy()
+
+	fqdnRayIP := utils.GenerateFQDNServiceName(ctx, *cluster, cluster.Namespace)
+	worker := cluster.Spec.WorkerGroupSpecs[0]
+	expectedSubdomain := cluster.Name + utils.DashSymbol + utils.HeadlessServiceSuffix
+	worker.Template.ObjectMeta.Name = "ray-worker-test"
+	podName := func(groupName string) string {
+		return cluster.Name + utils.DashSymbol + string(rayv1.WorkerNode) + utils.DashSymbol + groupName + utils.DashSymbol + utils.FormatInt32(0)
+	}
+
+	t.Run("fqdn on, indexing off, multi-host", func(t *testing.T) {
+		// FQDN assigns replica indices itself and reuses slice labels so each host
+		// gets a unique hostname even when RayMultiHostIndexing is off.
+		features.SetFeatureGateDuringTest(t, features.RayMultiHostIndexing, false)
+		features.SetFeatureGateDuringTest(t, features.RayClusterWorkerFQDN, true)
+
+		w := worker.DeepCopy()
+		w.NumOfHosts = 4
+		groupReplicaName := utils.GenerateRayWorkerReplicaGroupName(w.GroupName)
+		pod := DefaultWorkerPodTemplate(ctx, *cluster, *w, podName(w.GroupName), fqdnRayIP, "6379", groupReplicaName, 0, 2)
+		assert.Equal(t, "small-group-0-2", pod.Spec.Hostname)
+		assert.Equal(t, expectedSubdomain, pod.Spec.Subdomain)
+		assert.Equal(t, "0", pod.Labels[utils.RayWorkerReplicaIndexKey])
+		assert.Equal(t, groupReplicaName, pod.Labels[utils.RayWorkerReplicaNameKey])
+		assert.Equal(t, "2", pod.Labels[utils.RayHostIndexKey])
+	})
+	t.Run("fqdn on, indexing off, single-host", func(t *testing.T) {
+		// Single-host FQDN is {group}-{replicaIndex}; slice labels are not set.
+		features.SetFeatureGateDuringTest(t, features.RayMultiHostIndexing, false)
+		features.SetFeatureGateDuringTest(t, features.RayClusterWorkerFQDN, true)
+
+		w := worker.DeepCopy()
+		w.NumOfHosts = 1
+		pod := DefaultWorkerPodTemplate(ctx, *cluster, *w, podName(w.GroupName), fqdnRayIP, "6379", "", 3, 0)
+		assert.Equal(t, "small-group-3", pod.Spec.Hostname)
+		assert.Equal(t, expectedSubdomain, pod.Spec.Subdomain)
+		assert.Equal(t, "3", pod.Labels[utils.RayWorkerReplicaIndexKey])
+		assert.Empty(t, pod.Labels[utils.RayWorkerReplicaNameKey])
+		assert.Empty(t, pod.Labels[utils.RayHostIndexKey])
+	})
+	t.Run("both off", func(t *testing.T) {
+		// Pre-indexing, pre-FQDN: no replica labels and no per-pod DNS names.
+		features.SetFeatureGateDuringTest(t, features.RayMultiHostIndexing, false)
+		features.SetFeatureGateDuringTest(t, features.RayClusterWorkerFQDN, false)
+
+		w := worker.DeepCopy()
+		w.NumOfHosts = 1
+		pod := DefaultWorkerPodTemplate(ctx, *cluster, *w, podName(w.GroupName), fqdnRayIP, "6379", "", 3, 0)
+		assert.Empty(t, pod.Spec.Hostname)
+		assert.Empty(t, pod.Spec.Subdomain)
+		assert.Empty(t, pod.Labels[utils.RayWorkerReplicaIndexKey])
+		assert.Empty(t, pod.Labels[utils.RayWorkerReplicaNameKey])
+		assert.Empty(t, pod.Labels[utils.RayHostIndexKey])
+	})
+	t.Run("indexing on, fqdn off, single-host", func(t *testing.T) {
+		// Production default for single-host: replica index only, no hostname/subdomain.
+		features.SetFeatureGateDuringTest(t, features.RayMultiHostIndexing, true)
+		features.SetFeatureGateDuringTest(t, features.RayClusterWorkerFQDN, false)
+
+		w := worker.DeepCopy()
+		w.NumOfHosts = 1
+		pod := DefaultWorkerPodTemplate(ctx, *cluster, *w, podName(w.GroupName), fqdnRayIP, "6379", "", 3, 0)
+		assert.Empty(t, pod.Spec.Hostname)
+		assert.Empty(t, pod.Spec.Subdomain)
+		assert.Equal(t, "3", pod.Labels[utils.RayWorkerReplicaIndexKey])
+		assert.Empty(t, pod.Labels[utils.RayWorkerReplicaNameKey])
+		assert.Empty(t, pod.Labels[utils.RayHostIndexKey])
+	})
+	t.Run("indexing on, fqdn off, multi-host", func(t *testing.T) {
+		// Production default for multi-host: slice labels only, no hostname/subdomain.
+		features.SetFeatureGateDuringTest(t, features.RayMultiHostIndexing, true)
+		features.SetFeatureGateDuringTest(t, features.RayClusterWorkerFQDN, false)
+
+		w := worker.DeepCopy()
+		w.NumOfHosts = 4
+		groupReplicaName := utils.GenerateRayWorkerReplicaGroupName(w.GroupName)
+		pod := DefaultWorkerPodTemplate(ctx, *cluster, *w, podName(w.GroupName), fqdnRayIP, "6379", groupReplicaName, 0, 2)
+		assert.Empty(t, pod.Spec.Hostname)
+		assert.Empty(t, pod.Spec.Subdomain)
+		assert.Equal(t, "0", pod.Labels[utils.RayWorkerReplicaIndexKey])
+		assert.Equal(t, groupReplicaName, pod.Labels[utils.RayWorkerReplicaNameKey])
+		assert.Equal(t, "2", pod.Labels[utils.RayHostIndexKey])
+	})
 }
 
 func containerPortExists(ports []corev1.ContainerPort, containerPort int32) error {

ray-operator/controllers/ray/common/service.go

@@ -295,7 +295,9 @@ func BuildServeService(ctx context.Context, rayService rayv1.RayService, rayClus
 	return serveService, nil
 }
 
-// BuildHeadlessService builds the headless service for workers in multi-host worker groups to communicate
+// BuildHeadlessServiceForRayCluster builds the headless service for worker Pods.
+// It enables stable per-worker Pod FQDNs via hostname + subdomain, and supports
+// peer communication between multi-host workers.
 func BuildHeadlessServiceForRayCluster(rayCluster rayv1.RayCluster) *corev1.Service {
 	name := rayCluster.Name + utils.DashSymbol + utils.HeadlessServiceSuffix
 	namespace := rayCluster.Namespace
@@ -318,8 +320,8 @@ func BuildHeadlessServiceForRayCluster(rayCluster rayv1.RayCluster) *corev1.Serv
 			ClusterIP: "None",
 			Selector:  selectorLabels,
 			Type:      corev1.ServiceTypeClusterIP,
-			// The headless worker service is used for peer communication between multi-host workers and should not be
-			// dependent on Proxy Actor placement to publish DNS addresses.
+			// Publish addresses even before Pods are Ready so peer DNS and per-Pod FQDNs
+			// are available without depending on Proxy Actor placement.
 			PublishNotReadyAddresses: true,
 		},
 	}

ray-operator/controllers/ray/raycluster_controller.go

@@ -896,36 +896,37 @@ func (r *RayClusterReconciler) reconcileServeService(ctx context.Context, instan
 
 // Return nil only when the headless service for multi-host worker groups is successfully created or already exists.
 func (r *RayClusterReconciler) reconcileHeadlessService(ctx context.Context, instance *rayv1.RayCluster) error {
-	// Check if there are worker groups with NumOfHosts > 1 in the cluster
-	isMultiHost := false
-	for _, workerGroup := range instance.Spec.WorkerGroupSpecs {
-		if workerGroup.NumOfHosts > 1 {
-			isMultiHost = true
-			break
+	// Check if Stable Worker FQDN is enabled
+	needHeadless := features.Enabled(features.RayClusterWorkerFQDN)
+
+	// If Stable Worker FQDN is not enabled, check if there are worker groups with NumOfHosts > 1 in the cluster
+	if !needHeadless {
+		for _, workerGroup := range instance.Spec.WorkerGroupSpecs {
+			if workerGroup.NumOfHosts > 1 {
+				needHeadless = true
+				break
+			}
 		}
 	}
+	if !needHeadless {
+		return nil
+	}
 
-	if isMultiHost {
-		services := corev1.ServiceList{}
-		options := common.RayClusterHeadlessServiceListOptions(instance)
-
-		if err := r.List(ctx, &services, options...); err != nil {
-			return err
-		}
-		// Check if there's an existing headless service in the cluster.
-		if len(services.Items) != 0 {
-			// service exists, do nothing
-			return nil
-		}
-		// Create headless tpu worker service if there's no existing one in the cluster.
-		headlessSvc := common.BuildHeadlessServiceForRayCluster(*instance)
+	services := corev1.ServiceList{}
+	options := common.RayClusterHeadlessServiceListOptions(instance)
 
-		if err := r.createService(ctx, headlessSvc, instance); err != nil {
-			return err
-		}
+	if err := r.List(ctx, &services, options...); err != nil {
+		return err
 	}
+	// Check if there's an existing headless service in the cluster.
+	if len(services.Items) != 0 {
+		// service exists, do nothing
+		return nil
+	}
+	// Create headless tpu worker service if there's no existing one in the cluster.
+	headlessSvc := common.BuildHeadlessServiceForRayCluster(*instance)
 
-	return nil
+	return r.createService(ctx, headlessSvc, instance)
 }
 
 func (r *RayClusterReconciler) reconcilePods(ctx context.Context, instance *rayv1.RayCluster) error {
@@ -1092,7 +1093,7 @@ func (r *RayClusterReconciler) reconcilePods(ctx context.Context, instance *rayv
 			continue
 		}
 
-		isRayMultiHostIndexing := worker.NumOfHosts > 1 && features.Enabled(features.RayMultiHostIndexing)
+		isRayMultiHostIndexing := worker.NumOfHosts > 1 && features.NeedsWorkerIndices()
 		if isRayMultiHostIndexing {
 			if err := r.reconcileMultiHostWorkerGroup(ctx, instance, &worker, workerPods.Items); err != nil {
 				return err
@@ -1168,9 +1169,9 @@ func (r *RayClusterReconciler) reconcilePods(ctx context.Context, instance *rayv
 
 		logger.Info("reconcilePods", "workerReplicas", numExpectedWorkerPods, "NumOfHosts", worker.NumOfHosts, "runningPods", len(runningPods.Items), "diff", diff)
 
-		// Support replica indices for single-host, multi-slice environments.
+		// Support replica indices for single-host workers (RayMultiHostIndexing and/or RayClusterWorkerFQDN).
 		validReplicaIndices := make(map[int]bool)
-		if features.Enabled(features.RayMultiHostIndexing) {
+		if features.NeedsWorkerIndices() {
 			for _, pod := range runningPods.Items {
 				if indexStr, ok := pod.Labels[utils.RayWorkerReplicaIndexKey]; ok {
 					if index, err := strconv.Atoi(indexStr); err == nil {
@@ -1189,7 +1190,7 @@ func (r *RayClusterReconciler) reconcilePods(ctx context.Context, instance *rayv
 					return fmt.Errorf("mTLS secrets not ready: %w", err)
 				}
 			}
-			if features.Enabled(features.RayMultiHostIndexing) {
+			if features.NeedsWorkerIndices() {
 				newReplicaIndex := 0
 				// create all workers of this group
 				for i := range diff {
@@ -1280,7 +1281,7 @@ func (r *RayClusterReconciler) deletePods(ctx context.Context, instance *rayv1.R
 }
 
 // reconcileMultiHostWorkerGroup handles reconciliation and Pod deletion for worker groups with NumOfHosts > 1 when
-// the RayMultihostIndexing feature is enabled. This function is responsible for:
+// NeedsWorkerIndices is true (RayMultiHostIndexing and/or RayClusterWorkerFQDN). This function is responsible for:
 // 1. Deleting incomplete or unhealthy multi-host groups atomically.
 // 2. Explicit deletes of entire multi-host groups for the autoscaler.
 // 3. Scale up/down of multi-host groups.