* Allow adding tags to EFS resources to configure IAM policy conditio…

…n based on tags.

* Add default tags and provide a command line option to provide more tags
* Update permissions and efs-utils version

Dockerfile

@@ -32,7 +32,7 @@ ENV EFS_CLIENT_SOURCE=$client_source
 RUN GOOS=${TARGETOS} GOARCH=${TARGETARCH} make aws-efs-csi-driver
 
 FROM amazonlinux:2.0.20210126.0
-RUN yum install amazon-efs-utils-1.26-3.amzn2.noarch -y
+RUN yum install amazon-efs-utils-1.28.2-1.amzn2.noarch -y
 
 # At image build time, static files installed by efs-utils in the config directory, i.e. CAs file, need
 # to be saved in another place so that the other stateful files created at runtime, i.e. private key for

charts/aws-efs-csi-driver/templates/serviceaccount.yaml

@@ -30,7 +30,7 @@ rules:
     verbs: ["get", "list", "watch"]
   - apiGroups: [""]
     resources: ["events"]
-    verbs: ["list", "watch", "create", "update", "patch"]
+    verbs: ["list", "watch", "create"]
   - apiGroups: ["storage.k8s.io"]
     resources: ["csinodes"]
     verbs: ["get", "list", "watch"]
@@ -40,6 +40,7 @@ rules:
   - apiGroups: ["coordination.k8s.io"]
     resources: ["leases"]
     verbs: ["get", "watch", "list", "delete", "update", "create"]
+
 ---
 
 kind: ClusterRoleBinding

charts/aws-efs-csi-driver/values.yaml

@@ -18,7 +18,7 @@ sidecars:
     tag: "v1.3.0"
   csiProvisionerImage:
     repository: k8s.gcr.io/sig-storage/csi-provisioner
-    tag: "v2.0.2"
+    tag: "v2.0.4"
 
 imagePullSecrets: []
 nameOverride: ""

cmd/main.go

@@ -41,6 +41,7 @@ func main() {
 		volMetricsFsRateLimit    = flag.Int("vol-metrics-fs-rate-limit", 5, "Volume metrics routines rate limiter per file system")
 		deleteAccessPointRootDir = flag.Bool("delete-access-point-root-dir", false,
 			"Opt in to delete access point root directory by DeleteVolume. By default, DeleteVolume will delete the access point behind Persistent Volume and deleting access point will not delete the access point root directory or its contents.")
+		tags = flag.String("tags", "", "Space separated key:value pairs which will be added as tags for EFS resources. For example, 'environment:prod region:us-east-1'")
 	)
 	klog.InitFlags(nil)
 	flag.Parse()
@@ -59,7 +60,7 @@ func main() {
 	if err != nil {
 		klog.Fatalln(err)
 	}
-	drv := driver.NewDriver(*endpoint, etcAmazonEfs, *efsUtilsStaticFilesPath, *volMetricsOptIn, *volMetricsRefreshPeriod, *volMetricsFsRateLimit, *deleteAccessPointRootDir)
+	drv := driver.NewDriver(*endpoint, etcAmazonEfs, *efsUtilsStaticFilesPath, *tags, *volMetricsOptIn, *volMetricsRefreshPeriod, *volMetricsFsRateLimit, *deleteAccessPointRootDir)
 	if err := drv.Run(); err != nil {
 		klog.Fatalln(err)
 	}

deploy/kubernetes/base/clusterrole-provisioner.yaml

@@ -17,7 +17,7 @@ rules:
     verbs: ["get", "list", "watch"]
   - apiGroups: [""]
     resources: ["events"]
-    verbs: ["list", "watch", "create", "update", "patch"]
+    verbs: ["list", "watch", "create"]
   - apiGroups: ["storage.k8s.io"]
     resources: ["csinodes"]
     verbs: ["get", "list", "watch"]

deploy/kubernetes/base/controller.yaml

@@ -56,7 +56,7 @@ spec:
             periodSeconds: 10
             failureThreshold: 5
         - name: csi-provisioner
-          image: k8s.gcr.io/sig-storage/csi-provisioner:v2.0.2
+          image: k8s.gcr.io/sig-storage/csi-provisioner:v2.0.4
           args:
             - --csi-address=$(ADDRESS)
             - --v=5

pkg/cloud/cloud.go

@@ -63,6 +63,7 @@ type AccessPointOptions struct {
 	Gid            int64
 	DirectoryPerms string
 	DirectoryPath  string
+	Tags           map[string]string
 }
 
 // Efs abstracts efs client(https://docs.aws.amazon.com/sdk-for-go/api/service/efs/)
@@ -109,6 +110,7 @@ func (c *cloud) GetMetadata() MetadataService {
 }
 
 func (c *cloud) CreateAccessPoint(ctx context.Context, volumeName string, accessPointOpts *AccessPointOptions) (accessPoint *AccessPoint, err error) {
+	efsTags := parseEfsTags(accessPointOpts.Tags)
 	createAPInput := &efs.CreateAccessPointInput{
 		ClientToken:  &volumeName,
 		FileSystemId: &accessPointOpts.FileSystemId,
@@ -124,6 +126,7 @@ func (c *cloud) CreateAccessPoint(ctx context.Context, volumeName string, access
 			},
 			Path: &accessPointOpts.DirectoryPath,
 		},
+		Tags: efsTags,
 	}
 
 	klog.V(5).Infof("Calling Create AP with input: %+v", *createAPInput)
@@ -234,3 +237,16 @@ func isAccessDenied(err error) bool {
 	}
 	return false
 }
