Support checkLimit for multiple pages (#235)

This change uses the Meta.Total field from the godo response when
listing volumes for account limit checking, thereby supporting limits
beyond the size of a single page.

We also refactor checkLimits to not return gRPC codes but leave that to
the calling function. This also allows us to distinguish between true
errors and limit violations, and better handle each case.

CHANGELOG.md

@@ -8,6 +8,8 @@
   [[GH-241]](https://github.com/digitalocean/csi-digitalocean/pull/241)
 * Check all snapshots for existence
   [[GH-240]](https://github.com/digitalocean/csi-digitalocean/pull/240)
+* Support checkLimit for multiple pages
+  [[GH-235]](https://github.com/digitalocean/csi-digitalocean/pull/235)
 * Return error when fetching the snapshot fails
   [[GH-233]](https://github.com/digitalocean/csi-digitalocean/pull/233)
 * Reject requests for block access type

driver/controller.go

@@ -18,6 +18,7 @@ package driver
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"strconv"
@@ -177,8 +178,14 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 	}
 
 	log.Info("checking volume limit")
-	if err := d.checkLimit(ctx); err != nil {
-		return nil, err
+	details, err := d.checkLimit(ctx)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to check volume limit: %s", err)
+	}
+	if details != nil {
+		return nil, status.Errorf(codes.ResourceExhausted,
+			"volume limit (%d) has been reached. Current number of volumes: %d. Please contact support.",
+			details.limit, details.numVolumes)
 	}
 
 	log.WithField("volume_req", volumeReq).Info("creating volume")
@@ -1012,41 +1019,54 @@ func (d *Driver) waitAction(ctx context.Context, log *logrus.Entry, volumeId str
 	return err
 }
 
-// checkLimit checks whether the user hit their volume limit to ensure.
-func (d *Driver) checkLimit(ctx context.Context) error {
+type limitDetails struct {
+	limit      int
+	numVolumes int
+}
+
+// checkLimit checks whether the user hit their account volume limit.
+func (d *Driver) checkLimit(ctx context.Context) (*limitDetails, error) {
 	// only one provisioner runs, we can make sure to prevent burst creation
 	d.readyMu.Lock()
 	defer d.readyMu.Unlock()
 
 	account, _, err := d.account.Get(ctx)
 	if err != nil {
-		return status.Errorf(codes.Internal,
-			"couldn't get account information to check volume limit: %s", err.Error())
+		return nil, fmt.Errorf("failed to get account information: %s", err)
 	}
 
 	// administrative accounts might have zero length limits, make sure to not check them
 	if account.VolumeLimit == 0 {
-		return nil //  hail to the king!
+		return nil, nil //  hail to the king!
 	}
 
-	// NOTE(arslan): the API returns the limit for *all* regions, so passing
-	// the region down as a parameter doesn't change the response.
-	// Nevertheless, this is something we should be aware of.
-	volumes, _, err := d.storage.ListVolumes(ctx, &godo.ListVolumeParams{
+	// The API returns the limit for *all* regions, so passing the region
+	// down as a parameter doesn't change the response. Nevertheless, this
+	// is something we should be aware of.
+	_, resp, err := d.storage.ListVolumes(ctx, &godo.ListVolumeParams{
 		Region: d.region,
+		ListOptions: &godo.ListOptions{
+			Page:    1,
+			PerPage: 1,
+		},
 	})
 	if err != nil {
-		return status.Errorf(codes.Internal,
-			"couldn't get fetch volume list to check volume limit: %s", err.Error())
+		return nil, fmt.Errorf("failed to list volumes: %s", err)
 	}
 
-	if account.VolumeLimit <= len(volumes) {
-		return status.Errorf(codes.ResourceExhausted,
-			"volume limit (%d) has been reached. Current number of volumes: %d. Please contact support.",
-			account.VolumeLimit, len(volumes))
+	if resp.Meta == nil {
+		// This should really never happen.
+		return nil, errors.New("no meta field available in list volumes response")
+	}
+	numVolumes := resp.Meta.Total
+	if account.VolumeLimit <= numVolumes {
+		return &limitDetails{
+			limit:      account.VolumeLimit,
+			numVolumes: numVolumes,
+		}, nil
 	}
 
-	return nil
+	return nil, nil
 }
 
 // toCSISnapshot converts a DO Snapshot struct into a csi.Snapshot struct

driver/controller_test.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"errors"
 	"net/http"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/digitalocean/godo"
+	"github.com/google/go-cmp/cmp"
 	"github.com/magiconair/properties/assert"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc/codes"
@@ -323,6 +325,65 @@ func TestCreateVolume(t *testing.T) {
 	}
 }
 
+func TestCheckLimit(t *testing.T) {
+	tests := []struct {
+		name        string
+		limit       int
+		numVolumes  int
+		wantErr     error
+		wantDetails *limitDetails
+	}{
+		{
+			name:       "limit insufficient",
+			limit:      25,
+			numVolumes: 30,
+			wantDetails: &limitDetails{
+				limit:      25,
+				numVolumes: 30,
+			},
+		},
+		{
+			name:        "limit sufficient",
+			limit:       100,
+			numVolumes:  25,
+			wantDetails: nil,
+		},
+		{
+			name:        "administrative account",
+			limit:       0,
+			numVolumes:  1000,
+			wantDetails: nil,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			storage := &fakeStorageDriver{
+				volumes: map[string]*godo.Volume{},
+			}
+			for i := 0; i < test.numVolumes; i++ {
+				storage.volumes[strconv.Itoa(i)] = &godo.Volume{}
+			}
+
+			d := Driver{
+				account: &fakeAccountDriver{
+					volumeLimit: test.limit,
+				},
+				storage: storage,
+			}
+
+			gotDetails, err := d.checkLimit(context.Background())
+			if err != nil {
+				t.Fatalf("got error: %s", err)
+			}
+
+			if diff := cmp.Diff(gotDetails, test.wantDetails, cmp.AllowUnexported(limitDetails{})); diff != "" {
+				t.Errorf("details mismatch (-got +want):\n%s", diff)
+			}
+		})
+	}
+}
+
 type fakeStorageAction struct {
 	*fakeStorageActionsDriver
 	storageGetValsFunc func(invocation int) (*godo.Action, *godo.Response, error)

driver/driver.go

@@ -205,8 +205,15 @@ func (d *Driver) Run() error {
 	// something is wrong in the logs. Only check if the driver is running with
 	// a token (i.e: controller)
 	if d.isController {
-		if err := d.checkLimit(context.Background()); err != nil {
-			d.log.WithError(err).Warn("CSI plugin will not function correctly, please resolve volume limit")
+		details, err := d.checkLimit(context.Background())
+		if err != nil {
+			return fmt.Errorf("failed to check volumes limits on startup: %s", err)
+		}
+		if details != nil {
+			d.log.WithFields(logrus.Fields{
+				"limit":       details.limit,
+				"num_volumes": details.numVolumes,
+			}).Warn("CSI plugin will not function correctly, please resolve volume limit")
 		}
 	}
 

driver/driver_test.go

@@ -94,10 +94,14 @@ func TestDriverSuite(t *testing.T) {
 	sanity.Test(t, cfg)
 }
 
-type fakeAccountDriver struct{}
+type fakeAccountDriver struct {
+	volumeLimit int
+}
 
 func (f *fakeAccountDriver) Get(context.Context) (*godo.Account, *godo.Response, error) {
-	return &godo.Account{}, godoResponse(), nil
+	return &godo.Account{
+		VolumeLimit: f.volumeLimit,
+	}, godoResponse(), nil
 }
 
 type fakeStorageDriver struct {
@@ -119,12 +123,16 @@ func (f *fakeStorageDriver) ListVolumes(ctx context.Context, param *godo.ListVol
 
 	if param != nil && param.ListOptions != nil && param.ListOptions.PerPage != 0 {
 		perPage := param.ListOptions.PerPage
-		vols := volumes[:perPage]
+		chunkSize := perPage
+		if len(volumes) < perPage {
+			chunkSize = len(volumes)
+		}
+		vols := volumes[:chunkSize]
 		for _, vol := range vols {
 			delete(f.volumes, vol.ID)
 		}
 
-		return vols, godoResponse(), nil
+		return vols, godoResponseWithMeta(len(volumes)), nil
 	}
 
 	if param.Name != "" {
@@ -135,10 +143,10 @@ func (f *fakeStorageDriver) ListVolumes(ctx context.Context, param *godo.ListVol
 			}
 		}
 
-		return filtered, godoResponse(), nil
+		return filtered, godoResponseWithMeta(len(filtered)), nil
 	}
 
-	return volumes, godoResponse(), nil
+	return volumes, godoResponseWithMeta(len(volumes)), nil
 }
 
 func (f *fakeStorageDriver) GetVolume(ctx context.Context, id string) (*godo.Volume, *godo.Response, error) {
@@ -183,7 +191,7 @@ func (f *fakeStorageDriver) ListSnapshots(ctx context.Context, volumeID string,
 		}
 	}
 
-	return snapshots, godoResponse(), nil
+	return snapshots, godoResponseWithMeta(len(snapshots)), nil
 }
 
 func (f *fakeStorageDriver) GetSnapshot(ctx context.Context, id string) (*godo.Snapshot, *godo.Response, error) {
@@ -247,7 +255,7 @@ func (f *fakeStorageActionsDriver) Get(ctx context.Context, volumeID string, act
 }
 
 func (f *fakeStorageActionsDriver) List(ctx context.Context, volumeID string, opt *godo.ListOptions) ([]godo.Action, *godo.Response, error) {
-	return nil, godoResponse(), nil
+	return nil, godoResponseWithMeta(0), nil
 }
 
 func (f *fakeStorageActionsDriver) Resize(ctx context.Context, volumeID string, sizeGigabytes int, regionSlug string) (*godo.Action, *godo.Response, error) {
@@ -332,7 +340,7 @@ func (f *fakeSnapshotsDriver) ListVolume(ctx context.Context, opts *godo.ListOpt
 		snapshots = append(snapshots, *snap)
 	}
 
-	return snapshots, godoResponse(), nil
+	return snapshots, godoResponseWithMeta(len(snapshots)), nil
 }
 
 func (f *fakeSnapshotsDriver) ListDroplet(context.Context, *godo.ListOptions) ([]godo.Snapshot, *godo.Response, error) {
@@ -399,9 +407,16 @@ func (f *fakeMounter) GetStatistics(volumePath string) (volumeStatistics, error)
 }
 
 func godoResponse() *godo.Response {
+	return godoResponseWithMeta(0)
+}
+
+func godoResponseWithMeta(total int) *godo.Response {
 	return &godo.Response{
 		Response: &http.Response{StatusCode: 200},
 		Rate:     godo.Rate{Limit: 10, Remaining: 10},
+		Meta: &godo.Meta{
+			Total: total,
+		},
 	}
 }
 

go.mod

@@ -3,10 +3,10 @@ module github.com/digitalocean/csi-digitalocean
 require (
 	github.com/container-storage-interface/spec v1.2.0
 	github.com/digitalocean/go-metadata v0.0.0-20180111002115-15bd36e5f6f7
-	github.com/digitalocean/godo v1.13.0
+	github.com/digitalocean/godo v1.29.0
 	github.com/golang/groupcache v0.0.0-20181024230925-c65c006176ff // indirect
 	github.com/golang/protobuf v1.3.1
-	github.com/google/go-querystring v0.0.0-20170111101155-53e6ce116135 // indirect
+	github.com/google/go-cmp v0.3.0
 	github.com/googleapis/gnostic v0.2.0 // indirect
 	github.com/imdario/mergo v0.3.6 // indirect
 	github.com/kubernetes-csi/csi-test v2.2.0+incompatible

go.sum

@@ -80,8 +80,8 @@ github.com/dgrijalva/jwt-go v0.0.0-20160705203006-01aeca54ebda/go.mod h1:E3ru+11
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/digitalocean/go-metadata v0.0.0-20180111002115-15bd36e5f6f7 h1:ESwakcvVdp1uaY19Bt8e3OHLqpqh1hS6tDRgAIa8m78=
 github.com/digitalocean/go-metadata v0.0.0-20180111002115-15bd36e5f6f7/go.mod h1:lNrzMwI4fx6xfzieyLEpYIJPLWjT/Sak4G/hIzGTEL4=
-github.com/digitalocean/godo v1.13.0 h1:+0rQUl0AkTmhKDeqP8Y1Z60cJzhC/1jPI6jWpWnLUr8=
-github.com/digitalocean/godo v1.13.0/go.mod h1:h6faOIcZ8lWIwNQ+DN7b3CgX4Kwby5T+nbpNqkUIozU=
+github.com/digitalocean/godo v1.29.0 h1:KgNNU0k9SZqVgn7m8NN9iDsq0+nluHBe8HR9QE0QVmA=
+github.com/digitalocean/godo v1.29.0/go.mod h1:iJnN9rVu6K5LioLxLimlq0uRI+y/eAQjROUmeU/r0hY=
 github.com/dnaeon/go-vcr v1.0.1/go.mod h1:aBB1+wY4s93YsC3HHjMBMrwTj2R9FHDzUr9KyGc8n1E=
 github.com/docker/distribution v2.7.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
 github.com/docker/docker v0.7.3-0.20190327010347-be7ac8be2ae0/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
@@ -174,8 +174,8 @@ github.com/google/certificate-transparency-go v1.0.21/go.mod h1:QeJfpSbVSfYc7RgB
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0 h1:crn/baboCvb5fXaQ0IJ1SGTsTVrWpDsCWC8EGETZijY=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
-github.com/google/go-querystring v0.0.0-20170111101155-53e6ce116135 h1:zLTLjkaOFEFIOxY5BWLFLwh+cL8vOBW4XJ2aqLE/Tf0=
-github.com/google/go-querystring v0.0.0-20170111101155-53e6ce116135/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
+github.com/google/go-querystring v1.0.0 h1:Xkwi/a1rcvNg1PPYe5vI8GbeBY/jrVuDX5ASuANWTrk=
+github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
 github.com/google/gofuzz v0.0.0-20161122191042-44d81051d367/go.mod h1:HP5RmnzzSNb993RKQDq4+1A4ia9nllfqcQFTQJedwGI=
 github.com/google/gofuzz v0.0.0-20170612174753-24818f796faf h1:+RRA9JqSOZFfKrOeqr2z77+8R2RKyh8PG66dcu1V0ck=
 github.com/google/gofuzz v0.0.0-20170612174753-24818f796faf/go.mod h1:HP5RmnzzSNb993RKQDq4+1A4ia9nllfqcQFTQJedwGI=