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

@@ -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

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
 	}

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) {