feature: implement necessary gRPC handlers of health-monitor

Author: fengzixu

Dockerfile

@@ -4,6 +4,6 @@ LABEL description="HostPath Driver"
 ARG binary=./bin/hostpathplugin
 
 # Add util-linux to get a new version of losetup.
-RUN apk add util-linux
+RUN apk add util-linux && apk update && apk upgrade
 COPY ${binary} /hostpathplugin
 ENTRYPOINT ["/hostpathplugin"]

cmd/hostpathplugin/main.go

@@ -36,6 +36,7 @@ var (
 	ephemeral         = flag.Bool("ephemeral", false, "publish volumes in ephemeral mode even if kubelet did not ask for it (only needed for Kubernetes 1.15)")
 	maxVolumesPerNode = flag.Int64("maxvolumespernode", 0, "limit of volumes per node")
 	showVersion       = flag.Bool("version", false, "Show version.")
+	kubeconfig        = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
 	// Set by the build process
 	version = ""
 )
@@ -58,10 +59,11 @@ func main() {
 }
 
 func handle() {
-	driver, err := hostpath.NewHostPathDriver(*driverName, *nodeID, *endpoint, *ephemeral, *maxVolumesPerNode, version)
+	driver, err := hostpath.NewHostPathDriver(*driverName, *nodeID, *endpoint, *ephemeral, *maxVolumesPerNode, version, *kubeconfig)
 	if err != nil {
 		fmt.Printf("Failed to initialize driver: %s", err.Error())
 		os.Exit(1)
 	}
+
 	driver.Run()
 }

go.mod

@@ -4,19 +4,19 @@ go 1.12
 
 require (
 	github.com/container-storage-interface/spec v1.3.0
-	github.com/gogo/protobuf v1.3.1 // indirect
 	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
-	github.com/golang/protobuf v1.3.2
+	github.com/golang/protobuf v1.4.2
 	github.com/google/gofuzz v1.2.0 // indirect
-	github.com/google/uuid v1.0.0 // indirect
+	github.com/imdario/mergo v0.3.11 // indirect
 	github.com/kubernetes-csi/csi-lib-utils v0.3.0
 	github.com/pborman/uuid v0.0.0-20180906182336-adf5a7427709
-	github.com/spf13/afero v1.2.2 // indirect
-	github.com/stretchr/testify v1.4.0 // indirect
-	golang.org/x/net v0.0.0-20190311183353-d8887717615a
-	google.golang.org/grpc v1.26.0
-	gopkg.in/inf.v0 v0.9.1 // indirect
-	k8s.io/apimachinery v0.0.0-20181110190943-2a7c93004028 // indirect
+	github.com/stretchr/testify v1.4.0
+	golang.org/x/net v0.0.0-20200707034311-ab3426394381
+	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e // indirect
+	google.golang.org/grpc v1.27.0
+	k8s.io/api v0.19.0
+	k8s.io/apimachinery v0.19.0
+	k8s.io/client-go v0.19.0
 	k8s.io/kubernetes v1.12.2
-	k8s.io/utils v0.0.0-20181102055113-1bd4f387aa67
+	k8s.io/utils v0.0.0-20201015054608-420da100c033
 )

go.sum

@@ -67,6 +67,7 @@ func NewControllerServer(ephemeral bool, nodeID string) *controllerServer {
 				csi.ControllerServiceCapability_RPC_LIST_VOLUMES,
 				csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
 				csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
+				csi.ControllerServiceCapability_RPC_VOLUME_CONDITION,
 			}),
 		nodeID: nodeID,
 	}
@@ -288,16 +289,26 @@ func (cs *controllerServer) ListVolumes(ctx context.Context, req *csi.ListVolume
 		hpVolume                           hostPathVolume
 	)
 	volumeIds := getSortedVolumeIDs()
+	if req.StartingToken == "" {
+		req.StartingToken = "1"
+	}
+
 	startIdx, err := strconv.ParseInt(req.StartingToken, 10, 32)
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, "The type of startingToken should be integer")
 	}
 
 	volumesLength = int64(len(volumeIds))
 	maxLength = int64(req.MaxEntries)
