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

Prefix job workloads with job- #591

Closed
wants to merge 1 commit into from
Closed

Prefix job workloads with job- #591

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
21 changes: 21 additions & 0 deletions pkg/controller/workload/common/workload_utils.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
/*
Copyright 2023 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package common

func GetNameForJob(jobName string) string {
return "job-" + jobName
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This can produce a name that is too long to pass validation. And if we cut the last characters we could have collisions. I would suggest using generateName.

Copy link
Contributor Author

@mimowo mimowo Feb 21, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see, makes sense, then probably we can park it for now and only use generateName for other frameworks, at least for now. WDYT?
Meaning we stay with the current workload naming for batch.Job.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sg

}
3 changes: 2 additions & 1 deletion pkg/controller/workload/job/job_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ import (

kueue "sigs.k8s.io/kueue/apis/kueue/v1alpha2"
"sigs.k8s.io/kueue/pkg/constants"
"sigs.k8s.io/kueue/pkg/controller/workload/common"
utilpriority "sigs.k8s.io/kueue/pkg/util/priority"
"sigs.k8s.io/kueue/pkg/workload"
)
Expand Down Expand Up @@ -525,7 +526,7 @@ func ConstructWorkloadFor(ctx context.Context, client client.Client,
job *batchv1.Job, scheme *runtime.Scheme) (*kueue.Workload, error) {
w := &kueue.Workload{
ObjectMeta: metav1.ObjectMeta{
Name: job.Name,
Name: common.GetNameForJob(job.Name),
Namespace: job.Namespace,
},
Spec: kueue.WorkloadSpec{
Expand Down
7 changes: 5 additions & 2 deletions test/e2e/e2e_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import (
"k8s.io/apimachinery/pkg/types"

kueue "sigs.k8s.io/kueue/apis/kueue/v1alpha2"
"sigs.k8s.io/kueue/pkg/controller/workload/common"
"sigs.k8s.io/kueue/pkg/util/testing"
"sigs.k8s.io/kueue/test/util"
)
Expand Down Expand Up @@ -60,9 +61,10 @@ var _ = ginkgo.Describe("Kueue", func() {
}
return *createdJob.Spec.Suspend
}, util.Timeout, util.Interval).Should(gomega.BeTrue())
wlLookupKey := types.NamespacedName{Name: common.GetNameForJob(lookupKey.Name), Namespace: ns.Name}
createdWorkload := &kueue.Workload{}
gomega.Eventually(func() bool {
if err := k8sClient.Get(ctx, lookupKey, createdWorkload); err != nil {
if err := k8sClient.Get(ctx, wlLookupKey, createdWorkload); err != nil {
return false
}
return apimeta.IsStatusConditionTrue(createdWorkload.Status.Conditions, kueue.WorkloadAdmitted)
Expand Down Expand Up @@ -106,8 +108,9 @@ var _ = ginkgo.Describe("Kueue", func() {
}
return !*createdJob.Spec.Suspend && createdJob.Status.Succeeded > 0
}, util.Timeout, util.Interval).Should(gomega.BeTrue())
wlLookupKey := types.NamespacedName{Name: common.GetNameForJob(lookupKey.Name), Namespace: ns.Name}
gomega.Eventually(func() bool {
if err := k8sClient.Get(ctx, lookupKey, createdWorkload); err != nil {
if err := k8sClient.Get(ctx, wlLookupKey, createdWorkload); err != nil {
return false
}
return apimeta.IsStatusConditionTrue(createdWorkload.Status.Conditions, kueue.WorkloadAdmitted) &&
Expand Down
61 changes: 32 additions & 29 deletions test/integration/controller/job/job_controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import (

kueue "sigs.k8s.io/kueue/apis/kueue/v1alpha2"
"sigs.k8s.io/kueue/pkg/constants"
"sigs.k8s.io/kueue/pkg/controller/workload/common"
workloadjob "sigs.k8s.io/kueue/pkg/controller/workload/job"
"sigs.k8s.io/kueue/pkg/util/pointer"
"sigs.k8s.io/kueue/pkg/util/testing"
Expand All @@ -44,12 +45,16 @@ const (
parallelism = 4
jobName = "test-job"
jobNamespace = "default"
jobKey = jobNamespace + "/" + jobName
labelKey = "cloud.provider.com/instance"
priorityClassName = "test-priority-class"
priorityValue = 10
)

var (
lookupKey = types.NamespacedName{Name: jobName, Namespace: jobNamespace}
wlLookupKey = types.NamespacedName{Name: common.GetNameForJob(jobName), Namespace: jobNamespace}
)

var ignoreConditionTimestamps = cmpopts.IgnoreFields(metav1.Condition{}, "LastTransitionTime")

// +kubebuilder:docs-gen:collapse=Imports
Expand All @@ -73,7 +78,6 @@ var _ = ginkgo.Describe("Job controller", func() {
gomega.Expect(k8sClient.Create(ctx, priorityClass)).Should(gomega.Succeed())
job := testing.MakeJob(jobName, jobNamespace).PriorityClass(priorityClassName).Obj()
gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
lookupKey := types.NamespacedName{Name: jobName, Namespace: jobNamespace}
createdJob := &batchv1.Job{}
gomega.Eventually(func() bool {
if err := k8sClient.Get(ctx, lookupKey, createdJob); err != nil {
Expand All @@ -85,7 +89,7 @@ var _ = ginkgo.Describe("Job controller", func() {
ginkgo.By("checking the workload is created without queue assigned")
createdWorkload := &kueue.Workload{}
gomega.Eventually(func() bool {
err := k8sClient.Get(ctx, lookupKey, createdWorkload)
err := k8sClient.Get(ctx, wlLookupKey, createdWorkload)
return err == nil
}, util.Timeout, util.Interval).Should(gomega.BeTrue())
gomega.Expect(createdWorkload.Spec.QueueName).Should(gomega.Equal(""), "The Workload shouldn't have .spec.queueName set")
Expand All @@ -100,15 +104,15 @@ var _ = ginkgo.Describe("Job controller", func() {
createdJob.Annotations = map[string]string{constants.QueueAnnotation: jobQueueName}
gomega.Expect(k8sClient.Update(ctx, createdJob)).Should(gomega.Succeed())
gomega.Eventually(func() bool {
if err := k8sClient.Get(ctx, lookupKey, createdWorkload); err != nil {
if err := k8sClient.Get(ctx, wlLookupKey, createdWorkload); err != nil {
return false
}
return createdWorkload.Spec.QueueName == jobQueueName
}, util.Timeout, util.Interval).Should(gomega.BeTrue())

ginkgo.By("checking a second non-matching workload is deleted")
secondWl, _ := workloadjob.ConstructWorkloadFor(ctx, k8sClient, createdJob, scheme.Scheme)
secondWl.Name = "second-workload"
secondWl.Name = common.GetNameForJob("second-workload")
secondWl.Spec.PodSets[0].Count = parallelism + 1
gomega.Expect(k8sClient.Create(ctx, secondWl)).Should(gomega.Succeed())
gomega.Eventually(func() error {
Expand All @@ -118,7 +122,7 @@ var _ = ginkgo.Describe("Job controller", func() {
}, util.Timeout, util.Interval).Should(testing.BeNotFoundError())
// check the original wl is still there
gomega.Consistently(func() bool {
err := k8sClient.Get(ctx, lookupKey, createdWorkload)
err := k8sClient.Get(ctx, wlLookupKey, createdWorkload)
return err == nil
}, util.ConsistentDuration, util.Interval).Should(gomega.BeTrue())
gomega.Eventually(func() bool {
Expand Down Expand Up @@ -158,7 +162,7 @@ var _ = ginkgo.Describe("Job controller", func() {
gomega.Expect(len(createdJob.Spec.Template.Spec.NodeSelector)).Should(gomega.Equal(1))
gomega.Expect(createdJob.Spec.Template.Spec.NodeSelector[labelKey]).Should(gomega.Equal(onDemandFlavor.Name))
gomega.Consistently(func() bool {
if err := k8sClient.Get(ctx, lookupKey, createdWorkload); err != nil {
if err := k8sClient.Get(ctx, wlLookupKey, createdWorkload); err != nil {
return false
}
return len(createdWorkload.Status.Conditions) == 0
Expand All @@ -176,13 +180,13 @@ var _ = ginkgo.Describe("Job controller", func() {
len(createdJob.Spec.Template.Spec.NodeSelector) == 0
}, util.Timeout, util.Interval).Should(gomega.BeTrue())
gomega.Eventually(func() bool {
ok, _ := testing.CheckLatestEvent(ctx, k8sClient, "DeletedWorkload", corev1.EventTypeNormal, fmt.Sprintf("Deleted not matching Workload: %v", jobKey))
ok, _ := testing.CheckLatestEvent(ctx, k8sClient, "DeletedWorkload", corev1.EventTypeNormal, fmt.Sprintf("Deleted not matching Workload: %v", wlLookupKey.String()))
return ok
}, util.Timeout, util.Interval).Should(gomega.BeTrue())

ginkgo.By("checking the workload is updated with new count")
gomega.Eventually(func() bool {
if err := k8sClient.Get(ctx, lookupKey, createdWorkload); err != nil {
if err := k8sClient.Get(ctx, wlLookupKey, createdWorkload); err != nil {
return false
}
return createdWorkload.Spec.PodSets[0].Count == newParallelism
Expand All @@ -208,7 +212,7 @@ var _ = ginkgo.Describe("Job controller", func() {
gomega.Expect(len(createdJob.Spec.Template.Spec.NodeSelector)).Should(gomega.Equal(1))
gomega.Expect(createdJob.Spec.Template.Spec.NodeSelector[labelKey]).Should(gomega.Equal(spotFlavor.Name))
gomega.Consistently(func() bool {
if err := k8sClient.Get(ctx, lookupKey, createdWorkload); err != nil {
if err := k8sClient.Get(ctx, wlLookupKey, createdWorkload); err != nil {
return false
}
return len(createdWorkload.Status.Conditions) == 0
Expand All @@ -224,7 +228,7 @@ var _ = ginkgo.Describe("Job controller", func() {
})
gomega.Expect(k8sClient.Status().Update(ctx, createdJob)).Should(gomega.Succeed())
gomega.Eventually(func() bool {
err := k8sClient.Get(ctx, lookupKey, createdWorkload)
err := k8sClient.Get(ctx, wlLookupKey, createdWorkload)
if err != nil || len(createdWorkload.Status.Conditions) == 0 {
return false
}
Expand All @@ -236,10 +240,11 @@ var _ = ginkgo.Describe("Job controller", func() {
ginkgo.When("The parent-workload annotation is used", func() {

var (
parentJobName = jobName + "-parent"
parentLookupKey = types.NamespacedName{Name: parentJobName, Namespace: jobNamespace}
childJobName = jobName + "-child"
childLookupKey = types.NamespacedName{Name: childJobName, Namespace: jobNamespace}
parentJobName = jobName + "-parent"
parentWlLookupKey = types.NamespacedName{Name: common.GetNameForJob(parentJobName), Namespace: jobNamespace}
childJobName = jobName + "-child"
childLookupKey = types.NamespacedName{Name: childJobName, Namespace: jobNamespace}
childWlLookupKey = types.NamespacedName{Name: common.GetNameForJob(childJobName), Namespace: jobNamespace}
)

ginkgo.It("Should suspend a job if the parent workload does not exist", func() {
Expand All @@ -262,7 +267,7 @@ var _ = ginkgo.Describe("Job controller", func() {
ginkgo.By("waiting for the parent workload to be created")
parentWorkload := &kueue.Workload{}
gomega.Eventually(func() error {
return k8sClient.Get(ctx, parentLookupKey, parentWorkload)
return k8sClient.Get(ctx, parentWlLookupKey, parentWorkload)
}, util.Timeout, util.Interval).Should(gomega.Succeed())

ginkgo.By("Creating the child job which uses the parent workload annotation")
Expand All @@ -272,7 +277,7 @@ var _ = ginkgo.Describe("Job controller", func() {
ginkgo.By("Checking that the child workload is not created")
childWorkload := &kueue.Workload{}
gomega.Consistently(func() bool {
return apierrors.IsNotFound(k8sClient.Get(ctx, childLookupKey, childWorkload))
return apierrors.IsNotFound(k8sClient.Get(ctx, childWlLookupKey, childWorkload))
}, util.ConsistentDuration, util.Interval).Should(gomega.BeTrue())
})

Expand All @@ -286,13 +291,13 @@ var _ = ginkgo.Describe("Job controller", func() {
gomega.Expect(k8sClient.Create(ctx, parentJob)).Should(gomega.Succeed())

ginkgo.By("Creating the child job with the parent-workload annotation")
childJob := testing.MakeJob(childJobName, jobNamespace).ParentWorkload(parentJobName).Obj()
childJob := testing.MakeJob(childJobName, jobNamespace).ParentWorkload(parentWlLookupKey.Name).Obj()
gomega.Expect(k8sClient.Create(ctx, childJob)).Should(gomega.Succeed())

ginkgo.By("waiting for the parent workload to be created")
parentWorkload := &kueue.Workload{}
gomega.Eventually(func() error {
return k8sClient.Get(ctx, parentLookupKey, parentWorkload)
return k8sClient.Get(ctx, parentWlLookupKey, parentWorkload)
}, util.Timeout, util.Interval).Should(gomega.Succeed())

ginkgo.By("admit the parent workload")
Expand All @@ -316,7 +321,7 @@ var _ = ginkgo.Describe("Job controller", func() {

ginkgo.By("Unset admission of the workload to suspend the job")
gomega.Eventually(func() error {
if err := k8sClient.Get(ctx, parentLookupKey, parentWorkload); err != nil {
if err := k8sClient.Get(ctx, parentWlLookupKey, parentWorkload); err != nil {
return err
}
parentWorkload.Spec.Admission = nil
Expand Down Expand Up @@ -347,21 +352,20 @@ var _ = ginkgo.Describe("Job controller for workloads with no queue set", func()
ginkgo.By("checking the workload is not created when queue name is not set")
job := testing.MakeJob(jobName, jobNamespace).Obj()
gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
lookupKey := types.NamespacedName{Name: jobName, Namespace: jobNamespace}
createdJob := &batchv1.Job{}
gomega.Expect(k8sClient.Get(ctx, lookupKey, createdJob)).Should(gomega.Succeed())

createdWorkload := &kueue.Workload{}
gomega.Consistently(func() bool {
return apierrors.IsNotFound(k8sClient.Get(ctx, lookupKey, createdWorkload))
return apierrors.IsNotFound(k8sClient.Get(ctx, wlLookupKey, createdWorkload))
}, util.ConsistentDuration, util.Interval).Should(gomega.BeTrue())

ginkgo.By("checking the workload is created when queue name is set")
jobQueueName := "test-queue"
createdJob.Annotations = map[string]string{constants.QueueAnnotation: jobQueueName}
gomega.Expect(k8sClient.Update(ctx, createdJob)).Should(gomega.Succeed())
gomega.Eventually(func() error {
return k8sClient.Get(ctx, lookupKey, createdWorkload)
return k8sClient.Get(ctx, wlLookupKey, createdWorkload)
}, util.Timeout, util.Interval).Should(gomega.Succeed())
})
ginkgo.When("The parent-workload annotation is used", func() {
Expand Down Expand Up @@ -417,14 +421,13 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", func() {
jobQueueName := "test-queue"
job.Annotations = map[string]string{constants.QueueAnnotation: jobQueueName}
gomega.Expect(k8sClient.Create(ctx, job)).Should(gomega.Succeed())
lookupKey := types.NamespacedName{Name: jobName, Namespace: jobNamespace}
createdJob := &batchv1.Job{}
gomega.Expect(k8sClient.Get(ctx, lookupKey, createdJob)).Should(gomega.Succeed())

ginkgo.By("Fetch the workload created for the job")
createdWorkload := &kueue.Workload{}
gomega.Eventually(func() error {
return k8sClient.Get(ctx, lookupKey, createdWorkload)
return k8sClient.Get(ctx, wlLookupKey, createdWorkload)
}, util.Timeout, util.Interval).Should(gomega.Succeed())

ginkgo.By("Admit the workload created for the job")
Expand All @@ -437,7 +440,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", func() {
}},
}
gomega.Expect(k8sClient.Update(ctx, createdWorkload)).Should(gomega.Succeed())
gomega.Expect(k8sClient.Get(ctx, lookupKey, createdWorkload)).Should(gomega.Succeed())
gomega.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).Should(gomega.Succeed())

ginkgo.By("Await for the job to be unsuspended")
gomega.Eventually(func() *bool {
Expand All @@ -455,7 +458,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", func() {
if podsReadyTestSpec.beforeCondition != nil {
ginkgo.By("Update the workload status")
gomega.Eventually(func() *metav1.Condition {
gomega.Expect(k8sClient.Get(ctx, lookupKey, createdWorkload)).Should(gomega.Succeed())
gomega.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).Should(gomega.Succeed())
return apimeta.FindStatusCondition(createdWorkload.Status.Conditions, kueue.WorkloadPodsReady)
}, util.Timeout, util.Interval).Should(gomega.BeComparableTo(podsReadyTestSpec.beforeCondition, ignoreConditionTimestamps))
}
Expand All @@ -470,7 +473,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", func() {
gomega.Eventually(func() error {
// the update may need to be retried due to a conflict as the workload gets
// also updated due to setting of the job status.
if err := k8sClient.Get(ctx, lookupKey, createdWorkload); err != nil {
if err := k8sClient.Get(ctx, wlLookupKey, createdWorkload); err != nil {
return err
}
createdWorkload.Spec.Admission = nil
Expand All @@ -480,7 +483,7 @@ var _ = ginkgo.Describe("Job controller when waitForPodsReady enabled", func() {

ginkgo.By("Verify the PodsReady condition is added")
gomega.Eventually(func() *metav1.Condition {
gomega.Expect(k8sClient.Get(ctx, lookupKey, createdWorkload)).Should(gomega.Succeed())
gomega.Expect(k8sClient.Get(ctx, wlLookupKey, createdWorkload)).Should(gomega.Succeed())
return apimeta.FindStatusCondition(createdWorkload.Status.Conditions, kueue.WorkloadPodsReady)
}, util.Timeout, util.Interval).Should(gomega.BeComparableTo(podsReadyTestSpec.wantCondition, ignoreConditionTimestamps))
},
Expand Down