Merge pull request #465 from adam-marek/amarek-pdb-queue

Enable PDB based gang scheduling with discrete queues.

Author: k8s-ci-robot

cmd/kube-batch/app/options/options.go

@@ -33,6 +33,18 @@ type ServerOption struct {
 	NamespaceAsQueue     bool
 	EnableLeaderElection bool
 	LockObjectNamespace  string
+	DefaultQueue         string
+}
+
+var (
+	opts *ServerOption
+)
+
+func Options() *ServerOption {
+	if opts == nil {
+		opts = &ServerOption{}
+	}
+	return opts
 }
 
 // NewServerOption creates a new CMServer with a default config.
@@ -49,6 +61,7 @@ func (s *ServerOption) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&s.SchedulerName, "scheduler-name", "kube-batch", "kube-batch will handle pods with the scheduler-name")
 	fs.StringVar(&s.SchedulerConf, "scheduler-conf", "", "The namespace and name of ConfigMap for scheduler configuration")
 	fs.StringVar(&s.SchedulePeriod, "schedule-period", "1s", "The period between each scheduling cycle")
+	fs.StringVar(&s.DefaultQueue, "default-queue", "", "The name of the queue to fall-back to instead of namespace name")
 	fs.BoolVar(&s.EnableLeaderElection, "leader-elect", s.EnableLeaderElection, "Start a leader election client and gain leadership before "+
 		"executing the main loop. Enable this when running replicated kube-batch for high availability")
 	fs.BoolVar(&s.NamespaceAsQueue, "enable-namespace-as-queue", true, "Make Namespace as Queue with weight one, "+

cmd/kube-batch/main.go

@@ -33,7 +33,7 @@ import (
 var logFlushFreq = pflag.Duration("log-flush-frequency", 5*time.Second, "Maximum number of seconds between log flushes")
 
 func main() {
-	s := options.NewServerOption()
+	s := options.Options()
 	s.AddFlags(pflag.CommandLine)
 
 	flag.InitFlags()

pkg/scheduler/api/job_info.go

@@ -23,6 +23,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 
+	"github.com/kubernetes-sigs/kube-batch/cmd/kube-batch/app/options"
 	arbcorev1 "github.com/kubernetes-sigs/kube-batch/pkg/apis/scheduling/v1alpha1"
 	"github.com/kubernetes-sigs/kube-batch/pkg/apis/utils"
 )
@@ -159,10 +160,17 @@ func (ji *JobInfo) SetPodGroup(pg *arbcorev1.PodGroup) {
 	ji.Namespace = pg.Namespace
 	ji.MinAvailable = pg.Spec.MinMember
 
-	if len(pg.Spec.Queue) == 0 {
-		ji.Queue = QueueID(pg.Namespace)
-	} else {
+	//set queue name based on the available information
+	//in the following priority order:
+	// 1. queue name from PodGroup spec (if available)
+	// 2. queue name from default-queue command line option (if specified)
+	// 3. namespace name
+	if len(pg.Spec.Queue) > 0 {
 		ji.Queue = QueueID(pg.Spec.Queue)
+	} else if len(options.Options().DefaultQueue) > 0 {
+		ji.Queue = QueueID(options.Options().DefaultQueue)
+	} else {
+		ji.Queue = QueueID(pg.Namespace)
 	}
 
 	ji.CreationTimestamp = pg.GetCreationTimestamp()
@@ -173,7 +181,11 @@ func (ji *JobInfo) SetPDB(pdb *policyv1.PodDisruptionBudget) {
 	ji.Name = pdb.Name
 	ji.MinAvailable = pdb.Spec.MinAvailable.IntVal
 	ji.Namespace = pdb.Namespace
-	ji.Queue = QueueID(pdb.Namespace)
+	if len(options.Options().DefaultQueue) == 0 {
+		ji.Queue = QueueID(pdb.Namespace)
+	} else {
+		ji.Queue = QueueID(options.Options().DefaultQueue)
+	}
 
 	ji.CreationTimestamp = pdb.GetCreationTimestamp()
 	ji.PDB = pdb