Merge pull request #591 from torredil/cherrypick5123

Update release-1.16 branch for release

Author: k8s-ci-robot

pkg/hostpath/controllerserver.go

@@ -97,12 +97,23 @@ func (hp *hostPath) CreateVolume(ctx context.Context, req *csi.CreateVolumeReque
 		requestedAccessType = state.MountAccess
 	}
 
+	capacity := int64(req.GetCapacityRange().GetRequiredBytes())
+	limit := int64(req.GetCapacityRange().GetLimitBytes())
+	if capacity < 0 || limit < 0 {
+		return nil, status.Error(codes.InvalidArgument, "cannot have negative capacity")
+	}
+	if limit > 0 && capacity > limit {
+		return nil, status.Error(codes.InvalidArgument, "capacity cannot exceed limit")
+	}
+	if capacity == 0 && limit > 0 {
+		capacity = limit
+	}
+
 	// Lock before acting on global state. A production-quality
 	// driver might use more fine-grained locking.
 	hp.mutex.Lock()
 	defer hp.mutex.Unlock()
 
-	capacity := int64(req.GetCapacityRange().GetRequiredBytes())
 	topologies := []*csi.Topology{}
 	if hp.config.EnableTopology {
 		topologies = append(topologies, &csi.Topology{Segments: map[string]string{TopologyKeyNode: hp.config.NodeID}})
@@ -114,7 +125,7 @@ func (hp *hostPath) CreateVolume(ctx context.Context, req *csi.CreateVolumeReque
 		// Since err is nil, it means the volume with the same name already exists
 		// need to check if the size of existing volume is the same as in new
 		// request
-		if exVol.VolSize < capacity {
+		if exVol.VolSize < capacity || (limit > 0 && exVol.VolSize > limit) {
 			return nil, status.Errorf(codes.AlreadyExists, "Volume with the same name: %s but with different size already exist", req.GetName())
 		}
 		if req.GetVolumeContentSource() != nil {
@@ -146,6 +157,8 @@ func (hp *hostPath) CreateVolume(ctx context.Context, req *csi.CreateVolumeReque
 
 	volumeID := uuid.NewUUID().String()
 	kind := req.GetParameters()[storageKind]
+	// This code does not check whether hp.createVolume rounds capacity up;
+	// a more robust driver would ensure any rounding does not exceed limit.
 	vol, err := hp.createVolume(volumeID, req.GetName(), capacity, requestedAccessType, false /* ephemeral */, kind)
 	if err != nil {
 		return nil, err
@@ -182,7 +195,7 @@ func (hp *hostPath) CreateVolume(ctx context.Context, req *csi.CreateVolumeReque
 	return &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
 			VolumeId:           volumeID,
-			CapacityBytes:      req.GetCapacityRange().GetRequiredBytes(),
+			CapacityBytes:      capacity,
 			VolumeContext:      req.GetParameters(),
 			ContentSource:      req.GetVolumeContentSource(),
 			AccessibleTopology: topologies,

pkg/hostpath/nodeserver.go

@@ -18,6 +18,7 @@ package hostpath
 
 import (
 	"fmt"
+	"math/rand"
 	"os"
 	"strings"
 
@@ -365,6 +366,11 @@ func (hp *hostPath) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest
 		resp.MaxVolumesPerNode = hp.config.AttachLimit
 	}
 
+	// if attach limit is -1, set a random MaxVolumesPerNode between 1 and 10
+	if hp.config.AttachLimit == -1 {
+		resp.MaxVolumesPerNode = int64(rand.Intn(10) + 1)
+	}
+
 	return resp, nil
 }
 

pkg/hostpath/snapshotmetadata.go

@@ -118,6 +118,7 @@ func (cb *fileBlockReader) getChangedBlockMetadata(ctx context.Context) ([]*csi.
 	blockIndex := cb.offset / cb.blockSize
 	sBuffer := make([]byte, cb.blockSize)
 	tBuffer := make([]byte, cb.blockSize)
+	zeroBlock := make([]byte, cb.blockSize)
 	eofBaseFile, eofTargetFile := false, false
 
 	changedBlocks := []*csi.BlockMetadata{}
@@ -143,15 +144,19 @@ func (cb *fileBlockReader) getChangedBlockMetadata(ctx context.Context) ([]*csi.
 			if eofTargetFile {
 				return changedBlocks, io.EOF
 			}
-			// if VARIABLE_LENGTH type is enabled, return blocks extend instead of individual blocks.
-			blockMetadata := createBlockMetadata(blockIndex, cb.blockSize)
-			if extendBlock(changedBlocks, csi.BlockMetadataType(cb.blockMetadataType), blockIndex, cb.blockSize) {
-				changedBlocks[len(changedBlocks)-1].SizeBytes += cb.blockSize
-				cb.offset += cb.blockSize
-				blockIndex++
-				continue
+			// return only allocated blocks.
+			if blockChanged(zeroBlock, tBuffer[:targetReadBytes]) {
+				// if VARIABLE_LENGTH type is enabled, return blocks extend instead of individual blocks.
+				blockMetadata := createBlockMetadata(blockIndex, cb.blockSize)
+				if extendBlock(changedBlocks, csi.BlockMetadataType(cb.blockMetadataType), blockIndex, cb.blockSize) {
+					changedBlocks[len(changedBlocks)-1].SizeBytes += cb.blockSize
+					cb.offset += cb.blockSize
+					blockIndex++
+					continue
+				}
+				changedBlocks = append(changedBlocks, blockMetadata)
 			}
-			changedBlocks = append(changedBlocks, blockMetadata)
+
 			cb.offset += cb.blockSize
 			blockIndex++
 			continue