feature: support external-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

@@ -63,5 +63,10 @@ func handle() {
 		fmt.Printf("Failed to initialize driver: %s", err.Error())
 		os.Exit(1)
 	}
-	driver.Run()
+
+	if err := driver.Run(); err != nil {
+		fmt.Printf("Failed to run driver: %s", err.Error())
+		os.Exit(1)
+
+	}
 }

go.mod

@@ -3,17 +3,18 @@ module github.com/kubernetes-csi/csi-driver-host-path
 go 1.12
 
 require (
-	github.com/container-storage-interface/spec v1.2.0
+	github.com/container-storage-interface/spec v1.3.0
 	github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
-	github.com/golang/protobuf v1.3.2
-	github.com/google/uuid v1.0.0 // indirect
+	github.com/golang/protobuf v1.4.2
+	github.com/google/gofuzz v1.2.0 // 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
-	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
+	gopkg.in/yaml.v2 v2.3.0 // indirect
+	k8s.io/client-go v0.19.1
 	k8s.io/kubernetes v1.12.2
-	k8s.io/utils v0.0.0-20181102055113-1bd4f387aa67
+	k8s.io/utils v0.0.0-20201104234853-8146046b121e
 )

go.sum

@@ -61,10 +61,13 @@ func NewControllerServer(ephemeral bool, nodeID string) *controllerServer {
 		caps: getControllerServiceCapabilities(
 			[]csi.ControllerServiceCapability_RPC_Type{
 				csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+				csi.ControllerServiceCapability_RPC_GET_VOLUME,
 				csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
 				csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
+				csi.ControllerServiceCapability_RPC_LIST_VOLUMES,
 				csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
 				csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
+				csi.ControllerServiceCapability_RPC_VOLUME_CONDITION,
 			}),
 		nodeID: nodeID,
 	}
