This is an automated email from the ASF dual-hosted git repository.
wilfred-s pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-k8shim.git
The following commit(s) were added to refs/heads/master by this push:
new 04c62413 [YUNIKORN-3329] Reject apps that overflow TG resources (#1053)
04c62413 is described below
commit 04c62413bccdf84d95af47323f149153eee0f9c4
Author: thc1006 <[email protected]>
AuthorDate: Wed Jul 29 11:35:58 2026 +1000
[YUNIKORN-3329] Reject apps that overflow TG resources (#1053)
GetTGResource computed the gang placeholder ask with unchecked int64
accessors. Because minMember and minResource come unbounded from the
task-groups pod annotation, the ask could wrap to a negative value that
is sent to the core as PlaceholderAsk.
Reject the annotation at the parse boundary (GetTaskGroupsFromAnnotation)
when a minResource or the aggregate is negative or would overflowi.
Reject also if the taskgroup sets CPU and its canonical-key vcore.
Signed-off-by: thc1006 <[email protected]>
Closes: #1053
Signed-off-by: Wilfred Spiegelenburg <[email protected]>
---
pkg/cache/utils.go | 59 ++++++++++++++++++++++++
pkg/cache/utils_test.go | 117 ++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 176 insertions(+)
diff --git a/pkg/cache/utils.go b/pkg/cache/utils.go
index f7c98c4d..2ad03151 100644
--- a/pkg/cache/utils.go
+++ b/pkg/cache/utils.go
@@ -21,11 +21,13 @@ package cache
import (
"encoding/json"
"fmt"
+ "math"
v1 "k8s.io/api/core/v1"
"github.com/apache/yunikorn-k8shim/pkg/common/constants"
"github.com/apache/yunikorn-k8shim/pkg/common/utils"
+ siCommon "github.com/apache/yunikorn-scheduler-interface/lib/go/common"
)
func GetTaskGroupsFromAnnotation(pod *v1.Pod) ([]TaskGroup, error) {
@@ -40,6 +42,7 @@ func GetTaskGroupsFromAnnotation(pod *v1.Pod) ([]TaskGroup,
error) {
return nil, err
}
// json.Unmarshal won't return error if name or MinMember is empty, but
will return error if MinResource is empty or error format.
+ totals := make(map[string]int64)
for _, taskGroup := range taskGroups {
if taskGroup.Name == "" {
return nil, fmt.Errorf("can't get taskGroup Name from
pod annotation, %s",
@@ -57,6 +60,62 @@ func GetTaskGroupsFromAnnotation(pod *v1.Pod) ([]TaskGroup,
error) {
return nil, fmt.Errorf("minMember cannot be negative,
%s",
taskGroupInfo)
}
+ if err := validateTaskGroupResources(taskGroup, totals); err !=
nil {
+ return nil, err
+ }
}
return taskGroups, nil
}
+
+// validateTaskGroupResources folds one task group's placeholder-ask
contribution
+// into totals (the canonical-key aggregate across all groups) and rejects an
ask
+// that cannot be a non-negative int64: a negative minResource, an int64
overflow
+// (the accessor, the minMember product, or the cross-group aggregate), or a
+// same-canonical-key collision. It mirrors GetTGResource, which emits
minMember
+// "pods" plus minMember*minResource per resource ("cpu" canonicalizes to
+// siCommon.CPU, "vcore"); a collision (cpu with vcore, or an explicit "pods")
is
+// rejected because GetTGResource would otherwise let one silently overwrite
the
+// other (cpu vs vcore by map-iteration order, an explicit "pods" over the
implicit count).
+func validateTaskGroupResources(taskGroup TaskGroup, totals map[string]int64)
error {
+ members := int64(taskGroup.MinMember)
+ // seen holds the canonical keys this task group has already
contributed, so a
+ // same-group collision is rejected instead of silently overwritten.
The implicit
+ // "pods" count claims the "pods" key up front.
+ seen := map[string]bool{"pods": true}
+ if totals["pods"] > math.MaxInt64-members {
+ return fmt.Errorf("aggregate placeholder request for \"pods\"
overflows int64 across taskGroups")
+ }
+ totals["pods"] += members
+
+ for resName, quantity := range taskGroup.MinResource {
+ if quantity.Sign() < 0 {
+ return fmt.Errorf("minResource %q in taskGroup %q
cannot be negative", resName, taskGroup.Name)
+ }
+ cpu := resName == v1.ResourceCPU.String()
+ canonical := resName
+ milliConvert := int64(1)
+ if cpu {
+ canonical = siCommon.CPU
+ milliConvert = 1000
+ }
+ if seen[canonical] {
+ return fmt.Errorf("minResource %q in taskGroup %q
collides with another resource under canonical key %q", resName,
taskGroup.Name, canonical)
+ }
+ seen[canonical] = true
+ // Reject before the int64 accessor or the minMember product
can overflow:
+ // members*value (in milli-units for cpu) must stay within
int64.
+ if quantity.CmpInt64(math.MaxInt64/(members*milliConvert)) > 0 {
+ return fmt.Errorf("minResource %q in taskGroup %q
overflows int64 when scaled by minMember %d", resName, taskGroup.Name, members)
+ }
+ value := quantity.Value()
+ if cpu {
+ value = quantity.MilliValue()
+ }
+ contribution := members * value
+ if totals[canonical] > math.MaxInt64-contribution {
+ return fmt.Errorf("aggregate placeholder request for %q
overflows int64 across taskGroups", canonical)
+ }
+ totals[canonical] += contribution
+ }
+ return nil
+}
diff --git a/pkg/cache/utils_test.go b/pkg/cache/utils_test.go
index 1e527399..2e988c5c 100644
--- a/pkg/cache/utils_test.go
+++ b/pkg/cache/utils_test.go
@@ -19,6 +19,9 @@
package cache
import (
+ "fmt"
+ "math"
+ "math/big"
"testing"
"gotest.tools/v3/assert"
@@ -26,7 +29,9 @@ import (
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "github.com/apache/yunikorn-k8shim/pkg/common"
"github.com/apache/yunikorn-k8shim/pkg/common/constants"
+ siCommon "github.com/apache/yunikorn-scheduler-interface/lib/go/common"
)
const (
@@ -217,3 +222,115 @@ func TestGetTaskGroupFromAnnotation(t *testing.T) {
assert.Equal(t, taskGroups2[0].MinResource["cpu"],
resource.MustParse("2"))
assert.Equal(t, taskGroups2[0].MinResource["memory"],
resource.MustParse("1Gi"))
}
+
+func TestGetTaskGroupsFromAnnotationValidatesResources(t *testing.T) {
+ pod := &v1.Pod{
+ ObjectMeta: metav1.ObjectMeta{Name: "test-pod-validate",
Namespace: "test", UID: "test-pod-UID-validate"},
+ }
+ rejected := []struct {
+ name string
+ anno string
+ }{
+ {"negative cpu",
`[{"name":"g","minMember":1,"minResource":{"cpu":"-1"}}]`},
+ {"cpu MilliValue overflow",
`[{"name":"g","minMember":1,"minResource":{"cpu":"9223372036854776"}}]`},
+ {"memory Value overflow",
`[{"name":"g","minMember":1,"minResource":{"memory":"9223372036854775808"}}]`},
+ {"minMember times minResource overflow",
`[{"name":"g","minMember":2000000000,"minResource":{"cpu":"5000000"}}]`},
+ {"aggregate overflow across taskGroups",
`[{"name":"a","minMember":1,"minResource":{"memory":"5E"}},{"name":"b","minMember":1,"minResource":{"memory":"5E"}}]`},
+ {"cpu and vcore canonical aggregate overflow",
`[{"name":"a","minMember":1,"minResource":{"cpu":"5P"}},{"name":"b","minMember":1,"minResource":{"vcore":"5E"}}]`},
+ {"cpu and vcore collide within a group",
`[{"name":"g","minMember":1,"minResource":{"cpu":"1","vcore":"1"}}]`},
+ {"explicit pods collides with implicit pods",
`[{"name":"a","minMember":2147483647,"minResource":{}},{"name":"b","minMember":1,"minResource":{"pods":"9223372036854775807"}}]`},
+ }
+ for _, tc := range rejected {
+ pod.Annotations =
map[string]string{constants.AnnotationTaskGroups: tc.anno}
+ tg, err := GetTaskGroupsFromAnnotation(pod)
+ assert.Assert(t, tg == nil, tc.name)
+ assert.Assert(t, err != nil, tc.name)
+ }
+ // A task group with representable, non-negative resources is accepted.
+ pod.Annotations = map[string]string{constants.AnnotationTaskGroups:
`[{"name":"g","minMember":2,"minResource":{"cpu":"1","memory":"1Gi"}}]`}
+ tg, err := GetTaskGroupsFromAnnotation(pod)
+ assert.NilError(t, err)
+ assert.Equal(t, len(tg), 1)
+}
+
+// TestValidatedTaskGroupsNeverProduceNegativePlaceholderAsk is the
caller-level
+// invariant: whenever GetTaskGroupsFromAnnotation accepts an annotation, the
+// placeholder ask that setTaskGroups builds (common.Add over GetTGResource)
must be
+// non-negative on every resource, including after cpu->vcore canonicalization
and
+// the implicit pods count. It exercises the cpu/vcore/pods key overlap that a
+// raw-key aggregate check would miss.
+func TestValidatedTaskGroupsNeverProduceNegativePlaceholderAsk(t *testing.T) {
+ strs := []string{"0", "1", "500m", "1.5", "2", "1Gi", "5E", "5P",
+ "9223372036854775807", "9223372036854775808",
"18446744073709551616"}
+ keys := []string{"cpu", "vcore", "memory", "pods", "nvidia.com/gpu"}
+ members := []int32{1, 2, 2000000000, math.MaxInt32}
+ pod := &v1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "p", Namespace: "n",
UID: "u"}}
+ accepted := 0
+ for _, k1 := range keys {
+ for _, k2 := range keys {
+ for _, s := range strs {
+ for _, m := range members {
+ anno := fmt.Sprintf(
+
`[{"name":"a","minMember":%d,"minResource":{%q:%q}},{"name":"b","minMember":1,"minResource":{%q:%q}}]`,
+ m, k1, s, k2, s)
+ pod.Annotations =
map[string]string{constants.AnnotationTaskGroups: anno}
+ tg, err :=
GetTaskGroupsFromAnnotation(pod)
+ if err != nil {
+ continue // rejected: fine
+ }
+ accepted++
+ // Replicate setTaskGroups exactly.
+ ask :=
common.NewResourceBuilder().Build()
+ for _, g := range tg {
+ ask = common.Add(ask,
common.GetTGResource(g.MinResource, int64(g.MinMember)))
+ }
+ // big.Int oracle:
GetTGResource/common.Add use unchecked int64, so an
+ // accepted annotation whose ask
overflowed would carry a wrong (possibly
+ // still non-negative) Value. Recompute
the exact ask in big.Int and require
+ // an exact per-key match, which
catches an overflow that wraps to any value
+ // and any drift between the
validator's canonicalization and GetTGResource's.
+ want := map[string]*big.Int{}
+ addWant := func(k string, v *big.Int) {
+ if want[k] == nil {
+ want[k] = new(big.Int)
+ }
+ want[k].Add(want[k], v)
+ }
+ for _, g := range tg {
+ m :=
big.NewInt(int64(g.MinMember))
+ addWant("pods", m)
+ for resName, q := range
g.MinResource {
+ canonical, acc :=
resName, q.Value()
+ if resName ==
v1.ResourceCPU.String() {
+ canonical, acc
= siCommon.CPU, q.MilliValue()
+ }
+ addWant(canonical,
new(big.Int).Mul(m, big.NewInt(acc)))
+ }
+ }
+ for name, q := range ask.Resources {
+ if w := want[name]; w == nil ||
w.Cmp(big.NewInt(q.Value)) != 0 {
+ t.Fatalf("placeholder
ask %s=%d does not match exact value %v (overflow) for annotation %s", name,
q.Value, want[name], anno)
+ }
+ }
+ }
+ }
+ }
+ }
+ t.Logf("accepted %d combos, all placeholder asks match the exact
value", accepted)
+ assert.Assert(t, accepted > 0, "no combos accepted; the validator may
be over-rejecting")
+}
+
+// TestValidateTaskGroupResourcesRejectsPodsAggregateOverflow drives the helper
+// directly with a totals accumulator already near MaxInt64.
GetTaskGroupsFromAnnotation
+// cannot reach this state (MinMember is int32 and the number of task groups
is bounded by
+// the 256KB pod-annotation limit, so the "pods" total maxes out around 1e13),
but the
+// accumulator's contract must still hold under composition: adding another
group's members
+// must be rejected rather than silently wrapped past MaxInt64. Without the
guard,
+// totals["pods"] += members would wrap negative and flow into a negative
PlaceholderAsk.
+func TestValidateTaskGroupResourcesRejectsPodsAggregateOverflow(t *testing.T) {
+ totals := map[string]int64{"pods": math.MaxInt64 - 5}
+ tg := TaskGroup{Name: "g", MinMember: 10, MinResource:
map[string]resource.Quantity{}}
+ err := validateTaskGroupResources(tg, totals)
+ assert.Assert(t, err != nil, "expected the pods aggregate overflow to
be rejected")
+ assert.Equal(t, totals["pods"], int64(math.MaxInt64-5), "totals must be
left unchanged when the group is rejected")
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
