cherry-pick important bugfix to release-1.8 branch for v1.8.x

cmd/controller-manager/app/options/options.go

@@ -19,22 +19,20 @@ package options
 import (
 	"fmt"
 	"os"
-	"time"
 
 	"github.com/spf13/pflag"
 
 	"volcano.sh/volcano/pkg/kube"
 )
 
 const (
-	defaultQPS                            = 50.0
-	defaultBurst                          = 100
-	defaultWorkers                        = 3
-	defaultMaxRequeueNum                  = 15
-	defaultSchedulerName                  = "volcano"
-	defaultHealthzAddress                 = ":11251"
-	defaultDetectionPeriodOfDependsOntask = 100 * time.Millisecond
-	defaultLockObjectNamespace            = "volcano-system"
+	defaultQPS                 = 50.0
+	defaultBurst               = 100
+	defaultWorkers             = 3
+	defaultMaxRequeueNum       = 15
+	defaultSchedulerName       = "volcano"
+	defaultHealthzAddress      = ":11251"
+	defaultLockObjectNamespace = "volcano-system"
 )
 
 // ServerOption is the main context object for the controllers.
@@ -60,11 +58,11 @@ type ServerOption struct {
 	// defaulting to 0.0.0.0:11252
 	HealthzBindAddress string
 	EnableHealthz      bool
-	// For dependent tasks, there is a detection cycle inside volcano
-	// It indicates how often to detect the status of dependent tasks
-	DetectionPeriodOfDependsOntask time.Duration
 	// To determine whether inherit owner's annotations for pods when create podgroup
 	InheritOwnerAnnotations bool
+	// WorkerThreadsForPG is the number of threads syncing podgroup operations
+	// The larger the number, the faster the podgroup processing, but requires more CPU load.
+	WorkerThreadsForPG uint32
 }
 
 type DecryptFunc func(c *ServerOption) error
@@ -94,9 +92,8 @@ func (s *ServerOption) AddFlags(fs *pflag.FlagSet) {
 	fs.IntVar(&s.MaxRequeueNum, "max-requeue-num", defaultMaxRequeueNum, "The number of times a job, queue or command will be requeued before it is dropped out of the queue")
 	fs.StringVar(&s.HealthzBindAddress, "healthz-address", defaultHealthzAddress, "The address to listen on for the health check server.")
 	fs.BoolVar(&s.EnableHealthz, "enable-healthz", false, "Enable the health check; it is false by default")
-	fs.DurationVar(&s.DetectionPeriodOfDependsOntask, "detection-period-of-dependson-task", defaultDetectionPeriodOfDependsOntask, "It indicates how often to detect the status of dependent tasks."+
-		"e.g. --detection-period-of-dependson-task=1s")
 	fs.BoolVar(&s.InheritOwnerAnnotations, "inherit-owner-annotations", true, "Enable inherit owner annotations for pods when create podgroup; it is enabled by default")
+	fs.Uint32Var(&s.WorkerThreadsForPG, "worker-threads-for-podgroup", 1, "The number of threads syncing podgroup operations. The larger the number, the faster the podgroup processing, but requires more CPU load.")
 }
 
 // CheckOptionOrDie checks the LockObjectNamespace.

cmd/controller-manager/app/options/options_test.go

@@ -46,15 +46,15 @@ func TestAddFlags(t *testing.T) {
 			QPS:        defaultQPS,
 			Burst:      200,
 		},
-		PrintVersion:                   false,
-		WorkerThreads:                  defaultWorkers,
-		SchedulerNames:                 []string{"volcano", "volcano2"},
-		MaxRequeueNum:                  defaultMaxRequeueNum,
-		HealthzBindAddress:             ":11251",
-		DetectionPeriodOfDependsOntask: defaultDetectionPeriodOfDependsOntask,
-		InheritOwnerAnnotations:        true,
-		EnableLeaderElection:           true,
-		LockObjectNamespace:            defaultLockObjectNamespace,
+		PrintVersion:            false,
+		WorkerThreads:           defaultWorkers,
+		SchedulerNames:          []string{"volcano", "volcano2"},
+		MaxRequeueNum:           defaultMaxRequeueNum,
+		HealthzBindAddress:      ":11251",
+		InheritOwnerAnnotations: true,
+		EnableLeaderElection:    true,
+		LockObjectNamespace:     defaultLockObjectNamespace,
+		WorkerThreadsForPG:      1,
 	}
 
 	if !reflect.DeepEqual(expected, s) {

cmd/controller-manager/app/server.go

@@ -38,7 +38,6 @@ import (
 	vcclientset "volcano.sh/apis/pkg/client/clientset/versioned"
 	"volcano.sh/volcano/cmd/controller-manager/app/options"
 	"volcano.sh/volcano/pkg/controllers/framework"
-	"volcano.sh/volcano/pkg/controllers/job"
 	"volcano.sh/volcano/pkg/kube"
 )
 
@@ -61,8 +60,6 @@ func Run(opt *options.ServerOption) error {
 		}
 	}
 
