Skip to content

Commit 53bdc4e

Browse files
mskanth972mskanth972
committed
Re-use Access point.
1 parent 158d6f7 commit 53bdc4e

File tree

8 files changed

+483
-83
lines changed

8 files changed

+483
-83
lines changed

‎docs/README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ The following CSI interfaces are implemented:
3737
| subPathPattern | | `/${.PV.name}` | true | The template used to construct the subPath under which each of the access points created under Dynamic Provisioning. Can be made up of fixed strings and limited variables, is akin to the 'subPathPattern' variable on the [nfs-subdir-external-provisioner](https://github.com/kubernetes-sigs/nfs-subdir-external-provisioner) chart. Supports `.PVC.name`,`.PVC.namespace` and `.PV.name` |
3838
| ensureUniqueDirectory | | true | true | **NOTE: Only set this to false if you're sure this is the behaviour you want**.<br/> Used when dynamic provisioning is enabled, if set to true, appends the a UID to the pattern specified in `subPathPattern` to ensure that access points will not accidentally point at the same directory. |
3939
| az | | "" | true | Used for cross-account mount. `az` under storage class parameter is optional. If specified, mount target associated with the az will be used for cross-account mount. If not specified, a random mount target will be picked for cross account mount |
40+
| reuseAccessPoint | | false | true | When set to true, it creates Accesspoint client-token from the provided PVC name. So that the AccessPoint can be re-used from a differen cluster if same PVC name and storageclass configuration are used. |
4041

4142
**Note**
4243
* Custom Posix group Id range for Access Point root directory must include both `gidRangeStart` and `gidRangeEnd` parameters. These parameters are optional only if both are omitted. If you specify one, the other becomes mandatory.

‎examples/kubernetes/dynamic_provisioning/specs/storageclass.yaml‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,4 +11,5 @@ parameters:
1111
gidRangeEnd: "2000" # optional
1212
basePath: "/dynamic_provisioning" # optional
1313
subPathPattern: "${.PVC.namespace}/${.PVC.name}" # optional
14-
ensureUniqueDirectory: "true" # optional
14+
ensureUniqueDirectory: "true" # optional
15+
reuseAccessPoint: "false" # optional

‎pkg/cloud/cloud.go‎

Lines changed: 57 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,9 @@ import (
3535
)
3636

3737
const (
38-
AccessDeniedException = "AccessDeniedException"
38+
AccessDeniedException = "AccessDeniedException"
39+
AccessPointAlreadyExists = "AccessPointAlreadyExists"
40+
PvcNameTagKey = "pvcName"
3941
)
4042

4143
var (
@@ -94,7 +96,7 @@ type Efs interface {
9496

9597
type Cloud interface {
9698
GetMetadata() MetadataService
97-
CreateAccessPoint(ctx context.Context, volumeName string, accessPointOpts *AccessPointOptions) (accessPoint *AccessPoint, err error)
99+
CreateAccessPoint(ctx context.Context, clientToken string, accessPointOpts *AccessPointOptions, reuseAccessPoint bool) (accessPoint *AccessPoint, err error)
98100
DeleteAccessPoint(ctx context.Context, accessPointId string) (err error)
99101
DescribeAccessPoint(ctx context.Context, accessPointId string) (accessPoint *AccessPoint, err error)
100102
ListAccessPoints(ctx context.Context, fileSystemId string) (accessPoints []*AccessPoint, err error)
@@ -161,10 +163,28 @@ func (c *cloud) GetMetadata() MetadataService {
161163
return c.metadata
162164
}
163165

164-
func (c *cloud) CreateAccessPoint(ctx context.Context, volumeName string, accessPointOpts *AccessPointOptions) (accessPoint *AccessPoint, err error) {
166+
func (c *cloud) CreateAccessPoint(ctx context.Context, clientToken string, accessPointOpts *AccessPointOptions, reuseAccessPoint bool) (accessPoint *AccessPoint, err error) {
165167
efsTags := parseEfsTags(accessPointOpts.Tags)
168+
169+
//if reuseAccessPoint is true, check for AP with same Root Directory exists in efs
170+
// if found reuse that AP
171+
if reuseAccessPoint {
172+
existingAP, err := c.findAccessPointByClientToken(ctx, clientToken, accessPointOpts)
173+
if err != nil {
174+
return nil, fmt.Errorf("failed to find access point: %v", err)
175+
}
176+
if existingAP != nil {
177+
//AP path already exists
178+
klog.V(2).Infof("Existing AccessPoint found : %+v", existingAP)
179+
return &AccessPoint{
180+
AccessPointId: existingAP.AccessPointId,
181+
FileSystemId: existingAP.FileSystemId,
182+
CapacityGiB: accessPointOpts.CapacityGiB,
183+
}, nil
184+
}
185+
}
166186
createAPInput := &efs.CreateAccessPointInput{
167-
ClientToken: &volumeName,
187+
ClientToken: &clientToken,
168188
FileSystemId: &accessPointOpts.FileSystemId,
169189
PosixUser: &efs.PosixUser{
170190
Gid: &accessPointOpts.Gid,
@@ -189,6 +209,7 @@ func (c *cloud) CreateAccessPoint(ctx context.Context, volumeName string, access
189209
}
190210
return nil, fmt.Errorf("Failed to create access point: %v", err)
191211
}
212+
klog.V(5).Infof("Create AP response : %+v", res)
192213

193214
return &AccessPoint{
194215
AccessPointId: *res.AccessPointId,
@@ -240,6 +261,38 @@ func (c *cloud) DescribeAccessPoint(ctx context.Context, accessPointId string) (
240261
}, nil
241262
}
242263

264+
func (c *cloud) findAccessPointByClientToken(ctx context.Context, clientToken string, accessPointOpts *AccessPointOptions) (accessPoint *AccessPoint, err error) {
265+
klog.V(5).Infof("AccessPointOptions to find AP : %+v", accessPointOpts)
266+
klog.V(2).Infof("ClientToken to find AP : %s", clientToken)
267+
describeAPInput := &efs.DescribeAccessPointsInput{
268+
FileSystemId: &accessPointOpts.FileSystemId,
269+
MaxResults: aws.Int64(1000),
270+
}
271+
res, err := c.efs.DescribeAccessPointsWithContext(ctx, describeAPInput)
272+
if err != nil {
273+
if isAccessDenied(err) {
274+
return
275+
}
276+
if isFileSystemNotFound(err) {
277+
return
278+
}
279+
err = fmt.Errorf("failed to list Access Points of efs = %s : %v", accessPointOpts.FileSystemId, err)
280+
return
281+
}
282+
for _, ap := range res.AccessPoints {
283+
// check if AP exists with same client token
284+
if aws.StringValue(ap.ClientToken) == clientToken {
285+
return &AccessPoint{
286+
AccessPointId: *ap.AccessPointId,
287+
FileSystemId: *ap.FileSystemId,
288+
AccessPointRootDir: *ap.RootDirectory.Path,
289+
}, nil
290+
}
291+
}
292+
klog.V(2).Infof("Access point does not exist")
293+
return nil, nil
294+
}
295+
243296
func (c *cloud) ListAccessPoints(ctx context.Context, fileSystemId string) (accessPoints []*AccessPoint, err error) {
244297
describeAPInput := &efs.DescribeAccessPointsInput{
245298
FileSystemId: &fileSystemId,

‎pkg/cloud/cloud_test.go‎

Lines changed: 129 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package cloud
33
import (
44
"context"
55
"errors"
6+
"reflect"
67
"testing"
78

89
"github.com/aws/aws-sdk-go/aws"
@@ -27,13 +28,14 @@ func TestCreateAccessPoint(t *testing.T) {
2728
directoryPerms = "0777"
2829
directoryPath = "/test"
2930
volName = "volName"
31+
clientToken = volName
3032
)
3133
testCases := []struct {
3234
name string
3335
testFunc func(t *testing.T)
3436
}{
3537
{
36-
name: "Success",
38+
name: "Success - AP does not exist",
3739
testFunc: func(t *testing.T) {
3840
mockCtl := gomock.NewController(t)
3941
mockEfs := mocks.NewMockEfs(mockCtl)
@@ -72,9 +74,63 @@ func TestCreateAccessPoint(t *testing.T) {
7274
},
7375
}
7476

77+
describeAPOutput := &efs.DescribeAccessPointsOutput{
78+
AccessPoints: nil,
79+
}
80+
7581
ctx := context.Background()
82+
mockEfs.EXPECT().DescribeAccessPointsWithContext(gomock.Eq(ctx), gomock.Any()).Return(describeAPOutput, nil)
7683
mockEfs.EXPECT().CreateAccessPointWithContext(gomock.Eq(ctx), gomock.Any()).Return(output, nil)
77-
res, err := c.CreateAccessPoint(ctx, volName, req)
84+
res, err := c.CreateAccessPoint(ctx, clientToken, req, true)
85+
86+
if err != nil {
87+
t.Fatalf("CreateAccessPointFailed is failed: %v", err)
88+
}
89+
90+
if res == nil {
91+
t.Fatal("Result is nil")
92+
}
93+
94+
if accessPointId != res.AccessPointId {
95+
t.Fatalf("AccessPointId mismatched. Expected: %v, Actual: %v", accessPointId, res.AccessPointId)
96+
}
97+
98+
if fsId != res.FileSystemId {
99+
t.Fatalf("FileSystemId mismatched. Expected: %v, Actual: %v", fsId, res.FileSystemId)
100+
}
101+
mockCtl.Finish()
102+
},
103+
},
104+
{
105+
name: "Success - AP already exists",
106+
testFunc: func(t *testing.T) {
107+
mockCtl := gomock.NewController(t)
108+
mockEfs := mocks.NewMockEfs(mockCtl)
109+
c := &cloud{
110+
efs: mockEfs,
111+
}
112+
113+
tags := make(map[string]string)
114+
tags["cluster"] = "efs"
115+
116+
req := &AccessPointOptions{
117+
FileSystemId: fsId,
118+
Uid: uid,
119+
Gid: gid,
120+
DirectoryPerms: directoryPerms,
121+
DirectoryPath: directoryPath,
122+
Tags: tags,
123+
}
124+
125+
describeAPOutput := &efs.DescribeAccessPointsOutput{
126+
AccessPoints: []*efs.AccessPointDescription{
127+
{AccessPointId: aws.String(accessPointId), FileSystemId: aws.String(fsId), ClientToken: aws.String(clientToken), RootDirectory: &efs.RootDirectory{Path: aws.String(directoryPath)}, Tags: []*efs.Tag{{Key: aws.String(PvcNameTagKey), Value: aws.String(volName)}}},
128+
},
129+
}
130+
131+
ctx := context.Background()
132+
mockEfs.EXPECT().DescribeAccessPointsWithContext(gomock.Eq(ctx), gomock.Any()).Return(describeAPOutput, nil)
133+
res, err := c.CreateAccessPoint(ctx, clientToken, req, true)
78134

79135
if err != nil {
80136
t.Fatalf("CreateAccessPointFailed is failed: %v", err)
@@ -108,10 +164,14 @@ func TestCreateAccessPoint(t *testing.T) {
108164
DirectoryPerms: directoryPerms,
109165
DirectoryPath: directoryPath,
110166
}
167+
describeAPOutput := &efs.DescribeAccessPointsOutput{
168+
AccessPoints: nil,
169+
}
111170

112171
ctx := context.Background()
172+
mockEfs.EXPECT().DescribeAccessPointsWithContext(gomock.Eq(ctx), gomock.Any()).Return(describeAPOutput, nil)
113173
mockEfs.EXPECT().CreateAccessPointWithContext(gomock.Eq(ctx), gomock.Any()).Return(nil, errors.New("CreateAccessPointWithContext failed"))
114-
_, err := c.CreateAccessPoint(ctx, volName, req)
174+
_, err := c.CreateAccessPoint(ctx, clientToken, req, true)
115175
if err == nil {
116176
t.Fatalf("CreateAccessPoint did not fail")
117177
}
@@ -135,7 +195,7 @@ func TestCreateAccessPoint(t *testing.T) {
135195

136196
ctx := context.Background()
137197
mockEfs.EXPECT().CreateAccessPointWithContext(gomock.Eq(ctx), gomock.Any()).Return(nil, awserr.New(AccessDeniedException, "Access Denied", errors.New("Access Denied")))
138-
_, err := c.CreateAccessPoint(ctx, volName, req)
198+
_, err := c.CreateAccessPoint(ctx, clientToken, req, false)
139199
if err == nil {
140200
t.Fatalf("CreateAccessPoint did not fail")
141201
}
@@ -862,3 +922,68 @@ func testResult(t *testing.T, funcName string, ret interface{}, err error, expec
862922
}
863923
}
864924
}
925+
926+
func Test_findAccessPointByPath(t *testing.T) {
927+
fsId := "testFsId"
928+
clientToken := "testPvcName"
929+
dirPath := "testPath"
930+
diffClientToken := aws.String("diff")
931+
932+
mockctl := gomock.NewController(t)
933+
defer mockctl.Finish()
934+
mockEfs := mocks.NewMockEfs(mockctl)
935+
936+
expectedSingleAP := &AccessPoint{
937+
AccessPointId: "testApId",
938+
AccessPointRootDir: dirPath,
939+
FileSystemId: fsId,
940+
}
941+
942+
type args struct {
943+
clientToken string
944+
accessPointOpts *AccessPointOptions
945+
}
946+
tests := []struct {
947+
name string
948+
args args
949+
prepare func(*mocks.MockEfs)
950+
wantAccessPoint *AccessPoint
951+
wantErr bool
952+
}{
953+
{name: "Expected_ClientToken_Not_Found", args: args{clientToken, &AccessPointOptions{FileSystemId: fsId, DirectoryPath: dirPath}}, prepare: func(mockEfs *mocks.MockEfs) {
954+
mockEfs.EXPECT().DescribeAccessPointsWithContext(gomock.Any(), gomock.Any()).Return(&efs.DescribeAccessPointsOutput{
955+
AccessPoints: []*efs.AccessPointDescription{{FileSystemId: aws.String(fsId), ClientToken: diffClientToken, AccessPointId: aws.String(expectedSingleAP.AccessPointId), RootDirectory: &efs.RootDirectory{Path: aws.String("differentPath")}}},
956+
}, nil)
957+
}, wantAccessPoint: nil, wantErr: false},
958+
{name: "Expected_Path_Found_In_Multiple_APs_And_One_AP_Filtered_By_ClientToken", args: args{clientToken, &AccessPointOptions{FileSystemId: fsId, DirectoryPath: dirPath}}, prepare: func(mockEfs *mocks.MockEfs) {
959+
mockEfs.EXPECT().DescribeAccessPointsWithContext(gomock.Any(), gomock.Any()).Return(&efs.DescribeAccessPointsOutput{
960+
AccessPoints: []*efs.AccessPointDescription{
961+
{FileSystemId: aws.String(fsId), ClientToken: diffClientToken, AccessPointId: aws.String("differentApId"), RootDirectory: &efs.RootDirectory{Path: aws.String(expectedSingleAP.AccessPointRootDir)}},
962+
{FileSystemId: aws.String(fsId), ClientToken: &clientToken, AccessPointId: aws.String(expectedSingleAP.AccessPointId), RootDirectory: &efs.RootDirectory{Path: aws.String(expectedSingleAP.AccessPointRootDir)}},
963+
},
964+
}, nil)
965+
}, wantAccessPoint: expectedSingleAP, wantErr: false},
966+
{name: "Fail_DescribeAccessPoints", args: args{clientToken, &AccessPointOptions{FileSystemId: fsId, DirectoryPath: dirPath}}, prepare: func(mockEfs *mocks.MockEfs) {
967+
mockEfs.EXPECT().DescribeAccessPointsWithContext(gomock.Any(), gomock.Any()).Return(nil, errors.New("access_denied"))
968+
}, wantAccessPoint: nil, wantErr: true},
969+
}
970+
for _, tt := range tests {
971+
t.Run(tt.name, func(t *testing.T) {
972+
c := &cloud{efs: mockEfs}
973+
ctx := context.Background()
974+
975+
if tt.prepare != nil {
976+
tt.prepare(mockEfs)
977+
}
978+
979+
gotAccessPoint, err := c.findAccessPointByClientToken(ctx, tt.args.clientToken, tt.args.accessPointOpts)
980+
if (err != nil) != tt.wantErr {
981+
t.Errorf("findAccessPointByClientToken() error = %v, wantErr %v", err, tt.wantErr)
982+
return
983+
}
984+
if !reflect.DeepEqual(gotAccessPoint, tt.wantAccessPoint) {
985+
t.Errorf("findAccessPointByClientToken() gotAccessPoint = %v, want %v", gotAccessPoint, tt.wantAccessPoint)
986+
}
987+
})
988+
}
989+
}

‎pkg/cloud/fakes.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,8 +27,8 @@ func (c *FakeCloudProvider) GetMetadata() MetadataService {
2727
return c.m
2828
}
2929

30-
func (c *FakeCloudProvider) CreateAccessPoint(ctx context.Context, volumeName string, accessPointOpts *AccessPointOptions) (accessPoint *AccessPoint, err error) {
31-
ap, exists := c.accessPoints[volumeName]
30+
func (c *FakeCloudProvider) CreateAccessPoint(ctx context.Context, clientToken string, accessPointOpts *AccessPointOptions, usePvcName bool) (accessPoint *AccessPoint, err error) {
31+
ap, exists := c.accessPoints[clientToken]
3232
if exists {
3333
if accessPointOpts.CapacityGiB == ap.CapacityGiB {
3434
return ap, nil
@@ -45,7 +45,7 @@ func (c *FakeCloudProvider) CreateAccessPoint(ctx context.Context, volumeName st
4545
CapacityGiB: accessPointOpts.CapacityGiB,
4646
}
4747

48-
c.accessPoints[volumeName] = ap
48+
c.accessPoints[clientToken] = ap
4949
return ap, nil
5050
}
5151

‎pkg/driver/controller.go‎

Lines changed: 28 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ package driver
1818

1919
import (
2020
"context"
21+
"crypto/sha256"
2122
"fmt"
2223
"github.com/google/uuid"
2324
"os"
@@ -56,6 +57,8 @@ const (
5657
SubPathPattern = "subPathPattern"
5758
TempMountPathPrefix = "/var/lib/csi/pv"
5859
Uid = "uid"
60+
ReuseAccessPointKey = "reuseAccessPoint"
61+
PvcNameKey = "csi.storage.k8s.io/pvc/name"
5962
)
6063

6164
var (
@@ -74,7 +77,25 @@ var (
7477

7578
func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
7679
klog.V(4).Infof("CreateVolume: called with args %+v", *req)
80+
81+
var reuseAccessPoint bool
82+
var err error
83+
volumeParams := req.GetParameters()
7784
volName := req.GetName()
85+
clientToken := volName
86+
87+
// if true, then use sha256 hash of pvcName as clientToken instead of PVC Id
88+
// This allows users to reconnect to the same AP from different k8s cluster
89+
if reuseAccessPointStr, ok := volumeParams[ReuseAccessPointKey]; ok {
90+
reuseAccessPoint, err = strconv.ParseBool(reuseAccessPointStr)
91+
if err != nil {
92+
return nil, status.Error(codes.InvalidArgument, "Invalid value for reuseAccessPoint parameter")
93+
}
94+
if reuseAccessPoint {
95+
clientToken = get64LenHash(volumeParams[PvcNameKey])
96+
klog.V(5).Infof("Client token : %s", clientToken)
97+
}
98+
}
7899
if volName == "" {
79100
return nil, status.Error(codes.InvalidArgument, "Volume name not provided")
80101
}
@@ -98,7 +119,6 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
98119
var (
99120
azName string
100121
basePath string
101-
err error
102122
gid int64
103123
gidMin int
104124
gidMax int
@@ -109,7 +129,6 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
109129
)
110130

111131
//Parse parameters
112-
volumeParams := req.GetParameters()
113132
if value, ok := volumeParams[ProvisioningMode]; ok {
114133
provisioningMode = value
115134
//TODO: Add FS provisioning mode check when implemented
@@ -287,7 +306,7 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
287306
accessPointsOptions.Gid = gid
288307
accessPointsOptions.DirectoryPath = rootDir
289308

290-
accessPointId, err := localCloud.CreateAccessPoint(ctx, volName, accessPointsOptions)
309+
accessPointId, err := localCloud.CreateAccessPoint(ctx, clientToken, accessPointsOptions, reuseAccessPoint)
291310
if err != nil {
292311
if err == cloud.ErrAccessDenied {
293312
return nil, status.Errorf(codes.Unauthenticated, "Access Denied. Please ensure you have the right AWS permissions: %v", err)
@@ -565,3 +584,9 @@ func validateEfsPathRequirements(proposedPath string) (bool, error) {
565584
return true, nil
566585
}
567586
}
587+
588+
func get64LenHash(text string) string {
589+
h := sha256.New()
590+
h.Write([]byte(text))
591+
return fmt.Sprintf("%x", h.Sum(nil))
592+
}

0 commit comments

Comments
 (0)