hostpaht snapshot implemention

Author: wackxu

pkg/cinder/deploy/kubernetes/csi-attacher-cinderplugin.yaml

@@ -30,7 +30,7 @@ spec:
       serviceAccount: csi-attacher
       containers:
         - name: csi-attacher
-          image: quay.io/k8scsi/csi-attacher:v0.2.0
+          image: quay.io/k8scsi/csi-attacher:v0.3.0
           args:
             - "--v=5"
             - "--csi-address=$(ADDRESS)"

pkg/cinder/deploy/kubernetes/csi-nodeplugin-cinderplugin.yaml

@@ -18,7 +18,7 @@ spec:
       hostNetwork: true
       containers:
         - name: driver-registrar
-          image: quay.io/k8scsi/driver-registrar:v0.2.0
+          image: quay.io/k8scsi/driver-registrar:v0.3.0
           args:
             - "--v=5"
             - "--csi-address=$(ADDRESS)"

pkg/cinder/deploy/kubernetes/csi-provisioner-cinderplugin.yaml

@@ -30,7 +30,7 @@ spec:
       serviceAccount: csi-provisioner
       containers:
         - name: csi-provisioner
-          image: quay.io/k8scsi/csi-provisioner:v0.2.0
+          image: quay.io/k8scsi/csi-provisioner:v0.3.0
           args:
             - "--provisioner=csi-cinderplugin"
             - "--csi-address=$(ADDRESS)"

pkg/cinder/driver.go

@@ -42,7 +42,7 @@ const (
 )
 
 var (
-	version = "0.2.0"
+	version = "0.3.0"
 )
 
 func NewDriver(nodeID, endpoint string, cloudconfig string) *driver {

pkg/csi-common/controllerserver-default.go

@@ -85,3 +85,15 @@ func (cs *DefaultControllerServer) ControllerGetCapabilities(ctx context.Context
 		Capabilities: cs.Driver.cap,
 	}, nil
 }
+
+func (cs *DefaultControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (cs *DefaultControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}
+
+func (cs *DefaultControllerServer) ListSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}

pkg/csi-common/driver_test.go

@@ -31,7 +31,7 @@ const (
 )
 
 var (
-	vendorVersion = "0.2.0"
+	vendorVersion = "0.3.0"
 )
 
 func NewFakeDriver() *CSIDriver {

pkg/csi-common/nodeserver-default.go

@@ -59,3 +59,7 @@ func (ns *DefaultNodeServer) NodeGetCapabilities(ctx context.Context, req *csi.N
 		},
 	}, nil
 }
+
+func (ns *DefaultNodeServer) NodeGetInfo(context.Context, *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) {
+	return nil, status.Error(codes.Unimplemented, "")
+}

pkg/flexadapter/flexadapter.go

