From d15fd7455ed6edbd137b687f8d028fbe28f9689a Mon Sep 17 00:00:00 2001
From: Sergey Berezansky <sergey@weka.io>
Date: Mon, 9 Sep 2024 13:26:46 +0300
Subject: [PATCH] feat(CSI-252): implement kubelet PVC stats

---
 go.mod                   |  1 +
 go.sum                   |  1 +
 pkg/wekafs/nodeserver.go | 70 ++++++++++++++++++++++++++++++++++++++--
 3 files changed, 69 insertions(+), 3 deletions(-)

diff --git a/go.mod b/go.mod
index 406c632ff..18f44402b 100644
--- a/go.mod
+++ b/go.mod
@@ -8,6 +8,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/go-version v1.7.0
 	github.com/kubernetes-csi/csi-lib-utils v0.19.0
+	github.com/pkg/errors v0.9.1
 	github.com/pkg/xattr v0.4.10
 	github.com/prometheus/client_golang v1.20.2
 	github.com/rs/zerolog v1.33.0
diff --git a/go.sum b/go.sum
index ece93c7da..c36b7427c 100644
--- a/go.sum
+++ b/go.sum
@@ -54,6 +54,7 @@ github.com/opencontainers/runc v1.1.13 h1:98S2srgG9vw0zWcDpFMn5TRrh8kLxa/5OFUstu
 github.com/opencontainers/runc v1.1.13/go.mod h1:R016aXacfp/gwQBYw2FDGa9m+n6atbLWrYY8hNMT/sA=
 github.com/opencontainers/runtime-spec v1.2.0 h1:z97+pHb3uELt/yiAWD691HNHQIF07bE7dzrbT927iTk=
 github.com/opencontainers/runtime-spec v1.2.0/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/xattr v0.4.10 h1:Qe0mtiNFHQZ296vRgUjRCoPHPqH7VdTOrZx3g0T+pGA=
 github.com/pkg/xattr v0.4.10/go.mod h1:di8WF84zAKk8jzR1UBTEWh9AUlIZZ7M/JNt8e9B6ktU=
diff --git a/pkg/wekafs/nodeserver.go b/pkg/wekafs/nodeserver.go
index 4190666ff..3a793940e 100644
--- a/pkg/wekafs/nodeserver.go
+++ b/pkg/wekafs/nodeserver.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"go.opentelemetry.io/otel"
@@ -32,6 +33,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"syscall"
 	"time"
 )
 
@@ -84,9 +86,68 @@ func (ns *NodeServer) NodeExpandVolume(ctx context.Context, request *csi.NodeExp
 	panic("implement me")
 }
 
-//goland:noinspection GoUnusedParameter
-func (ns *NodeServer) NodeGetVolumeStats(ctx context.Context, request *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
-	panic("implement me")
+func (ns *NodeServer) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
+	volumeID := req.GetVolumeId()
+	volumePath := req.GetVolumePath()
+
+	// Validate request fields
+	if volumeID == "" || volumePath == "" {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID and path must be provided")
+	}
+
+	// Check if the volume path exists
+	if !PathIsWekaMount(ctx, volumePath) {
+		return nil, status.Error(codes.NotFound, "Volume path not found")
+	}
+
+	// Validate Weka volume ID
+	if err := validateVolumeId(volumeID); err != nil {
+		return nil, status.Error(codes.InvalidArgument, errors.Wrap(err, "invalid volume ID").Error())
+	}
+
+	capacityBytes, usedBytes, availableBytes, err := getVolumeStats(volumePath)
+	if err != nil {
+		return &csi.NodeGetVolumeStatsResponse{
+			Usage: nil,
+			VolumeCondition: &csi.VolumeCondition{
+				Abnormal: true,
+				Message:  "Failed to fetch volume stats for volume",
+			},
+		}, status.Errorf(codes.Internal, "Failed to get stats for volume %s: %v", volumeID, err)
+	}
+	// Prepare response
+	return &csi.NodeGetVolumeStatsResponse{
+		Usage: []*csi.VolumeUsage{
+			{
+				Unit:      csi.VolumeUsage_BYTES,
+				Total:     capacityBytes,
+				Used:      usedBytes,
+				Available: availableBytes,
+			},
+		},
+		VolumeCondition: &csi.VolumeCondition{
+			Abnormal: false,
+			Message:  "volume is healthy",
+		},
+	}, nil
+}
+
+// getVolumeStats fetches filesystem statistics for the mounted volume path.
+func getVolumeStats(volumePath string) (capacityBytes, usedBytes, availableBytes int64, err error) {
+	var stat syscall.Statfs_t
+
+	// Use Statfs to get filesystem statistics for the volume path
+	err = syscall.Statfs(volumePath, &stat)
+	if err != nil {
+		return 0, 0, 0, err
+	}
+
+	// Calculate capacity, available, and used space in bytes
+	capacityBytes = int64(stat.Blocks) * int64(stat.Bsize)
+	availableBytes = int64(stat.Bavail) * int64(stat.Bsize)
+	usedBytes = capacityBytes - availableBytes
+
+	return capacityBytes, usedBytes, availableBytes, nil
 }
 
 func NewNodeServer(nodeId string, maxVolumesPerNode int64, api *ApiStore, mounter AnyMounter, config *DriverConfig) *NodeServer {
@@ -94,6 +155,9 @@ func NewNodeServer(nodeId string, maxVolumesPerNode int64, api *ApiStore, mounte
 	return &NodeServer{
 		caps: getNodeServiceCapabilities(
 			[]csi.NodeServiceCapability_RPC_Type{
+				csi.NodeServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER,
+				csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,
+				csi.NodeServiceCapability_RPC_VOLUME_CONDITION,
 				//csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
 			},
 		),