Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add unit tests for raycluster_controller reconcilePods function #219

Merged
merged 9 commits into from
Apr 5, 2022
Merged

Add unit tests for raycluster_controller reconcilePods function #219

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
82c90ee
Write raycluster_controller reconcilePods unit tests
Apr 4, 2022
86258fe
Improve unit tests code
Waynegates Apr 4, 2022
b0c6de8
Fix lint issue and Improve unit tests code
Waynegates Apr 4, 2022
6e3db6c
Run goimports
Waynegates Apr 4, 2022
37fb342
Fix a bug of workersToDelete update and update unit tests
Waynegates Apr 4, 2022
be89699
Fix a bug of workersToDelete update and update unit tests
Waynegates Apr 4, 2022
f34632e
Update unit tests log
Waynegates Apr 4, 2022
84d5e09
Update workersToDelete local var to avoid unuseful kube api server de…
Waynegates Apr 5, 2022
0138e28
Remove code of updating instance spec workersToDelete
Waynegates Apr 5, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 22 additions & 4 deletions ray-operator/controllers/raycluster_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@ func (r *RayClusterReconciler) reconcilePods(instance *rayiov1alpha1.RayCluster)
}
}
// Reconcile worker pods now
for index, worker := range instance.Spec.WorkerGroupSpecs {
for _, worker := range instance.Spec.WorkerGroupSpecs {
workerPods := corev1.PodList{}
filterLabels = client.MatchingLabels{common.RayClusterLabelKey: instance.Name, common.RayNodeGroupLabelKey: worker.GroupName}
if err := r.List(context.TODO(), &workerPods, client.InNamespace(instance.Namespace), filterLabels); err != nil {
Expand All @@ -248,6 +248,7 @@ func (r *RayClusterReconciler) reconcilePods(instance *rayiov1alpha1.RayCluster)
runningPods.Items = append(runningPods.Items, aPod)
}
}
r.updateLocalWorkersToDelete(&worker, runningPods.Items)
diff := *worker.Replicas - int32(len(runningPods.Items))

if PrioritizeWorkersToDelete {
Expand All @@ -265,7 +266,7 @@ func (r *RayClusterReconciler) reconcilePods(instance *rayiov1alpha1.RayCluster)
}
log.Info("reconcilePods", "unable to delete worker ", pod.Name)
} else {
diff--
diff++
r.Recorder.Eventf(instance, v1.EventTypeNormal, "Deleted", "Deleted pod %s", pod.Name)
}
}
Expand Down Expand Up @@ -304,7 +305,6 @@ func (r *RayClusterReconciler) reconcilePods(instance *rayiov1alpha1.RayCluster)
}
r.Recorder.Eventf(instance, v1.EventTypeNormal, "Deleted", "Deleted pod %s", pod.Name)
}
instance.Spec.WorkerGroupSpecs[index].ScaleStrategy.WorkersToDelete = []string{}
continue
} else {
// diff < 0 and not the same absolute value as int32(len(worker.ScaleStrategy.WorkersToDelete)
Expand All @@ -326,7 +326,6 @@ func (r *RayClusterReconciler) reconcilePods(instance *rayiov1alpha1.RayCluster)
}
r.Recorder.Eventf(instance, v1.EventTypeNormal, "Deleted", "Deleted pod %s", pod.Name)
}
instance.Spec.WorkerGroupSpecs[index].ScaleStrategy.WorkersToDelete = []string{}

// remove the remaining pods not part of the scaleStrategy
i := 0
Expand Down Expand Up @@ -361,6 +360,25 @@ func (r *RayClusterReconciler) reconcilePods(instance *rayiov1alpha1.RayCluster)
return nil
}

func (r *RayClusterReconciler) updateLocalWorkersToDelete(worker *rayiov1alpha1.WorkerGroupSpec, runningItems []v1.Pod) {
var actualWorkersToDelete []string
itemMap := make(map[string]int)

// Create a map for quick lookup.
for _, item := range runningItems {
itemMap[item.Name] = 1
}

// Build actualWorkersToDelete to only include running items.
for _, workerToDelete := range worker.ScaleStrategy.WorkersToDelete {
if _, ok := itemMap[workerToDelete]; ok {
actualWorkersToDelete = append(actualWorkersToDelete, workerToDelete)
}
}

worker.ScaleStrategy.WorkersToDelete = actualWorkersToDelete
}

func (r *RayClusterReconciler) createHeadIngress(ingress *networkingv1.Ingress, instance *rayiov1alpha1.RayCluster) error {
// making sure the name is valid
ingress.Name = utils.CheckName(ingress.Name)
Expand Down
Loading