@@ -38,7 +38,7 @@ type flexAdapter struct {
 }
 
 var (
-	version = "0.2.0"
+	version = "0.3.0"
 )
 
 func New() *flexAdapter {

pkg/hostpath/controllerserver.go

@@ -19,6 +19,7 @@ package hostpath
 import (
 	"fmt"
 	"os"
+	"time"
 
 	"github.com/golang/glog"
 	"github.com/pborman/uuid"
@@ -28,11 +29,13 @@ import (
 
 	"github.com/container-storage-interface/spec/lib/go/csi/v0"
 	"github.com/kubernetes-csi/drivers/pkg/csi-common"
+	utilexec "k8s.io/utils/exec"
 )
 
 const (
 	deviceID           = "deviceID"
 	provisionRoot      = "/tmp/"
+	snapshotRoot       = "/tmp/"
 	maxStorageCapacity = tib
 )
 
@@ -136,3 +139,151 @@ func (cs *controllerServer) ValidateVolumeCapabilities(ctx context.Context, req
 	}
 	return &csi.ValidateVolumeCapabilitiesResponse{Supported: true, Message: ""}, nil
 }
+
+// CreateSnapshot uses tar command to create snapshot for hostpath volume. The tar command can quickly create
+// archives of entire directories. The host image must have "tar" binaries in /bin, /usr/sbin, or /usr/bin.
+func (cs *controllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateSnapshotRequest) (*csi.CreateSnapshotResponse, error) {
+	if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT); err != nil {
+		glog.V(3).Infof("invalid create snapshot req: %v", req)
+		return nil, err
+	}
+
+	// Check arguments
+	if len(req.GetSourceVolumeId()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "SourceVolumeId missing in request")
+	}
+	if len(req.GetName()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Name missing in request")
+	}
+
+	// Need to check for already existing snapshot name, and if found check for the
+	// requested sourceVolumeId and sourceVolumeId of snapshot that has been created.
+	if exSnap, err := getSnapshotByName(req.GetName()); err == nil {
+		// Since err is nil, it means the snapshot with the same name already exists need
+		// to check if the sourceVolumeId of existing snapshot is the same as in new request.
+		if exSnap.VolID == req.GetSourceVolumeId() {
+			// same snapshot has been created.
+			return &csi.CreateSnapshotResponse{
+				Snapshot: &csi.Snapshot{
+					Id:             exSnap.Id,
+					SourceVolumeId: exSnap.VolID,
+					CreatedAt:      exSnap.CreateAt,
+					Status:         exSnap.Status,
+				},
+			}, nil
+		}
+		return nil, status.Error(codes.AlreadyExists, fmt.Sprintf("snapshot with the same name: %s but with different SourceVolumeId already exist", req.GetName()))
+	}
+
+	volumeID := req.GetSourceVolumeId()
+	hostPathVolume, ok := hostPathVolumes[volumeID]
+	if !ok {
+		return nil, status.Error(codes.Internal, "volumeID is not exist")
+	}
+
+	snapshotID := uuid.NewUUID().String()
+	createAt := time.Now().UnixNano()
+	volPath := hostPathVolume.VolPath
+	file := snapshotRoot + snapshotID + ".tgz"
+	args := []string{"czf", file, "-C", volPath, "."}
+	executor := utilexec.New()
+	out, err := executor.Command("tar", args...).CombinedOutput()
+	if err != nil {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("failed create snapshot: %v: %s", err, out))
+	}
+
+	glog.V(4).Infof("create volume snapshot %s", file)
+	snapshot := hostPathSnapshot{}
+	snapshot.Name = req.GetName()
+	snapshot.Id = snapshotID
+	snapshot.VolID = volumeID
+	snapshot.Path = file
+	snapshot.CreateAt = createAt
+	snapshot.Status = &csi.SnapshotStatus{
+		Type:    csi.SnapshotStatus_READY,
+		Details: fmt.Sprint("Successfully create snapshot"),
+	}
+
+	hostPathVolumeSnapshots[snapshotID] = snapshot
+
+	return &csi.CreateSnapshotResponse{
+		Snapshot: &csi.Snapshot{
+			Id:             snapshot.Id,
+			SourceVolumeId: snapshot.VolID,
+			CreatedAt:      snapshot.CreateAt,
+			Status:         snapshot.Status,
+		},
+	}, nil
+}
+
+func (cs *controllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteSnapshotRequest) (*csi.DeleteSnapshotResponse, error) {
+	// Check arguments
+	if len(req.GetSnapshotId()) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "Snapshot ID missing in request")
+	}
+
+	if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT); err != nil {
+		glog.V(3).Infof("invalid delete snapshot req: %v", req)
+		return nil, err
+	}
+	snapshotID := req.SnapshotId
+	glog.V(4).Infof("deleting volume %s", snapshotID)
+	path := snapshotRoot + snapshotID + ".tgz"
+	os.RemoveAll(path)
+	delete(hostPathVolumeSnapshots, snapshotID)
+	return &csi.DeleteSnapshotResponse{}, nil
+}
+
+func (cs *controllerServer) ListSnapshots(ctx context.Context, req *csi.ListSnapshotsRequest) (*csi.ListSnapshotsResponse, error) {
+	if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS); err != nil {
+		glog.V(3).Infof("invalid list snapshot req: %v", req)
+		return nil, err
+	}
+
+	var snapshots []hostPathSnapshot
+
+	// case 1: SnapshotId is not empty, return snapshots that match the snapshot id.
+	if len(req.GetSnapshotId()) != 0 {
+		snapshotID := req.SnapshotId
+		if snapshot, ok := hostPathVolumeSnapshots[snapshotID]; ok {
+			snapshots = append(snapshots, snapshot)
+			// jump to the result processing.
+			goto result
+		}
+	}
+
+	// case 2: SourceVolumeId is not empty, return snapshots that match the source volume id.
+	if len(req.SourceVolumeId) != 0 {
+		for _, snapshot := range hostPathVolumeSnapshots {
+			if snapshot.VolID == req.SourceVolumeId {
+				snapshots = append(snapshots, snapshot)
+				// jump to the result processing.
+				goto result
+			}
+		}
+	}
+
+	// case 3: no parameter is set, so we return all the snapshots.
+	for _, snapshot := range hostPathVolumeSnapshots {
+		snapshots = append(snapshots, snapshot)
+	}
+
+result:
+	var entries []*csi.ListSnapshotsResponse_Entry
+	for _, snap := range snapshots {
+		entries = append(entries, &csi.ListSnapshotsResponse_Entry{
+			Snapshot: &csi.Snapshot{
+				Id:             snap.Id,
+				SourceVolumeId: snap.VolID,
+				CreatedAt:      snap.CreateAt,
+				Status:         snap.Status,
+			},
+		})
+	}
+
+	rsp := &csi.ListSnapshotsResponse{
+		Entries: entries,
+	}
+
+	return rsp, nil
+}

