driver: make sure to handle concurrent calls to the droplet

The DigitalOcean API doesn't process actions for a given resource in
async manner. Calls needs to be done sequentially. This means, if there
is an ongoing process, say an "attach" process, another "attach" or
"detach" will fail.

We return a `ABORTED` error, which makes sure the csi-attacher to retry
again in the next backoff tick (see:
https://github.com/kubernetes-csi/external-attacher/blob/5a4cc3b1af15966cdb9b8ca1af6ab3adade877b0/pkg/controller/csi_handler.go#L307
and
https://github.com/kubernetes-csi/external-attacher/blob/5a4cc3b1af15966cdb9b8ca1af6ab3adade877b0/pkg/connection/connection.go#L223-L227
)

CHANGELOG.md

@@ -9,6 +9,8 @@
   gracefully. We're not very strict anymore in cases we don't need to be, but
   we're also returning a better error in cases we need to be.
   [[GH-60]](https://github.com/digitalocean/csi-digitalocean/pull/60)
+* Fix attaching multiple volumes to a single pod
+  [[GH-61]](https://github.com/digitalocean/csi-digitalocean/pull/61)
 
 ## v0.1.3 - August 3rd 2018
 

driver/controller.go

@@ -202,12 +202,24 @@ func (d *Driver) ControllerPublishVolume(ctx context.Context, req *csi.Controlle
 	action, resp, err := d.doClient.StorageActions.Attach(ctx, req.VolumeId, dropletID)
 	if err != nil {
 		// don't do anything if attached
-		if (resp != nil && resp.StatusCode == http.StatusUnprocessableEntity) || strings.Contains(err.Error(), "This volume is already attached") {
-			ll.WithFields(logrus.Fields{
-				"error": err,
-				"resp":  resp,
-			}).Warn("assuming volume is attached already")
-			return &csi.ControllerPublishVolumeResponse{}, nil
+		if resp != nil && resp.StatusCode == http.StatusUnprocessableEntity {
+			if strings.Contains(err.Error(), "This volume is already attached") {
+				ll.WithFields(logrus.Fields{
+					"error": err,
+					"resp":  resp,
+				}).Warn("assuming volume is attached already")
+				return &csi.ControllerPublishVolumeResponse{}, nil
+			}
+
+			if strings.Contains(err.Error(), "Droplet already has a pending event") {
+				ll.WithFields(logrus.Fields{
+					"error": err,
+					"resp":  resp,
+				}).Warn("droplet is not able to detach the volume")
+				// sending an abort makes sure the csi-attacher retries with the next backoff tick
+				return nil, status.Errorf(codes.Aborted, "volume %q couldn't be attached. droplet %d is in process of another action",
+					req.VolumeId, dropletID)
+			}
 		}
 		return nil, err
 	}
@@ -263,12 +275,24 @@ func (d *Driver) ControllerUnpublishVolume(ctx context.Context, req *csi.Control
 
 	action, resp, err := d.doClient.StorageActions.DetachByDropletID(ctx, req.VolumeId, dropletID)
 	if err != nil {
-		if (resp != nil && resp.StatusCode == http.StatusUnprocessableEntity) || strings.Contains(err.Error(), "Attachment not found") {
-			ll.WithFields(logrus.Fields{
-				"error": err,
-				"resp":  resp,
-			}).Warn("assuming volume is detached already")
-			return &csi.ControllerUnpublishVolumeResponse{}, nil
+		if resp != nil && resp.StatusCode == http.StatusUnprocessableEntity {
+			if strings.Contains(err.Error(), "Attachment not found") {
+				ll.WithFields(logrus.Fields{
+					"error": err,
+					"resp":  resp,
+				}).Warn("assuming volume is detached already")
+				return &csi.ControllerUnpublishVolumeResponse{}, nil
+			}
+
+			if strings.Contains(err.Error(), "Droplet already has a pending event") {
+				ll.WithFields(logrus.Fields{
+					"error": err,
+					"resp":  resp,
+				}).Warn("droplet is not able to detach the volume")
+				// sending an abort makes sure the csi-attacher retries with the next backoff tick
+				return nil, status.Errorf(codes.Aborted, "volume %q couldn't be detached. droplet %d is in process of another action",
+					req.VolumeId, dropletID)
+			}
 		}
 		return nil, err
 	}