From 98c12a80f6b7b8d10eb26f44d0195fd406c85f70 Mon Sep 17 00:00:00 2001
From: Timo Reimann <treimann@digitalocean.com>
Date: Thu, 23 Apr 2020 09:49:35 +0200
Subject: [PATCH] Enable tests for node volume attachment limits

The csi-test sanity package ships with off-by-default tests to validate
per-node attachment limits. This change toggles the corresponding test
configuration flag to enable the tests.

The change requires modifying our fake driver to return a 422 HTTP error
code when the limit is exceeded. As a consequence, we also need to
customize the IdempotentCount test setting which parameterizes the
'should be idempotent' test that creates the given number of volumes in
sequence. The default value of 10 causes our (fake) limit to be
exceeded, which is why we tune it down to 5.

The test also revealed that we missed to handle the case where the node
volume attachment limit is exceed during ControllerPublishVolume. We
extend our error handling to identify this case and return an
RESOURCE_EXHAUSTED code accordingly.
---
 CHANGELOG.md          |  2 ++
 driver/controller.go  |  8 ++++++
 driver/driver_test.go | 67 +++++++++++++++++++++++++++++++++++++++++--
 3 files changed, 75 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2f0e4915..e3961fc7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,7 @@
 ## unreleased
 
+* Handle per-node volume limit exceeding error during ControllerPublishVolume
+  [[GH-303]](https://github.com/digitalocean/csi-digitalocean/pull/303)
 * Build using Go 1.14
   [[GH-302]](https://github.com/digitalocean/csi-digitalocean/pull/302)
 * Fix ListSnapshots paging
diff --git a/driver/controller.go b/driver/controller.go
index 0007518e..0bcea6b8 100644
--- a/driver/controller.go
+++ b/driver/controller.go
@@ -63,6 +63,10 @@ const (
 	// doAPITimeout sets the timeout we will use when communicating with the
 	// Digital Ocean API. NOTE: some queries inherit the context timeout
 	doAPITimeout = 10 * time.Second
+
+	// maxVolumesPerDropletErrorMessage is the error message returned by the DO
+	// API when the per-droplet volume limit would be exceeded.
+	maxVolumesPerDropletErrorMessage = "cannot attach more than 7 volumes to a single Droplet"
 )
 
 var (
@@ -362,6 +366,10 @@ func (d *Driver) ControllerPublishVolume(ctx context.Context, req *csi.Controlle
 				// sending an abort makes sure the csi-attacher retries with the next backoff tick
 				return nil, status.Errorf(codes.Aborted, "cannot attach because droplet %d has pending action for volume %q", dropletID, req.VolumeId)
 			}
+
+			if strings.Contains(err.Error(), maxVolumesPerDropletErrorMessage) {
+				return nil, status.Errorf(codes.ResourceExhausted, err.Error())
+			}
 		}
 		return nil, err
 	}
diff --git a/driver/driver_test.go b/driver/driver_test.go
index cacb1ee9..23cf6779 100644
--- a/driver/driver_test.go
+++ b/driver/driver_test.go
@@ -96,6 +96,8 @@ func TestDriverSuite(t *testing.T) {
 
 	cfg := sanity.NewTestConfig()
 	cfg.Address = endpoint
+	cfg.IdempotentCount = 5
+	cfg.TestNodeVolumeAttachLimit = true
 	sanity.Test(t, cfg)
 
 	cancel()
@@ -248,11 +250,72 @@ type fakeStorageActionsDriver struct {
 }
 
 func (f *fakeStorageActionsDriver) Attach(ctx context.Context, volumeID string, dropletID int) (*godo.Action, *godo.Response, error) {
-	return nil, godoResponse(), nil
+	resp := godoResponse()
+
+	if _, ok := f.volumes[volumeID]; !ok {
+		resp.Response = &http.Response{
+			StatusCode: http.StatusNotFound,
+		}
+		return nil, resp, errors.New("volume was not found")
+	}
+
+	droplet, ok := f.droplets[dropletID]
+	if !ok {
+		resp.Response = &http.Response{
+			StatusCode: http.StatusNotFound,
+		}
+		return nil, resp, errors.New("droplet was not found")
+	}
+
+	if len(droplet.VolumeIDs) >= maxVolumesPerNode {
+		resp.Response = &http.Response{
+			StatusCode: http.StatusUnprocessableEntity,
+		}
+		return nil, resp, errors.New(maxVolumesPerDropletErrorMessage)
+	}
+	droplet.VolumeIDs = append(droplet.VolumeIDs, volumeID)
+
+	return nil, resp, nil
 }
 
 func (f *fakeStorageActionsDriver) DetachByDropletID(ctx context.Context, volumeID string, dropletID int) (*godo.Action, *godo.Response, error) {
-	return nil, godoResponse(), nil
+	resp := godoResponse()
+
+	if _, ok := f.volumes[volumeID]; !ok {
+		resp.Response = &http.Response{
+			StatusCode: http.StatusNotFound,
+		}
+		return nil, resp, errors.New("volume was not found")
+	}
+
+	droplet, ok := f.droplets[dropletID]
+	if !ok {
+		resp.Response = &http.Response{
+			StatusCode: http.StatusNotFound,
+		}
+		return nil, resp, errors.New("droplet was not found")
+	}
+
+	var found bool
+	var updatedAttachedVolIDs []string
+	for _, attachedVolID := range droplet.VolumeIDs {
+		if attachedVolID == volumeID {
+			found = true
+		} else {
+			updatedAttachedVolIDs = append(updatedAttachedVolIDs, attachedVolID)
+		}
+	}
+
+	if !found {
+		resp.Response = &http.Response{
+			StatusCode: http.StatusNotFound,
+		}
+		return nil, resp, errors.New("volume is not attached to droplet")
+	}
+
+	droplet.VolumeIDs = updatedAttachedVolIDs
+
+	return nil, resp, nil
 }
 
 func (f *fakeStorageActionsDriver) Get(ctx context.Context, volumeID string, actionID int) (*godo.Action, *godo.Response, error) {