func (s *service) LinkVolumeToVolume(ctx context.Context, symID string, vol *types.Volume, tgtDevID, snapID string, reqID string, pmaxClient pmax.Pmax) error {
	log.Debugf("Creating snapshot %s on %s and linking it to %s", snapID, vol.VolumeID, tgtDevID)
	var TTL int64 = 1
	snapInfo, err := s.CreateSnapshotFromVolume(ctx, symID, vol, snapID, TTL, reqID, pmaxClient)
	if err != nil {
		if strings.Contains(err.Error(), "The maximum number of sessions has been exceeded for the specified Source device") {
			return status.Errorf(codes.FailedPrecondition, "Failed to create snapshot: %s", err.Error())
		}
		return err
	}
	err = s.LinkVolumeToSnapshot(ctx, symID, vol.VolumeID, tgtDevID, snapID, reqID, pmaxClient)
	if err != nil {
		return err
	}
	var cleanReq snapCleanupRequest
	cleanReq.snapshotID = snapInfo.SnapshotName
	cleanReq.symmetrixID = symID
	cleanReq.volumeID = vol.VolumeID
	cleanReq.requestID = reqID
	snapCleaner.requestCleanup(&cleanReq)
	return nil
}
LinkVolumeToVolume attaches the newly created volume
to a temporary snapshot created from the source volume
package service
import (
	"container/heap"
	"context"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"
	pmax "github.com/dell/gopowermax"
	types "github.com/dell/gopowermax/types/v90"
	log "github.com/sirupsen/logrus"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)