-	job.SetDetectionPeriodOfDependsOntask(opt.DetectionPeriodOfDependsOntask)
-
 	run := startControllers(config, opt)
 
 	if !opt.EnableLeaderElection {
@@ -127,6 +124,7 @@ func startControllers(config *rest.Config, opt *options.ServerOption) func(ctx c
 	controllerOpt.VolcanoClient = vcclientset.NewForConfigOrDie(config)
 	controllerOpt.SharedInformerFactory = informers.NewSharedInformerFactory(controllerOpt.KubeClient, 0)
 	controllerOpt.InheritOwnerAnnotations = opt.InheritOwnerAnnotations
+	controllerOpt.WorkerThreadsForPG = opt.WorkerThreadsForPG
 
 	return func(ctx context.Context) {
 		framework.ForeachController(func(c framework.Controller) {

docs/design/Enhance-Generate-PodGroup-OwnerReferences-for-Normal-Pod.md

@@ -68,7 +68,7 @@ When controller need create a podgroup, it will check the pod `ownerReferences`
 
 ### DiscoveryClient & DynamicClient
 From the `ownerReferences`, we can get or create a GVK and owner resource name. 
-Discovery client is foucs on the k8s resources, it can get GVR from GVK. When get GVR and owner resource name, we can use Dynamic client to get the owner resource's `ownerReferences`.
+Discovery client is focused on the k8s resources, it can get GVR from GVK. When get GVR and owner resource name, we can use Dynamic client to get the owner resource's `ownerReferences`.
 
 ### RBAC
 

installer/helm/chart/volcano/templates/controllers.yaml

@@ -20,7 +20,7 @@ rules:
     verbs: ["create", "get", "list", "watch", "delete"]
   - apiGroups: ["batch.volcano.sh"]
     resources: ["jobs"]
-    verbs: ["get", "list", "watch", "update", "delete"]
+    verbs: ["create", "get", "list", "watch", "update", "delete"]
   - apiGroups: ["batch.volcano.sh"]
     resources: ["jobs/status", "jobs/finalizers"]
     verbs: ["update", "patch"]
@@ -54,6 +54,9 @@ rules:
   - apiGroups: ["flow.volcano.sh"]
     resources: ["jobflows", "jobtemplates"]
     verbs: ["get", "list", "watch", "create", "delete", "update"]
+  - apiGroups: [ "flow.volcano.sh" ]
+    resources: [ "jobflows/status", "jobs/finalizers","jobtemplates/status", "jobtemplates/finalizers" ]
+    verbs: [ "update", "patch" ]
   - apiGroups: ["scheduling.k8s.io"]
     resources: ["priorityclasses"]
     verbs: ["get", "list", "watch", "create", "delete"]

installer/volcano-development.yaml

@@ -8527,7 +8527,7 @@ rules:
     verbs: ["create", "get", "list", "watch", "delete"]
   - apiGroups: ["batch.volcano.sh"]
     resources: ["jobs"]
-    verbs: ["get", "list", "watch", "update", "delete"]
+    verbs: ["create", "get", "list", "watch", "update", "delete"]
   - apiGroups: ["batch.volcano.sh"]
     resources: ["jobs/status", "jobs/finalizers"]
     verbs: ["update", "patch"]
@@ -8561,6 +8561,9 @@ rules:
   - apiGroups: ["flow.volcano.sh"]
     resources: ["jobflows", "jobtemplates"]
     verbs: ["get", "list", "watch", "create", "delete", "update"]
+  - apiGroups: [ "flow.volcano.sh" ]
+    resources: [ "jobflows/status", "jobs/finalizers","jobtemplates/status", "jobtemplates/finalizers" ]
+    verbs: [ "update", "patch" ]
   - apiGroups: ["scheduling.k8s.io"]
     resources: ["priorityclasses"]
     verbs: ["get", "list", "watch", "create", "delete"]

pkg/controllers/framework/interface.go

@@ -33,6 +33,7 @@ type ControllerOption struct {
 	MaxRequeueNum         int
 
 	InheritOwnerAnnotations bool
+	WorkerThreadsForPG      uint32
 }
 
 // Controller is the interface of all controllers.

pkg/controllers/job/job_controller_actions.go

@@ -28,7 +28,6 @@ import (
 	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
 	quotav1 "k8s.io/apiserver/pkg/quota/v1"
 	"k8s.io/klog/v2"
 
@@ -378,7 +377,16 @@ func (cc *jobcontroller) syncJob(jobInfo *apis.JobInfo, updateStatus state.Updat
 		go func(taskName string, podToCreateEachTask []*v1.Pod) {
 			taskIndex := jobhelpers.GetTasklndexUnderJob(taskName, job)
 			if job.Spec.Tasks[taskIndex].DependsOn != nil {
-				cc.waitDependsOnTaskMeetCondition(taskName, taskIndex, podToCreateEachTask, job)
+				if !cc.waitDependsOnTaskMeetCondition(taskName, taskIndex, podToCreateEachTask, job) {
+					klog.V(3).Infof("Job %s/%s depends on task not ready", job.Name, job.Namespace)
+					// release wait group
+					for _, pod := range podToCreateEachTask {
+						go func(pod *v1.Pod) {
+							defer waitCreationGroup.Done()
+						}(pod)
+					}
+					return
+				}
 			}
 
 			for _, pod := range podToCreateEachTask {
@@ -477,31 +485,29 @@ func (cc *jobcontroller) syncJob(jobInfo *apis.JobInfo, updateStatus state.Updat
 	return nil
 }
 
-func (cc *jobcontroller) waitDependsOnTaskMeetCondition(taskName string, taskIndex int, podToCreateEachTask []*v1.Pod, job *batch.Job) {
+func (cc *jobcontroller) waitDependsOnTaskMeetCondition(taskName string, taskIndex int, podToCreateEachTask []*v1.Pod, job *batch.Job) bool {
 	if job.Spec.Tasks[taskIndex].DependsOn == nil {
-		return
+		return true
 	}
-
 	dependsOn := *job.Spec.Tasks[taskIndex].DependsOn
 	if len(dependsOn.Name) > 1 && dependsOn.Iteration == batch.IterationAny {
-		wait.PollInfinite(detectionPeriodOfDependsOntask, func() (bool, error) {
-			for _, task := range dependsOn.Name {
-				if cc.isDependsOnPodsReady(task, job) {
-					return true, nil
-				}
+		// any ready to create task, return true
+		for _, task := range dependsOn.Name {
+			if cc.isDependsOnPodsReady(task, job) {
+				return true
 			}
-			return false, nil
-		})
-	} else {
-		for _, dependsOnTask := range dependsOn.Name {
-			wait.PollInfinite(detectionPeriodOfDependsOntask, func() (bool, error) {
-				if cc.isDependsOnPodsReady(dependsOnTask, job) {
-					return true, nil
-				}
-				return false, nil
-			})
+		}
+		// all not ready to skip create task, return false
+		return false
+	}
+	for _, dependsOnTask := range dependsOn.Name {
+		// any not ready to skip create task, return false
+		if !cc.isDependsOnPodsReady(dependsOnTask, job) {
+			return false
 		}
 	}
+	// all ready to create task, return true
+	return true
 }
 
 func (cc *jobcontroller) isDependsOnPodsReady(task string, job *batch.Job) bool {

pkg/controllers/job/job_controller_actions_test.go

@@ -20,12 +20,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"reflect"
-	"testing"
-
 	"github.com/agiledragon/gomonkey/v2"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"reflect"
+	"testing"
 
 	"volcano.sh/apis/pkg/apis/batch/v1alpha1"
 	schedulingapi "volcano.sh/apis/pkg/apis/scheduling/v1beta1"
@@ -261,6 +260,109 @@ func TestSyncJobFunc(t *testing.T) {
 			Plugins:      []string{"svc", "ssh", "env"},
 			ExpectVal:    nil,
 		},
+		{
+			Name: "SyncJob with dependsOn job can't find the dependent task",
+			/*
+				Work dependsOn Master task, preempt actions causes controller deadlock
+				   controller                master,work             scheduler
+				       |                           |  <---preempt-----  |
+				       |                           |  <---kill work---  |
+				       | ----watch work kill---->  |                    |
+					   |                           |  <---kill master-- |
+				       | ----create work pods--->  |                    |
+				       | --wait master running-->  |                    |
+				       |                           |                    |
+					   | ---watch master kill--->  |                    |
+				       | --push master to queue->  |                    |
+				       | -wait process to create-> |                    |
+			*/
+			Job: &v1alpha1.Job{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:            "job1",
+					Namespace:       namespace,
+					ResourceVersion: "100",
+					UID:             "e7f18111-1cec-11ea-b688-fa163ec79500",
+				},
+				Spec: v1alpha1.JobSpec{
+					Tasks: []v1alpha1.TaskSpec{
+						{
+							Name:     "master",
+							Replicas: 1,
+							Template: v1.PodTemplateSpec{
+								ObjectMeta: metav1.ObjectMeta{
+									Name:      "pods",
+									Namespace: namespace,
+								},
+								Spec: v1.PodSpec{
+									Containers: []v1.Container{
+										{
+											Name: "Containers",
+										},
+									},
+								},
+							},
+						},
+						{
+							Name:     "work",
+							Replicas: 3,
+							Template: v1.PodTemplateSpec{
+								ObjectMeta: metav1.ObjectMeta{
+									Name:      "pods",
+									Namespace: namespace,
+								},
+								Spec: v1.PodSpec{
+									Containers: []v1.Container{
+										{
+											Name: "Containers",
+										},
+									},
+								},
+							},
+							DependsOn: &v1alpha1.DependsOn{
+								Name: []string{"master"},
+							},
+						},
+					},
+				},
+				Status: v1alpha1.JobStatus{
+					State: v1alpha1.JobState{
+						Phase: v1alpha1.Pending,
+					},
+				},
+			},
+			PodGroup: &schedulingapi.PodGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "job1-e7f18111-1cec-11ea-b688-fa163ec79500",
+					Namespace: namespace,
+				},
+				Spec: schedulingapi.PodGroupSpec{
+					MinResources:  &v1.ResourceList{},
+					MinTaskMember: map[string]int32{},
+				},
+				Status: schedulingapi.PodGroupStatus{
+					Phase: schedulingapi.PodGroupInqueue,
+				},
+			},
+			PodRetainPhase: state.PodRetainPhaseNone,
+			UpdateStatus:   nil,
+			JobInfo: &apis.JobInfo{
+				Namespace: namespace,
+				Name:      "jobinfo1",
+				Pods: map[string]map[string]*v1.Pod{
+					"work": {
+						"job1-work-0": buildPod(namespace, "job1-work-0", v1.PodRunning, nil),
+						"job1-work-1": buildPod(namespace, "job1-work-1", v1.PodRunning, nil),
+					},
+				},
+			},
+			Pods: map[string]*v1.Pod{
+				"job1-work-0": buildPod(namespace, "job1-work-0", v1.PodRunning, nil),
+				"job1-work-1": buildPod(namespace, "job1-work-1", v1.PodRunning, nil),
+			},
+			TotalNumPods: 4,
+			Plugins:      []string{"svc", "ssh", "env"},
+			ExpectVal:    nil,
+		},
 	}
 	for i, testcase := range testcases {
 

pkg/controllers/job/job_controller_handler.go

@@ -427,7 +427,7 @@ func (cc *jobcontroller) updatePodGroup(oldObj, newObj interface{}) {
 	_, err := cc.cache.Get(jobcache.JobKeyByName(newPG.Namespace, jobNameKey))
 	if err != nil && newPG.Annotations != nil {
 		klog.Warningf(
-			"Failed to find job in cache by PodGroup, this may not be a PodGroup for volcano job.")
+			"Failed to find job in cache by PodGroup(%s/%s), this may not be a PodGroup for volcano job.", newPG.Namespace, newPG.Name)
 	}
 
 	if newPG.Status.Phase != oldPG.Status.Phase {

pkg/controllers/job/job_controller_util.go

@@ -18,7 +18,6 @@ package job
 
 import (
 	"fmt"
-	"time"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -33,8 +32,6 @@ import (
 	jobhelpers "volcano.sh/volcano/pkg/controllers/job/helpers"
 )
 
-var detectionPeriodOfDependsOntask time.Duration
-
 // MakePodName append podname,jobname,taskName and index and returns the string.
 func MakePodName(jobName string, taskName string, index int) string {
 	return fmt.Sprintf(jobhelpers.PodNameFmt, jobName, taskName, index)
@@ -258,7 +255,3 @@ func isControlledBy(obj metav1.Object, gvk schema.GroupVersionKind) bool {
 	}
 	return false
 }
-
-func SetDetectionPeriodOfDependsOntask(period time.Duration) {
-	detectionPeriodOfDependsOntask = period
-}