+
+func parseEfsTags(tagMap map[string]string) []*efs.Tag {
+	efsTags := []*efs.Tag{}
+	for k, v := range tagMap {
+		key := k
+		value := v
+		efsTags = append(efsTags, &efs.Tag{
+			Key:   &key,
+			Value: &value,
+		})
+	}
+	return efsTags
+}

pkg/driver/controller.go

@@ -42,6 +42,8 @@ const (
 	DefaultGidMax       = 7000000
 	RootDirPrefix       = "efs-csi-ap"
 	TempMountPathPrefix = "/var/lib/csi/pv"
+	DefaultTagKey       = "efs.csi.aws.com/cluster"
+	DefaultTagValue     = "true"
 )
 
 var (
@@ -92,8 +94,21 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		return nil, status.Errorf(codes.InvalidArgument, "Missing %v parameter", ProvisioningMode)
 	}
 
+	// Create tags
+	tags := map[string]string{
+		DefaultTagKey: DefaultTagValue,
+	}
+
+	// Append input tags to default tag
+	if len(d.tags) != 0 {
+		for k, v := range d.tags {
+			tags[k] = v
+		}
+	}
+
 	accessPointsOptions := &cloud.AccessPointOptions{
 		CapacityGiB: volSize,
+		Tags:        tags,
 	}
 
 	if value, ok := volumeParams[FsId]; ok {

pkg/driver/driver.go

@@ -19,6 +19,7 @@ package driver
 import (
 	"context"
 	"net"
+	"strings"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc"
@@ -46,9 +47,10 @@ type Driver struct {
 	volStatter               VolStatter
 	gidAllocator             GidAllocator
 	deleteAccessPointRootDir bool
+	tags                     map[string]string
 }
 
-func NewDriver(endpoint, efsUtilsCfgPath, efsUtilsStaticFilesPath string, volMetricsOptIn bool, volMetricsRefreshPeriod float64, volMetricsFsRateLimit int, deleteAccessPointRootDir bool) *Driver {
+func NewDriver(endpoint, efsUtilsCfgPath, efsUtilsStaticFilesPath, tags string, volMetricsOptIn bool, volMetricsRefreshPeriod float64, volMetricsFsRateLimit int, deleteAccessPointRootDir bool) *Driver {
 	cloud, err := cloud.NewCloud()
 	if err != nil {
 		klog.Fatalln(err)
@@ -69,6 +71,7 @@ func NewDriver(endpoint, efsUtilsCfgPath, efsUtilsStaticFilesPath string, volMet
 		volMetricsFsRateLimit:    volMetricsFsRateLimit,
 		gidAllocator:             NewGidAllocator(),
 		deleteAccessPointRootDir: deleteAccessPointRootDir,
+		tags:                     parseTagsFromStr(strings.TrimSpace(tags)),
 	}
 }
 
@@ -124,3 +127,23 @@ func (d *Driver) Run() error {
 	klog.Infof("Listening for connections on address: %#v", listener.Addr())
 	return d.srv.Serve(listener)
 }
+
+func parseTagsFromStr(tagStr string) map[string]string {
+	defer func() {
+		if r := recover(); r != nil {
+			klog.Errorf("Failed to parse input tag string: %v", tagStr)
+		}
+	}()
+
+	m := make(map[string]string)
+	if tagStr == "" {
+		klog.Infof("Did not find any input tags.")
+		return m
+	}
+	tagsSplit := strings.Split(tagStr, " ")
+	for _, pair := range tagsSplit {
+		p := strings.Split(pair, ":")
+		m[p[0]] = p[1]
+	}
+	return m
+}

pkg/driver/efs_watch_dog.go

@@ -26,7 +26,7 @@ import (
 	"k8s.io/klog"
 )
 
-// https://github.com/aws/efs-utils/blob/v1.26.3/dist/efs-utils.conf
+// https://github.com/aws/efs-utils/blob/v1.28.2/dist/efs-utils.conf
 const (
 	efsUtilsConfigTemplate = `
 [DEFAULT]
@@ -78,6 +78,14 @@ tls_cert_renewal_interval_min = 60
 
 [client-info] 
 source={{.EfsClientSource}}
+
+[cloudwatch-log]
+# enabled = true
+log_group_name = /aws/efs/utils
+
+# Possible values are : 1, 3, 5, 7, 14, 30, 60, 90, 120, 150, 180, 365, 400, 545, 731, 1827, and 3653
+# Comment this config to prevent log deletion
+retention_in_days = 14
 `
 
 	efsUtilsConfigFileName = "efs-utils.conf"

pkg/driver/efs_watch_dog_test.go

@@ -72,6 +72,14 @@ tls_cert_renewal_interval_min = 60
 
 [client-info] 
 source=k8s
+
+[cloudwatch-log]
+# enabled = true
+log_group_name = /aws/efs/utils
+
+# Possible values are : 1, 3, 5, 7, 14, 30, 60, 90, 120, 150, 180, 365, 400, 545, 731, 1827, and 3653
+# Comment this config to prevent log deletion
+retention_in_days = 14
 `
 	configFileName = "efs-utils.conf"
 )

test/e2e/cloud.go

@@ -159,6 +159,59 @@ func (c *cloud) DeleteFileSystem(fileSystemId string) error {
 	return nil
 }
 
+func (c *cloud) CreateAccessPoint(fileSystemId, clusterName string) (string, error) {
+	tags := []*efs.Tag{
+		{
+			Key:   aws.String("efs.csi.aws.com/cluster"),
+			Value: aws.String("true"),
+		},
+	}
+
+	request := &efs.CreateAccessPointInput{
+		ClientToken:  &clusterName,
+		FileSystemId: &fileSystemId,
+		PosixUser: &efs.PosixUser{
+			Gid: aws.Int64(1000),
+			Uid: aws.Int64(1000),
+		},
+		RootDirectory: &efs.RootDirectory{
+			CreationInfo: &efs.CreationInfo{
+				OwnerGid:    aws.Int64(1000),
+				OwnerUid:    aws.Int64(1000),
+				Permissions: aws.String("0777"),
+			},
+			Path: aws.String("/integ-test"),
+		},
+		Tags: tags,
+	}
+
+	var accessPointId *string
+	response, err := c.efsclient.CreateAccessPoint(request)
+	if err != nil {
+		return "", err
+	}
+
+	accessPointId = response.AccessPointId
+	err = c.ensureAccessPointStatus(*accessPointId, "available")
+	if err != nil {
+		return "", err
+	}
+
+	return aws.StringValue(accessPointId), nil
+}
+
+func (c *cloud) DeleteAccessPoint(accessPointId string) error {
+	request := &efs.DeleteAccessPointInput{
+		AccessPointId: &accessPointId,
+	}
+
+	_, err := c.efsclient.DeleteAccessPoint(request)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 // getSecurityGroup returns the node security group ID given cluster name
 func (c *cloud) getSecurityGroup(clusterName string) (string, error) {
 	// First assume the cluster was installed by kops then fallback to EKS
@@ -299,6 +352,28 @@ func (c *cloud) ensureFileSystemStatus(fileSystemId, status string) error {
 	}
 }
 
+func (c *cloud) ensureAccessPointStatus(accessPointId, status string) error {
+	request := &efs.DescribeAccessPointsInput{
+		AccessPointId: aws.String(accessPointId),
+	}
+
+	for {
+		response, err := c.efsclient.DescribeAccessPoints(request)
+		if err != nil {
+			return err
+		}
+
+		if len(response.AccessPoints) == 0 {
+			return errors.New("no access point found")
+		}
+
+		if *response.AccessPoints[0].LifeCycleState == status {
+			return nil
+		}
+		time.Sleep(time.Second)
+	}
+}
+
 func (c *cloud) ensureNoMountTarget(fileSystemId string) error {
 	request := &efs.DescribeFileSystemsInput{
 		FileSystemId: aws.String(fileSystemId),

test/e2e/e2e.go

@@ -3,6 +3,8 @@ package e2e
 import (
 	"context"
 	"fmt"
+	"os"
+	"os/exec"
 	"time"
 
 	"github.com/onsi/ginkgo"
@@ -157,6 +159,47 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
 
 		FileSystemId = id
 		ginkgo.By(fmt.Sprintf("Created EFS filesystem %q in region %q for cluster %q", FileSystemId, Region, ClusterName))
+		//Sleep 3 minutes to allow file system bootstrapping
+		ginkgo.By("Sleeping 3 minutes to allow file system bootstrapping")
+		time.Sleep(3 * time.Minute)
+		ginkgo.By("Done sleeping. Now proceeding to test access point creation")
+
+		accessPointId, err := c.CreateAccessPoint(FileSystemId, ClusterName)
+		if err != nil {
+			framework.ExpectNoError(err, "creating access point")
+		}
+		ginkgo.By(fmt.Sprintf("Created access point %q", accessPointId))
+		ginkgo.By("Now proceeding to mount using access point")
+
+		target := "accessPointTest"
+		if err = makeDir(target); err != nil {
+			framework.ExpectNoError(err, fmt.Sprintf("Failed to create temp directory %q", target))
+		}
+		defer os.RemoveAll(target)
+
+		// mount access point
+		command := exec.Command("/bin/sh", "-c", "mount", "-t", "efs", "-o", "tls,accesspoint="+accessPointId, FileSystemId, target)
+		if err = command.Run(); err != nil {
+			framework.ExpectNoError(err, "Failed to mount using access point")
+		}
+
+		//Allow access point bootstrapping
+		ginkgo.By("Sleeping 3 minutes to allow mount bootstrapping")
+		time.Sleep(3 * time.Minute)
+		ginkgo.By("Done sleeping. Now proceeding interact with access point")
+
+		// interact with access point by creating a directory
+		command = exec.Command("/bin/sh", "-c", "mkdir -p accessPointTest/a1")
+		if err = command.Run(); err != nil {
+			framework.ExpectNoError(err, "Failed to create directory in file system")
+		}
+
+		ginkgo.By("Access Point interaction succeeded, now proceeding to delete access point...")
+		err = c.DeleteAccessPoint(accessPointId)
+		if err != nil {
+			framework.ExpectNoError(err, "deleting access point")
+		}
+		ginkgo.By(fmt.Sprintf("Deleted access point %q", accessPointId))
 		deleteFileSystem = true
 	} else {
 		ginkgo.By(fmt.Sprintf("Using already-created EFS file system %q", FileSystemId))
@@ -429,3 +472,13 @@ func makeEFSPV(name, path string, volumeAttributes map[string]string) *v1.Persis
 		},
 	}
 }
+
+func makeDir(path string) error {
+	err := os.MkdirAll(path, os.FileMode(0777))
+	if err != nil {
+		if !os.IsExist(err) {
+			return err
+		}
+	}
+	return nil
+}