Merge remote-tracking branch 'origin/master' into add_pv_labels

Author: Sneha-at

cmd/gce-pd-csi-driver/main.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/deviceutils"
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 	metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
 	driver "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-pd-csi-driver"
@@ -41,8 +42,11 @@ var (
 	runNodeService       = flag.Bool("run-node-service", true, "If set to false then the CSI driver does not activate its node service (default: true)")
 	httpEndpoint         = flag.String("http-endpoint", "", "The TCP network address where the prometheus metrics endpoint will listen (example: `:8080`). The default is empty string, which means metrics endpoint is disabled.")
 	metricsPath          = flag.String("metrics-path", "/metrics", "The HTTP path where prometheus metrics will be exposed. Default is `/metrics`.")
+	grpcLogCharCap       = flag.Int("grpc-log-char-cap", 10000, "The maximum amount of characters logged for every grpc responses")
 
-	extraVolumeLabelsStr = flag.String("extra-labels", "", "Extra labels to attach to each PD created. It is a comma separated list of key value pairs like '<key1>=<value1>,<key2>=<value2>'. See https://cloud.google.com/compute/docs/labeling-resources for details")
+	errorBackoffInitialDurationMs = flag.Int("backoff-initial-duration-ms", 200, "The amount of ms for the initial duration of the backoff condition for controller publish/unpublish CSI operations. Default is 200.")
+	errorBackoffMaxDurationMs     = flag.Int("backoff-max-duration-ms", 300000, "The amount of ms for the max duration of the backoff condition for controller publish/unpublish CSI operations. Default is 300000 (5m).")
+	extraVolumeLabelsStr          = flag.String("extra-labels", "", "Extra labels to attach to each PD created. It is a comma separated list of key value pairs like '<key1>=<value1>,<key2>=<value2>'. See https://cloud.google.com/compute/docs/labeling-resources for details")
 
 	attachDiskBackoffDuration = flag.Duration("attach-disk-backoff-duration", 5*time.Second, "Duration for attachDisk backoff")
 	attachDiskBackoffFactor   = flag.Float64("attach-disk-backoff-factor", 0.0, "Factor for attachDisk backoff")
@@ -122,7 +126,9 @@ func handle() {
 		if err != nil {
 			klog.Fatalf("Failed to get cloud provider: %w", err)
 		}
-		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider)
+		initialBackoffDuration := time.Duration(*errorBackoffInitialDurationMs) * time.Millisecond
+		maxBackoffDuration := time.Duration(*errorBackoffMaxDurationMs) * time.Microsecond
+		controllerServer = driver.NewControllerServer(gceDriver, cloudProvider, initialBackoffDuration, maxBackoffDuration)
 	} else if *cloudConfigFilePath != "" {
 		klog.Warningf("controller service is disabled but cloud config given - it has no effect")
 	}
