diff --git a/ray-operator/controllers/ray/raycluster_controller.go b/ray-operator/controllers/ray/raycluster_controller.go index 869a78a82be..5d1227368e2 100644 --- a/ray-operator/controllers/ray/raycluster_controller.go +++ b/ray-operator/controllers/ray/raycluster_controller.go @@ -615,45 +615,145 @@ func (r *RayClusterReconciler) reconcileHeadService(ctx context.Context, instanc return err } - // Check if there's existing head service in the cluster. - if len(services.Items) != 0 { - if len(services.Items) == 1 { - logger.Info("reconcileHeadService", "1 head service found", services.Items[0].Name) - return nil - } - // This should never happen. This protects against the case that users manually create service with the same label. - if len(services.Items) > 1 { - logger.Info("reconcileHeadService", "Duplicate head service found", services.Items) - return fmt.Errorf("%d head service found %v", len(services.Items), services.Items) - } - } else { - // Create head service if there's no existing one in the cluster. - labels := make(map[string]string) - if val, ok := instance.Spec.HeadGroupSpec.Template.ObjectMeta.Labels[utils.KubernetesApplicationNameLabelKey]; ok { - labels[utils.KubernetesApplicationNameLabelKey] = val - } - annotations := make(map[string]string) - // TODO (kevin85421): KubeRay has already exposed the entire head service (#1040) to users. - // We may consider deprecating this field when we bump the CRD version. - maps.Copy(annotations, instance.Spec.HeadServiceAnnotations) - headSvc, err := common.BuildServiceForHeadPod(ctx, *instance, labels, annotations) - if err != nil { - return err + // This should never happen. This protects against the case that users manually create service with the same label. + if len(services.Items) > 1 { + logger.Info("reconcileHeadService", "Duplicate head service found", services.Items) + return fmt.Errorf("%d head service found %v", len(services.Items), services.Items) + } + + annotations := make(map[string]string) + // TODO (kevin85421): KubeRay has already exposed the entire head service (#1040) to users. + // We may consider deprecating this field when we bump the CRD version. + maps.Copy(annotations, instance.Spec.HeadServiceAnnotations) + selectorOverrides, err := r.headServiceSelectorOverrides(ctx, instance) + if err != nil { + return err + } + headSvc, err := common.BuildServiceForHeadPod(ctx, *instance, selectorOverrides, annotations) + if err != nil { + return err + } + + // The head service already exists. Its selector is derived from head Pod template labels that + // users are allowed to change, so it has to be reconciled instead of accepted as is. + if len(services.Items) == 1 { + logger.Info("reconcileHeadService", "1 head service found", services.Items[0].Name) + return r.updateHeadService(ctx, instance, &services.Items[0], headSvc) + } + + // Create head service if there's no existing one in the cluster. + // TODO (kevin85421): Provide a detailed and actionable error message. For example, which port is missing? + if len(headSvc.Spec.Ports) == 0 { + logger.Info("Ray head service does not have any ports set up.", "serviceSpecification", headSvc.Spec) + return fmt.Errorf("ray head service does not have any ports set up. Service specification: %v", headSvc.Spec) + } + + return r.createService(ctx, headSvc, instance) +} + +// headServiceSelectorOverrides returns the labels that are allowed to change the head service +// selector. Only the two app.kubernetes.io keys qualify: BuildServiceForHeadPod keys the selector +// off HeadServiceLabels, and of those, ray.io/cluster and ray.io/node-type are refused by labelPod, +// while ray.io/identifier is what the operator uses to find this service again. +// +// The values are read from the head Pod that is running right now, not from the head Pod template. +// KubeRay never relabels a running Pod, and the default upgrade strategy does not recreate one, so a +// selector built from an edited template would stop matching the Pod that is currently serving +// traffic. Once that Pod restarts it comes back carrying the template labels, and the selector +// follows it, which is the case issue #2564 reports. A Pod that is terminating is skipped because +// its replacement will come back with the template labels. Before any head Pod exists there is +// nothing to select yet, so the template is the only source available. +// +// A cluster is only supposed to have one head Pod, and reconcilePods refuses to clean up extras on +// its own, so the oldest one wins when several are alive. List order is not specified, and without +// a tiebreak the selector could flip from one reconcile to the next and rewrite the service every +// time. The oldest Pod is also the one that has been serving traffic. +func (r *RayClusterReconciler) headServiceSelectorOverrides(ctx context.Context, instance *rayv1.RayCluster) (map[string]string, error) { + sourceLabels := instance.Spec.HeadGroupSpec.Template.ObjectMeta.Labels + + headPods := corev1.PodList{} + if err := r.List(ctx, &headPods, common.RayClusterHeadPodsAssociationOptions(instance).ToListOptions()...); err != nil { + return nil, err + } + var servingHeadPod *corev1.Pod + for i := range headPods.Items { + headPod := &headPods.Items[i] + if headPod.DeletionTimestamp != nil { + continue } - // TODO (kevin85421): Provide a detailed and actionable error message. For example, which port is missing? - if len(headSvc.Spec.Ports) == 0 { - logger.Info("Ray head service does not have any ports set up.", "serviceSpecification", headSvc.Spec) - return fmt.Errorf("ray head service does not have any ports set up. Service specification: %v", headSvc.Spec) + if servingHeadPod == nil || isOlderHeadPod(headPod, servingHeadPod) { + servingHeadPod = headPod } + } + if servingHeadPod != nil { + sourceLabels = servingHeadPod.Labels + } - if err := r.createService(ctx, headSvc, instance); err != nil { - return err + overrides := make(map[string]string, 2) + for _, key := range []string{utils.KubernetesApplicationNameLabelKey, utils.KubernetesCreatedByLabelKey} { + if val, ok := sourceLabels[key]; ok { + overrides[key] = val } } + return overrides, nil +} + +// isOlderHeadPod reports whether a was created before b, falling back to the name so that two Pods +// created within the same clock tick still produce a stable answer. +func isOlderHeadPod(a, b *corev1.Pod) bool { + if !a.CreationTimestamp.Equal(&b.CreationTimestamp) { + return a.CreationTimestamp.Before(&b.CreationTimestamp) + } + return a.Name < b.Name +} + +// updateHeadService keeps the parts of an existing head service that the operator owns in sync with +// the RayCluster. Everything else on the live object is left alone, including ClusterIP (immutable), +// ports, service type, and any label or annotation another tool added after creation. +func (r *RayClusterReconciler) updateHeadService(ctx context.Context, instance *rayv1.RayCluster, existingSvc, desiredSvc *corev1.Service) error { + logger := ctrl.LoggerFrom(ctx) + + if !headServiceNeedsUpdate(existingSvc, desiredSvc) { + return nil + } + logger.Info("reconcileHeadService", "head service updated", existingSvc.Name, "selector", desiredSvc.Spec.Selector) + if err := r.Update(ctx, existingSvc); err != nil { + r.Recorder.Eventf(instance, nil, corev1.EventTypeWarning, string(utils.FailedToUpdateService), string(utils.UpdateAction), "Failed updating service %s/%s, %v", existingSvc.Namespace, existingSvc.Name, err) + return err + } + r.Recorder.Eventf(instance, nil, corev1.EventTypeNormal, string(utils.UpdatedService), string(utils.UpdateAction), "Updated service %s/%s", existingSvc.Namespace, existingSvc.Name) return nil } +// headServiceNeedsUpdate copies the operator owned fields of desiredSvc onto existingSvc and reports +// whether anything changed. +func headServiceNeedsUpdate(existingSvc, desiredSvc *corev1.Service) bool { + updated := false + + // The selector belongs entirely to the operator: it is the only thing that decides whether the + // head Pod is reachable through this service, so a stale or hand edited value is replaced. + if !maps.Equal(existingSvc.Spec.Selector, desiredSvc.Spec.Selector) { + existingSvc.Spec.Selector = maps.Clone(desiredSvc.Spec.Selector) + updated = true + } + + // Labels are merged key by key rather than replaced. Labels the operator never wrote, such as + // those added by Helm or Argo CD, are not ours to delete. + for key, value := range desiredSvc.Labels { + if existingSvc.Labels[key] == value { + continue + } + if existingSvc.Labels == nil { + existingSvc.Labels = make(map[string]string, len(desiredSvc.Labels)) + } + existingSvc.Labels[key] = value + updated = true + } + + return updated +} + // reconcileGCSStoragePVC provisions the persistent volume backing the embedded // RocksDB GCS store. It is a no-op unless GCS FT uses the embedded backend. // diff --git a/ray-operator/controllers/ray/raycluster_controller_test.go b/ray-operator/controllers/ray/raycluster_controller_test.go index d868e2e9d94..e0298429978 100644 --- a/ray-operator/controllers/ray/raycluster_controller_test.go +++ b/ray-operator/controllers/ray/raycluster_controller_test.go @@ -31,6 +31,7 @@ import ( "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/tools/events" "k8s.io/client-go/util/retry" @@ -344,6 +345,83 @@ var _ = Context("Inside the default namespace", func() { return rayCluster.Status.Head.ServiceIP, err }, time.Second*3, time.Millisecond*500).Should(Equal("1.1.1.1"), "Should be able to see the rayCluster.Status.Head.ServiceIP: %v", rayCluster.Status.Head.ServiceIP) }) + + It("The head service should select the head Pod", func() { + Expect(headPods.Items).To(HaveLen(1)) + headSvc := corev1.ServiceList{} + Expect(k8sClient.List(ctx, &headSvc, common.RayClusterHeadServiceListOptions(rayCluster)...)).To(Succeed()) + Expect(headSvc.Items).To(HaveLen(1)) + Expect(labels.SelectorFromSet(headSvc.Items[0].Spec.Selector).Matches(labels.Set(headPods.Items[0].Labels))).To(BeTrue(), + "head service selector %v should select head Pod labels %v", headSvc.Items[0].Spec.Selector, headPods.Items[0].Labels) + }) + + It("The head service should keep selecting the head Pod that is still running", func() { + // Regression guard for the window before the head Pod restarts. KubeRay never relabels a + // running Pod, and the default upgrade strategy does not recreate one either, so a selector + // taken straight from the template would stop matching the Pod that is serving traffic. + Expect(k8sClient.List(ctx, &headPods, headFilters...)).To(Succeed()) + Expect(headPods.Items).To(HaveLen(1)) + livePodLabels := labels.Set(headPods.Items[0].Labels) + Expect(livePodLabels[utils.KubernetesApplicationNameLabelKey]).To(Equal("myapp")) + + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + Eventually( + getResourceFunc(ctx, client.ObjectKey{Name: rayCluster.Name, Namespace: namespace}, rayCluster), + time.Second*3, time.Millisecond*500).Should(Succeed(), "rayCluster: %v", rayCluster) + rayCluster.Spec.HeadGroupSpec.Template.Labels[utils.KubernetesApplicationNameLabelKey] = "myapp-live-rename" + return k8sClient.Update(ctx, rayCluster) + }) + Expect(err).NotTo(HaveOccurred(), "Failed to update RayCluster") + + // The head Pod is deliberately left running, so its labels stay as they were. + Consistently(func() string { + if err := k8sClient.List(ctx, &headPods, headFilters...); err != nil || len(headPods.Items) != 1 { + return "" + } + return headPods.Items[0].Labels[utils.KubernetesApplicationNameLabelKey] + }, time.Second*3, time.Millisecond*250).Should(Equal("myapp"), "the running head Pod should not be relabeled") + + Consistently(func() bool { + headSvc := corev1.ServiceList{} + if err := k8sClient.List(ctx, &headSvc, common.RayClusterHeadServiceListOptions(rayCluster)...); err != nil || len(headSvc.Items) != 1 { + return false + } + return labels.SelectorFromSet(headSvc.Items[0].Spec.Selector).Matches(livePodLabels) + }, time.Second*3, time.Millisecond*250).Should(BeTrue(), "head service must keep selecting the head Pod that is still running") + }) + + It("The head service should follow a renamed app.kubernetes.io/name label", func() { + // See https://github.com/ray-project/kuberay/issues/2564. KubeRay does not relabel a running + // head Pod, but a head Pod that restarts comes back with the new template label, while the + // head service used to keep the value it was created with and stop selecting anything. + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + Eventually( + getResourceFunc(ctx, client.ObjectKey{Name: rayCluster.Name, Namespace: namespace}, rayCluster), + time.Second*3, time.Millisecond*500).Should(Succeed(), "rayCluster: %v", rayCluster) + rayCluster.Spec.HeadGroupSpec.Template.Labels[utils.KubernetesApplicationNameLabelKey] = "myapp-renamed" + return k8sClient.Update(ctx, rayCluster) + }) + Expect(err).NotTo(HaveOccurred(), "Failed to update RayCluster") + + // Restart the head Pod the way an eviction or a node drain would. + Expect(headPods.Items).To(HaveLen(1)) + oldHeadPod := headPods.Items[0] + Expect(k8sClient.Delete(ctx, &oldHeadPod, &client.DeleteOptions{GracePeriodSeconds: ptr.To[int64](0)})).To(Succeed()) + Eventually(func() string { + if err := k8sClient.List(ctx, &headPods, headFilters...); err != nil || len(headPods.Items) != 1 { + return "" + } + return headPods.Items[0].Labels[utils.KubernetesApplicationNameLabelKey] + }, time.Second*5, time.Millisecond*500).Should(Equal("myapp-renamed"), "the restarted head Pod should carry the new label") + + Eventually(func() bool { + headSvc := corev1.ServiceList{} + if err := k8sClient.List(ctx, &headSvc, common.RayClusterHeadServiceListOptions(rayCluster)...); err != nil || len(headSvc.Items) != 1 { + return false + } + return labels.SelectorFromSet(headSvc.Items[0].Spec.Selector).Matches(labels.Set(headPods.Items[0].Labels)) + }, time.Second*5, time.Millisecond*500).Should(BeTrue(), "head service should select the restarted head Pod") + }) }) Describe("RayCluster with invalid overridden ray.io/cluster labels", Ordered, func() { diff --git a/ray-operator/controllers/ray/raycluster_controller_unit_test.go b/ray-operator/controllers/ray/raycluster_controller_unit_test.go index e41a841dba7..698b1f5ab1d 100644 --- a/ray-operator/controllers/ray/raycluster_controller_unit_test.go +++ b/ray-operator/controllers/ray/raycluster_controller_unit_test.go @@ -1097,6 +1097,333 @@ func TestReconcileHeadService(t *testing.T) { require.Error(t, err, "Reconciler should report an error when there are two head services") } +// newHeadServiceReconciler returns a reconciler backed by a fake client that already knows about +// the given RayCluster, plus any other objects the test wants to put in front of it. +func newHeadServiceReconciler(cluster *rayv1.RayCluster, objects ...runtime.Object) *RayClusterReconciler { + newScheme := runtime.NewScheme() + _ = rayv1.AddToScheme(newScheme) + _ = corev1.AddToScheme(newScheme) + + runtimeObjects := append([]runtime.Object{cluster}, objects...) + fakeClient := clientFake.NewClientBuilder().WithScheme(newScheme).WithRuntimeObjects(runtimeObjects...).Build() + return &RayClusterReconciler{ + Client: fakeClient, + Recorder: &events.FakeRecorder{}, + Scheme: scheme.Scheme, + rayClusterScaleExpectation: expectations.NewRayClusterScaleExpectation(fakeClient), + } +} + +// getHeadService returns the one head service the operator owns for the given cluster. +func getHeadService(ctx context.Context, t *testing.T, c client.Client, cluster *rayv1.RayCluster) *corev1.Service { + t.Helper() + services := corev1.ServiceList{} + require.NoError(t, c.List(ctx, &services, common.RayClusterHeadServiceListOptions(cluster)...)) + require.Len(t, services.Items, 1, "expected exactly one head service") + return &services.Items[0] +} + +// TestReconcileHeadServiceSelectorDrift covers https://github.com/ray-project/kuberay/issues/2564. +// The head service selector takes the value of app.kubernetes.io/name from the head Pod template, +// so renaming that label leaves the selector pointing at the old value and the head Pod stops +// being reachable through the service. +func TestReconcileHeadServiceSelectorDrift(t *testing.T) { + setupTest(t) + + ctx := context.TODO() + cluster := testRayCluster.DeepCopy() + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels = map[string]string{ + utils.KubernetesApplicationNameLabelKey: "ray-converters-2", + } + + r := newHeadServiceReconciler(cluster) + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc := getHeadService(ctx, t, r.Client, cluster) + require.Equal(t, "ray-converters-2", svc.Spec.Selector[utils.KubernetesApplicationNameLabelKey]) + + // A label some other tool (Helm, Argo CD, a mesh injector) adds to the service after creation. + // The operator owns the keys it stamps, not the whole label map, so this one has to survive. + svc.Labels["argocd.argoproj.io/instance"] = "converters" + require.NoError(t, r.Update(ctx, svc)) + + // The user renames the head Pod label, so the head Pod now carries ray-converters-3. + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels[utils.KubernetesApplicationNameLabelKey] = "ray-converters-3" + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc = getHeadService(ctx, t, r.Client, cluster) + assert.Equal(t, "ray-converters-3", svc.Spec.Selector[utils.KubernetesApplicationNameLabelKey], + "head service selector should follow the head Pod label") + assert.Equal(t, "ray-converters-3", svc.Labels[utils.KubernetesApplicationNameLabelKey], + "head service labels should follow the head Pod label") + assert.Equal(t, "converters", svc.Labels["argocd.argoproj.io/instance"], + "labels the operator does not own should be left alone") + + // The three labels the operator uses to find the service again must not move, otherwise the + // next reconcile creates a duplicate. + assert.Equal(t, cluster.Name, svc.Spec.Selector[utils.RayClusterLabelKey]) + assert.Equal(t, string(rayv1.HeadNode), svc.Spec.Selector[utils.RayNodeTypeLabelKey]) + assert.Equal(t, utils.CheckLabel(utils.GenerateIdentifier(cluster.Name, rayv1.HeadNode)), svc.Spec.Selector[utils.RayIDLabelKey]) + + // Reconciling an in-sync service should not write to the API server. + resourceVersion := svc.ResourceVersion + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + svc = getHeadService(ctx, t, r.Client, cluster) + assert.Equal(t, resourceVersion, svc.ResourceVersion, "reconcile should be a no-op once the head service matches") +} + +// runningHeadPod builds a head Pod labelled the way labelPod would label it, so a test can put a +// head Pod that is already running in front of the reconciler. +func runningHeadPod(cluster *rayv1.RayCluster, appName string) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: cluster.Name + "-head-abcde", + Namespace: cluster.Namespace, + Labels: map[string]string{ + utils.RayNodeLabelKey: "yes", + utils.RayClusterLabelKey: cluster.Name, + utils.RayNodeTypeLabelKey: string(rayv1.HeadNode), + utils.RayNodeGroupLabelKey: utils.RayNodeHeadGroupLabelValue, + utils.RayIDLabelKey: utils.CheckLabel(utils.GenerateIdentifier(cluster.Name, rayv1.HeadNode)), + utils.KubernetesApplicationNameLabelKey: appName, + utils.KubernetesCreatedByLabelKey: utils.ComponentName, + }, + }, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, + } +} + +// TestReconcileHeadServiceKeepsSelectingRunningHeadPod guards the window between a template label +// edit and the head Pod restart. KubeRay never relabels a running Pod, and the default upgrade +// strategy does not recreate one, so taking the selector straight from the template would leave the +// head Pod that is serving traffic unreachable through its own service. +func TestReconcileHeadServiceKeepsSelectingRunningHeadPod(t *testing.T) { + setupTest(t) + + ctx := context.TODO() + cluster := testRayCluster.DeepCopy() + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels = map[string]string{ + utils.KubernetesApplicationNameLabelKey: "ray-converters-2", + } + + headPod := runningHeadPod(cluster, "ray-converters-2") + r := newHeadServiceReconciler(cluster, headPod) + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc := getHeadService(ctx, t, r.Client, cluster) + require.True(t, labels.SelectorFromSet(svc.Spec.Selector).Matches(labels.Set(headPod.Labels)), + "the freshly created head service should select the head Pod") + + // The user renames the label on the template. The running head Pod keeps the old value. + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels[utils.KubernetesApplicationNameLabelKey] = "ray-converters-3" + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc = getHeadService(ctx, t, r.Client, cluster) + assert.True(t, labels.SelectorFromSet(svc.Spec.Selector).Matches(labels.Set(headPod.Labels)), + "selector %v must keep selecting the running head Pod %v", svc.Spec.Selector, headPod.Labels) + assert.Equal(t, "ray-converters-2", svc.Spec.Selector[utils.KubernetesApplicationNameLabelKey], + "the selector should track the head Pod, not the template") +} + +// TestReconcileHeadServicePrefersOldestHeadPod pins the tiebreak when more than one head Pod is +// alive, which reconcilePods refuses to clean up on its own. List order is not specified, so +// without a tiebreak the selector could flip between reconciles and rewrite the service each time. +// The oldest Pod is the one that has been serving traffic, so it is the one that must not be +// orphaned. +func TestReconcileHeadServicePrefersOldestHeadPod(t *testing.T) { + setupTest(t) + + ctx := context.TODO() + cluster := testRayCluster.DeepCopy() + + oldest := runningHeadPod(cluster, "serving-app") + oldest.Name = "zzz-oldest-head" + oldest.CreationTimestamp = metav1.NewTime(time.Now().Add(-time.Hour)) + + newer := runningHeadPod(cluster, "intruder-app") + newer.Name = "aaa-newer-head" + newer.CreationTimestamp = metav1.NewTime(time.Now()) + + r := newHeadServiceReconciler(cluster, oldest, newer) + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc := getHeadService(ctx, t, r.Client, cluster) + assert.Equal(t, "serving-app", svc.Spec.Selector[utils.KubernetesApplicationNameLabelKey], + "the oldest head Pod should decide the selector regardless of list order") + + // The choice must not depend on which reconcile pass we are in. + for i := 0; i < 5; i++ { + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + } + svc = getHeadService(ctx, t, r.Client, cluster) + assert.Equal(t, "serving-app", svc.Spec.Selector[utils.KubernetesApplicationNameLabelKey], + "repeated reconciles should not flip the selector") +} + +// TestReconcileHeadServiceFollowsRestartedHeadPod is the other half of issue #2564. Once the head +// Pod restarts it comes back carrying the new template label, and the selector has to follow it. +func TestReconcileHeadServiceFollowsRestartedHeadPod(t *testing.T) { + setupTest(t) + + ctx := context.TODO() + cluster := testRayCluster.DeepCopy() + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels = map[string]string{ + utils.KubernetesApplicationNameLabelKey: "ray-converters-2", + } + + oldHeadPod := runningHeadPod(cluster, "ray-converters-2") + r := newHeadServiceReconciler(cluster, oldHeadPod) + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + // The user renames the label, then the head Pod restarts and comes back with the new value. + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels[utils.KubernetesApplicationNameLabelKey] = "ray-converters-3" + require.NoError(t, r.Delete(ctx, oldHeadPod)) + newHeadPod := runningHeadPod(cluster, "ray-converters-3") + require.NoError(t, r.Create(ctx, newHeadPod)) + + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc := getHeadService(ctx, t, r.Client, cluster) + assert.True(t, labels.SelectorFromSet(svc.Spec.Selector).Matches(labels.Set(newHeadPod.Labels)), + "selector %v should select the restarted head Pod %v", svc.Spec.Selector, newHeadPod.Labels) + assert.Equal(t, "ray-converters-3", svc.Spec.Selector[utils.KubernetesApplicationNameLabelKey]) +} + +// TestReconcileHeadServiceOverriddenCreatedByLabel checks the sibling of the bug above. Both +// app.kubernetes.io labels in the head service selector can be overridden on the head Pod +// template (see labelPod), so both have to be read from it, not just app.kubernetes.io/name. +func TestReconcileHeadServiceOverriddenCreatedByLabel(t *testing.T) { + setupTest(t) + + ctx := context.TODO() + cluster := testRayCluster.DeepCopy() + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels = map[string]string{ + utils.KubernetesCreatedByLabelKey: "my-platform", + } + + r := newHeadServiceReconciler(cluster) + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc := getHeadService(ctx, t, r.Client, cluster) + assert.Equal(t, "my-platform", svc.Spec.Selector[utils.KubernetesCreatedByLabelKey], + "selector should use the head Pod value of app.kubernetes.io/created-by") + assert.Equal(t, utils.ApplicationName, svc.Spec.Selector[utils.KubernetesApplicationNameLabelKey], + "labels the user did not override keep their default value") +} + +// TestReconcileHeadServiceIgnoresProtectedLabelOverrides guards the identity labels. labelPod +// refuses to take ray.io/cluster, ray.io/node-type and ray.io/group from the user template, so a +// user setting them must not be able to steer the head service selector away from the head Pod. +func TestReconcileHeadServiceIgnoresProtectedLabelOverrides(t *testing.T) { + setupTest(t) + + ctx := context.TODO() + cluster := testRayCluster.DeepCopy() + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels = map[string]string{ + utils.RayClusterLabelKey: "some-other-cluster", + utils.RayNodeTypeLabelKey: string(rayv1.WorkerNode), + } + + r := newHeadServiceReconciler(cluster) + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc := getHeadService(ctx, t, r.Client, cluster) + assert.Equal(t, cluster.Name, svc.Spec.Selector[utils.RayClusterLabelKey]) + assert.Equal(t, string(rayv1.HeadNode), svc.Spec.Selector[utils.RayNodeTypeLabelKey]) +} + +// TestReconcileHeadServiceSelectorDriftWithCustomHeadService repeats the drift case for a cluster +// that ships its own HeadGroupSpec.HeadService. +func TestReconcileHeadServiceSelectorDriftWithCustomHeadService(t *testing.T) { + setupTest(t) + + ctx := context.TODO() + cluster := testRayCluster.DeepCopy() + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels = map[string]string{ + utils.KubernetesApplicationNameLabelKey: "ray-converters-2", + } + cluster.Spec.HeadGroupSpec.HeadService = &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "custom-head-svc", + Labels: map[string]string{"my-team": "converters"}, + }, + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, + }, + } + + r := newHeadServiceReconciler(cluster) + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc := getHeadService(ctx, t, r.Client, cluster) + require.Equal(t, "ray-converters-2", svc.Spec.Selector[utils.KubernetesApplicationNameLabelKey]) + require.Equal(t, "converters", svc.Labels["my-team"]) + + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels[utils.KubernetesApplicationNameLabelKey] = "ray-converters-3" + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc = getHeadService(ctx, t, r.Client, cluster) + assert.Equal(t, "ray-converters-3", svc.Spec.Selector[utils.KubernetesApplicationNameLabelKey]) + assert.Equal(t, "ray-converters-3", svc.Labels[utils.KubernetesApplicationNameLabelKey]) + assert.Equal(t, "converters", svc.Labels["my-team"], "labels from the user provided HeadService should survive") + assert.Equal(t, "custom-head-svc", svc.Name, "the update should not rename the service") +} + +// TestReconcileHeadServiceReplacesHandEditedSelector pins the asymmetry between the selector and +// the labels. The selector is replaced whole, so a key someone added by hand goes away, while +// labels are merged and keep keys the operator never wrote. +func TestReconcileHeadServiceReplacesHandEditedSelector(t *testing.T) { + setupTest(t) + + ctx := context.TODO() + cluster := testRayCluster.DeepCopy() + + r := newHeadServiceReconciler(cluster) + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc := getHeadService(ctx, t, r.Client, cluster) + svc.Spec.Selector["hand-edited"] = "yes" + svc.Labels["hand-edited"] = "yes" + require.NoError(t, r.Update(ctx, svc)) + + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc = getHeadService(ctx, t, r.Client, cluster) + assert.NotContains(t, svc.Spec.Selector, "hand-edited", + "the operator owns the whole selector, so an extra key is removed") + assert.Equal(t, "yes", svc.Labels["hand-edited"], + "labels are merged, so a key the operator never wrote is kept") +} + +// TestReconcileHeadServiceLeavesUnmanagedSpecAlone pins the scope of the update: the selector is +// operator owned, the rest of the service spec is not touched by drift reconciliation. +func TestReconcileHeadServiceLeavesUnmanagedSpecAlone(t *testing.T) { + setupTest(t) + + ctx := context.TODO() + cluster := testRayCluster.DeepCopy() + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels = map[string]string{ + utils.KubernetesApplicationNameLabelKey: "ray-converters-2", + } + + r := newHeadServiceReconciler(cluster) + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc := getHeadService(ctx, t, r.Client, cluster) + svc.Spec.ClusterIP = "10.0.0.42" + svc.Annotations = map[string]string{"service.beta.kubernetes.io/aws-load-balancer-type": "nlb"} + require.NoError(t, r.Update(ctx, svc)) + + cluster.Spec.HeadGroupSpec.Template.ObjectMeta.Labels[utils.KubernetesApplicationNameLabelKey] = "ray-converters-3" + require.NoError(t, r.reconcileHeadService(ctx, cluster)) + + svc = getHeadService(ctx, t, r.Client, cluster) + assert.Equal(t, "ray-converters-3", svc.Spec.Selector[utils.KubernetesApplicationNameLabelKey]) + assert.Equal(t, "10.0.0.42", svc.Spec.ClusterIP, "ClusterIP is immutable and should never be rewritten") + assert.Equal(t, "nlb", svc.Annotations["service.beta.kubernetes.io/aws-load-balancer-type"], + "annotations outside HeadServiceAnnotations should be left alone") +} + func TestReconcileHeadlessService(t *testing.T) { setupTest(t)