capacity: read-only Capacity struct

The previous approach of adapting remaining capacity while adding or
removing volumes was unnecessarily complex. When we accept that we
need to sum up allocated space repeatedly, the remaining code becomes
simpler.

Author: pohly

pkg/hostpath/capacity.go

@@ -22,15 +22,12 @@ import (
 	"fmt"
 	"strings"
 
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/api/resource"
 )
 
 // Capacity simulates linear storage of certain types ("fast",
-// "slow"). When volumes of those types get created, they must
-// allocate storage (which can fail!) and that storage must
-// be freed again when volumes get destroyed.
+// "slow"). To calculate the amount of allocated space, the size of
+// all currently existing volumes of the same kind is summed up.
 //
 // Available capacity is configurable with a command line flag
 // -capacity <type>=<size> where <type> is a string and <size>
@@ -63,63 +60,6 @@ func (c *Capacity) String() string {
 
 var _ flag.Value = &Capacity{}
 
-// Alloc reserves a certain amount of bytes. Errors are
-// usable as result of gRPC calls. Empty kind means
-// that any large enough one is fine.
-func (c *Capacity) Alloc(kind string, size int64) (actualKind string, err error) {
-	requested := *resource.NewQuantity(size, resource.BinarySI)
-
-	if kind == "" {
-		for k, quantity := range *c {
-			if quantity.Value() >= size {
-				kind = k
-				break
-			}
-		}
-		// Still nothing?
-		if kind == "" {
-			available := c.Check("")
-			return "", status.Error(codes.ResourceExhausted,
-				fmt.Sprintf("not enough capacity: have %s, need %s", available.String(), requested.String()))
-		}
-	}
-
-	available, ok := (*c)[kind]
-	if !ok {
-		return "", status.Error(codes.InvalidArgument, fmt.Sprintf("unknown capacity kind: %q", kind))
-	}
-	if available.Cmp(requested) < 0 {
-		return "", status.Error(codes.ResourceExhausted,
-			fmt.Sprintf("not enough capacity of kind %q: have %s, need %s", kind, available.String(), requested.String()))
-	}
-	available.Sub(requested)
-	(*c)[kind] = available
-	return kind, nil
-}
-
-// Free returns capacity reserved earlier with Alloc.
-func (c *Capacity) Free(kind string, size int64) {
-	available := (*c)[kind]
-	available.Add(*resource.NewQuantity(size, resource.BinarySI))
-	(*c)[kind] = available
-}
-
-// Check reports available capacity for a certain kind.
-// If empty, it reports the maximum capacity.
-func (c *Capacity) Check(kind string) resource.Quantity {
-	if kind != "" {
-		quantity := (*c)[kind]
-		return quantity
-	}
-	available := resource.Quantity{}
-	for _, q := range *c {
-		if q.Cmp(available) >= 0 {
-			available = q
-		}
-	}
-	return available
-}
-
 // Enabled returns true if capacities are configured.
 func (c *Capacity) Enabled() bool {
 	return len(*c) > 0

pkg/hostpath/controllerserver.go

@@ -363,9 +363,13 @@ func (hp *hostPath) GetCapacity(ctx context.Context, req *csi.GetCapacityRequest
 	// Without configured capacity, we just have the maximum size.
 	available := hp.config.MaxVolumeSize
 	if hp.config.Capacity.Enabled() {
+		// Empty "kind" will return "zero capacity". There is no fallback
+		// to some arbitrary kind here because in practice it always should
+		// be set.
 		kind := req.GetParameters()[storageKind]
-		quantity := hp.config.Capacity.Check(kind)
-		available = quantity.Value()
+		quantity := hp.config.Capacity[kind]
+		allocated := hp.sumVolumeSizes(kind)
+		available = quantity.Value() - allocated
 	}
 	maxVolumeSize := hp.config.MaxVolumeSize
 	if maxVolumeSize < available {

pkg/hostpath/hostpath.go

@@ -33,6 +33,7 @@ import (
 	timestamp "github.com/golang/protobuf/ptypes/timestamp"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"k8s.io/apimachinery/pkg/api/resource"
 	fs "k8s.io/kubernetes/pkg/volume/util/fs"
 	"k8s.io/kubernetes/pkg/volume/util/volumepathhandler"
 	utilexec "k8s.io/utils/exec"
@@ -219,12 +220,6 @@ func (hp *hostPath) discoveryExistingVolumes() error {
 		if err != nil {
 			return err
 		}
-
-		if hpv.Kind != "" && hp.config.Capacity.Enabled() {
-			if _, err := hp.config.Capacity.Alloc(hpv.Kind, hpv.VolSize); err != nil {
-				return fmt.Errorf("existing volume(s) do not match new capacity configuration: %v", err)
-			}
-		}
 		hp.volumes[hpv.VolID] = *hpv
 	}
 
@@ -281,17 +276,26 @@ func (hp *hostPath) createVolume(volID, name string, cap int64, volAccessType ac
 		return nil, status.Errorf(codes.OutOfRange, "Requested capacity %d exceeds maximum allowed %d", cap, hp.config.MaxVolumeSize)
 	}
 	if hp.config.Capacity.Enabled() {
-		actualKind, err := hp.config.Capacity.Alloc(kind, cap)
-		if err != nil {
-			return nil, err
-		}
-		// Free the capacity in case of any error - either a volume gets created or it doesn't.
-		defer func() {
-			if finalErr != nil {
-				hp.config.Capacity.Free(actualKind, cap)
+		if kind == "" {
+			// Pick some kind with sufficient remaining capacity.
+			for k, c := range hp.config.Capacity {
+				if hp.sumVolumeSizes(k)+cap <= c.Value() {
+					kind = k
+					break
+				}
 			}
-		}()
-		kind = actualKind
+		}
+		if kind == "" {
+			// Still nothing?!
+			return nil, status.Errorf(codes.OutOfRange, "requested capacity %d of arbitrary storage exceeds all remaining capacity", cap)
+		}
+		used := hp.sumVolumeSizes(kind)
+		available := hp.config.Capacity[kind]
+		if used+cap > available.Value() {
+
+			return nil, status.Errorf(codes.OutOfRange, "requested capacity %d exceeds remaining capacity for %q, %s out of %s already used",
+				cap, kind, resource.NewQuantity(used, resource.BinarySI).String(), available.String())
+		}
 	} else if kind != "" {
 		return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("capacity tracking disabled, specifying kind %q is invalid", kind))
 	}
@@ -383,14 +387,20 @@ func (hp *hostPath) deleteVolume(volID string) error {
 	if err := os.RemoveAll(path); err != nil && !os.IsNotExist(err) {
 		return err
 	}
-	if hp.config.Capacity.Enabled() {
-		hp.config.Capacity.Free(vol.Kind, vol.VolSize)
-	}
 	delete(hp.volumes, volID)
 	glog.V(4).Infof("deleted hostpath volume: %s = %+v", volID, vol)
 	return nil
 }
 
+func (hp *hostPath) sumVolumeSizes(kind string) (sum int64) {
+	for _, volume := range hp.volumes {
+		if volume.Kind == kind {
+			sum += volume.VolSize
+		}
+	}
+	return
+}
+
 // hostPathIsEmpty is a simple check to determine if the specified hostpath directory
 // is empty or not.
 func hostPathIsEmpty(p string) (bool, error) {