@@ -134,7 +140,7 @@ func handle() {
 		if err != nil {
 			klog.Fatalf("Failed to get safe mounter: %w", err)
 		}
-		deviceUtils := mountmanager.NewDeviceUtils()
+		deviceUtils := deviceutils.NewDeviceUtils()
 		statter := mountmanager.NewStatter(mounter)
 		meta, err := metadataservice.NewMetadataService()
 		if err != nil {
@@ -160,5 +166,5 @@ func handle() {
 	gce.WaitForOpBackoff.Steps = *waitForOpBackoffSteps
 	gce.WaitForOpBackoff.Cap = *waitForOpBackoffCap
 
-	gceDriver.Run(*endpoint)
+	gceDriver.Run(*endpoint, *grpcLogCharCap)
 }

creds/cloud-sa.json

@@ -12,7 +12,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package mountmanager
+package deviceutils
 
 import (
 	"fmt"
@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 	pathutils "k8s.io/utils/path"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/resizefs"
 )
 
 const (
@@ -84,6 +85,9 @@ type DeviceUtils interface {
 	// DisableDevice performs necessary disabling prior to a device being
 	// detached from a node. The path is that from GetDiskByIdPaths.
 	DisableDevice(devicePath string) error
+
+	// Resize returns whether or not a device needs resizing.
+	Resize(resizer resizefs.Resizefs, devicePath string, deviceMountPath string) (bool, error)
 }
 
 type deviceUtils struct {
@@ -274,6 +278,10 @@ func (m *deviceUtils) VerifyDevicePath(devicePaths []string, deviceName string)
 	return devicePath, nil
 }
 
+func (m *deviceUtils) Resize(resizer resizefs.Resizefs, devicePath string, deviceMountPath string) (bool, error) {
+	return resizer.Resize(devicePath, deviceMountPath)
+}
+
 // getDevFsSerial returns the serial number of the /dev/* path at devFsPath.
 // If devFsPath does not start with a known prefix, returns the empty string.
 func getDevFsSerial(devFsPath string) (string, error) {

pkg/mount-manager/device-utils.go renamed to pkg/deviceutils/device-utils.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package mountmanager
+package deviceutils
 
 import (
 	"fmt"

pkg/mount-manager/device-utils_linux.go renamed to pkg/deviceutils/device-utils_linux.go

@@ -1,4 +1,4 @@
-package mountmanager
+package deviceutils
 
 import (
 	"testing"

pkg/mount-manager/device-utils_test.go renamed to pkg/deviceutils/device-utils_test.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package mountmanager
+package deviceutils
 
 func (_ *deviceUtils) DisableDevice(devicePath string) error {
 	// No disabling is necessary on windows.

pkg/mount-manager/device-utils_windows.go renamed to pkg/deviceutils/device-utils_windows.go

@@ -12,7 +12,9 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package mountmanager
+package deviceutils
+
+import "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/resizefs"
 
 type fakeDeviceUtils struct {
 }
@@ -38,3 +40,7 @@ func (_ *fakeDeviceUtils) DisableDevice(devicePath string) error {
 	// No-op for testing.
 	return nil
 }
+
+func (_ *fakeDeviceUtils) Resize(resizer resizefs.Resizefs, devicePath string, deviceMountPath string) (bool, error) {
+	return false, nil
+}

pkg/mount-manager/fake-device-utils.go renamed to pkg/deviceutils/fake-device-utils.go

@@ -37,11 +37,6 @@ import (
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 )
 
-const (
-	errorBackoffInitialDuration = 200 * time.Millisecond
-	errorBackoffMaxDuration     = 5 * time.Minute
-)
-
 type GCEControllerServer struct {
 	Driver        *GCEDriver
 	CloudProvider gce.GCECompute
@@ -517,7 +512,7 @@ func (gceCS *GCEControllerServer) executeControllerPublishVolume(ctx context.Con
 
 	attached, err := diskIsAttachedAndCompatible(deviceName, instance, volumeCapability, readWrite)
 	if err != nil {
-		return nil, status.Error(codes.AlreadyExists, fmt.Sprintf("Disk %v already published to node %v but incompatbile: %v", volKey.Name, nodeID, err))
+		return nil, status.Error(codes.AlreadyExists, fmt.Sprintf("Disk %v already published to node %v but incompatible: %v", volKey.Name, nodeID, err))
 	}
 	if attached {
 		// Volume is attached to node. Success!
@@ -1596,8 +1591,8 @@ func pickRandAndConsecutive(slice []string, n int) ([]string, error) {
 	return ret, nil
 }
 
-func newCsiErrorBackoff() *csiErrorBackoff {
-	return &csiErrorBackoff{flowcontrol.NewBackOff(errorBackoffInitialDuration, errorBackoffMaxDuration)}
+func newCsiErrorBackoff(initialDuration, errorBackoffMaxDuration time.Duration) *csiErrorBackoff {
+	return &csiErrorBackoff{flowcontrol.NewBackOff(initialDuration, errorBackoffMaxDuration)}
 }
 
 func (_ *csiErrorBackoff) backoffId(nodeId, volumeId string) csiErrorBackoffId {

pkg/gce-pd-csi-driver/controller.go

@@ -2149,6 +2149,8 @@ type backoffTesterConfig struct {
 }
 
 func newFakeCsiErrorBackoff(tc *clock.FakeClock) *csiErrorBackoff {
+	errorBackoffInitialDuration := 200 * time.Millisecond
+	errorBackoffMaxDuration := 5 * time.Minute
 	return &csiErrorBackoff{flowcontrol.NewFakeBackOff(errorBackoffInitialDuration, errorBackoffMaxDuration, tc)}
 }
 

pkg/gce-pd-csi-driver/controller_test.go

@@ -16,18 +16,22 @@ package gceGCEDriver
 
 import (
 	"fmt"
+	"time"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/klog/v2"
 	"k8s.io/mount-utils"
 	common "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/deviceutils"
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 	metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
 	mountmanager "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
 )
 
+var maxLogChar int
+
 type GCEDriver struct {
 	name              string
 	vendorVersion     string
@@ -137,7 +141,7 @@ func NewIdentityServer(gceDriver *GCEDriver) *GCEIdentityServer {
 	}
 }
 
-func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, deviceUtils mountmanager.DeviceUtils, meta metadataservice.MetadataService, statter mountmanager.Statter) *GCENodeServer {
+func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, deviceUtils deviceutils.DeviceUtils, meta metadataservice.MetadataService, statter mountmanager.Statter) *GCENodeServer {
 	return &GCENodeServer{
 		Driver:          gceDriver,
 		Mounter:         mounter,
@@ -148,17 +152,19 @@ func NewNodeServer(gceDriver *GCEDriver, mounter *mount.SafeFormatAndMount, devi
 	}
 }
 
-func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute) *GCEControllerServer {
+func NewControllerServer(gceDriver *GCEDriver, cloudProvider gce.GCECompute, errorBackoffInitialDuration, errorBackoffMaxDuration time.Duration) *GCEControllerServer {
 	return &GCEControllerServer{
 		Driver:        gceDriver,
 		CloudProvider: cloudProvider,
 		seen:          map[string]int{},
 		volumeLocks:   common.NewVolumeLocks(),
-		errorBackoff:  newCsiErrorBackoff(),
+		errorBackoff:  newCsiErrorBackoff(errorBackoffInitialDuration, errorBackoffMaxDuration),
 	}
 }
 
-func (gceDriver *GCEDriver) Run(endpoint string) {
+func (gceDriver *GCEDriver) Run(endpoint string, grpcLogCharCap int) {
+	maxLogChar = grpcLogCharCap
+
 	klog.V(4).Infof("Driver: %v", gceDriver.name)
 	//Start the nonblocking GRPC
 	s := NewNonBlockingGRPCServer()

pkg/gce-pd-csi-driver/gce-pd-driver.go

@@ -16,6 +16,7 @@ package gceGCEDriver
 
 import (
 	"testing"
+	"time"
 
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
 )
@@ -43,7 +44,10 @@ func initBlockingGCEDriver(t *testing.T, cloudDisks []*gce.CloudDisk, readyToExe
 func initGCEDriverWithCloudProvider(t *testing.T, cloudProvider gce.GCECompute) *GCEDriver {
 	vendorVersion := "test-vendor"
 	gceDriver := GetGCEDriver()
-	controllerServer := NewControllerServer(gceDriver, cloudProvider)
+	errorBackoffInitialDuration := 200 * time.Millisecond
+	errorBackoffMaxDuration := 5 * time.Minute
+
+	controllerServer := NewControllerServer(gceDriver, cloudProvider, errorBackoffInitialDuration, errorBackoffMaxDuration)
 	err := gceDriver.SetupGCEDriver(driver, vendorVersion, nil, nil, controllerServer, nil)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)

pkg/gce-pd-csi-driver/gce-pd-driver_test.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/mount-utils"
 
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/deviceutils"
 	metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
 	mountmanager "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/resizefs"
@@ -38,7 +39,7 @@ import (
 type GCENodeServer struct {
 	Driver          *GCEDriver
 	Mounter         *mount.SafeFormatAndMount
-	DeviceUtils     mountmanager.DeviceUtils
+	DeviceUtils     deviceutils.DeviceUtils
 	VolumeStatter   mountmanager.Statter
 	MetadataService metadataservice.MetadataService
 
@@ -343,10 +344,9 @@ func (ns *GCENodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStage
 	// Part 4: Resize filesystem.
 	// https://github.com/kubernetes/kubernetes/issues/94929
 	resizer := resizefs.NewResizeFs(ns.Mounter)
-	_, err = resizer.Resize(devicePath, stagingTargetPath)
+	_, err = ns.DeviceUtils.Resize(resizer, devicePath, stagingTargetPath)
 	if err != nil {
-		return nil, status.Error(codes.Internal, fmt.Sprintf("error when resizing volume %s: %v", volumeID, err))
-
+		return nil, status.Error(codes.Internal, fmt.Sprintf("error when resizing volume %s from device '%s' at path '%s': %v", volumeID, devicePath, stagingTargetPath, err))
 	}
 
 	klog.V(4).Infof("NodeStageVolume succeeded on %v to %s", volumeID, stagingTargetPath)
@@ -508,7 +508,7 @@ func (ns *GCENodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpa
 	resizer := resizefs.NewResizeFs(ns.Mounter)
 	_, err = resizer.Resize(devicePath, volumePath)
 	if err != nil {
-		return nil, status.Error(codes.Internal, fmt.Sprintf("error when resizing volume %s: %v", volKey.String(), err))
+		return nil, status.Error(codes.Internal, fmt.Sprintf("error when resizing volume %s from device '%s' at path '%s': %v", volKey.String(), devicePath, volumePath, err))
 
 	}
 

pkg/gce-pd-csi-driver/node.go

@@ -19,16 +19,18 @@ import (
 	"context"
 	"fmt"
 	"io/ioutil"
-	"k8s.io/utils/exec"
-	testingexec "k8s.io/utils/exec/testing"
 	"os"
 	"path/filepath"
 	"testing"
 
+	"k8s.io/utils/exec"
+	testingexec "k8s.io/utils/exec/testing"
+
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/mount-utils"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/deviceutils"
 	metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
 	mountmanager "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
 )
@@ -38,14 +40,14 @@ const defaultTargetPath = "/mnt/test"
 const defaultStagingPath = "/staging"
 
 func getTestGCEDriver(t *testing.T) *GCEDriver {
-	return getCustomTestGCEDriver(t, mountmanager.NewFakeSafeMounter(), mountmanager.NewFakeDeviceUtils(), metadataservice.NewFakeService())
+	return getCustomTestGCEDriver(t, mountmanager.NewFakeSafeMounter(), deviceutils.NewFakeDeviceUtils(), metadataservice.NewFakeService())
 }
 
 func getTestGCEDriverWithCustomMounter(t *testing.T, mounter *mount.SafeFormatAndMount) *GCEDriver {
-	return getCustomTestGCEDriver(t, mounter, mountmanager.NewFakeDeviceUtils(), metadataservice.NewFakeService())
+	return getCustomTestGCEDriver(t, mounter, deviceutils.NewFakeDeviceUtils(), metadataservice.NewFakeService())
 }
 
-func getCustomTestGCEDriver(t *testing.T, mounter *mount.SafeFormatAndMount, deviceUtils mountmanager.DeviceUtils, metaService metadataservice.MetadataService) *GCEDriver {
+func getCustomTestGCEDriver(t *testing.T, mounter *mount.SafeFormatAndMount, deviceUtils deviceutils.DeviceUtils, metaService metadataservice.MetadataService) *GCEDriver {
 	gceDriver := GetGCEDriver()
 	nodeServer := NewNodeServer(gceDriver, mounter, deviceUtils, metaService, mountmanager.NewFakeStatter(mounter))
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nodeServer)
@@ -58,7 +60,7 @@ func getCustomTestGCEDriver(t *testing.T, mounter *mount.SafeFormatAndMount, dev
 func getTestBlockingGCEDriver(t *testing.T, readyToExecute chan chan struct{}) *GCEDriver {
 	gceDriver := GetGCEDriver()
 	mounter := mountmanager.NewFakeSafeBlockingMounter(readyToExecute)
-	nodeServer := NewNodeServer(gceDriver, mounter, mountmanager.NewFakeDeviceUtils(), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter))
+	nodeServer := NewNodeServer(gceDriver, mounter, deviceutils.NewFakeDeviceUtils(), metadataservice.NewFakeService(), mountmanager.NewFakeStatter(mounter))
 	err := gceDriver.SetupGCEDriver(driver, "test-vendor", nil, nil, nil, nodeServer)
 	if err != nil {
 		t.Fatalf("Failed to setup GCE Driver: %v", err)

pkg/gce-pd-csi-driver/node_test.go

@@ -69,7 +69,11 @@ func logGRPC(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, h
 	if err != nil {
 		klog.Errorf("%s returned with error: %w", info.FullMethod, err)
 	} else {
-		klog.V(4).Infof("%s returned with response: %s", info.FullMethod, resp)
+		cappedStr := fmt.Sprintf("%v", resp)
+		if len(cappedStr) > maxLogChar {
+			cappedStr = cappedStr[:maxLogChar] + fmt.Sprintf(" [response body too large, log capped to %d chars]", maxLogChar)
+		}
+		klog.V(4).Infof("%s returned with response: %s", info.FullMethod, cappedStr)
 	}
 	return resp, err
 }

pkg/gce-pd-csi-driver/utils.go

@@ -26,8 +26,8 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/deviceutils"
 	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
-	mountmanager "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
 	testutils "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/test/e2e/utils"
 	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/test/remote"
 
@@ -122,7 +122,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		}()
 
 		// MESS UP THE symlink
-		devicePaths := mountmanager.NewDeviceUtils().GetDiskByIdPaths(volName, "")
+		devicePaths := deviceutils.NewDeviceUtils().GetDiskByIdPaths(volName, "")
 		for _, devicePath := range devicePaths {
 			err = testutils.RmAll(instance, devicePath)
 			Expect(err).To(BeNil(), "failed to remove /dev/by-id folder")
@@ -194,7 +194,7 @@ var _ = Describe("GCE PD CSI Driver", func() {
 		}()
 
 		// DELETE THE symlink
-		devicePaths := mountmanager.NewDeviceUtils().GetDiskByIdPaths(volName, "")
+		devicePaths := deviceutils.NewDeviceUtils().GetDiskByIdPaths(volName, "")
 		for _, devicePath := range devicePaths {
 			err = testutils.RmAll(instance, devicePath)
 			Expect(err).To(BeNil(), "failed to remove /dev/by-id folder")