pkg/hostpath/hostpath.go

@@ -52,11 +52,21 @@ type hostPathVolume struct {
 	VolPath string `json:"volPath"`
 }
 
+type hostPathSnapshot struct {
+	Name     string              `json:"name"`
+	Id       string              `json:"id"`
+	VolID    string              `json:"volID"`
+	Path     string              `json:"path"`
+	CreateAt int64               `json:"createAt"`
+	Status   *csi.SnapshotStatus `json:"status"`
+}
+
 var hostPathVolumes map[string]hostPathVolume
+var hostPathVolumeSnapshots map[string]hostPathSnapshot
 
 var (
 	hostPathDriver *hostPath
-	vendorVersion  = "0.2.0"
+	vendorVersion  = "0.3.0"
 )
 
 func init() {
@@ -93,7 +103,12 @@ func (hp *hostPath) Run(driverName, nodeID, endpoint string) {
 	if hp.driver == nil {
 		glog.Fatalln("Failed to initialize CSI Driver.")
 	}
-	hp.driver.AddControllerServiceCapabilities([]csi.ControllerServiceCapability_RPC_Type{csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME})
+	hp.driver.AddControllerServiceCapabilities(
+		[]csi.ControllerServiceCapability_RPC_Type{
+			csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+			csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
+			csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
+		})
 	hp.driver.AddVolumeCapabilityAccessModes([]csi.VolumeCapability_AccessMode_Mode{csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER})
 
 	// Create GRPC servers
@@ -121,3 +136,12 @@ func getVolumeByName(volName string) (hostPathVolume, error) {
 	}
 	return hostPathVolume{}, fmt.Errorf("volume name %s does not exit in the volumes list", volName)
 }
+
+func getSnapshotByName(name string) (hostPathSnapshot, error) {
+	for _, snapshot := range hostPathVolumeSnapshots {
+		if snapshot.Name == name {
+			return snapshot, nil
+		}
+	}
+	return hostPathSnapshot{}, fmt.Errorf("snapshot name %s does not exit in the snapshots list", name)
+}