@@ -277,7 +280,75 @@ func (cs *controllerServer) GetCapacity(ctx context.Context, req *csi.GetCapacit
 }
 
 func (cs *controllerServer) ListVolumes(ctx context.Context, req *csi.ListVolumesRequest) (*csi.ListVolumesResponse, error) {
-	return nil, status.Error(codes.Unimplemented, "")
+	volumeRes := &csi.ListVolumesResponse{
+		Entries: []*csi.ListVolumesResponse_Entry{},
+	}
+
+	var (
+		startIdx, volumesLength, maxLength int64
+		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)
+
+	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: %s Volume: %t", hpVolume.VolName, healthy)
+		volumeRes.Entries = append(volumeRes.Entries, &csi.ListVolumesResponse_Entry{
+			Volume: &csi.Volume{
+				VolumeId:      hpVolume.VolID,
+				CapacityBytes: hpVolume.VolSize,
+			},
+			Status: &csi.ListVolumesResponse_VolumeStatus{
+				PublishedNodeIds: []string{hpVolume.NodeID},
+				VolumeCondition: &csi.VolumeCondition{
+					Abnormal: !healthy,
+					Message:  msg,
+				},
+			},
+		})
+	}
+
+	glog.V(5).Infof("Volumes are: %+v", *volumeRes)
+	return volumeRes, nil
+}
+
+func (cs *controllerServer) ControllerGetVolume(ctx context.Context, req *csi.ControllerGetVolumeRequest) (*csi.ControllerGetVolumeResponse, error) {
+	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: %s Volume: %t", 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
@@ -312,7 +383,7 @@ func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 				Snapshot: &csi.Snapshot{
 					SnapshotId:     exSnap.Id,
 					SourceVolumeId: exSnap.VolID,
-					CreationTime:   &exSnap.CreationTime,
+					CreationTime:   exSnap.CreationTime,
 					SizeBytes:      exSnap.SizeBytes,
 					ReadyToUse:     exSnap.ReadyToUse,
 				},
@@ -352,7 +423,7 @@ func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 	snapshot.Id = snapshotID
 	snapshot.VolID = volumeID
 	snapshot.Path = file
-	snapshot.CreationTime = *creationTime
+	snapshot.CreationTime = creationTime
 	snapshot.SizeBytes = hostPathVolume.VolSize
 	snapshot.ReadyToUse = true
 
@@ -362,7 +433,7 @@ func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 		Snapshot: &csi.Snapshot{
 			SnapshotId:     snapshot.Id,
 			SourceVolumeId: snapshot.VolID,
-			CreationTime:   &snapshot.CreationTime,
+			CreationTime:   snapshot.CreationTime,
 			SizeBytes:      snapshot.SizeBytes,
 			ReadyToUse:     snapshot.ReadyToUse,
 		},
@@ -423,7 +494,7 @@ func (cs *controllerServer) ListSnapshots(ctx context.Context, req *csi.ListSnap
 		snapshot := csi.Snapshot{
 			SnapshotId:     snap.Id,
 			SourceVolumeId: snap.VolID,
-			CreationTime:   &snap.CreationTime,
+			CreationTime:   snap.CreationTime,
 			SizeBytes:      snap.SizeBytes,
 			ReadyToUse:     snap.ReadyToUse,
 		}
@@ -531,7 +602,7 @@ func convertSnapshot(snap hostPathSnapshot) *csi.ListSnapshotsResponse {
 			Snapshot: &csi.Snapshot{
 				SnapshotId:     snap.Id,
 				SourceVolumeId: snap.VolID,
-				CreationTime:   &snap.CreationTime,
+				CreationTime:   snap.CreationTime,
 				SizeBytes:      snap.SizeBytes,
 				ReadyToUse:     snap.ReadyToUse,
 			},

pkg/hostpath/controllerserver.go

@@ -0,0 +1,201 @@
+package hostpath
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+
+	"github.com/golang/glog"
+	fs "k8s.io/kubernetes/pkg/volume/util/fs"
+)
+
+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          = "/csi-data-dir/"
+	podVolumeTargetPath       = "/var/lib/kubelet/pods"
+	csiSignOfVolumeTargetPath = "kubernetes.io~csi/pvc"
+)
+
+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)
+	for _, path := range []string{"/bin", "/usr/sbin", "/usr/bin"} {
+		binPath := filepath.Join(path, binary)
+		if _, err := os.Stat(binPath); err != nil {
+			continue
+		}
+
+		return binPath
+	}
+
+	return ""
+}
+
+func getSourcePath(volumeHandle string) string {
+	return fmt.Sprintf("%s%s", sourcePathPrefix, volumeHandle)
+}
+
+func checkSourcePathExist(volumeHandle string) (bool, error) {
+	sourcePath := getSourcePath(volumeHandle)
+	glog.V(3).Infof("Volume: %s Source path is: %s", volumeHandle, sourcePath)
+	_, err := os.Stat(sourcePath)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return false, nil
+		}
+
+		return false, err
+	}
+
+	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) {
+	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")
+	}
+
+	mountInfos, err := parseMountInfo([]byte(out))
+	if err != nil {
+		return false, fmt.Errorf("failed to parse the mount infos: %+v", err)
+	}
+
+	mountInfosOfPod := MountPointInfo{}
+	for _, mountInfo := range mountInfos {
+		if mountInfo.Target == podVolumeTargetPath {
+			mountInfosOfPod = mountInfo
+			break
+		}
+	}
+
+	for _, mountInfo := range mountInfosOfPod.ContainerFileSystem {
+		if !strings.Contains(mountInfo.Source, sourcePath) {
+			continue
+		}
+
+		_, err = os.Stat(mountInfo.Target)
+		if err != nil {
+			if os.IsNotExist(err) {
+				return false, nil
+			}
+
+			return false, err
+		}
+
+		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, 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
+}
+
+func doHealthCheck(volumeHandle string) (bool, string) {
+	spExist, err := checkSourcePathExist(volumeHandle)
+	if err != nil {
+		return false, err.Error()
+	}
+
+	if !spExist {
+		return false, "The source path of the volume doesn't exist"
+	}
+
+	mpExist, err := checkMountPointExist(volumeHandle)
+	if err != nil {
+		return false, err.Error()
+	}
+
+	if !mpExist {
+		return false, "The volume isn't mounted"
+	}
+
+	capValid, err := checkPVCapacityValid(volumeHandle)
+	if err != nil {
+		return false, err.Error()
+	}
+
+	if !capValid {
+		return false, "The capacity of volume is greater than actual storage"
+	}
+
+	available, err := checkPVUsage(volumeHandle)
+	if err != nil {
+		return false, err.Error()
+	}
+
+	if !available {
+		return false, "The free space of the volume is insufficient"
+	}
+
+	return true, ""
+}