/* Copyright 2017 The Kubernetes Authors. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. */ package hostpath import ( "fmt" "math" "os" "sort" "strconv" "github.com/golang/protobuf/ptypes" "github.com/golang/glog" "github.com/pborman/uuid" "golang.org/x/net/context" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "github.com/container-storage-interface/spec/lib/go/csi" "github.com/kubernetes-csi/drivers/pkg/csi-common" utilexec "k8s.io/utils/exec" ) const ( deviceID = "deviceID" provisionRoot = "/tmp/" snapshotRoot = "/tmp/" maxStorageCapacity = tib ) type controllerServer struct { *csicommon.DefaultControllerServer } func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) { if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME); err != nil { glog.V(3).Infof("invalid create volume req: %v", req) return nil, err } // Check arguments if len(req.GetName()) == 0 { return nil, status.Error(codes.InvalidArgument, "Name missing in request") } if req.GetVolumeCapabilities() == nil { return nil, status.Error(codes.InvalidArgument, "Volume Capabilities missing in request") } // Need to check for already existing volume name, and if found // check for the requested capacity and already allocated capacity if exVol, err := getVolumeByName(req.GetName()); err == nil { // Since err is nil, it means the volume with the same name already exists // need to check if the size of exisiting volume is the same as in new // request if exVol.VolSize >= int64(req.GetCapacityRange().GetRequiredBytes()) { // exisiting volume is compatible with new request and should be reused. // TODO (sbezverk) Do I need to make sure that RBD volume still exists? return &csi.CreateVolumeResponse{ Volume: &csi.Volume{ VolumeId: exVol.VolID, CapacityBytes: int64(exVol.VolSize), VolumeContext: req.GetParameters(), }, }, nil } return nil, status.Error(codes.AlreadyExists, fmt.Sprintf("Volume with the same name: %s but with different size already exist", req.GetName())) } // Check for maximum available capacity capacity := int64(req.GetCapacityRange().GetRequiredBytes()) if capacity >= maxStorageCapacity { return nil, status.Errorf(codes.OutOfRange, "Requested capacity %d exceeds maximum allowed %d", capacity, maxStorageCapacity) } volumeID := uuid.NewUUID().String() path := provisionRoot + volumeID err := os.MkdirAll(path, 0777) if err != nil { glog.V(3).Infof("failed to create volume: %v", err) return nil, err } if req.GetVolumeContentSource() != nil { contentSource := req.GetVolumeContentSource() if contentSource.GetSnapshot() != nil { snapshotId := contentSource.GetSnapshot().GetSnapshotId() snapshot, ok := hostPathVolumeSnapshots[snapshotId] if !ok { return nil, status.Errorf(codes.NotFound, "cannot find snapshot %v", snapshotId) } if snapshot.ReadyToUse != true { return nil, status.Errorf(codes.Internal, "Snapshot %v is not yet ready to use.", snapshotId) } snapshotPath := snapshot.Path args := []string{"zxvf", snapshotPath, "-C", path} executor := utilexec.New() out, err := executor.Command("tar", args...).CombinedOutput() if err != nil { return nil, status.Error(codes.Internal, fmt.Sprintf("failed pre-populate data for volume: %v: %s", err, out)) } } } glog.V(4).Infof("create volume %s", path) hostPathVol := hostPathVolume{} hostPathVol.VolName = req.GetName() hostPathVol.VolID = volumeID hostPathVol.VolSize = capacity hostPathVol.VolPath = path hostPathVolumes[volumeID] = hostPathVol return &csi.CreateVolumeResponse{ Volume: &csi.Volume{ VolumeId: volumeID, CapacityBytes: req.GetCapacityRange().GetRequiredBytes(), VolumeContext: req.GetParameters(), }, }, nil } func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) { // Check arguments if len(req.GetVolumeId()) == 0 { return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request") } if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME); err != nil { glog.V(3).Infof("invalid delete volume req: %v", req) return nil, err } volumeID := req.VolumeId glog.V(4).Infof("deleting volume %s", volumeID) path := provisionRoot + volumeID os.RemoveAll(path) delete(hostPathVolumes, volumeID) return &csi.DeleteVolumeResponse{}, nil } func (cs *controllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) { return cs.DefaultControllerServer.ValidateVolumeCapabilities(ctx, req) } // 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 } if len(req.GetName()) == 0 { return nil, status.Error(codes.InvalidArgument, "Name missing in request") } // Check arguments if len(req.GetSourceVolumeId()) == 0 { return nil, status.Error(codes.InvalidArgument, "SourceVolumeId 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{ SnapshotId: exSnap.Id, SourceVolumeId: exSnap.VolID, CreationTime: &exSnap.CreationTime, SizeBytes: exSnap.SizeBytes, ReadyToUse: exSnap.ReadyToUse, }, }, 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() creationTime := ptypes.TimestampNow() 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.CreationTime = *creationTime snapshot.SizeBytes = hostPathVolume.VolSize snapshot.ReadyToUse = true hostPathVolumeSnapshots[snapshotID] = snapshot return &csi.CreateSnapshotResponse{ Snapshot: &csi.Snapshot{ SnapshotId: snapshot.Id, SourceVolumeId: snapshot.VolID, CreationTime: &snapshot.CreationTime, SizeBytes: snapshot.SizeBytes, ReadyToUse: snapshot.ReadyToUse, }, }, 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.GetSnapshotId() 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 } // 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 { return convertSnapshot(snapshot), nil } } // case 2: SourceVolumeId is not empty, return snapshots that match the source volume id. if len(req.GetSourceVolumeId()) != 0 { for _, snapshot := range hostPathVolumeSnapshots { if snapshot.VolID == req.SourceVolumeId { return convertSnapshot(snapshot), nil } } } var snapshots []csi.Snapshot // case 3: no parameter is set, so we return all the snapshots. sortedKeys := make([]string, 0) for k := range hostPathVolumeSnapshots { sortedKeys = append(sortedKeys, k) } sort.Strings(sortedKeys) for _, key := range sortedKeys { snap := hostPathVolumeSnapshots[key] snapshot := csi.Snapshot{ SnapshotId: snap.Id, SourceVolumeId: snap.VolID, CreationTime: &snap.CreationTime, SizeBytes: snap.SizeBytes, ReadyToUse: snap.ReadyToUse, } snapshots = append(snapshots, snapshot) } var ( ulenSnapshots = int32(len(snapshots)) maxEntries = req.MaxEntries startingToken int32 ) if v := req.StartingToken; v != "" { i, err := strconv.ParseUint(v, 10, 32) if err != nil { return nil, status.Errorf( codes.Aborted, "startingToken=%d !< int32=%d", startingToken, math.MaxUint32) } startingToken = int32(i) } if startingToken > ulenSnapshots { return nil, status.Errorf( codes.Aborted, "startingToken=%d > len(snapshots)=%d", startingToken, ulenSnapshots) } // Discern the number of remaining entries. rem := ulenSnapshots - startingToken // If maxEntries is 0 or greater than the number of remaining entries then // set maxEntries to the number of remaining entries. if maxEntries == 0 || maxEntries > rem { maxEntries = rem } var ( i int j = startingToken entries = make( []*csi.ListSnapshotsResponse_Entry, maxEntries) ) for i = 0; i < len(entries); i++ { entries[i] = &csi.ListSnapshotsResponse_Entry{ Snapshot: &snapshots[j], } j++ } var nextToken string if j < ulenSnapshots { nextToken = fmt.Sprintf("%d", j) } return &csi.ListSnapshotsResponse{ Entries: entries, NextToken: nextToken, }, nil } func convertSnapshot(snap hostPathSnapshot) *csi.ListSnapshotsResponse { entries := []*csi.ListSnapshotsResponse_Entry{ { Snapshot: &csi.Snapshot{ SnapshotId: snap.Id, SourceVolumeId: snap.VolID, CreationTime: &snap.CreationTime, SizeBytes: snap.SizeBytes, ReadyToUse: snap.ReadyToUse, }, }, } rsp := &csi.ListSnapshotsResponse{ Entries: entries, } return rsp }