github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-08-09 03:57:41 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #119291 from mimowo/use-jobctx-for-first-pending

Browse Source 
Pass Job context down to firstPendingIndexes
This commit is contained in:
Kubernetes Prow Robot 2023-07-13 09:57:05 -07:00 committed by GitHub
commit 4fa97eae1b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
pkg/controller/job/indexed_job_utils.go
View File

@ -190,19 +190,19 @@ func succeededIndexesFromString(logger klog.Logger, completedIndexes string, com
// firstPendingIndexes returns `count` indexes less than `completions` that are // firstPendingIndexes returns `count` indexes less than `completions` that are
// not covered by `activePods` or `succeededIndexes`. // not covered by `activePods` or `succeededIndexes`.
func firstPendingIndexes(activePods []*v1.Pod, succeededIndexes orderedIntervals, count, completions int) []int { func firstPendingIndexes(jobCtx *syncJobCtx, count, completions int) []int {
if count == 0 { if count == 0 {
return nil return nil
} }
active := sets.New[int]() active := sets.New[int]()
for _, p := range activePods { for _, p := range jobCtx.activePods {
ix := getCompletionIndex(p.Annotations) ix := getCompletionIndex(p.Annotations)
if ix != unknownCompletionIndex { if ix != unknownCompletionIndex {
active.Insert(ix) active.Insert(ix)
} }
} }
result := make([]int, 0, count) result := make([]int, 0, count)
nonPending := succeededIndexes.withOrderedIndexes(sets.List(active)) nonPending := jobCtx.succeededIndexes.withOrderedIndexes(sets.List(active))
// The following algorithm is bounded by len(nonPending) and count. // The following algorithm is bounded by len(nonPending) and count.
candidate := 0 candidate := 0
for _, sInterval := range nonPending { for _, sInterval := range nonPending {

7
pkg/controller/job/indexed_job_utils_test.go
View File

@ -313,8 +313,11 @@ func TestFirstPendingIndexes(t *testing.T) {
} }
for name, tc := range cases { for name, tc := range cases {
t.Run(name, func(t *testing.T) { t.Run(name, func(t *testing.T) {
pods := hollowPodsWithIndexPhase(tc.activePods) jobCtx := &syncJobCtx{
got := firstPendingIndexes(pods, tc.succeededIndexes, tc.cnt, tc.completions) activePods: hollowPodsWithIndexPhase(tc.activePods),
succeededIndexes: tc.succeededIndexes,
}
got := firstPendingIndexes(jobCtx, tc.cnt, tc.completions)
if diff := cmp.Diff(tc.want, got); diff != "" { if diff := cmp.Diff(tc.want, got); diff != "" {
t.Errorf("Wrong first pending indexes (-want,+got):\n%s", diff) t.Errorf("Wrong first pending indexes (-want,+got):\n%s", diff)
} }

2
pkg/controller/job/job_controller.go
View File

@ -1461,7 +1461,7 @@ func (jm *Controller) manageJob(ctx context.Context, job *batch.Job, jobCtx *syn
var indexesToAdd []int var indexesToAdd []int
if isIndexedJob(job) { if isIndexedJob(job) {
indexesToAdd = firstPendingIndexes(jobCtx.activePods, jobCtx.succeededIndexes, int(diff), int(*job.Spec.Completions)) indexesToAdd = firstPendingIndexes(jobCtx, int(diff), int(*job.Spec.Completions))
diff = int32(len(indexesToAdd)) diff = int32(len(indexesToAdd))
} }
active += diff active += diff