const (
	TempSnap       = "CSI_TEMP_SNAP"
	Defined        = "Defined"
	Link           = "Link"
	Unlink         = "Unlink"
	Rename         = "Rename"
	MaxUnlinkCount = 5
)
var cleanupStarted = false
var licenseCached = false
var symmRepCapabilities *types.SymReplicationCapabilities
var mutex sync.Mutex
var snapCleaner *snapCleanupWorker
type SnapSession struct {
	Source     string
	Name       string
	Generation int64
	Expired    bool
	Target     []types.SnapTarget
}
type snapCleanupWorker struct {
	PollingInterval time.Duration
	Mutex           sync.Mutex
	Queue           snapCleanupQueue
	MaxRetries      int
}
type snapCleanupRequest struct {
	symmetrixID string
	snapshotID  string
	volumeID    string
	requestID   string
	retries     int
}
type snapCleanupQueue []*snapCleanupRequest
func (q snapCleanupQueue) Len() int {
	return len(q)
}
func (q snapCleanupQueue) Less(i, j int) bool {
	return false
}
func (q snapCleanupQueue) Swap(i, j int) {
	q[i], q[j] = q[j], q[i]
}
func (q *snapCleanupQueue) Push(x interface{}) {
	req := x.(*snapCleanupRequest)
	n := len(*q)
	*q = append(*q, req)
	heap.Fix(q, n)
}
func (q *snapCleanupQueue) Pop() interface{} {
	old := *q
	n := len(old)
	req := old[n-1]
	*q = old[0 : n-1]
	return *req
}
func (req *snapCleanupRequest) fields() map[string]interface{} {
	fields := map[string]interface{}{
		"SymmetrixID": req.symmetrixID,
		"SnapshotID":  req.snapshotID,
		"VolumeID":    req.volumeID,
		"RequestID":   req.requestID,
	}
	return fields
}
func (scw *snapCleanupWorker) getQueueLen() int {
	scw.Mutex.Lock()
	defer scw.Mutex.Unlock()
	return len(scw.Queue)
}
func (scw *snapCleanupWorker) requestCleanup(req *snapCleanupRequest) {
	fields := req.fields()
	scw.Mutex.Lock()
	defer scw.Mutex.Unlock()
	for i := range scw.Queue {
		if scw.Queue[i].snapshotID == req.snapshotID && scw.Queue[i].symmetrixID == req.symmetrixID {
			log.Warningf("Snapshot ID: %s already present in the deletion queue", req.snapshotID)
			return
		}
	}
	heap.Push(&scw.Queue, req)
	log.WithFields(fields).Debug("Queued for Deletion")
}
func (scw *snapCleanupWorker) queueForRetry(req *snapCleanupRequest) {
	scw.Mutex.Lock()
	defer scw.Mutex.Unlock()
	heap.Push(&scw.Queue, req)
}
func (scw *snapCleanupWorker) removeItem() *snapCleanupRequest {
	scw.Mutex.Lock()
	defer scw.Mutex.Unlock()
	if len(scw.Queue) == 0 {
		return nil
	}
	reqx := heap.Pop(&scw.Queue)
	req := reqx.(snapCleanupRequest)
	return &req
}
func (s *service) UnlinkTargets(ctx context.Context, symID, srcDevID string, pmaxClient pmax.Pmax) error {
	SrcSession, _, err := s.GetSnapSessions(ctx, symID, srcDevID, pmaxClient)
	if err != nil {
		return err
	}
	if SrcSession != nil {
		err := s.UnlinkSnapshot(ctx, symID, &SrcSession[0], MaxUnlinkCount, pmaxClient)
		if err != nil {
			log.Error("UnlinkSnapshot failed for target session:" + srcDevID)
			return err
		}
	}
	return nil
}
func (s *service) IsSnapshotLicensed(ctx context.Context, symID string, pmaxClient pmax.Pmax) (err error) {
	if _, err := pmaxClient.IsAllowedArray(symID); err != nil {
		return err
	}
	mutex.Lock()
	defer mutex.Unlock()
	if licenseCached == false {
		symmRepCapabilities, err = pmaxClient.GetReplicationCapabilities(ctx)
		if err != nil {
			return err
		}
		licenseCached = true
		log.Infof("License information with Powermax %s is cached", symID)
	}
	for i := range symmRepCapabilities.SymmetrixCapability {
		if symmRepCapabilities.SymmetrixCapability[i].SymmetrixID == symID {
			if symmRepCapabilities.SymmetrixCapability[i].SnapVxCapable {
				return nil
			}
			return fmt.Errorf("PowerMax array (%s) doesn't have Snapshot license", symID)
		}
	}
	return fmt.Errorf("PowerMax array (%s) is not being managed by Unisphere", symID)
}
func (s *service) UnlinkAndTerminate(ctx context.Context, symID, deviceID, snapID string, pmaxClient pmax.Pmax) error {
	var noOfSnapsOnSrc int
	SrcSessions, TgtSession, err := s.GetSnapSessions(ctx, symID, deviceID, pmaxClient)
	if err != nil {
		return err
	}
	noOfSnapsOnSrc = len(SrcSessions)
	if TgtSession == nil && noOfSnapsOnSrc == 0 {
		log.Debugf("Couldn't find any snapshot on %s", deviceID)
		return fmt.Errorf("Couldn't find any source or target session for %s", deviceID)
	}
	if TgtSession != nil {
		err = s.UnlinkSnapshot(ctx, symID, TgtSession, 0, pmaxClient)
		if err != nil {
			log.Error("UnlinkSnapshot failed for target session" + TgtSession.Source)
			return err
		}
	}
	log.Debugf("Source sesion length (%d) : Snapshot name (%s)", noOfSnapsOnSrc, snapID)
	if noOfSnapsOnSrc > 0 {
		sort.Slice(SrcSessions[:], func(i, j int) bool {
			return SrcSessions[i].Generation > SrcSessions[j].Generation
		})
		for i := range SrcSessions {
			if SrcSessions[i].Name == snapID || snapID == "" {
				err = s.UnlinkSnapshot(ctx, symID, &SrcSessions[i], 0, pmaxClient)
				if err != nil {
					log.Error("UnlinkSnapshot failed for source session: " + SrcSessions[i].Source)
					return err
				}
				if SrcSessions[i].Expired {
					continue
				}
				err = s.TerminateSnapshot(ctx, symID, SrcSessions[i].Source, SrcSessions[i].Name, pmaxClient)
				if err != nil {
					log.Error("Failed to terminate snapshot (%s)" + SrcSessions[i].Name)
					return fmt.Errorf("Failed to terminate snapshot - Error(%s)", err.Error())
				}
				noOfSnapsOnSrc--
			}
		}
	}
	if noOfSnapsOnSrc == 0 {
		var vol *types.Volume
		vol, err = pmaxClient.GetVolumeByID(ctx, symID, deviceID)
		if err != nil {
			log.Errorf("Failed to find source snapshot volume. Error (%s) ", err.Error())
			return nil
		}
		if s.isSourceTaggedToDelete(vol.VolumeIdentifier) {
			err = s.MarkVolumeForDeletion(ctx, symID, vol, pmaxClient)
			if err != nil {
				log.Error("MarkVolumeForDeletion failed with error - ", err.Error())
			}
		}
	}
	return nil
}
func (s *service) UnlinkSnapshot(ctx context.Context, symID string, snapSession *SnapSession, maxUnlinkCount int, pmaxClient pmax.Pmax) (err error) {
	if snapSession.Target == nil {
		return
	}
	var counter int
	for _, target := range snapSession.Target {
		if target.Defined {
			TargetList := []types.VolumeList{{Name: target.Target}}
			SourceList := []types.VolumeList{{Name: snapSession.Source}}
			log.Debugf("Executing Unlink on (%s) with source (%v) target (%v)", snapSession.Name, SourceList, TargetList)
			err = pmaxClient.ModifySnapshotS(ctx, symID, SourceList, TargetList, snapSession.Name, Unlink, "", snapSession.Generation)
			if err != nil {
				if strings.Contains(err.Error(), "The Device(s) is (are) already in the desired state or mode") {
					log.Debugf("Unlink on (%s) with source (%v) target (%v) is already done", snapSession.Name, SourceList, TargetList)
					return nil
				}
				return err
			}
			if maxUnlinkCount != 0 {
				counter++
				if counter == maxUnlinkCount {
					log.Debugf("Max Unlink count reached")
					break
				}
			}
		} else {
			return fmt.Errorf("Not all the targets are in Defined state")
		}
	}
	return nil
}
func (s *service) TerminateSnapshot(ctx context.Context, symID string, srcDev string, snapID string, pmaxClient pmax.Pmax) (err error) {
	snap, err := pmaxClient.GetSnapshotInfo(ctx, symID, srcDev, snapID)
	if err != nil || snap.VolumeSnapshotSource == nil {
		log.Info("Snapshot is already deleted: " + snapID)
		return nil
	}
	err = s.RemoveSnapshot(ctx, symID, srcDev, snapID, 0, pmaxClient)
	if err != nil {
		return err
	}
	return nil
}
func (s *service) RemoveSnapshot(ctx context.Context, symID string, srcDev string, snapID string, Generation int64, pmaxClient pmax.Pmax) (err error) {
	log.Info(fmt.Sprintf("Deleting snapshot (%s) with generation (%d)", snapID, Generation))
	sourceVolumes := []types.VolumeList{}
	sourceVolumes = append(sourceVolumes, types.VolumeList{Name: srcDev})
	err = pmaxClient.DeleteSnapshotS(ctx, symID, snapID, sourceVolumes, Generation)
	if err != nil {
		return fmt.Errorf("DeleteSnapshot failed with error (%s)", err.Error())
	}
	return nil
}
func (s *service) IsVolumeInSnapSession(ctx context.Context, symID, deviceID string, pmaxClient pmax.Pmax) (source, target bool, err error) {
	vol, err := pmaxClient.GetVolumeByID(ctx, symID, deviceID)
	if err != nil {
		return false, false, err
	}
	return vol.SnapSource, vol.SnapTarget, nil
}
func (s *service) GetSnapSessions(ctx context.Context, symID, deviceID string, pmaxClient pmax.Pmax) (srcSession []SnapSession, tgtSession *SnapSession, err error) {
	snapInfo, err := pmaxClient.GetVolumeSnapInfo(ctx, symID, deviceID)
	if err != nil {
		log.Errorf("GetVolumeSnapInfo failed for (%s): (%s)\n", deviceID, err.Error())
		return
	}
	log.Debugf("For Volume (%s), Snap Info: %v\n", deviceID, snapInfo)
	for _, volumeSnapshotSource := range snapInfo.VolumeSnapshotSource {
		snapSession := SnapSession{
			Source:     deviceID,
			Generation: volumeSnapshotSource.Generation,
			Name:       volumeSnapshotSource.SnapshotName,
			Expired:    volumeSnapshotSource.Expired,
		}
		for _, targets := range volumeSnapshotSource.LinkedVolumes {
			snapTgt := types.SnapTarget{
				Target:  targets.TargetDevice,
				CpMode:  targets.Copy,
				Defined: targets.Defined}
			snapSession.Target = append(snapSession.Target, snapTgt)
		}
		srcSession = append(srcSession, snapSession)
	}
	if snapInfo.VolumeSnapshotLink != nil &&
		len(snapInfo.VolumeSnapshotLink) > 0 {
		var pVolInfo *types.VolumeResultPrivate
		pVolInfo, err = pmaxClient.GetPrivVolumeByID(ctx, symID, deviceID)
		if err != nil {
			log.Errorf("GetPrivVolumeByID failed for (%s): (%s)\n", deviceID, err.Error())
			return
		}
		log.Debugf("For Volume (%s), Priv Vol Info: %v\n", deviceID, pVolInfo)
		if &pVolInfo.TimeFinderInfo != nil &&
			pVolInfo.TimeFinderInfo.SnapVXTgt {
			snapSession := pVolInfo.TimeFinderInfo.SnapVXSession[0].TargetSourceSnapshotGenInfo
			if snapSession != nil {
				tgtSession = &SnapSession{
					Name:       snapSession.SnapshotName,
					Expired:    snapSession.Expired,
					Generation: snapSession.Generation,
					Source:     snapSession.SourceDevice,
					Target: []types.SnapTarget{{
						Target:  snapSession.TargetDevice,
						CpMode:  snapInfo.VolumeSnapshotLink[0].Copy,
						Defined: snapInfo.VolumeSnapshotLink[0].Defined}},
				}
			}
		}
	}
	return
}
func (s *service) LinkVolumeToSnapshot(ctx context.Context, symID, srcDevID, tgtDevID, snapID string, reqID string, pmaxClient pmax.Pmax) (err error) {
	lockHandle := fmt.Sprintf("%s%s", srcDevID, symID)
	lockNum := RequestLock(lockHandle, reqID)
	defer ReleaseLock(lockHandle, reqID, lockNum)
	_, err = pmaxClient.GetSnapshotInfo(ctx, symID, srcDevID, snapID)
	if err != nil {
		return err
	}
	sourceList := []types.VolumeList{}
	targetList := []types.VolumeList{}
	sourceList = append(sourceList, types.VolumeList{Name: srcDevID})
	targetList = append(targetList, types.VolumeList{Name: tgtDevID})
	err = pmaxClient.ModifySnapshotS(ctx, symID, sourceList, targetList, snapID, Link, "", 0)
	if err != nil {
		if strings.Contains(err.Error(), "The maximum number of sessions has been exceeded for the specified Source device") {
			return status.Errorf(codes.FailedPrecondition, "Failed to link volumes: %s", err.Error())
		}
		return err
	}
	return nil
}