driver: check volume limit

driver/controller.go

@@ -135,6 +135,11 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 		return nil, status.Error(codes.AlreadyExists, "invalid volume capabilities requested. Only SINGLE_NODE_WRITER is supported ('accessModes.ReadWriteOnce' on Kubernetes)")
 	}
 
+	ll.Info("checking volume limit")
+	if err := d.checkLimit(ctx); err != nil {
+		return nil, err
+	}
+
 	ll.WithField("volume_req", volumeReq).Info("creating volume")
 	vol, _, err := d.doClient.Storage.CreateVolume(ctx, volumeReq)
 	if err != nil {
@@ -623,6 +628,40 @@ func (d *Driver) waitAction(ctx context.Context, volumeId string, actionId int)
 	}
 }
 
+// checkLimit checks whether the user hit their volume limit to ensure.
+func (d *Driver) checkLimit(ctx context.Context) error {
+	// only one provisioner runs, we can make sure to prevent burst creation
+	d.readyMu.Lock()
+	defer d.readyMu.Unlock()
+
+	account, _, err := d.doClient.Account.Get(ctx)
+	if err != nil {
+		return status.Errorf(codes.Internal,
+			"couldn't get account information to check volume limit: %s", err.Error())
+	}
+
+	// administrative accounts might have zero length limits, make sure to not check them
+	if account.VolumeLimit == 0 {
+		return nil //  hail to the king!
+	}
+
+	volumes, _, err := d.doClient.Storage.ListVolumes(ctx, &godo.ListVolumeParams{
+		Region: d.region,
+	})
+	if err != nil {
+		return status.Errorf(codes.Internal,
+			"couldn't get fetch volume list to check volume limit: %s", err.Error())
+	}
+
+	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))
+	}
+
+	return nil
+}
+
 // validateCapabilities validates the requested capabilities. It returns false
 // if it doesn't satisfy the currently supported modes of DigitalOcean Block
 // Storage

driver/driver.go

@@ -147,6 +147,12 @@ func (d *Driver) Run() error {
 		return resp, err
 	}
 
+	// warn the user, it'll not propagate to the user but at least we see if
+	// something is wrong in the logs
+	if err := d.checkLimit(context.Background()); err != nil {
+		d.log.WithError(err).Warn("CSI plugin will not function correctly, please resolve volume limit")
+	}
+
 	d.srv = grpc.NewServer(grpc.UnaryInterceptor(errHandler))
 	csi.RegisterIdentityServer(d.srv, d)
 	csi.RegisterControllerServer(d.srv, d)

driver/driver_test.go

@@ -135,6 +135,19 @@ func (f *fakeAPI) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// return empty response to account, assuming there is no volume limit
+	if strings.HasPrefix(r.URL.Path, "/v2/account") {
+		var resp = struct {
+			Account *godo.Account
+		}{
+			Account: &godo.Account{},
+		}
+
+		_ = json.NewEncoder(w).Encode(&resp)
+		return
+	}
+
+	// rest is /v2/volumes related
 	switch r.Method {
 	case "GET":
 		// A list call