implement distributed resizing

README.md

@@ -57,6 +57,8 @@ Note that the external-resizer does not scale with more replicas. Only one exter
 
 * `--timeout <duration>`: Timeout of all calls to CSI driver. It should be set to value that accommodates majority of `ControllerExpandVolume` calls. 10 seconds is used by default.
 
+* `--node-deployment`: Enables the resizer sidecar to handle resize operations for the volumes local to the node on which it is deployed. Off by default.
+
 * `-kube-api-burst <int>` : Burst to use while communicating with the kubernetes apiserver. Defaults to 10. (default 10).
 
 * `-kube-api-qps <float>` : QPS to use while communicating with the kubernetes apiserver. Defaults to 5.0. (default 5).
@@ -97,6 +99,16 @@ Note that the external-resizer does not scale with more replicas. Only one exter
 
 * All glog / klog arguments are supported, such as `-v <log level>` or `-alsologtostderr`.
 
+
+### Distributed Resizing
+
+The distributed resizing feature is provided to handle resize operations for local volumes. To use this functionality, the resizer sidecar should be deployed along with the csi driver on each node so that every node manages the resize operations only for the volumes local to that node. This feature can be enabled by setting the following command line option to true:
+
+* `--node-deployment`: Enables the resizer sidecar to handle resize operations for the volumes local to the node on which it is deployed. Off by default.
+
+Other than this, the NODE_NAME environment variable must be set where the CSI resizer sidecar is deployed. The value of NODE_NAME should be the name of the node where the sidecar is running.
+
+
 ### HTTP endpoint
 
 The external-resizer optionally exposes an HTTP endpoint at address:port specified by `--http-endpoint` argument. When set, these two paths are exposed:

cmd/csi-resizer/main.go

