Move matching labels to association.go

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

@@ -100,6 +100,16 @@ func RayClusterWorkerPodsAssociationOptions(instance *rayv1.RayCluster) Associat
 	}
 }
 
+func RayClusterRedisPodsAssociationOptions(instance *rayv1.RayCluster) AssociationOptions {
+	return AssociationOptions{
+		client.InNamespace(instance.Namespace),
+		client.MatchingLabels{
+			utils.RayClusterLabelKey:  instance.Name,
+			utils.RayNodeTypeLabelKey: string(rayv1.RedisCleanupNode),
+		},
+	}
+}
+
 func RayClusterGroupPodsAssociationOptions(instance *rayv1.RayCluster, group string) AssociationOptions {
 	return AssociationOptions{
 		client.InNamespace(instance.Namespace),

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

@@ -292,3 +292,40 @@ func TestGetRayClusterHeadPod(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, ret, headPod)
 }
+
+func TestRayClusterRedisPodsAssociationOptions(t *testing.T) {
+	// Create a new scheme
+	newScheme := runtime.NewScheme()
+	_ = rayv1.AddToScheme(newScheme)
+	_ = corev1.AddToScheme(newScheme)
+
+	instance := &rayv1.RayCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "raycluster-example",
+			Namespace: "default",
+		},
+	}
+
+	_ = &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "redis-pod",
+			Namespace: instance.ObjectMeta.Namespace,
+			Labels: map[string]string{
+				utils.RayClusterLabelKey:  instance.Name,
+				utils.RayNodeTypeLabelKey: string(rayv1.RedisCleanupNode),
+			},
+		},
+	}
+
+	expected := []client.ListOption{
+		client.InNamespace(instance.ObjectMeta.Namespace),
+		client.MatchingLabels(map[string]string{
+			utils.RayClusterLabelKey:  instance.Name,
+			utils.RayNodeTypeLabelKey: string(rayv1.RedisCleanupNode),
+		}),
+	}
+
+	listOpts := RayClusterRedisPodsAssociationOptions(instance).ToListOptions()
+
+	assert.Equal(t, expected, listOpts)
+}

ray-operator/controllers/ray/raycluster_controller.go

@@ -286,9 +286,9 @@ func (r *RayClusterReconciler) rayClusterReconcile(ctx context.Context, instance
 			}
 
 			// We can start the Redis cleanup process now because the head Pod has been terminated.
-			filterLabels := client.MatchingLabels{utils.RayClusterLabelKey: instance.Name, utils.RayNodeTypeLabelKey: string(rayv1.RedisCleanupNode)}
+			filterLabels := common.RayClusterRedisPodsAssociationOptions(instance).ToListOptions()
 			redisCleanupJobs := batchv1.JobList{}
-			if err := r.List(ctx, &redisCleanupJobs, client.InNamespace(instance.Namespace), filterLabels); err != nil {
+			if err := r.List(ctx, &redisCleanupJobs, filterLabels...); err != nil {
 				return ctrl.Result{RequeueAfter: DefaultRequeueDuration}, err
 			}
 
@@ -481,8 +481,8 @@ func (r *RayClusterReconciler) reconcileIngress(ctx context.Context, instance *r
 func (r *RayClusterReconciler) reconcileRouteOpenShift(ctx context.Context, instance *rayv1.RayCluster) error {
 	logger := ctrl.LoggerFrom(ctx)
 	headRoutes := routev1.RouteList{}
-	filterLabels := client.MatchingLabels{utils.RayClusterLabelKey: instance.Name}
-	if err := r.List(ctx, &headRoutes, client.InNamespace(instance.Namespace), filterLabels); err != nil {
+	filterLabels := common.RayClusterAllPodsAssociationOptions(instance).ToListOptions()
+	if err := r.List(ctx, &headRoutes, filterLabels...); err != nil {
 		return err
 	}
 
@@ -512,8 +512,8 @@ func (r *RayClusterReconciler) reconcileRouteOpenShift(ctx context.Context, inst
 func (r *RayClusterReconciler) reconcileIngressKubernetes(ctx context.Context, instance *rayv1.RayCluster) error {
 	logger := ctrl.LoggerFrom(ctx)
 	headIngresses := networkingv1.IngressList{}
-	filterLabels := client.MatchingLabels{utils.RayClusterLabelKey: instance.Name}
-	if err := r.List(ctx, &headIngresses, client.InNamespace(instance.Namespace), filterLabels); err != nil {
+	filterLabels := common.RayClusterAllPodsAssociationOptions(instance).ToListOptions()
+	if err := r.List(ctx, &headIngresses, filterLabels...); err != nil {
 		return err
 	}
 
@@ -544,9 +544,9 @@ func (r *RayClusterReconciler) reconcileIngressKubernetes(ctx context.Context, i
 func (r *RayClusterReconciler) reconcileHeadService(ctx context.Context, instance *rayv1.RayCluster) error {
 	logger := ctrl.LoggerFrom(ctx)
 	services := corev1.ServiceList{}
-	filterLabels := client.MatchingLabels{utils.RayClusterLabelKey: instance.Name, utils.RayNodeTypeLabelKey: string(rayv1.HeadNode)}
+	filterLabels := common.RayClusterHeadPodsAssociationOptions(instance).ToListOptions()
 
-	if err := r.List(ctx, &services, client.InNamespace(instance.Namespace), filterLabels); err != nil {
+	if err := r.List(ctx, &services, filterLabels...); err != nil {
 		return err
 	}
 
@@ -1280,8 +1280,8 @@ func (r *RayClusterReconciler) calculateStatus(ctx context.Context, instance *ra
 	newInstance.Status.ObservedGeneration = newInstance.ObjectMeta.Generation
 
 	runtimePods := corev1.PodList{}
-	filterLabels := client.MatchingLabels{utils.RayClusterLabelKey: newInstance.Name}
-	if err := r.List(ctx, &runtimePods, client.InNamespace(newInstance.Namespace), filterLabels); err != nil {
+	filterLabels := common.RayClusterAllPodsAssociationOptions(newInstance).ToListOptions()
+	if err := r.List(ctx, &runtimePods, filterLabels...); err != nil {
 		return nil, err
 	}
 
@@ -1450,11 +1450,8 @@ func (r *RayClusterReconciler) updateEndpoints(ctx context.Context, instance *ra
 	// We assume we can find the right one by filtering Services with appropriate label selectors
 	// and picking the first one. We may need to select by name in the future if the Service naming is stable.
 	rayHeadSvc := corev1.ServiceList{}
-	filterLabels := client.MatchingLabels{
-		utils.RayClusterLabelKey:  instance.Name,
-		utils.RayNodeTypeLabelKey: "head",
-	}
-	if err := r.List(ctx, &rayHeadSvc, client.InNamespace(instance.Namespace), filterLabels); err != nil {
+	filterLabels := common.RayClusterAllPodsAssociationOptions(instance).ToListOptions()
+	if err := r.List(ctx, &rayHeadSvc, filterLabels...); err != nil {
 		return err
 	}