-	for index := startIdx - 1; index < volumesLength || index < maxLength; index++ {
+
+	if maxLength > volumesLength || maxLength <= 0 {
+		maxLength = volumesLength
+	}
+
+	for index := startIdx - 1; index < volumesLength && index < maxLength; index++ {
 		hpVolume = hostPathVolumes[volumeIds[index]]
 		healthy, msg := doHealthCheck(volumeIds[index])
+		glog.V(3).Infof("Healthy state: %+v Volume: %+v", hpVolume.VolName, healthy)
 		volumeRes.Entries = append(volumeRes.Entries, &csi.ListVolumesResponse_Entry{
 			Volume: &csi.Volume{
 				VolumeId:      hpVolume.VolID,
@@ -312,11 +323,32 @@ func (cs *controllerServer) ListVolumes(ctx context.Context, req *csi.ListVolume
 			},
 		})
 	}
+
+	glog.V(5).Infof("Volumes are: %+v", *volumeRes)
 	return volumeRes, nil
 }
 
 func (cs *controllerServer) ControllerGetVolume(ctx context.Context, req *csi.ControllerGetVolumeRequest) (*csi.ControllerGetVolumeResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
+	volume, ok := hostPathVolumes[req.GetVolumeId()]
+	if !ok {
+		return nil, status.Error(codes.NotFound, "The volume not found")
+	}
+
+	healthy, msg := doHealthCheck(req.GetVolumeId())
+	glog.V(3).Infof("Healthy state: %+v Volume: %+v", volume.VolName, healthy)
+	return &csi.ControllerGetVolumeResponse{
+		Volume: &csi.Volume{
+			VolumeId:      volume.VolID,
+			CapacityBytes: volume.VolSize,
+		},
+		Status: &csi.ControllerGetVolumeResponse_VolumeStatus{
+			PublishedNodeIds: []string{volume.NodeID},
+			VolumeCondition: &csi.VolumeCondition{
+				Abnormal: !healthy,
+				Message:  msg,
+			},
+		},
+	}, nil
 }
 
 // getSnapshotPath returns the full path to where the snapshot is stored

pkg/hostpath/controllerserver.go

@@ -1,21 +1,40 @@
 package hostpath
 
 import (
+	"encoding/json"
 	"fmt"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"strings"
 
+	"github.com/golang/glog"
 	fs "k8s.io/kubernetes/pkg/volume/util/fs"
-	utilexec "k8s.io/utils/exec"
 )
 
 const (
 	// sourcePathPrefix should be specified by the csi-data-dir volume.
 	// You can change it by modifying the volume path of csi-data-dir volume.
-	sourcePathPrefix = "/var/lib/csi-hostpath-data/"
+	sourcePathPrefix    = "/csi-data-dir/"
+	podVolumeTargetPath = "/var/lib/kubelet/pods"
 )
 
+type MountPointInfo struct {
+	Target              string           `json:"target"`
+	Source              string           `json:"source"`
+	FsType              string           `json:"fstype"`
+	Options             string           `json:"options"`
+	ContainerFileSystem []MountPointInfo `json:"children,omitempty"`
+}
+
+type ContainerFileSystem struct {
+	Children []MountPointInfo `json:"children"`
+}
+
+type FileSystems struct {
+	Filsystem []ContainerFileSystem `json:"filesystems"`
+}
+
 func locateCommandPath(commandName string) string {
 	// default to root
 	binary := filepath.Join("/", commandName)
@@ -36,7 +55,8 @@ func getSourcePath(volumeHandle string) string {
 }
 
 func checkSourcePathExist(volumeHandle string) (bool, error) {
-	sourcePath := getSnapshotPath(volumeHandle)
+	sourcePath := getSourcePath(volumeHandle)
+	glog.V(3).Infof("Volume: %+v Source path is: %+v", volumeHandle, sourcePath)
 	_, err := os.Stat(sourcePath)
 	if err != nil {
 		if os.IsNotExist(err) {
@@ -49,60 +69,93 @@ func checkSourcePathExist(volumeHandle string) (bool, error) {
 	return true, nil
 }
 
+func parseMountInfo(originalMountInfo []byte) ([]MountPointInfo, error) {
+	fs := FileSystems{
+		Filsystem: make([]ContainerFileSystem, 0),
+	}
+
+	if err := json.Unmarshal(originalMountInfo, &fs); err != nil {
+		return nil, err
+	}
+
+	if len(fs.Filsystem) <= 0 {
+		return nil, fmt.Errorf("failed to get mount info")
+	}
+
+	return fs.Filsystem[0].Children, nil
+}
+
 func checkMountPointExist(sourcePath string) (bool, error) {
-	exec := utilexec.New()
-	args := []string{"-l", "-o TARGET,SOURCE"}
-	out, err := exec.Command(locateCommandPath("findmnt"), args...).CombinedOutput()
+	cmdPath := locateCommandPath("findmnt")
+	out, err := exec.Command(cmdPath, "--json").CombinedOutput()
 	if err != nil {
+		glog.V(3).Infof("failed to execute command: %+v", cmdPath)
 		return false, err
 	}
 
 	if len(out) < 1 {
 		return false, fmt.Errorf("mount point info is nil")
 	}
 
-	mountInfo := strings.Split(string(out), " ")
-	if len(mountInfo) != 2 {
-		return false, nil
+	mountInfos, err := parseMountInfo([]byte(out))
+	if err != nil {
+		return false, fmt.Errorf("failed to parse the mount infos: %+v", err)
 	}
 
-	mountPoint := mountInfo[0]
-	sp := mountInfo[1]
-
-	if !strings.Contains(sp, sourcePath) {
-		return false, nil
+	mountInfosOfPod := MountPointInfo{}
+	for _, mountInfo := range mountInfos {
+		if mountInfo.Target == podVolumeTargetPath {
+			mountInfosOfPod = mountInfo
+			break
+		}
 	}
 
-	_, err = os.Stat(mountPoint)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return false, nil
+	for _, mountInfo := range mountInfosOfPod.ContainerFileSystem {
+		if !strings.Contains(mountInfo.Source, sourcePath) {
+			continue
 		}
 
-		return false, err
+		_, err = os.Stat(mountInfo.Target)
+		if err != nil {
+			if os.IsNotExist(err) {
+				return false, nil
+			}
+
+			return false, err
+		}
+
+		return true, nil
 	}
 
-	return true, nil
+	return false, nil
 }
 
 func checkPVCapacityValid(volumeHandle string) (bool, error) {
 	sourcePath := getSourcePath(volumeHandle)
 	_, fscapacity, _, _, _, _, err := fs.FsInfo(sourcePath)
 	if err != nil {
-		return false, err
+		return false, fmt.Errorf("failed to get capacity info: %+v", err)
 	}
 
 	volumeCapacity := hostPathVolumes[volumeHandle].VolSize
+	glog.V(3).Infof("volume capacity: %+v fs capacity:%+v", volumeCapacity, fscapacity)
 	return fscapacity >= volumeCapacity, nil
 }
 
+func getPVCapacity(volumeHandle string) (int64, int64, int64, error) {
+	sourcePath := getSourcePath(volumeHandle)
+	fsavailable, fscapacity, fsused, _, _, _, err := fs.FsInfo(sourcePath)
+	return fscapacity, fsused, fsavailable, err
+}
+
 func checkPVUsage(volumeHandle string) (bool, error) {
 	sourcePath := getSourcePath(volumeHandle)
 	fsavailable, _, _, _, _, _, err := fs.FsInfo(sourcePath)
 	if err != nil {
 		return false, err
 	}
 
+	glog.V(3).Infof("fs available: %+v", fsavailable)
 	return fsavailable > 0, nil
 }