prevent volume leaks when using non-existing tags

pkg/csi_driver/controller.go

@@ -285,6 +285,14 @@ func (s *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolu
 		return nil, file.StatusError(err)
 	}
 
+	// If tags are used, check if they exist
+	if tags, ok := req.GetParameters()[cloud.ParameterKeyResourceTags]; ok {
+		_, err = s.config.tagManager.ValidateResourceTags(ctx, "CreateVolumeRequest", tags)
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		}
+	}
+
 	if filer != nil {
 		klog.V(4).Infof("Found existing instance %+v, current instance %+v\n", filer, newFiler)
 		// Instance already exists, check if it meets the request

pkg/csi_driver/controller_test.go

@@ -451,12 +451,13 @@ func TestCreateVolume(t *testing.T) {
 		},
 	}
 	cases := []struct {
-		name            string
-		req             *csi.CreateVolumeRequest
-		resp            *csi.CreateVolumeResponse
-		expectErr       bool
-		features        *GCFSDriverFeatureOptions
-		expectedOptions []*file.NfsExportOptions
+		name                   string
+		req                    *csi.CreateVolumeRequest
+		resp                   *csi.CreateVolumeResponse
+		expectErr              bool
+		features               *GCFSDriverFeatureOptions
+		expectedOptions        []*file.NfsExportOptions
+		validateTagsShouldFail bool
 	}{
 		{
 			name: "valid defaults",
@@ -593,6 +594,29 @@ func TestCreateVolume(t *testing.T) {
 			},
 			expectErr: true,
 		},
+		{
+			name: "validating resource tags fails",
+			req: &csi.CreateVolumeRequest{
+				Name: testCSIVolume,
+				VolumeCapabilities: []*csi.VolumeCapability{
+					{
+						AccessType: &csi.VolumeCapability_Mount{
+							Mount: &csi.VolumeCapability_MountVolume{},
+						},
+						AccessMode: &csi.VolumeCapability_AccessMode{
+							Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+						},
+					},
+				},
+				Parameters: map[string]string{
+					"network":                      "test",
+					cloud.ParameterKeyResourceTags: "kubernetes/invalid/tag",
+				},
+			},
+			expectErr:              true,
+			features:               features,
+			validateTagsShouldFail: true,
+		},
 		// TODO: create failed
 		// TODO: instance already exists error
 		// TODO: instance already exists invalid
@@ -782,6 +806,19 @@ func TestCreateVolume(t *testing.T) {
 	for _, test := range cases {
 		cs := initTestController(t).(*controllerServer)
 		cs.config.features = test.features
+
+		if tags, ok := test.req.GetParameters()[cloud.ParameterKeyResourceTags]; ok {
+			if test.validateTagsShouldFail {
+				cs.config.tagManager.(*cloud.FakeTagServiceManager).
+					On("ValidateResourceTags", context.TODO(), tags).
+					Return(nil, fmt.Errorf("mock failure: invalid resource tags"))
+			} else {
+				cs.config.tagManager.(*cloud.FakeTagServiceManager).
+					On("ValidateResourceTags", context.TODO(), tags).
+					Return(nil, nil)
+			}
+		}
+
 		cs.config.tagManager.(*cloud.FakeTagServiceManager).
 			On("AttachResourceTags", context.TODO(), cloud.FilestoreInstance, testCSIVolume, testLocation, test.req.GetName(), test.req.GetParameters()).
 			Return(nil)
@@ -797,6 +834,15 @@ func TestCreateVolume(t *testing.T) {
 		if test.expectErr && err == nil {
 			t.Errorf("test %q failed; got success", test.name)
 		}
+		if test.validateTagsShouldFail {
+			serverError, ok := status.FromError(err)
+			if !ok {
+				t.Fatalf("test %q failed: Could not get error status code from err: %v", test.name, err)
+			}
+			if serverError.Code() != codes.InvalidArgument {
+				t.Errorf("test %q failed: Expected error code: %v, got: %v. err: %v", test.name, codes.InvalidArgument, serverError.Code(), err)
+			}
+		}
 		if !cmp.Equal(resp, test.resp, protocmp.Transform()) {
 			t.Errorf("test %q failed: got resp %+v, expected %+v, diff: %s", test.name, resp, test.resp, cmp.Diff(resp, test.resp, protocmp.Transform()))
 		}