Add featuregates for volcano capabilities

Signed-off-by: Xuzheng Chang <[email protected]>

go.mod

@@ -19,6 +19,7 @@ require (
  github.com/prometheus/common v0.32.1
  github.com/spf13/cobra v1.4.0
  github.com/spf13/pflag v1.0.5
+ github.com/stretchr/testify v1.7.0
  go.uber.org/automaxprocs v1.4.0
  golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd
  golang.org/x/time v0.0.0-20220609170525-579cf78fd858
@@ -78,6 +79,7 @@ require (
  github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
  github.com/opencontainers/go-digest v1.0.0 // indirect
  github.com/opencontainers/selinux v1.10.0 // indirect
+ github.com/pmezard/go-difflib v1.0.0 // indirect
  github.com/prometheus/client_model v0.2.0 // indirect
  github.com/prometheus/procfs v0.7.3 // indirect
  golang.org/x/mod v0.8.0 // indirect

pkg/controllers/job/job_controller.go

@@ -24,6 +24,7 @@ import (
 
  v1 "k8s.io/api/core/v1"
  "k8s.io/apimachinery/pkg/util/wait"
+ utilfeature "k8s.io/apiserver/pkg/util/feature"
  "k8s.io/client-go/informers"
  coreinformers "k8s.io/client-go/informers/core/v1"
  kubeschedulinginformers "k8s.io/client-go/informers/scheduling/v1"
@@ -48,10 +49,12 @@ import (
  batchlister "volcano.sh/apis/pkg/client/listers/batch/v1alpha1"
  buslister "volcano.sh/apis/pkg/client/listers/bus/v1alpha1"
  schedulinglisters "volcano.sh/apis/pkg/client/listers/scheduling/v1beta1"
+
  "volcano.sh/volcano/pkg/controllers/apis"
  jobcache "volcano.sh/volcano/pkg/controllers/cache"
  "volcano.sh/volcano/pkg/controllers/framework"
  "volcano.sh/volcano/pkg/controllers/job/state"
+ "volcano.sh/volcano/pkg/features"
 )
 
 func init() {
@@ -151,39 +154,43 @@ func (cc *jobcontroller) Initialize(opt *framework.ControllerOption) error {
 
  factory := informerfactory.NewSharedInformerFactory(cc.vcClient, 0)
  cc.vcInformerFactory = factory
- cc.jobInformer = factory.Batch().V1alpha1().Jobs()
- cc.jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
- AddFunc: cc.addJob,
- UpdateFunc: cc.updateJob,
- DeleteFunc: cc.deleteJob,
- })
- cc.jobLister = cc.jobInformer.Lister()
- cc.jobSynced = cc.jobInformer.Informer().HasSynced
-
- cc.cmdInformer = factory.Bus().V1alpha1().Commands()
- cc.cmdInformer.Informer().AddEventHandler(
- cache.FilteringResourceEventHandler{
- FilterFunc: func(obj interface{}) bool {
- switch v := obj.(type) {
- case *busv1alpha1.Command:
- if v.TargetObject != nil &&
- v.TargetObject.APIVersion == batchv1alpha1.SchemeGroupVersion.String() &&
- v.TargetObject.Kind == "Job" {
- return true
- }
+ if utilfeature.DefaultFeatureGate.Enabled(features.WorkLoadSupport) {
+ cc.jobInformer = factory.Batch().V1alpha1().Jobs()
+ cc.jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+ AddFunc: cc.addJob,
+ UpdateFunc: cc.updateJob,
+ DeleteFunc: cc.deleteJob,
+ })
+ cc.jobLister = cc.jobInformer.Lister()
+ cc.jobSynced = cc.jobInformer.Informer().HasSynced
+ }
 
- return false
- default:
- return false
- }
- },
- Handler: cache.ResourceEventHandlerFuncs{
- AddFunc: cc.addCommand,
+ if utilfeature.DefaultFeatureGate.Enabled(features.QueueCommandSync) {
+ cc.cmdInformer = factory.Bus().V1alpha1().Commands()
+ cc.cmdInformer.Informer().AddEventHandler(
+ cache.FilteringResourceEventHandler{
+ FilterFunc: func(obj interface{}) bool {
+ switch v := obj.(type) {
+ case *busv1alpha1.Command:
+ if v.TargetObject != nil &&
+ v.TargetObject.APIVersion == batchv1alpha1.SchemeGroupVersion.String() &&
+ v.TargetObject.Kind == "Job" {
+ return true
+ }
+
+ return false
+ default:
+ return false
+ }
+ },
+ Handler: cache.ResourceEventHandlerFuncs{
+ AddFunc: cc.addCommand,
+ },
  },
- },
- )
- cc.cmdLister = cc.cmdInformer.Lister()
- cc.cmdSynced = cc.cmdInformer.Informer().HasSynced
+ )
+  cc.cmdLister = cc.cmdInformer.Lister()
+  cc.cmdSynced = cc.cmdInformer.Informer().HasSynced
+ }
 
  cc.podInformer = sharedInformers.Core().V1().Pods()
  cc.podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -210,9 +217,11 @@ func (cc *jobcontroller) Initialize(opt *framework.ControllerOption) error {
  cc.pgLister = cc.pgInformer.Lister()
  cc.pgSynced = cc.pgInformer.Informer().HasSynced
 
- cc.pcInformer = sharedInformers.Scheduling().V1().PriorityClasses()
- cc.pcLister = cc.pcInformer.Lister()
- cc.pcSynced = cc.pcInformer.Informer().HasSynced
+ if utilfeature.DefaultFeatureGate.Enabled(features.PriorityClass) {
+ cc.pcInformer = sharedInformers.Scheduling().V1().PriorityClasses()
+ cc.pcLister = cc.pcInformer.Lister()
+ cc.pcSynced = cc.pcInformer.Informer().HasSynced
+ }
 
  cc.queueInformer = factory.Scheduling().V1beta1().Queues()
  cc.queueLister = cc.queueInformer.Lister()

pkg/controllers/podgroup/pg_controller.go

@@ -18,6 +18,7 @@ package podgroup
 
 import (
  "k8s.io/apimachinery/pkg/util/wait"
+ utilfeature "k8s.io/apiserver/pkg/util/feature"
  "k8s.io/client-go/informers"
  appinformers "k8s.io/client-go/informers/apps/v1"
  coreinformers "k8s.io/client-go/informers/core/v1"
@@ -34,6 +35,7 @@ import (
  schedulinginformer "volcano.sh/apis/pkg/client/informers/externalversions/scheduling/v1beta1"
  schedulinglister "volcano.sh/apis/pkg/client/listers/scheduling/v1beta1"
  "volcano.sh/volcano/pkg/controllers/framework"
+ "volcano.sh/volcano/pkg/features"
  commonutil "volcano.sh/volcano/pkg/util"
 )
 
@@ -103,13 +105,14 @@ func (pg *pgcontroller) Initialize(opt *framework.ControllerOption) error {
  pg.pgLister = pg.pgInformer.Lister()
  pg.pgSynced = pg.pgInformer.Informer().HasSynced
 
- pg.rsInformer = pg.informerFactory.Apps().V1().ReplicaSets()
- pg.rsSynced = pg.rsInformer.Informer().HasSynced
- pg.rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
- AddFunc: pg.addReplicaSet,
- UpdateFunc: pg.updateReplicaSet,
- })
-
+ if utilfeature.DefaultFeatureGate.Enabled(features.WorkLoadSupport) {
+ pg.rsInformer = pg.informerFactory.Apps().V1().ReplicaSets()
+ pg.rsSynced = pg.rsInformer.Informer().HasSynced
+ pg.rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+ AddFunc: pg.addReplicaSet,
+ UpdateFunc: pg.updateReplicaSet,
+ })
+ }
  return nil
 }
 

