Skip to content

K
k3s
Commit 8d72f66e authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub
Browse files
Options

Merge pull request #32129 from jsafrane/refactor-controller-startup

Automatic merge from submit-queue Refactor volume controller parameters into a structure `persistentvolumecontroller.NewPersistentVolumeController` has 11 arguments now, put them into a structure. Also, rename `NewPersistentVolumeController` to `NewController`, `persistentvolume` is already name of the package. Fixes #30219
parents 45723e07 a54c9e28
Hide whitespace changes
Inline Side-by-side
Showing with 75 additions and 81 deletions
cmd/kube-controller-manager/app/controllermanager.go
View file @ 8d72f66e
......@@ -425,19 +425,16 @@ func StartControllers(s *options.CMServer, kubeconfig *restclient.Config, stop <
if err != nil {
glog.Fatalf("An backward-compatible provisioner could not be created: %v, but one was expected. Provisioning will not work. This functionality is considered an early Alpha version.", err)
}
volumeController := persistentvolumecontroller.NewPersistentVolumeController(
client("persistent-volume-binder"),
s.PVClaimBinderSyncPeriod.Duration,
alphaProvisioner,
ProbeControllerVolumePlugins(cloud, s.VolumeConfiguration),
cloud,
s.ClusterName,
nil, // volumeSource
nil, // claimSource
nil, // classSource
nil, // eventRecorder
s.VolumeConfiguration.EnableDynamicProvisioning,
)
params := persistentvolumecontroller.ControllerParameters{
KubeClient: client("persistent-volume-binder"),
SyncPeriod: s.PVClaimBinderSyncPeriod.Duration,
AlphaProvisioner: alphaProvisioner,
VolumePlugins: ProbeControllerVolumePlugins(cloud, s.VolumeConfiguration),
Cloud: cloud,
ClusterName: s.ClusterName,
EnableDynamicProvisioning: s.VolumeConfiguration.EnableDynamicProvisioning,
}
volumeController := persistentvolumecontroller.NewController(params)
volumeController.Run(wait.NeverStop)
time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
......
contrib/mesos/pkg/controllermanager/controllermanager.go
View file @ 8d72f66e
......@@ -294,19 +294,16 @@ func (s *CMServer) Run(_ []string) error {
if err != nil {
glog.Fatalf("An backward-compatible provisioner could not be created: %v, but one was expected. Provisioning will not work. This functionality is considered an early Alpha version.", err)
}
volumeController := persistentvolumecontroller.NewPersistentVolumeController(
clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "persistent-volume-binder")),
s.PVClaimBinderSyncPeriod.Duration,
alphaProvisioner,
kubecontrollermanager.ProbeControllerVolumePlugins(cloud, s.VolumeConfiguration),
cloud,
s.ClusterName,
nil, // volumeSource
nil, // claimSource
nil, // classSource
nil, // eventRecorder
s.VolumeConfiguration.EnableDynamicProvisioning,
)
params := persistentvolumecontroller.ControllerParameters{
KubeClient: clientset.NewForConfigOrDie(restclient.AddUserAgent(kubeconfig, "persistent-volume-binder")),
SyncPeriod: s.PVClaimBinderSyncPeriod.Duration,
AlphaProvisioner: alphaProvisioner,
VolumePlugins: kubecontrollermanager.ProbeControllerVolumePlugins(cloud, s.VolumeConfiguration),
Cloud: cloud,
ClusterName: s.ClusterName,
EnableDynamicProvisioning: s.VolumeConfiguration.EnableDynamicProvisioning,
}
volumeController := persistentvolumecontroller.NewController(params)
volumeController.Run(wait.NeverStop)
var rootCA []byte
......
pkg/controller/volume/persistentvolume/framework_test.go
View file @ 8d72f66e
......@@ -594,19 +594,18 @@ func newTestController(kubeClient clientset.Interface, volumeSource, claimSource
if classSource == nil {
classSource = fcache.NewFakeControllerSource()
}
ctrl := NewPersistentVolumeController(
kubeClient,
5*time.Second, // sync period
nil, // alpha provisioner
[]vol.VolumePlugin{}, // recyclers
nil, // cloud
"",
volumeSource,
claimSource,
classSource,
record.NewFakeRecorder(1000), // event recorder
enableDynamicProvisioning,
)
params := ControllerParameters{
KubeClient: kubeClient,
SyncPeriod: 5 * time.Second,
VolumePlugins: []vol.VolumePlugin{},
VolumeSource: volumeSource,
ClaimSource: claimSource,
ClassSource: classSource,
EventRecorder: record.NewFakeRecorder(1000),
EnableDynamicProvisioning: enableDynamicProvisioning,
}
ctrl := NewController(params)
// Speed up the test
ctrl.createProvisionedPVInterval = 5 * time.Millisecond
......
pkg/controller/volume/persistentvolume/pv_controller_base.go
View file @ 8d72f66e
......@@ -43,77 +43,84 @@ import (
// process PV/PVC added/updated/deleted events. The real binding, provisioning,
// recycling and deleting is done in pv_controller.go
// NewPersistentVolumeController creates a new PersistentVolumeController
func NewPersistentVolumeController(
kubeClient clientset.Interface,
syncPeriod time.Duration,
alphaProvisioner vol.ProvisionableVolumePlugin,
volumePlugins []vol.VolumePlugin,
cloud cloudprovider.Interface,
clusterName string,
volumeSource, claimSource, classSource cache.ListerWatcher,
eventRecorder record.EventRecorder,
enableDynamicProvisioning bool,
) *PersistentVolumeController {
// ControllerParameters contains arguments for creation of a new
// PersistentVolume controller.
type ControllerParameters struct {
KubeClient clientset.Interface
SyncPeriod time.Duration
AlphaProvisioner vol.ProvisionableVolumePlugin
VolumePlugins []vol.VolumePlugin
Cloud cloudprovider.Interface
ClusterName string
VolumeSource, ClaimSource, ClassSource cache.ListerWatcher
EventRecorder record.EventRecorder
EnableDynamicProvisioning bool
}
// NewController creates a new PersistentVolume controller
func NewController(p ControllerParameters) *PersistentVolumeController {
eventRecorder := p.EventRecorder
if eventRecorder == nil {
broadcaster := record.NewBroadcaster()
broadcaster.StartRecordingToSink(&unversioned_core.EventSinkImpl{Interface: kubeClient.Core().Events("")})
broadcaster.StartRecordingToSink(&unversioned_core.EventSinkImpl{Interface: p.KubeClient.Core().Events("")})
eventRecorder = broadcaster.NewRecorder(api.EventSource{Component: "persistentvolume-controller"})
}
controller := &PersistentVolumeController{
volumes: newPersistentVolumeOrderedIndex(),
claims: cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc),
kubeClient: kubeClient,
kubeClient: p.KubeClient,
eventRecorder: eventRecorder,
runningOperations: goroutinemap.NewGoRoutineMap(false /* exponentialBackOffOnError */),
cloud: cloud,
enableDynamicProvisioning: enableDynamicProvisioning,
clusterName: clusterName,
cloud: p.Cloud,
enableDynamicProvisioning: p.EnableDynamicProvisioning,
clusterName: p.ClusterName,
createProvisionedPVRetryCount: createProvisionedPVRetryCount,
createProvisionedPVInterval: createProvisionedPVInterval,
alphaProvisioner: alphaProvisioner,
alphaProvisioner: p.AlphaProvisioner,
}
controller.volumePluginMgr.InitPlugins(volumePlugins, controller)
controller.volumePluginMgr.InitPlugins(p.VolumePlugins, controller)
if controller.alphaProvisioner != nil {
if err := controller.alphaProvisioner.Init(controller); err != nil {
glog.Errorf("PersistentVolumeController: error initializing alpha provisioner plugin: %v", err)
}
}
volumeSource := p.VolumeSource
if volumeSource == nil {
volumeSource = &cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
return kubeClient.Core().PersistentVolumes().List(options)
return p.KubeClient.Core().PersistentVolumes().List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
return kubeClient.Core().PersistentVolumes().Watch(options)
return p.KubeClient.Core().PersistentVolumes().Watch(options)
},
}
}
controller.volumeSource = volumeSource
claimSource := p.ClaimSource
if claimSource == nil {
claimSource = &cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
return kubeClient.Core().PersistentVolumeClaims(api.NamespaceAll).List(options)
return p.KubeClient.Core().PersistentVolumeClaims(api.NamespaceAll).List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
return kubeClient.Core().PersistentVolumeClaims(api.NamespaceAll).Watch(options)
return p.KubeClient.Core().PersistentVolumeClaims(api.NamespaceAll).Watch(options)
},
}
}
controller.claimSource = claimSource
classSource := p.ClassSource
if classSource == nil {
classSource = &cache.ListWatch{
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
return kubeClient.Storage().StorageClasses().List(options)
return p.KubeClient.Storage().StorageClasses().List(options)
},
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
return kubeClient.Storage().StorageClasses().Watch(options)
return p.KubeClient.Storage().StorageClasses().Watch(options)
},
}
}
......@@ -122,7 +129,7 @@ func NewPersistentVolumeController(
_, controller.volumeController = cache.NewIndexerInformer(
volumeSource,
&api.PersistentVolume{},
syncPeriod,
p.SyncPeriod,
cache.ResourceEventHandlerFuncs{
AddFunc: controller.addVolume,
UpdateFunc: controller.updateVolume,
......@@ -133,7 +140,7 @@ func NewPersistentVolumeController(
_, controller.claimController = cache.NewInformer(
claimSource,
&api.PersistentVolumeClaim{},
syncPeriod,
p.SyncPeriod,
cache.ResourceEventHandlerFuncs{
AddFunc: controller.addClaim,
UpdateFunc: controller.updateClaim,
......@@ -148,7 +155,7 @@ func NewPersistentVolumeController(
classSource,
&storage.StorageClass{},
controller.classes,
syncPeriod,
p.SyncPeriod,
)
return controller
}
......
test/integration/persistentvolumes/persistent_volumes_test.go
View file @ 8d72f66e
......@@ -1124,20 +1124,14 @@ func createClients(ns *api.Namespace, t *testing.T, s *httptest.Server, syncPeri
}
plugins := []volume.VolumePlugin{plugin}
cloud := &fake_cloud.FakeCloud{}
syncPeriod = getSyncPeriod(syncPeriod)
ctrl := persistentvolumecontroller.NewPersistentVolumeController(
binderClient,
syncPeriod,
nil, // alpha provisioner
plugins,
cloud,
"", // cluster name
nil, // volumeSource
nil, // claimSource
nil, // classSource
nil, // eventRecorder
true) // enableDynamicProvisioning
ctrl := persistentvolumecontroller.NewController(
persistentvolumecontroller.ControllerParameters{
KubeClient: binderClient,
SyncPeriod: getSyncPeriod(syncPeriod),
VolumePlugins: plugins,
Cloud: cloud,
EnableDynamicProvisioning: true,
})
watchPV, err := testClient.PersistentVolumes().Watch(api.ListOptions{})
if err != nil {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment