Merge pull request #1997 from torredil/csi-refactor-mounter-node-v2

Fix relationship between node service and mounter interface

cmd/main.go

@@ -28,6 +28,7 @@ import (
 	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/cloud/metadata"
 	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/driver"
 	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/metrics"
+	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/mounter"
 	flag "github.com/spf13/pflag"
 	"k8s.io/component-base/featuregate"
 	logsapi "k8s.io/component-base/logs/api/v1"
@@ -132,19 +133,33 @@ func main() {
 		r.InitializeMetricsHandler(options.HttpEndpoint, "/metrics")
 	}
 
+	cfg := metadata.MetadataServiceConfig{
+		EC2MetadataClient: metadata.DefaultEC2MetadataClient,
+		K8sAPIClient:      metadata.DefaultKubernetesAPIClient,
+	}
+
 	region := os.Getenv("AWS_REGION")
+	var md metadata.MetadataService
+	var metadataErr error
+
 	if region == "" {
 		klog.V(5).InfoS("[Debug] Retrieving region from metadata service")
-		cfg := metadata.MetadataServiceConfig{
-			EC2MetadataClient: metadata.DefaultEC2MetadataClient,
-			K8sAPIClient:      metadata.DefaultKubernetesAPIClient,
-		}
-		metadata, metadataErr := metadata.NewMetadataService(cfg, region)
+		md, metadataErr = metadata.NewMetadataService(cfg, region)
 		if metadataErr != nil {
 			klog.ErrorS(metadataErr, "Could not determine region from any metadata service. The region can be manually supplied via the AWS_REGION environment variable.")
-			panic(err)
+			panic(metadataErr)
+		}
+		region = md.GetRegion()
+	}
+
+	if md == nil {
+		if options.Mode == driver.NodeMode || options.Mode == driver.AllMode {
+			md, metadataErr = metadata.NewMetadataService(cfg, region)
+			if metadataErr != nil {
+				klog.ErrorS(metadataErr, "failed to initialize metadata service")
+				klog.FlushAndExit(klog.ExitFlushTimeout, 1)
+			}
 		}
-		region = metadata.GetRegion()
 	}
 
 	cloud, err := cloud.NewCloud(region, options.AwsSdkDebugLog, options.UserAgentExtra, options.Batching)
@@ -153,7 +168,17 @@ func main() {
 		klog.FlushAndExit(klog.ExitFlushTimeout, 1)
 	}
 
-	drv, err := driver.NewDriver(cloud, &options)
+	m, err := mounter.NewNodeMounter()
+	if err != nil {
+		panic(err)
+	}
+
+	k8sClient, err := cfg.K8sAPIClient()
+	if err != nil {
+		klog.V(2).InfoS("Failed to setup k8s client")
+	}
+
+	drv, err := driver.NewDriver(cloud, &options, m, md, k8sClient)
 	if err != nil {
 		klog.ErrorS(err, "failed to create driver")
 		klog.FlushAndExit(klog.ExitFlushTimeout, 1)

hack/update-mockgen.sh

@@ -21,7 +21,7 @@ BIN="$(dirname "$(realpath "${BASH_SOURCE[0]}")")/../bin"
 # Source-based mocking for internal interfaces
 "${BIN}/mockgen" -package cloud -destination=./pkg/cloud/mock_cloud.go -source pkg/cloud/interface.go
 "${BIN}/mockgen" -package metadata -destination=./pkg/cloud/metadata/mock_metadata.go -source pkg/cloud/metadata/interface.go
-"${BIN}/mockgen" -package driver -destination=./pkg/driver/mock_mount.go -source pkg/driver/mount.go
+"${BIN}/mockgen" -package mounter -destination=./pkg/mounter/mock_mount.go -source pkg/mounter/mount.go
 "${BIN}/mockgen" -package mounter -destination=./pkg/mounter/mock_mount_windows.go -source pkg/mounter/safe_mounter_windows.go
 "${BIN}/mockgen" -package cloud -destination=./pkg/cloud/mock_ec2.go -source pkg/cloud/ec2_interface.go EC2API
 
@@ -30,8 +30,3 @@ BIN="$(dirname "$(realpath "${BASH_SOURCE[0]}")")/../bin"
 "${BIN}/mockgen" -package driver -destination=./pkg/driver/mock_k8s_corev1.go k8s.io/client-go/kubernetes/typed/core/v1 CoreV1Interface,NodeInterface
 "${BIN}/mockgen" -package driver -destination=./pkg/driver/mock_k8s_storagev1.go k8s.io/client-go/kubernetes/typed/storage/v1 VolumeAttachmentInterface,StorageV1Interface
 "${BIN}/mockgen" -package driver -destination=./pkg/driver/mock_k8s_storagev1_csinode.go k8s.io/client-go/kubernetes/typed/storage/v1 CSINodeInterface
-
-# Fixes "Mounter Type cannot implement 'Mounter' as it has a non-exported method and is defined in a different package"
-# See https://github.com/kubernetes/mount-utils/commit/a20fcfb15a701977d086330b47b7efad51eb608e for context.
-sed -i '/type MockMounter struct {/a \\tmount_utils.Interface' pkg/driver/mock_mount.go
-sed -i '/type MockProxyMounter struct {/a \\tmount.Interface' pkg/mounter/mock_mount_windows.go

pkg/driver/constants.go

@@ -32,12 +32,6 @@ const (
 	VolumeAttributePartition = "partition"
 )
 
-// constants of disk partition suffix
-const (
-	diskPartitionSuffix     = ""
-	nvmeDiskPartitionSuffix = "p"
-)
-
 // constants of keys in volume parameters
 const (
 	// VolumeTypeKey represents key for volume type
@@ -162,12 +156,6 @@ const (
 	DefaultModifyVolumeRequestHandlerTimeout = 2 * time.Second
 )
 
-// constants for disk block size
-const (
-	//DefaultBlockSize represents the default block size (4KB)
-	DefaultBlockSize = 4096
-)
-
 // constants for fstypes
 const (
 	// FSTypeExt2 represents the ext2 filesystem type

pkg/driver/driver.go

@@ -24,9 +24,12 @@ import (
 	"github.com/awslabs/volume-modifier-for-k8s/pkg/rpc"
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/cloud"
+	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/cloud/metadata"
+	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/mounter"
 	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/util"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 )
 
@@ -56,12 +59,12 @@ const (
 
 type Driver struct {
 	controller *ControllerService
-	nodeService
-	srv     *grpc.Server
-	options *Options
+	node       *NodeService
+	srv        *grpc.Server
+	options    *Options
 }
 
-func NewDriver(c cloud.Cloud, o *Options) (*Driver, error) {
+func NewDriver(c cloud.Cloud, o *Options, m mounter.Mounter, md metadata.MetadataService, k kubernetes.Interface) (*Driver, error) {
 	klog.InfoS("Driver Information", "Driver", DriverName, "Version", driverVersion)
 
 	if err := ValidateDriverOptions(o); err != nil {
@@ -76,10 +79,10 @@ func NewDriver(c cloud.Cloud, o *Options) (*Driver, error) {
 	case ControllerMode:
 		driver.controller = NewControllerService(c, o)
 	case NodeMode:
-		driver.nodeService = newNodeService(o)
+		driver.node = NewNodeService(o, md, m, k)
 	case AllMode:
 		driver.controller = NewControllerService(c, o)
-		driver.nodeService = newNodeService(o)
+		driver.node = NewNodeService(o, md, m, k)
 	default:
 		return nil, fmt.Errorf("unknown mode: %s", o.Mode)
 	}
@@ -122,10 +125,10 @@ func (d *Driver) Run() error {
 		csi.RegisterControllerServer(d.srv, d.controller)
 		rpc.RegisterModifyServer(d.srv, d.controller)
 	case NodeMode:
-		csi.RegisterNodeServer(d.srv, d)
+		csi.RegisterNodeServer(d.srv, d.node)
 	case AllMode:
 		csi.RegisterControllerServer(d.srv, d.controller)
-		csi.RegisterNodeServer(d.srv, d)
+		csi.RegisterNodeServer(d.srv, d.node)
 		rpc.RegisterModifyServer(d.srv, d.controller)
 	default:
 		return fmt.Errorf("unknown mode: %s", d.options.Mode)