pkg/controllers/queue/queue_controller.go

@@ -27,6 +27,7 @@ import (
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  utilruntime "k8s.io/apimachinery/pkg/util/runtime"
  "k8s.io/apimachinery/pkg/util/wait"
+ utilfeature "k8s.io/apiserver/pkg/util/feature"
  "k8s.io/client-go/kubernetes"
  corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
  "k8s.io/client-go/tools/cache"
@@ -46,6 +47,7 @@ import (
  "volcano.sh/volcano/pkg/controllers/apis"
  "volcano.sh/volcano/pkg/controllers/framework"
  queuestate "volcano.sh/volcano/pkg/controllers/queue/state"
+ "volcano.sh/volcano/pkg/features"
 )
 
 func init() {
@@ -137,22 +139,24 @@ func (c *queuecontroller) Initialize(opt *framework.ControllerOption) error {
  DeleteFunc: c.deletePodGroup,
  })
 
- c.cmdInformer = factory.Bus().V1alpha1().Commands()
- c.cmdInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
- FilterFunc: func(obj interface{}) bool {
- switch v := obj.(type) {
- case *busv1alpha1.Command:
- return IsQueueReference(v.TargetObject)
- default:
- return false
- }
- },
- Handler: cache.ResourceEventHandlerFuncs{
- AddFunc: c.addCommand,
- },
- })
- c.cmdLister = c.cmdInformer.Lister()
- c.cmdSynced = c.cmdInformer.Informer().HasSynced
+ if utilfeature.DefaultFeatureGate.Enabled(features.QueueCommandSync) {
+ c.cmdInformer = factory.Bus().V1alpha1().Commands()
+ c.cmdInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+ FilterFunc: func(obj interface{}) bool {
+ switch v := obj.(type) {
+ case *busv1alpha1.Command:
+ return IsQueueReference(v.TargetObject)
+ default:
+ return false
+ }
+ },
+ Handler: cache.ResourceEventHandlerFuncs{
+ AddFunc: c.addCommand,
+ },
+ })
+ c.cmdLister = c.cmdInformer.Lister()
+ c.cmdSynced = c.cmdInformer.Informer().HasSynced
+ }
 
  queuestate.SyncQueue = c.syncQueue
  queuestate.OpenQueue = c.openQueue

