Merge pull request #3649 from lowang-bh/fixEnableErrCache

don't enable error cache if task role spec is empty

pkg/scheduler/actions/allocate/allocate.go

@@ -31,10 +31,14 @@ import (
 
 type Action struct {
 	session *framework.Session
+	// configured flag for error cache
+	enablePredicateErrorCache bool
 }
 
 func New() *Action {
-	return &Action{}
+	return &Action{
+		enablePredicateErrorCache: true, // default to enable it
+	}
 }
 
 func (alloc *Action) Name() string {
@@ -43,10 +47,17 @@ func (alloc *Action) Name() string {
 
 func (alloc *Action) Initialize() {}
 
+func (alloc *Action) parseArguments(ssn *framework.Session) {
+	arguments := framework.GetArgOfActionFromConf(ssn.Configurations, alloc.Name())
+	arguments.GetBool(&alloc.enablePredicateErrorCache, conf.EnablePredicateErrCacheKey)
+}
+
 func (alloc *Action) Execute(ssn *framework.Session) {
 	klog.V(5).Infof("Enter Allocate ...")
 	defer klog.V(5).Infof("Leaving Allocate ...")
 
+	alloc.parseArguments(ssn)
+
 	// the allocation for pod may have many stages
 	// 1. pick a queue named Q (using ssn.QueueOrderFn)
 	// 2. pick a job named J from Q (using ssn.JobOrderFn)
@@ -204,7 +215,7 @@ func (alloc *Action) allocateResourcesForTasks(tasks *util.PriorityQueue, job *a
 			break
 		}
 
-		predicateNodes, fitErrors := ph.PredicateNodes(task, allNodes, alloc.predicate, true)
+		predicateNodes, fitErrors := ph.PredicateNodes(task, allNodes, alloc.predicate, alloc.enablePredicateErrorCache)
 		if len(predicateNodes) == 0 {
 			job.NodesFitErrors[task.UID] = fitErrors
 			// Assume that all left tasks are allocatable, but can not meet gang-scheduling min member,

pkg/scheduler/actions/allocate/allocate_test.go

@@ -24,6 +24,7 @@ import (
 	"testing"
 
 	"github.com/agiledragon/gomonkey/v2"
+	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
@@ -53,6 +54,16 @@ func TestMain(m *testing.M) {
 	os.Exit(m.Run())
 }
 
+func TestParseArgs(t *testing.T) {
+	test := uthelper.TestCommonStruct{Name: "set cache false"}
+
+	action := New()
+	test.RegisterSession(nil, []conf.Configuration{{Name: action.Name(),
+		Arguments: map[string]interface{}{conf.EnablePredicateErrCacheKey: false}}})
+	test.Run([]framework.Action{action})
+	assert.False(t, action.enablePredicateErrorCache)
+}
+
 func TestAllocate(t *testing.T) {
 	plugins := map[string]framework.PluginBuilder{
 		drf.PluginName:        drf.New,

pkg/scheduler/actions/backfill/backfill.go

@@ -22,15 +22,20 @@ import (
 	"k8s.io/klog/v2"
 
 	"volcano.sh/volcano/pkg/scheduler/api"
+	"volcano.sh/volcano/pkg/scheduler/conf"
 	"volcano.sh/volcano/pkg/scheduler/framework"
 	"volcano.sh/volcano/pkg/scheduler/metrics"
 	"volcano.sh/volcano/pkg/scheduler/util"
 )
 
-type Action struct{}
+type Action struct {
+	enablePredicateErrorCache bool
+}
 
 func New() *Action {
-	return &Action{}
+	return &Action{
+		enablePredicateErrorCache: true, // default to enable it
+	}
 }
 
 func (backfill *Action) Name() string {
@@ -39,10 +44,17 @@ func (backfill *Action) Name() string {
 
 func (backfill *Action) Initialize() {}
 
+func (backfill *Action) parseArguments(ssn *framework.Session) {
+	arguments := framework.GetArgOfActionFromConf(ssn.Configurations, backfill.Name())
+	arguments.GetBool(&backfill.enablePredicateErrorCache, conf.EnablePredicateErrCacheKey)
+}
+
 func (backfill *Action) Execute(ssn *framework.Session) {
 	klog.V(5).Infof("Enter Backfill ...")
 	defer klog.V(5).Infof("Leaving Backfill ...")
 
+	backfill.parseArguments(ssn)
+
 	predicateFunc := ssn.PredicateForAllocateAction
 
 	// TODO (k82cn): When backfill, it's also need to balance between Queues.
@@ -62,7 +74,7 @@ func (backfill *Action) Execute(ssn *framework.Session) {
 			break
 		}
 
-		predicateNodes, fitErrors := ph.PredicateNodes(task, ssn.NodeList, predicateFunc, true)
+		predicateNodes, fitErrors := ph.PredicateNodes(task, ssn.NodeList, predicateFunc, backfill.enablePredicateErrorCache)
 		if len(predicateNodes) == 0 {
 			job.NodesFitErrors[task.UID] = fitErrors
 			break

pkg/scheduler/actions/preempt/preempt.go

@@ -22,15 +22,20 @@ import (
 	"k8s.io/klog/v2"
 
 	"volcano.sh/volcano/pkg/scheduler/api"
+	"volcano.sh/volcano/pkg/scheduler/conf"
 	"volcano.sh/volcano/pkg/scheduler/framework"
 	"volcano.sh/volcano/pkg/scheduler/metrics"
 	"volcano.sh/volcano/pkg/scheduler/util"
 )
 
-type Action struct{}
+type Action struct {
+	enablePredicateErrorCache bool
+}
 
 func New() *Action {
-	return &Action{}
+	return &Action{
+		enablePredicateErrorCache: true,
+	}
 }
 
 func (pmpt *Action) Name() string {
@@ -39,10 +44,17 @@ func (pmpt *Action) Name() string {
 
 func (pmpt *Action) Initialize() {}
 
+func (pmpt *Action) parseArguments(ssn *framework.Session) {
+	arguments := framework.GetArgOfActionFromConf(ssn.Configurations, pmpt.Name())
+	arguments.GetBool(&pmpt.enablePredicateErrorCache, conf.EnablePredicateErrCacheKey)
+}
+
 func (pmpt *Action) Execute(ssn *framework.Session) {
 	klog.V(5).Infof("Enter Preempt ...")
 	defer klog.V(5).Infof("Leaving Preempt ...")
 
+	pmpt.parseArguments(ssn)
+
 	preemptorsMap := map[api.QueueID]*util.PriorityQueue{}
 	preemptorTasks := map[api.JobID]*util.PriorityQueue{}
 
@@ -116,7 +128,7 @@ func (pmpt *Action) Execute(ssn *framework.Session) {
 
 				preemptor := preemptorTasks[preemptorJob.UID].Pop().(*api.TaskInfo)
 
-				assigned, err = preempt(ssn, stmt, preemptor, func(task *api.TaskInfo) bool {
+				assigned, err = pmpt.preempt(ssn, stmt, preemptor, func(task *api.TaskInfo) bool {
 					// Ignore non running task.
 					if !api.PreemptableStatus(task.Status) {
 						return false
@@ -176,7 +188,7 @@ func (pmpt *Action) Execute(ssn *framework.Session) {
 				preemptor := preemptorTasks[job.UID].Pop().(*api.TaskInfo)
 
 				stmt := framework.NewStatement(ssn)
-				assigned, err := preempt(ssn, stmt, preemptor, func(task *api.TaskInfo) bool {
+				assigned, err := pmpt.preempt(ssn, stmt, preemptor, func(task *api.TaskInfo) bool {
 					// Ignore non running task.
 					if !api.PreemptableStatus(task.Status) {
 						return false
@@ -212,7 +224,7 @@ func (pmpt *Action) Execute(ssn *framework.Session) {
 
 func (pmpt *Action) UnInitialize() {}
 
-func preempt(
+func (pmpt *Action) preempt(
 	ssn *framework.Session,
 	stmt *framework.Statement,
 	preemptor *api.TaskInfo,
@@ -228,7 +240,7 @@ func preempt(
 	predicateFn := ssn.PredicateForPreemptAction
 	// we should filter out those nodes that are UnschedulableAndUnresolvable status got in allocate action
 	allNodes := ssn.GetUnschedulableAndUnresolvableNodesForTask(preemptor)
-	predicateNodes, _ := predicateHelper.PredicateNodes(preemptor, allNodes, predicateFn, true)
+	predicateNodes, _ := predicateHelper.PredicateNodes(preemptor, allNodes, predicateFn, pmpt.enablePredicateErrorCache)
 
 	nodeScores := util.PrioritizeNodes(preemptor, predicateNodes, ssn.BatchNodeOrderFn, ssn.NodeOrderMapFn, ssn.NodeOrderReduceFn)
 

pkg/scheduler/conf/constants.go

@@ -0,0 +1,6 @@
+package conf
+
+const (
+	// EnablePredicateErrCacheKey is the key whether predicate error cache is enabled
+	EnablePredicateErrCacheKey = "predicateErrorCacheEnable"
+)

pkg/scheduler/util/predicate_helper.go

@@ -25,6 +25,13 @@ func (ph *predicateHelper) PredicateNodes(task *api.TaskInfo, nodes []*api.NodeI
 	var errorLock sync.RWMutex
 	fe := api.NewFitErrors()
 
+	// don't enable error cache if task's TaskRole is empty, because different pods with empty TaskRole will all
+	// have the same taskGroupID, and one pod predicate failed, all other pods will also be failed
+	// see issue: https://github.com/volcano-sh/volcano/issues/3527
+	if len(task.TaskRole) == 0 {
+		enableErrorCache = false
+	}
+
 	allNodes := len(nodes)
 	if allNodes == 0 {
 		return make([]*api.NodeInfo, 0), fe