add implement for CSI plugin

Gopkg.toml

@@ -24,25 +24,30 @@
 #   go-tests = true
 #   unused-packages = true
 
-# TODO: Change to a concrete v1.14 tag after klog is included.
 [[constraint]]
-  branch = "master"
   name = "k8s.io/api"
+  version = "kubernetes-1.14.0-alpha.1"
 
-# TODO: Change to a concrete v1.14 tag after klog is included.
 [[constraint]]
-  branch = "master"
   name = "k8s.io/apimachinery"
+  version = "kubernetes-1.14.0-alpha.1"
 
-#    TODO: Change to a concrete v1.14 tag after klog is included.
 [[constraint]]
-  branch = "master"
   name = "k8s.io/client-go"
+  version = "kubernetes-1.14.0-alpha.1"
 
 [[constraint]]
   name = "k8s.io/klog"
   version = "0.1.0"
 
+[[constraint]]
+  branch = "master"
+  name = "github.com/container-storage-interface/spec"
+
+[[constraint]]
+  name = "github.com/kubernetes-csi/csi-lib-utils"
+  version = "0.2.0"
+
 [prune]
    non-go = true
   go-tests = true

cmd/csi-resizer/main.go

@@ -24,6 +24,8 @@ import (
 	"github.com/kubernetes-csi/external-resizer/pkg/resizer"
 	"github.com/kubernetes-csi/external-resizer/pkg/util"
 
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/informers"
 	"k8s.io/klog"
 )
 
@@ -33,6 +35,9 @@ var (
 	resyncPeriod = flag.Duration("resync-period", time.Minute*10, "Resync period for cache")
 	workers      = flag.Int("workers", 10, "Concurrency to process multiple resize requests")
 
+	csiAddress = flag.String("csi-address", "/run/csi/socket", "Address of the CSI driver socket.")
+	csiTimeout = flag.Duration("csiTimeout", 15*time.Second, "Timeout for waiting for CSI driver socket.")
+
 	enableLeaderElection      = flag.Bool("leader-election", false, "Enable leader election.")
 	leaderElectionIdentity    = flag.String("leader-election-identity", "", "Unique identity of this resizer. Typically name of the pod where the resizer runs.")
 	leaderElectionNamespace   = flag.String("leader-election-namespace", "kube-system", "Namespace where this resizer runs.")
@@ -58,13 +63,20 @@ func main() {
 	flag.Set("logtostderr", "true")
 	flag.Parse()
 
-	resizerName := "csi/example-resizer"
-
 	kubeClient, err := util.NewK8sClient(*master, *kubeConfig)
 	if err != nil {
 		klog.Fatal(err.Error())
 	}
 
+	informerFactory := informers.NewSharedInformerFactory(kubeClient, *resyncPeriod)
+
+	csiResizer, err := resizer.NewCSIResizer(*csiAddress, *csiTimeout, kubeClient, informerFactory)
+	if err != nil {
+		klog.Fatal(err.Error())
+	}
+
+	resizerName := csiResizer.Name()
+
 	var leaderElectionConfig *util.LeaderElectionConfig
 	if *enableLeaderElection {
 		if leaderElectionIdentity == nil || *leaderElectionIdentity == "" {
@@ -80,6 +92,8 @@ func main() {
 		}
 	}
 
-	rc := controller.NewResizeController(resizerName, resizer.New(), kubeClient, *resyncPeriod)
+	rc := controller.NewResizeController(resizerName, csiResizer, kubeClient, *resyncPeriod, informerFactory)
+
+	informerFactory.Start(wait.NeverStop)
 	rc.Run(*workers, leaderElectionConfig)
 }

pkg/controller/controller.go

@@ -48,25 +48,24 @@ type ResizeController interface {
 }
 
 type resizeController struct {
-	name            string
-	resizer         resizer.Resizer
-	kubeClient      kubernetes.Interface
-	claimQueue      workqueue.RateLimitingInterface
-	eventRecorder   record.EventRecorder
-	pvLister        corelisters.PersistentVolumeLister
-	pvSynced        cache.InformerSynced
-	pvcLister       corelisters.PersistentVolumeClaimLister
-	pvcSynced       cache.InformerSynced
-	informerFactory informers.SharedInformerFactory
+	name          string
+	resizer       resizer.Resizer
+	kubeClient    kubernetes.Interface
+	claimQueue    workqueue.RateLimitingInterface
+	eventRecorder record.EventRecorder
+	pvLister      corelisters.PersistentVolumeLister
+	pvSynced      cache.InformerSynced
+	pvcLister     corelisters.PersistentVolumeClaimLister
+	pvcSynced     cache.InformerSynced
 }
 
 // NewResizeController returns a ResizeController.
 func NewResizeController(
 	name string,
 	resizer resizer.Resizer,
 	kubeClient kubernetes.Interface,
-	resyncPeriod time.Duration) ResizeController {
-	informerFactory := informers.NewSharedInformerFactory(kubeClient, resyncPeriod)
+	resyncPeriod time.Duration,
+	informerFactory informers.SharedInformerFactory) ResizeController {
 	pvInformer := informerFactory.Core().V1().PersistentVolumes()
 	pvcInformer := informerFactory.Core().V1().PersistentVolumeClaims()
 
@@ -80,16 +79,15 @@ func NewResizeController(
 		workqueue.DefaultControllerRateLimiter(), fmt.Sprintf("%s-pvc", name))
 
 	ctrl := &resizeController{
-		name:            name,
-		resizer:         resizer,
-		kubeClient:      kubeClient,
-		pvLister:        pvInformer.Lister(),
-		pvSynced:        pvInformer.Informer().HasSynced,
-		pvcLister:       pvcInformer.Lister(),
-		pvcSynced:       pvcInformer.Informer().HasSynced,
-		claimQueue:      claimQueue,
-		eventRecorder:   eventRecorder,
-		informerFactory: informerFactory,
+		name:          name,
+		resizer:       resizer,
+		kubeClient:    kubeClient,
+		pvLister:      pvInformer.Lister(),
+		pvSynced:      pvInformer.Informer().HasSynced,
+		pvcLister:     pvcInformer.Lister(),
+		pvcSynced:     pvcInformer.Informer().HasSynced,
+		claimQueue:    claimQueue,
+		eventRecorder: eventRecorder,
 	}
 
 	// Add a resync period as the PVC's request size can be resized again when we handling
@@ -166,7 +164,6 @@ func (ctrl *resizeController) Run(
 
 		stopCh := ctx.Done()
 
-		ctrl.informerFactory.Start(stopCh)
 		if !cache.WaitForCacheSync(stopCh, ctrl.pvSynced, ctrl.pvcSynced) {
 			klog.Errorf("Cannot sync pv/pvc caches")
 			return