pkg/features/volcano_features.go

@@ -0,0 +1,57 @@
+/*
+ Copyright 2023 The Volcano Authors.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package features
+
+import (
+ "k8s.io/apimachinery/pkg/util/runtime"
+ utilfeature "k8s.io/apiserver/pkg/util/feature"
+ "k8s.io/component-base/featuregate"
+)
+
+const (
+ // WorkLoadSupport can cache and operate workload resource, Deployment/Replicas/ReplicationController/StatefulSet resources currently.
+ WorkLoadSupport featuregate.Feature = "WorkLoadSupport"
+
+ // VolcanoJobSupport can identify and schedule volcano job.
+ VolcanoJobSupport featuregate.Feature = "VolcanoJobSupport"
+
+ // QueueCommandSync supports queue command sync.
+ QueueCommandSync featuregate.Feature = "QueueCommandSync"
+
+ // PriorityClass to provide the capacity of preemption at pod group level.
+ PriorityClass featuregate.Feature = "PriorityClass"
+
+ // CSIStorage tracking of available storage capacity that CSI drivers provide
+ CSIStorage featuregate.Feature = "CSIStorage"
+
+ // ResourceTopology supports resources like cpu/memory topology aware.
+ ResourceTopology featuregate.Feature = "ResourceTopology"
+)
+
+func init() {
+ runtime.Must(utilfeature.DefaultMutableFeatureGate.Add(defaultVolcanoFeatureGates))
+}
+
+var defaultVolcanoFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{
+ WorkLoadSupport: {Default: true, PreRelease: featuregate.Alpha},
+ VolcanoJobSupport: {Default: true, PreRelease: featuregate.Alpha},
+ QueueCommandSync: {Default: true, PreRelease: featuregate.Alpha},
+ PriorityClass: {Default: true, PreRelease: featuregate.Alpha},
+ // CSIStorage is explicitly set to false by default.
+ CSIStorage: {Default: false, PreRelease: featuregate.Alpha},
+ ResourceTopology: {Default: true, PreRelease: featuregate.Alpha},
+}

pkg/scheduler/cache/cache.go

@@ -33,6 +33,7 @@ import (
  utilruntime "k8s.io/apimachinery/pkg/util/runtime"
  "k8s.io/apimachinery/pkg/util/sets"
  "k8s.io/apimachinery/pkg/util/wait"
+ utilfeature "k8s.io/apiserver/pkg/util/feature"
  "k8s.io/client-go/informers"
  infov1 "k8s.io/client-go/informers/core/v1"
  schedv1 "k8s.io/client-go/informers/scheduling/v1"
@@ -60,6 +61,7 @@ import (
  vcinformerv1 "volcano.sh/apis/pkg/client/informers/externalversions/scheduling/v1beta1"
 
  "volcano.sh/volcano/cmd/scheduler/app/options"
+ "volcano.sh/volcano/pkg/features"
  schedulingapi "volcano.sh/volcano/pkg/scheduler/api"
  volumescheduling "volcano.sh/volcano/pkg/scheduler/capabilities/volumebinding"
  "volcano.sh/volcano/pkg/scheduler/metrics"
@@ -501,9 +503,11 @@ func newSchedulerCache(config *rest.Config, schedulerNames []string, defaultQueu
  // `SelectorSpread` and `PodTopologySpread` plugins uses the following four so far.
  informerFactory.Core().V1().Namespaces().Informer()
  informerFactory.Core().V1().Services().Informer()
- informerFactory.Core().V1().ReplicationControllers().Informer()
- informerFactory.Apps().V1().ReplicaSets().Informer()
- informerFactory.Apps().V1().StatefulSets().Informer()
+ if utilfeature.DefaultFeatureGate.Enabled(features.WorkLoadSupport) {
+ informerFactory.Core().V1().ReplicationControllers().Informer()
+ informerFactory.Apps().V1().ReplicaSets().Informer()
+ informerFactory.Apps().V1().StatefulSets().Informer()
+ }
 
  // create informer for node information
  sc.nodeInformer = informerFactory.Core().V1().Nodes()
@@ -561,11 +565,11 @@ func newSchedulerCache(config *rest.Config, schedulerNames []string, defaultQueu
  DeleteFunc: sc.DeleteCSINode,
  },
  )
- sc.csiDriverInformer = informerFactory.Storage().V1().CSIDrivers()
- sc.csiStorageCapacityInformer = informerFactory.Storage().V1beta1().CSIStorageCapacities()
 
  var capacityCheck *volumescheduling.CapacityCheck
- if options.ServerOpts.EnableCSIStorage {
+ if options.ServerOpts.EnableCSIStorage && utilfeature.DefaultFeatureGate.Enabled(features.CSIStorage) {
+ sc.csiDriverInformer = informerFactory.Storage().V1().CSIDrivers()
+ sc.csiStorageCapacityInformer = informerFactory.Storage().V1beta1().CSIStorageCapacities()
  capacityCheck = &volumescheduling.CapacityCheck{
  CSIDriverInformer: sc.csiDriverInformer,
  CSIStorageCapacityInformer: sc.csiStorageCapacityInformer,
@@ -621,7 +625,7 @@ func newSchedulerCache(config *rest.Config, schedulerNames []string, defaultQueu
  },
  })
 
- if options.ServerOpts.EnablePriorityClass {
+ if options.ServerOpts.EnablePriorityClass && utilfeature.DefaultFeatureGate.Enabled(features.PriorityClass) {
  sc.pcInformer = informerFactory.Scheduling().V1().PriorityClasses()
  sc.pcInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  AddFunc: sc.AddPriorityClass,
@@ -677,12 +681,14 @@ func newSchedulerCache(config *rest.Config, schedulerNames []string, defaultQueu
  DeleteFunc: sc.DeleteQueueV1beta1,
  })
 
- sc.cpuInformer = vcinformers.Nodeinfo().V1alpha1().Numatopologies()
- sc.cpuInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
- AddFunc: sc.AddNumaInfoV1alpha1,
- UpdateFunc: sc.UpdateNumaInfoV1alpha1,
- DeleteFunc: sc.DeleteNumaInfoV1alpha1,
- })
+ if utilfeature.DefaultFeatureGate.Enabled(features.ResourceTopology) {
+ sc.cpuInformer = vcinformers.Nodeinfo().V1alpha1().Numatopologies()
+ sc.cpuInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+ AddFunc: sc.AddNumaInfoV1alpha1,
+ UpdateFunc: sc.UpdateNumaInfoV1alpha1,
+ DeleteFunc: sc.DeleteNumaInfoV1alpha1,
+ })
+ }
  return sc
 }
 

pkg/scheduler/capabilities/volumebinding/volume_binding.go

@@ -26,6 +26,7 @@ import (
  v1 "k8s.io/api/core/v1"
  apierrors "k8s.io/apimachinery/pkg/api/errors"
  "k8s.io/apimachinery/pkg/runtime"
+ utilfeature "k8s.io/apiserver/pkg/util/feature"
  corelisters "k8s.io/client-go/listers/core/v1"
  "k8s.io/component-helpers/storage/ephemeral"
  "k8s.io/klog/v2"
@@ -37,6 +38,7 @@ import (
  "k8s.io/kubernetes/pkg/scheduler/framework/plugins/names"
 
  "volcano.sh/volcano/cmd/scheduler/app/options"
+ "volcano.sh/volcano/pkg/features"
 )
 
 const (
@@ -377,7 +379,7 @@ func New(plArgs runtime.Object, fh framework.Handle, fts feature.Features) (fram
  storageClassInformer := fh.SharedInformerFactory().Storage().V1().StorageClasses()
  csiNodeInformer := fh.SharedInformerFactory().Storage().V1().CSINodes()
  var capacityCheck *CapacityCheck
- if options.ServerOpts.EnableCSIStorage {
+ if options.ServerOpts.EnableCSIStorage && utilfeature.DefaultFeatureGate.Enabled(features.CSIStorage) {
  capacityCheck = &CapacityCheck{
  CSIDriverInformer: fh.SharedInformerFactory().Storage().V1().CSIDrivers(),
  CSIStorageCapacityInformer: fh.SharedInformerFactory().Storage().V1beta1().CSIStorageCapacities(),