Skip to content

Commit

Permalink
Prefix job workloads with job-
Browse files Browse the repository at this point in the history
  • Loading branch information
@mimowo
mimowo committed Feb 21, 2023
1 parent 9f058c5 commit fdbbced
Show file tree
Hide file tree
Showing 4 changed files with 60 additions and 33 deletions.
21 changes: 21 additions & 0 deletions pkg/controller/workload/common/workload_utils.go
View file
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
}
3 changes: 2 additions & 1 deletion pkg/controller/workload/job/job_controller.go
View file
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
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
62 changes: 32 additions & 30 deletions test/integration/controller/job/job_controller_test.go
View file
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 Expand Up @@ -791,7 +794,6 @@ var _ = ginkgo.Describe("Job controller interacting with scheduler", func() {
gomega.Expect(k8sClient.Create(ctx, prodJob)).Should(gomega.Succeed())

ginkgo.By("job should be suspend")
lookupKey := types.NamespacedName{Name: prodJob.Name, Namespace: prodJob.Namespace}
createdProdJob := &batchv1.Job{}
gomega.Eventually(func() *bool {
gomega.Expect(k8sClient.Get(ctx, lookupKey, createdProdJob)).Should(gomega.Succeed())
Expand Down

0 comments on commit fdbbced

Please sign in to comment.