github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-01-06 07:57:35 +00:00
Code Issues Projects Releases Wiki Activity

Automatically install CRDs during controller init

Browse Source
This commit is contained in:
saad-ali
2018-08-27 01:31:05 -07:00
parent 7d673cb8f0
commit fdeb895d25
11 changed files with 128 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
pkg/controller/volume/attachdetach/BUILD
View File

@@ -19,6 +19,7 @@ go_library(
"//pkg/controller/volume/attachdetach/reconciler:go_default_library",
"//pkg/controller/volume/attachdetach/statusupdater:go_default_library",
"//pkg/controller/volume/attachdetach/util:go_default_library",
"//pkg/features:go_default_library",
"//pkg/util/mount:go_default_library",
"//pkg/volume:go_default_library",
"//pkg/volume/util:go_default_library",
@@ -26,11 +27,15 @@ go_library(
"//pkg/volume/util/volumepathhandler:go_default_library",
"//staging/src/k8s.io/api/authentication/v1:go_default_library",
"//staging/src/k8s.io/api/core/v1:go_default_library",
"//staging/src/k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1:go_default_library",
"//staging/src/k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
"//staging/src/k8s.io/client-go/informers/core/v1:go_default_library",
"//staging/src/k8s.io/client-go/kubernetes:go_default_library",
"//staging/src/k8s.io/client-go/kubernetes/scheme:go_default_library",
@@ -39,6 +44,7 @@ go_library(
"//staging/src/k8s.io/client-go/tools/cache:go_default_library",
"//staging/src/k8s.io/client-go/tools/record:go_default_library",
"//staging/src/k8s.io/client-go/util/workqueue:go_default_library",
"//staging/src/k8s.io/csi-api/pkg/apis/csi/v1alpha1:go_default_library",
"//staging/src/k8s.io/csi-api/pkg/client/clientset/versioned:go_default_library",
"//vendor/github.com/golang/glog:go_default_library",
],
@@ -54,6 +60,7 @@ go_test(
"//pkg/controller/volume/attachdetach/testing:go_default_library",
"//pkg/volume:go_default_library",
"//staging/src/k8s.io/api/core/v1:go_default_library",
"//staging/src/k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",

89
pkg/controller/volume/attachdetach/attach_detach_controller.go
View File

@@ -21,16 +21,21 @@ package attachdetach
import (
"fmt"
"net"
"reflect"
"time"
"github.com/golang/glog"
authenticationv1 "k8s.io/api/authentication/v1"
"k8s.io/api/core/v1"
apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
utilfeature "k8s.io/apiserver/pkg/util/feature"
coreinformers "k8s.io/client-go/informers/core/v1"
clientset "k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/scheme"
@@ -39,7 +44,8 @@ import (
kcache "k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/record"
"k8s.io/client-go/util/workqueue"
csiclientset "k8s.io/csi-api/pkg/client/clientset/versioned"
csiapiv1alpha1 "k8s.io/csi-api/pkg/apis/csi/v1alpha1"
csiclient "k8s.io/csi-api/pkg/client/clientset/versioned"
"k8s.io/kubernetes/pkg/cloudprovider"
"k8s.io/kubernetes/pkg/controller"
"k8s.io/kubernetes/pkg/controller/volume/attachdetach/cache"
@@ -48,6 +54,7 @@ import (
"k8s.io/kubernetes/pkg/controller/volume/attachdetach/reconciler"
"k8s.io/kubernetes/pkg/controller/volume/attachdetach/statusupdater"
"k8s.io/kubernetes/pkg/controller/volume/attachdetach/util"
"k8s.io/kubernetes/pkg/features"
"k8s.io/kubernetes/pkg/util/mount"
"k8s.io/kubernetes/pkg/volume"
volumeutil "k8s.io/kubernetes/pkg/volume/util"
@@ -97,7 +104,8 @@ type AttachDetachController interface {
// NewAttachDetachController returns a new instance of AttachDetachController.
func NewAttachDetachController(
kubeClient clientset.Interface,
csiClient csiclientset.Interface,
csiClient csiclient.Interface,
crdClient apiextensionsclient.Interface,
podInformer coreinformers.PodInformer,
nodeInformer coreinformers.NodeInformer,
pvcInformer coreinformers.PersistentVolumeClaimInformer,
@@ -125,6 +133,7 @@ func NewAttachDetachController(
adc := &attachDetachController{
kubeClient: kubeClient,
csiClient: csiClient,
crdClient: crdClient,
pvcLister: pvcInformer.Lister(),
pvcsSynced: pvcInformer.Informer().HasSynced,
pvLister: pvInformer.Lister(),
@@ -138,6 +147,11 @@ func NewAttachDetachController(
pvcQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "pvcs"),
}
// Install required CSI CRDs on API server
if utilfeature.DefaultFeatureGate.Enabled(features.CSICrdAutoInstall) {
adc.installCRDs()
}
if err := adc.volumePluginMgr.InitPlugins(plugins, prober, adc); err != nil {
return nil, fmt.Errorf("Could not initialize volume plugins for Attach/Detach Controller: %+v", err)
}
@@ -240,9 +254,13 @@ type attachDetachController struct {
// the API server.
kubeClient clientset.Interface
// csiClient is the csi.storage.k8s.io API client used by volumehost to communicate with
// the API server.
csiClient csiclientset.Interface
// csiClient is the client used to read/write csi.storage.k8s.io API objects
// from the API server.
csiClient csiclient.Interface
// crdClient is the client used to read/write apiextensions.k8s.io objects
// from the API server.
crdClient apiextensionsclient.Interface
// pvcLister is the shared PVC lister used to fetch and store PVC
// objects from the API server. It is shared with other controllers and
@@ -649,6 +667,65 @@ func (adc *attachDetachController) processVolumesInUse(
}
}
// installCRDs creates the specified CustomResourceDefinition for the CSIDrivers object.
func (adc *attachDetachController) installCRDs() error {
crd := &apiextensionsv1beta1.CustomResourceDefinition{
ObjectMeta: metav1.ObjectMeta{
Name: csiapiv1alpha1.CsiDriverResourcePlural + "." + csiapiv1alpha1.GroupName,
},
Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{
Group: csiapiv1alpha1.GroupName,
Version: csiapiv1alpha1.SchemeGroupVersion.Version,
Scope: apiextensionsv1beta1.ClusterScoped,
Names: apiextensionsv1beta1.CustomResourceDefinitionNames{
Plural: csiapiv1alpha1.CsiDriverResourcePlural,
Kind: reflect.TypeOf(csiapiv1alpha1.CSIDriver{}).Name(),
},
},
}
res, err := adc.crdClient.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd)
if err != nil && !apierrors.IsAlreadyExists(err) {
glog.Errorf("failed to create CSIDrivers CRD: %#v, err: %#v",
res, err)
} else if apierrors.IsAlreadyExists(err) {
glog.Warningf("CSIDrivers CRD already exists: %#v, err: %#v",
res, err)
} else {
glog.Infof("CSIDrivers CRD created successfully: %#v",
res)
}
crd = &apiextensionsv1beta1.CustomResourceDefinition{
ObjectMeta: metav1.ObjectMeta{
Name: csiapiv1alpha1.CsiNodeInfoResourcePlural + "." + csiapiv1alpha1.GroupName,
},
Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{
Group: csiapiv1alpha1.GroupName,
Version: csiapiv1alpha1.SchemeGroupVersion.Version,
Scope: apiextensionsv1beta1.ClusterScoped,
Names: apiextensionsv1beta1.CustomResourceDefinitionNames{
Plural: csiapiv1alpha1.CsiNodeInfoResourcePlural,
Kind: reflect.TypeOf(csiapiv1alpha1.CSINodeInfo{}).Name(),
},
},
}
res, err = adc.crdClient.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd)
if err != nil && !apierrors.IsAlreadyExists(err) {
glog.Errorf("failed to create CSINodeInfo CRD: %#v, err: %#v",
res, err)
} else if apierrors.IsAlreadyExists(err) {
glog.Warningf("CSINodeInfo CRD already exists: %#v, err: %#v",
res, err)
} else {
glog.Infof("CSINodeInfo CRD created successfully: %#v",
res)
}
return nil
}
// VolumeHost implementation
// This is an unfortunate requirement of the current factoring of volume plugin
// initializing code. It requires kubelet specific methods used by the mounting
@@ -759,6 +836,6 @@ func (adc *attachDetachController) GetEventRecorder() record.EventRecorder {
return adc.recorder
}
func (adc *attachDetachController) GetCSIClient() csiclientset.Interface {
func (adc *attachDetachController) GetCSIClient() csiclient.Interface {
return adc.csiClient
}

7
pkg/controller/volume/attachdetach/attach_detach_controller_test.go
View File

@@ -22,6 +22,7 @@ import (
"time"
"k8s.io/api/core/v1"
fakeapiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/types"
@@ -36,11 +37,13 @@ func Test_NewAttachDetachController_Positive(t *testing.T) {
// Arrange
fakeKubeClient := controllervolumetesting.CreateTestClient()
informerFactory := informers.NewSharedInformerFactory(fakeKubeClient, controller.NoResyncPeriodFunc())
fakeApiExtensionsClient := fakeapiextensionsclient.NewSimpleClientset()
// Act
_, err := NewAttachDetachController(
fakeKubeClient,
nil, /* csiClient */
fakeApiExtensionsClient, /* crdClient */
informerFactory.Core().V1().Pods(),
informerFactory.Core().V1().Nodes(),
informerFactory.Core().V1().PersistentVolumeClaims(),
@@ -146,6 +149,7 @@ func Test_AttachDetachControllerRecovery(t *testing.T) {
func attachDetachRecoveryTestCase(t *testing.T, extraPods1 []*v1.Pod, extraPods2 []*v1.Pod) {
fakeKubeClient := controllervolumetesting.CreateTestClient()
fakeApiExtensionsClient := fakeapiextensionsclient.NewSimpleClientset()
informerFactory := informers.NewSharedInformerFactory(fakeKubeClient, time.Second*1)
//informerFactory := informers.NewSharedInformerFactory(fakeKubeClient, time.Second*1)
plugins := controllervolumetesting.CreateTestPlugin()
@@ -215,7 +219,8 @@ func attachDetachRecoveryTestCase(t *testing.T, extraPods1 []*v1.Pod, extraPods2
// Create the controller
adcObj, err := NewAttachDetachController(
fakeKubeClient,
nil,
nil, /* csiClient */
fakeApiExtensionsClient, /* crdClient */
informerFactory.Core().V1().Pods(),
informerFactory.Core().V1().Nodes(),
informerFactory.Core().V1().PersistentVolumeClaims(),