Implement metrics for cephfs CSI driver

Signed-off-by: Humble Chirammal <hchiramm@redhat.com>

pkg/cephfs/nodeserver.go

@@ -28,6 +28,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/klog"
+	"k8s.io/kubernetes/pkg/volume"
 )
 
 // NodeServer struct of ceph CSI driver with supported methods of CSI
@@ -308,6 +309,89 @@ func (ns *NodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag
 	return &csi.NodeUnstageVolumeResponse{}, nil
 }
 
+func (ns *NodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
+
+	var err error
+	targetPath := req.GetVolumePath()
+	if targetPath == "" {
+		err = fmt.Errorf("targetpath %v is empty", targetPath)
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+	volID := req.GetVolumeId()
+	/* TODO: Map the volumeID to the targetpath.
+
+	we need secret to connect to the ceph cluster to get the volumeID from volume
+	Name, however `secret` field/option is not available  in NodeGetVolumeStats spec,
+	Below issue covers this request and once its available, we can do the validation
+	as per the spec.
+	https://github.com/container-storage-interface/spec/issues/371
+
+	*/
+
+	isMnt, err := isMountPoint(targetPath)
+
+	if err != nil {
+		klog.Errorf("stat failed on path: %v, error: %v", targetPath, err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	if isMnt {
+		klog.Infof("cephfs: volume %s is already mounted to %s", volID, targetPath)
+		cephfsProvider := volume.NewMetricsStatFS(targetPath)
+		volMetrics, volMetErr := cephfsProvider.GetMetrics()
+		if volMetErr != nil {
+			return nil, status.Error(codes.Internal, volMetErr.Error())
+		}
+
+		available, ok := (*(volMetrics.Available)).AsInt64()
+		if !ok {
+			klog.Errorf("failed to fetch available bytes")
+		}
+		capacity, ok := (*(volMetrics.Capacity)).AsInt64()
+		if !ok {
+			klog.Errorf("failed to fetch capacity bytes")
+		}
+		used, ok := (*(volMetrics.Used)).AsInt64()
+		if !ok {
+			klog.Errorf("failed to fetch used bytes")
+		}
+		inodes, ok := (*(volMetrics.Inodes)).AsInt64()
+		if !ok {
+			klog.Errorf("failed to fetch available inodes")
+		}
+		inodesFree, ok := (*(volMetrics.InodesFree)).AsInt64()
+		if !ok {
+			klog.Errorf("failed to fetch free inodes")
+		}
+
+		inodesUsed, ok := (*(volMetrics.InodesUsed)).AsInt64()
+		if !ok {
+			klog.Errorf("failed to fetch used inodes")
+		}
+		return &csi.NodeGetVolumeStatsResponse{
+			Usage: []*csi.VolumeUsage{
+				{
+					Available: available,
+					Total:     capacity,
+					Used:      used,
+					Unit:      1,
+				},
+				{
+					Available: inodesFree,
+					Total:     inodes,
+					Used:      inodesUsed,
+					Unit:      2,
+				},
+			},
+		}, nil
+
+	}
+
+	klog.Infof("cephfs: successfully got stats for volume %s to %s", volID, targetPath)
+
+	return &csi.NodeGetVolumeStatsResponse{}, nil
+}
+
 // NodeGetCapabilities returns the supported capabilities of the node server
 func (ns *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetCapabilitiesRequest) (*csi.NodeGetCapabilitiesResponse, error) {
 	return &csi.NodeGetCapabilitiesResponse{
@@ -319,6 +403,13 @@ func (ns *NodeServer) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetC
 					},
 				},
 			},
+			{
+				Type: &csi.NodeServiceCapability_Rpc{
+					Rpc: &csi.NodeServiceCapability_RPC{
+						Type: csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,
+					},
+				},
+			},
 		},
 	}, nil
 }