Job: Implement metSucceededIndexesCriteria function instead of isSubs…

…etOf function Signed-off-by: Yuki Iwai <yuki.iwai.tz@gmail.com>
kubernetes · Feb 29, 2024 · af26f57 · af26f57
1 parent 73b8903
commit af26f57
Show file tree

Hide file tree

Showing 4 changed files with 213 additions and 130 deletions.
diff --git a/pkg/controller/job/indexed_job_utils.go b/pkg/controller/job/indexed_job_utils.go
@@ -201,42 +201,6 @@ func (oi orderedIntervals) has(ix int) bool {
 	return oi[hi].First <= ix
 }
 
-func (oi orderedIntervals) isSubsetOf(superIntervals orderedIntervals) (int, bool) {
-	if len(oi) == 0 && len(superIntervals) == 0 {
-		return 0, true
-	} else if len(oi) == 0 {
-		return superIntervals.total(), true
-	} else if len(superIntervals) == 0 {
-		return oi.total(), true
-	}
-	var diff int
-	for _, sub := range oi {
-		found := false
-		var contains int
-		for _, super := range superIntervals {
-			// The range of sub exists inside the range of super.
-			// (super.First <= sub.First <= super.Last) AND (super.First <= sub.Last <= super.First)
-			if (sub.First >= super.First && sub.First <= super.Last) &&
-				(sub.Last >= super.First && sub.Last <= super.Last) {
-				found = true
-				break
-			}
-			// The range of sub overlaps the range of super at First OR Last.
-			// (super.First <= sub.First <= super.Last) OR (super.First <= sub.Last <= super.First)
-			// Otherwise and the range of sub completely overlaps the range of super.
-			// (sub.First <= super.First AND sub.Last >= super.Last)
-			if (sub.First >= super.First && sub.First <= super.Last) || (sub.Last >= super.First && sub.Last <= super.Last) ||
-				(sub.First <= super.First && sub.Last >= super.Last) {
-				contains += min(sub.Last, super.Last) - max(sub.First, super.First) + 1
-			}
-		}
-		if !found {
-			diff += (sub.Last - sub.First + 1) - contains
-		}
-	}
-	return diff, diff == 0
-}
-
 func parseIndexesFromString(logger klog.Logger, indexesStr string, completions int) orderedIntervals {
 	if indexesStr == "" {
 		return nil

diff --git a/pkg/controller/job/indexed_job_utils_test.go b/pkg/controller/job/indexed_job_utils_test.go
@@ -690,89 +690,6 @@ func TestIntervalsHaveIndex(t *testing.T) {
 	}
 }
 
-func TestIntervalsIsSubsetOfSuperIntervals(t *testing.T) {
-	cases := map[string]struct {
-		sub          orderedIntervals
-		super        orderedIntervals
-		wantDiff     int
-		wantIsSubset bool
-	}{
-		"sub is proper subset of super": {
-			sub:          []interval{{2, 4}, {6, 8}},
-			super:        []interval{{1, 5}, {6, 9}},
-			wantIsSubset: true,
-		},
-		"sub is proper subset of super; sub equals super": {
-			sub:          []interval{{2, 4}, {6, 8}},
-			super:        []interval{{2, 4}, {6, 9}},
-			wantIsSubset: true,
-		},
-		"sub is subset of super": {
-			sub:          []interval{{2, 4}, {8, 12}},
-			super:        []interval{{2, 5}, {7, 15}},
-			wantIsSubset: true,
-		},
-		"sub is subset of super; sub is an empty set": {
-			sub:          []interval{},
-			super:        []interval{{1, 3}},
-			wantIsSubset: true,
-			wantDiff:     3,
-		},
-		"sub is subset of super; super is an empty set": {
-			sub:          []interval{{2, 4}},
-			super:        []interval{},
-			wantIsSubset: true,
-			wantDiff:     3,
-		},
-		"sub is subset of super; sub and super are an empty set": {
-			sub:          []interval{},
-			super:        []interval{},
-			wantIsSubset: true,
-		},
-		"sub isn't subset of super": {
-			sub:      []interval{{2, 4}, {6, 8}},
-			super:    []interval{{1, 3}, {7, 9}},
-			wantDiff: 2,
-		},
-		"sub isn't subset of super; all sub elements aren't included in super": {
-			sub:      []interval{{10, 12}, {14, 16}},
-			super:    []interval{{1, 3}, {5, 7}},
-			wantDiff: 6,
-		},
-		"sub isn't subset of super; sub overlaps super at first": {
-			sub:      []interval{{1, 3}, {5, 7}},
-			super:    []interval{{2, 4}, {6, 8}},
-			wantDiff: 2,
-		},
-		"sub isn't subset of super; sub overlaps super at last": {
-			sub:      []interval{{2, 4}, {6, 9}},
-			super:    []interval{{1, 3}, {5, 7}},
-			wantDiff: 3,
-		},
-		"sub isn't subset of super; sub completely overlaps super": {
-			sub:      []interval{{0, 4}, {6, 9}},
-			super:    []interval{{1, 3}, {7, 8}},
-			wantDiff: 4,
-		},
-		"sub isn't subset of super; sub overlaps multiple super at last": {
-			sub:      []interval{{0, 6}, {8, 9}},
-			super:    []interval{{1, 3}, {5, 9}},
-			wantDiff: 2,
-		},
-	}
-	for name, tc := range cases {
-		t.Run(name, func(t *testing.T) {
-			gotDiff, gotIsSubset := tc.sub.isSubsetOf(tc.super)
-			if tc.wantIsSubset != gotIsSubset {
-				t.Errorf("Unexpected isSubset\nwant:%v, got:%v", tc.wantIsSubset, gotIsSubset)
-			}
-			if tc.wantDiff != gotDiff {
-				t.Errorf("Unexpected diff\nwant:%d, got:%d", tc.wantDiff, gotDiff)
-			}
-		})
-	}
-}
-
 func TestFirstPendingIndexes(t *testing.T) {
 	cases := map[string]struct {
 		cnt              int

diff --git a/pkg/controller/job/success_policy.go b/pkg/controller/job/success_policy.go
@@ -43,8 +43,7 @@ func metSuccessPolicy(logger klog.Logger, successPolicy *batch.SuccessPolicy, co
 			if len(requiredIndexes) == 0 {
 				continue
 			}
-			diff, met := requiredIndexes.isSubsetOf(succeededIndexes)
-			if met || criteria.SucceededCount != nil && (requiredIndexes.total()-diff) >= int(*criteria.SucceededCount) {
+			if metSucceededIndexesCriteria(requiredIndexes, succeededIndexes, criteria.SucceededCount) {
 				return fmt.Sprintf("%s %d", criteriaMetMsg, index), true
 			}
 		} else if criteria.SucceededCount != nil && succeededIndexes.total() >= int(*criteria.SucceededCount) {
@@ -53,3 +52,42 @@ func metSuccessPolicy(logger klog.Logger, successPolicy *batch.SuccessPolicy, co
 	}
 	return "", false
 }
+
+func metSucceededIndexesCriteria(requiredIndexes, succeededIndexes orderedIntervals, succeededCount *int32) bool {
+	if len(requiredIndexes) == 0 || len(succeededIndexes) == 0 {
+		return false
+	}
+	var contains, succeededPointer int
+	for _, criteria := range requiredIndexes {
+		for succeededPointer < len(succeededIndexes) {
+			// The range of criteria exists inside the range of succeededIndexes.
+			// 	(succeededIndexes[pointer].First <= criteria.First <= succeededIndexes[pointer].Last) AND
+			//	(succeededIndexes[pointer].First <= criteria.Last <= succeededIndexes[pointer].First)
+			if (criteria.First >= succeededIndexes[succeededPointer].First && criteria.First <= succeededIndexes[succeededPointer].Last) &&
+				(criteria.Last >= succeededIndexes[succeededPointer].First && criteria.Last <= succeededIndexes[succeededPointer].Last) {
+				contains += criteria.Last - criteria.First + 1
+				break
+			}
+			// The range of criteria overlaps the range of succeededIndexes at First OR Last.
+			// 	(succeededIndexes[pointer].First <= criteria.First <= succeededIndexes[pointer].Last) OR
+			//	(succeededIndexes[pointer].First <= criteria.Last <= succeededIndexes[pointer].First)
+			// Otherwise and the range of criteria completely overlaps the range of succeededIndexes.
+			// 	(criteria.First <= succeededIndexes[pointer].First AND criteria.Last >= succeededIndexes[pointer].Last)
+			if (criteria.First >= succeededIndexes[succeededPointer].First && criteria.First <= succeededIndexes[succeededPointer].Last) ||
+				(criteria.Last >= succeededIndexes[succeededPointer].First && criteria.Last <= succeededIndexes[succeededPointer].Last) ||
+				(criteria.First <= succeededIndexes[succeededPointer].First && criteria.Last >= succeededIndexes[succeededPointer].Last) {
+				contains += min(criteria.Last, succeededIndexes[succeededPointer].Last) - max(criteria.First, succeededIndexes[succeededPointer].First) + 1
+			}
+
+			if succeededIndexes[succeededPointer].Last < criteria.Last {
+				succeededPointer++
+			} else if succeededIndexes[succeededPointer].Last == criteria.Last {
+				succeededPointer++
+				break
+			} else {
+				break
+			}
+		}
+	}
+	return contains == requiredIndexes.total() || (succeededCount != nil && contains >= int(*succeededCount))
+}