Skip to content

Commit

Permalink
Add featuregates for volcano capabilities
Browse files Browse the repository at this point in the history
Signed-off-by: Xuzheng Chang <changxuzheng@huawei.com>
  • Loading branch information
Monokaix committed Aug 16, 2023
1 parent 39c6f02 commit 76c70f2
Show file tree
Hide file tree
Showing 7 changed files with 158 additions and 73 deletions.
77 changes: 43 additions & 34 deletions pkg/controllers/job/job_controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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() {
Expand Down Expand Up @@ -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{
Expand All @@ -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()
Expand Down
17 changes: 10 additions & 7 deletions pkg/controllers/podgroup/pg_controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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"
)

Expand Down Expand Up @@ -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
}

Expand Down
36 changes: 20 additions & 16 deletions pkg/controllers/queue/queue_controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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() {
Expand Down Expand Up @@ -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
Expand Down
57 changes: 57 additions & 0 deletions pkg/features/volcano_features.go
Original file line number Diff line number Diff line change
@@ -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 **K8s native 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},
}
32 changes: 19 additions & 13 deletions pkg/scheduler/cache/cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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
}

Expand Down
4 changes: 3 additions & 1 deletion pkg/scheduler/capabilities/volumebinding/volume_binding.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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 (
Expand Down Expand Up @@ -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(),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,21 +21,24 @@ import (
"reflect"
"testing"

"volcano.sh/volcano/cmd/scheduler/app/options"

"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
v1 "k8s.io/api/core/v1"
storagev1 "k8s.io/api/storage/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/wait"
utilfeature "k8s.io/apiserver/pkg/util/feature"
"k8s.io/client-go/informers"
"k8s.io/client-go/kubernetes/fake"
featuregatetesting "k8s.io/component-base/featuregate/testing"
"k8s.io/kubernetes/pkg/scheduler/apis/config"
"k8s.io/kubernetes/pkg/scheduler/framework"
"k8s.io/kubernetes/pkg/scheduler/framework/plugins/feature"
"k8s.io/kubernetes/pkg/scheduler/framework/runtime"

"volcano.sh/volcano/cmd/scheduler/app/options"
"volcano.sh/volcano/pkg/features"
)

var (
Expand Down Expand Up @@ -596,6 +599,7 @@ func TestVolumeBinding(t *testing.T) {
options.ServerOpts = &options.ServerOption{
EnableCSIStorage: true,
}
defer featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.CSIStorage, true)()
for _, item := range table {
t.Run(item.name, func(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
Expand Down

0 comments on commit 76c70f2

Please sign in to comment.