Merge pull request #293 from dr0pdb/nodegetstats

feat: Implement node get stats

Author: k8s-ci-robot

pkg/blob/blob.go

@@ -157,6 +157,7 @@ func (d *Driver) Run(endpoint, kubeconfig string, testBool bool) {
 
 	d.AddNodeServiceCapabilities([]csi.NodeServiceCapability_RPC_Type{
 		csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
+		csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,
 		csi.NodeServiceCapability_RPC_UNKNOWN,
 	})
 

pkg/blob/nodeserver.go

@@ -30,6 +30,7 @@ import (
 
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
+	"k8s.io/kubernetes/pkg/volume"
 	"k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/utils/mount"
 
@@ -252,8 +253,69 @@ func (d *Driver) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (
 }
 
 // NodeGetVolumeStats get volume stats
-func (d *Driver) NodeGetVolumeStats(ctx context.Context, in *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
+func (d *Driver) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
+	if len(req.VolumeId) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "NodeGetVolumeStats volume ID was empty")
+	}
+	if len(req.VolumePath) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "NodeGetVolumeStats volume path was empty")
+	}
+
+	_, err := os.Stat(req.VolumePath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, status.Errorf(codes.NotFound, "path %s does not exist", req.VolumePath)
+		}
+		return nil, status.Errorf(codes.Internal, "failed to stat file %s: %v", req.VolumePath, err)
+	}
+
+	volumeMetrics, err := volume.NewMetricsStatFS(req.VolumePath).GetMetrics()
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to get metrics: %v", err)
+	}
+
+	available, ok := volumeMetrics.Available.AsInt64()
+	if !ok {
+		return nil, status.Errorf(codes.Internal, "failed to transform volume available size(%v)", volumeMetrics.Available)
+	}
+	capacity, ok := volumeMetrics.Capacity.AsInt64()
+	if !ok {
+		return nil, status.Errorf(codes.Internal, "failed to transform volume capacity size(%v)", volumeMetrics.Capacity)
+	}
+	used, ok := volumeMetrics.Used.AsInt64()
+	if !ok {
+		return nil, status.Errorf(codes.Internal, "failed to transform volume used size(%v)", volumeMetrics.Used)
+	}
+
+	inodesFree, ok := volumeMetrics.InodesFree.AsInt64()
+	if !ok {
+		return nil, status.Errorf(codes.Internal, "failed to transform disk inodes free(%v)", volumeMetrics.InodesFree)
+	}
+	inodes, ok := volumeMetrics.Inodes.AsInt64()
+	if !ok {
+		return nil, status.Errorf(codes.Internal, "failed to transform disk inodes(%v)", volumeMetrics.Inodes)
+	}
+	inodesUsed, ok := volumeMetrics.InodesUsed.AsInt64()
+	if !ok {
+		return nil, status.Errorf(codes.Internal, "failed to transform disk inodes used(%v)", volumeMetrics.InodesUsed)
+	}
+
+	return &csi.NodeGetVolumeStatsResponse{
+		Usage: []*csi.VolumeUsage{
+			{
+				Unit:      csi.VolumeUsage_BYTES,
+				Available: available,
+				Total:     capacity,
+				Used:      used,
+			},
+			{
+				Unit:      csi.VolumeUsage_INODES,
+				Available: inodesFree,
+				Total:     inodes,
+				Used:      inodesUsed,
+			},
+		},
+	}, nil
 }
 
 // NodeExpandVolume node expand volume

pkg/blob/nodeserver_test.go

@@ -415,13 +415,51 @@ func TestNewSafeMounter(t *testing.T) {
 }
 
 func TestNodeGetVolumeStats(t *testing.T) {
+	nonexistedPath := "/not/a/real/directory"
+	fakePath := "/tmp/fake-volume-path"
+
+	tests := []struct {
+		desc        string
+		req         csi.NodeGetVolumeStatsRequest
+		expectedErr error
+	}{
+		{
+			desc:        "[Error] Volume ID missing",
+			req:         csi.NodeGetVolumeStatsRequest{VolumePath: targetTest},
+			expectedErr: status.Error(codes.InvalidArgument, "NodeGetVolumeStats volume ID was empty"),
+		},
+		{
+			desc:        "[Error] VolumePath missing",
+			req:         csi.NodeGetVolumeStatsRequest{VolumeId: "vol_1"},
+			expectedErr: status.Error(codes.InvalidArgument, "NodeGetVolumeStats volume path was empty"),
+		},
+		{
+			desc:        "[Error] Incorrect volume path",
+			req:         csi.NodeGetVolumeStatsRequest{VolumePath: nonexistedPath, VolumeId: "vol_1"},
+			expectedErr: status.Errorf(codes.NotFound, "path /not/a/real/directory does not exist"),
+		},
+		{
+			desc:        "[Success] Standard success",
+			req:         csi.NodeGetVolumeStatsRequest{VolumePath: fakePath, VolumeId: "vol_1"},
+			expectedErr: nil,
+		},
+	}
+
+	// Setup
+	_ = makeDir(fakePath)
 	d := NewFakeDriver()
-	req := csi.NodeGetVolumeStatsRequest{}
-	resp, err := d.NodeGetVolumeStats(context.Background(), &req)
-	assert.Nil(t, resp)
-	if !reflect.DeepEqual(err, status.Error(codes.Unimplemented, "")) {
-		t.Errorf("Unexpected error: %v", err)
+
+	for _, test := range tests {
+		_, err := d.NodeGetVolumeStats(context.Background(), &test.req)
+		//t.Errorf("[debug] error: %v\n metrics: %v", err, metrics)
+		if !reflect.DeepEqual(err, test.expectedErr) {
+			t.Errorf("desc: %v, expected error: %v, actual error: %v", test.desc, test.expectedErr, err)
+		}
 	}
+
+	// Clean up
+	err := os.RemoveAll(fakePath)
+	assert.NoError(t, err)
 }
 
 func TestNodeExpandVolume(t *testing.T) {

test/integration/run-test.sh

@@ -62,6 +62,10 @@ if [[ "$cloud" != "AzureChinaCloud" ]]; then
   csc node publish --endpoint "$endpoint" --cap 1,block --staging-target-path "$staging_target_path" --target-path "$target_path" "$volumeid"
   sleep 2
 
+  echo "node stats test:"
+  csc node stats --endpoint "$endpoint" "$volumeid:$target_path:$staging_target_path"
+  sleep 2
+
   echo "unpublish volume test:"
   csc node unpublish --endpoint "$endpoint" --target-path "$target_path" "$volumeid"
   sleep 2