@@ -75,6 +75,8 @@ var (
 
 	handleVolumeInUseError = flag.Bool("handle-volume-inuse-error", true, "Flag to turn on/off capability to handle volume in use error in resizer controller. Defaults to true if not set.")
 
+	enableNodeDeployment = flag.Bool("node-deployment", false, "Enables deploying the sidecar controller together with a CSI driver on nodes to manage resizing for node-local volumes.")
+
 	featureGates map[string]bool
 
 	version = "unknown"
@@ -104,6 +106,19 @@ func main() {
 		klog.Fatal(err)
 	}
 
+	// If distributed resizing is enabled and leaderElection is also set to true, return
+	if *enableNodeDeployment && *enableLeaderElection {
+		klog.Error("Leader election cannot happen when node-deployment is set to true")
+		os.Exit(1)
+	}
+
+	if *enableNodeDeployment {
+		node := os.Getenv("NODE_NAME")
+		if node == "" {
+			klog.Fatal("The NODE_NAME environment variable must be set when using --node-deployment.")
+		}
+	}
+
 	var config *rest.Config
 	var err error
 	if *master != "" || *kubeConfig != "" {
@@ -156,7 +171,8 @@ func main() {
 		*timeout,
 		kubeClient,
 		informerFactory,
-		driverName)
+		driverName,
+		*enableNodeDeployment)
 	if err != nil {
 		klog.Fatal(err.Error())
 	}

pkg/controller/controller_test.go

@@ -233,7 +233,7 @@ func TestController(t *testing.T) {
 		pvcInformer := informerFactory.Core().V1().PersistentVolumeClaims()
 		podInformer := informerFactory.Core().V1().Pods()
 
-		csiResizer, err := resizer.NewResizerFromClient(client, 15*time.Second, kubeClient, informerFactory, driverName)
+		csiResizer, err := resizer.NewResizerFromClient(client, 15*time.Second, kubeClient, informerFactory, driverName, false)
 		if err != nil {
 			t.Fatalf("Test %s: Unable to create resizer: %v", test.Name, err)
 		}
@@ -365,7 +365,7 @@ func TestResizePVC(t *testing.T) {
 		pvcInformer := informerFactory.Core().V1().PersistentVolumeClaims()
 		podInformer := informerFactory.Core().V1().Pods()
 
-		csiResizer, err := resizer.NewResizerFromClient(client, 15*time.Second, kubeClient, informerFactory, driverName)
+		csiResizer, err := resizer.NewResizerFromClient(client, 15*time.Second, kubeClient, informerFactory, driverName, false)
 		if err != nil {
 			t.Fatalf("Test %s: Unable to create resizer: %v", test.Name, err)
 		}

pkg/controller/expand_and_recover_test.go

@@ -117,7 +117,7 @@ func TestExpandAndRecover(t *testing.T) {
 
 			kubeClient, informerFactory := fakeK8s(initialObjects)
 
-			csiResizer, err := resizer.NewResizerFromClient(client, 15*time.Second, kubeClient, informerFactory, driverName)
+			csiResizer, err := resizer.NewResizerFromClient(client, 15*time.Second, kubeClient, informerFactory, driverName, false)
 			if err != nil {
 				t.Fatalf("Test %s: Unable to create resizer: %v", test.name, err)
 			}

pkg/resizer/csi_resizer.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"strconv"
 	"time"
 
@@ -47,7 +48,7 @@ func NewResizerFromClient(
 	timeout time.Duration,
 	k8sClient kubernetes.Interface,
 	informerFactory informers.SharedInformerFactory,
-	driverName string) (Resizer, error) {
+	driverName string, enableNodeDeployment bool) (Resizer, error) {
 
 	supportControllerService, err := supportsPluginControllerService(csiClient, timeout)
 	if err != nil {
@@ -81,18 +82,20 @@ func NewResizerFromClient(
 	}
 
 	return &csiResizer{
-		name:    driverName,
-		client:  csiClient,
-		timeout: timeout,
+		name:                 driverName,
+		client:               csiClient,
+		timeout:              timeout,
+		enableNodeDeployment: enableNodeDeployment,
 
 		k8sClient: k8sClient,
 	}, nil
 }
 
 type csiResizer struct {
-	name    string
-	client  csi.Client
-	timeout time.Duration
+	name                 string
+	client               csi.Client
+	timeout              time.Duration
+	enableNodeDeployment bool
 
 	k8sClient kubernetes.Interface
 }
@@ -105,6 +108,7 @@ func (r *csiResizer) Name() string {
 // Resizer will resize the volume if it is CSI volume or is migration enabled in-tree volume
 func (r *csiResizer) CanSupport(pv *v1.PersistentVolume, pvc *v1.PersistentVolumeClaim) bool {
 	resizerName := pvc.Annotations[util.VolumeResizerKey]
+	selectedNode := pvc.Annotations[util.VolumeSelectedNodeKey]
 	// resizerName will be CSI driver name when CSI migration is enabled
 	// otherwise, it will be in-tree plugin name
 	// r.name is the CSI driver name, return true only when they match
@@ -119,7 +123,11 @@ func (r *csiResizer) CanSupport(pv *v1.PersistentVolume, pvc *v1.PersistentVolum
 		return false
 	}
 	if source.Driver != r.name {
-		klog.V(4).Infof("Skip resize PV %s for resizer %s", pv.Name, source.Driver)
+		klog.V(4).Infof("Skip resize PV %s for resizer %s non-matching driver", pv.Name, source.Driver)
+		return false
+	}
+	if r.enableNodeDeployment && selectedNode != os.Getenv("NODE_NAME") {
+		klog.V(4).Infof("Skip resize PV %s for resizer %s non-matching node", pv.Name, source.Driver)
 		return false
 	}
 	return true

pkg/resizer/csi_resizer_test.go

@@ -68,7 +68,7 @@ func TestNewResizer(t *testing.T) {
 		client := csi.NewMockClient("mock", c.SupportsNodeResize, c.SupportsControllerResize, c.SupportsPluginControllerService, c.SupportsControllerSingleNodeMultiWriter)
 		driverName := "mock-driver"
 		k8sClient, informerFactory := fakeK8s()
-		resizer, err := NewResizerFromClient(client, 0, k8sClient, informerFactory, driverName)
+		resizer, err := NewResizerFromClient(client, 0, k8sClient, informerFactory, driverName, false)
 		if err != c.Error {
 			t.Errorf("Case %d: Unexpected error: wanted %v, got %v", i, c.Error, err)
 		}
@@ -160,7 +160,7 @@ func TestResizeMigratedPV(t *testing.T) {
 			client := csi.NewMockClient(driverName, true, true, true, true)
 			client.SetCheckMigratedLabel()
 			k8sClient, informerFactory := fakeK8s()
-			resizer, err := NewResizerFromClient(client, 0, k8sClient, informerFactory, driverName)
+			resizer, err := NewResizerFromClient(client, 0, k8sClient, informerFactory, driverName, false)
 			if err != nil {
 				t.Fatalf("Failed to create resizer: %v", err)
 			}
@@ -428,7 +428,7 @@ func TestCanSupport(t *testing.T) {
 			driverName := tc.driverName
 			client := csi.NewMockClient(driverName, true, true, true, true)
 			k8sClient, informerFactory := fakeK8s()
-			resizer, err := NewResizerFromClient(client, 0, k8sClient, informerFactory, driverName)
+			resizer, err := NewResizerFromClient(client, 0, k8sClient, informerFactory, driverName, false)
 			if err != nil {
 				t.Fatalf("Failed to create resizer: %v", err)
 			}

pkg/util/events.go

@@ -27,5 +27,6 @@ const (
 const (
 	// If CSI migration is enabled, the value will be CSI driver name
 	// Otherwise, it will be in-tree storage plugin name
-	VolumeResizerKey = "volume.kubernetes.io/storage-resizer"
+	VolumeResizerKey      = "volume.kubernetes.io/storage-resizer"
+	VolumeSelectedNodeKey = "volume.kubernetes.io/selected-node"
 )