func (c *Client) GenJobFilterTag(sourceTags, targetTags []string, sourceURL, targetURL *utils.RepoURL, sourceSecurity, targetSecurity configs.Security, wg *sync.WaitGroup) {
	tagChan := make(chan string, len(sourceTags))
	wg.Add(1)
	go func() {
		defer wg.Done()
		for _, tag := range sourceTags {
			tagChan <- tag
		}
		close(tagChan)
	}()
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for tag := range tagChan {
				urlPair := &URLPair{
					source: sourceURL.GetURL() + ":" + tag,
					target: targetURL.GetURL() + ":" + tag,
				}
				log.Debugf("handle tag %s", urlPair.source)
				if utils.IsContain(targetTags, tag) {
					if !c.config.FlagConf.Config.TagExistOverridden {
						log.Warnf("Skip push image, target image %s/%s:%s already exist, flag \"--tag-exist-overridden\" is set so skip", targetURL.GetRegistry(), targetURL.GetRepoWithNamespace(), tag)
						continue
					}
					imageSource, err := transfer.NewImageSource(sourceURL.GetRegistry(), sourceURL.GetRepoWithNamespace(),
						tag, sourceSecurity.Username, sourceSecurity.Password, sourceSecurity.Insecure)
					if err != nil {
						log.Errorf("generate %s image source error: %v", sourceURL.GetURL(), err)
						c.PutAFailedGenNormalURLPair(urlPair)
						continue
					}
					imageTarget, err := transfer.NewImageTarget(targetURL.GetRegistry(), targetURL.GetRepoWithNamespace(), tag, targetSecurity.Username, targetSecurity.Password, targetSecurity.Insecure)
					if err != nil {
						log.Errorf("generate %s image target error: %v", targetURL.GetURL(), err)
						c.PutAFailedGenNormalURLPair(urlPair)
						continue
					}
					sourceDigest, err := imageSource.GetImageDigest()
					if err != nil {
						log.Errorf("Failed to get source image digest from %s/%s:%s error: %v", imageSource.GetRegistry(), imageSource.GetRepository(), tag, err)
						c.PutAFailedGenNormalURLPair(urlPair)
						continue
					}
					targetDigest, err := imageTarget.GetImageDigest()
					if err != nil {
						log.Errorf("Failed to get target image digest from %s/%s:%s error: %v", imageTarget.GetRegistry(), imageTarget.GetRepository(), tag, err)
						c.PutAFailedGenNormalURLPair(urlPair)
						continue
					}
					if sourceDigest == targetDigest {
						log.Infof("Skip push image, target image %s/%s:%s already exist and has same digest %s", imageTarget.GetRegistry(), imageTarget.GetRepository(), imageTarget.GetTag(), sourceDigest)
						continue
					}
					if targetDigest != "" {
						log.Warnf("Target image %s/%s:%s already exist, target digest %s to be override as source digest %s", imageTarget.GetRegistry(), imageTarget.GetRepository(), imageTarget.GetTag(), targetDigest, sourceDigest)
					}
				}
				log.Infof("put normal url pair %v", urlPair)
				c.PutNormalURLPair(urlPair)
			}
		}()
	}
}
GenJobFilterTag is hornor by TagExistOverridden policy, skip generate job if tag in target and tag digest is same
package imagetransfer
import (
	"container/list"
	"fmt"
	"strings"
	"sync"
	"time"
	"tkestack.io/image-transfer/configs"
	"tkestack.io/image-transfer/pkg/apis/ccrapis"
	"tkestack.io/image-transfer/pkg/apis/tcrapis"
	"tkestack.io/image-transfer/pkg/image-transfer/options"
	"tkestack.io/image-transfer/pkg/log"
	"tkestack.io/image-transfer/pkg/transfer"
	"tkestack.io/image-transfer/pkg/utils"
)
type Client struct {
	jobList *list.List
	urlPairList *list.List
	normalURLPairList *list.List
	failedJobList              *list.List
	failedJobGenerateList      *list.List
	failedGenNormalURLPairList *list.List
	config *configs.Configs
	urlPairFinished bool
	jobListMutex                    sync.Mutex
	urlPairListMutex                sync.Mutex
	normalURLPairListMutex          sync.Mutex
	failedJobListMutex              sync.Mutex
	failedJobGenerateListMutex      sync.Mutex
	failedGenNormalURLPairListMutex sync.Mutex
	urlPairFinishedMutex            sync.Mutex
}
type URLPair struct {
	source string
	target string
}
func (c *Client) Run() error {
	if c.config.FlagConf.Config.CCRToTCR {
		return c.CCRToTCRTransfer()
	}
	return c.NormalTransfer(c.config.ImageList, nil, nil, nil)
}
func (c *Client) CCRToTCRTransfer() error {
	ccrClient := ccrapis.NewCCRAPIClient()
	ccrNs, err := ccrClient.GetAllNamespaceByName(c.config.Secret, c.config.FlagConf.Config.CCRRegion)
	log.Debugf("ccr namespaces is %s", ccrNs)
	if err != nil {
		log.Errorf("Get ccr ns returned error: %s", err)
		return err
	}
	tcrClient := tcrapis.NewTCRAPIClient()
	tcrNs, tcrID, err := tcrClient.GetAllNamespaceByName(c.config.Secret,
		c.config.FlagConf.Config.TCRRegion, c.config.FlagConf.Config.TCRName)
	log.Debugf("tcr namespaces is %s", tcrNs)
	if err != nil {
		log.Errorf("Get tcr ns returned error: %s", err)
		return err
	}
	failedNsList, err := c.CreateTcrNs(tcrClient, ccrNs, tcrNs, c.config.Secret, c.config.FlagConf.Config.TCRRegion, tcrID)
	if err != nil {
		log.Errorf("CreateTcrNs error: %s", err)
		return err
	}
	if len(failedNsList) != 0 {
		log.Infof("some ccr namespace create failed in tcr, retry Create Tcr Ns.")
		for times := 0; times < c.config.FlagConf.Config.RetryNums && len(failedNsList) != 0; times++ {
			tmpFailedNsList, err := c.RetryCreateTcrNs(tcrClient, failedNsList,
				c.config.Secret, c.config.FlagConf.Config.TCRRegion)
			if err != nil {
				continue
			} else {
				failedNsList = tmpFailedNsList
			}
		}
	}
	if len(failedNsList) != 0 {
		log.Warnf("some ccr namespace create failed in tcr: %s", failedNsList)
	}
	repoChan, err := c.GenerateCcrToTcrRules(failedNsList, ccrClient, c.config.Secret, c.config.FlagConf.Config.CCRRegion,
		c.config.FlagConf.Config.TCRRegion, c.config.FlagConf.Config.TCRName)
	if err != nil {
		return err
	}
	return c.NormalTransfer(nil, ccrClient, tcrClient, repoChan)
}
func (c *Client) GenerateCcrToTcrRules(failedNsList []string, ccrClient *ccrapis.CCRAPIClient,
	secret map[string]configs.Secret, ccrRegion string, tcrRegion string, tcrName string) (chan string, error) {
	secretID, secretKey, err := ccrapis.GetCcrSecret(secret)
	if err != nil {
		log.Errorf("GetCcrSecret error: %s", err)
		return nil, err
	}
	resp, err := ccrClient.DescribeRepositoryOwnerPersonal(secretID, secretKey, ccrRegion, 0, 1)
	if err != nil {
		log.Errorf("get ccr repo count error, %s", err)
		return nil, fmt.Errorf("get ccr repo count error %s", err)
	}
	totalRepo := *resp.Response.Data.TotalCount
	log.Debugf("total repo is %d", totalRepo)
	repoChan := make(chan string, totalRepo)
	err = ccrClient.GetAllCcrRepo(secret, ccrRegion, failedNsList, tcrRegion, tcrName, repoChan, totalRepo)
	if err != nil {
		log.Errorf("get ccr repo to tcr rules failed: %s", err)
		return nil, err
	}
	return repoChan, nil
}
func (c *Client) RetryCreateTcrNs(tcrClient *tcrapis.TCRAPIClient, retryList []string,
	secret map[string]configs.Secret, region string) ([]string, error) {
	var failedList []string
	secretID, secretKey, err := tcrapis.GetTcrSecret(secret)
	if err != nil {
		log.Errorf("GetTcrSecret error: %s", err)
		return failedList, err
	}
	tcrNs, tcrID, err := tcrClient.GetAllNamespaceByName(c.config.Secret,
		c.config.FlagConf.Config.TCRRegion, c.config.FlagConf.Config.TCRName)
	log.Debugf("tcr namespaces is %s", tcrNs)
	if err != nil {
		log.Errorf("retry create tcr ns, get tcr ns error: ", err)
		return nil, err
	}
	for _, ns := range retryList {
		if !utils.IsContain(tcrNs, ns) {
			_, err := tcrClient.CreateNamespace(secretID, secretKey, region, tcrID, ns)
			if err != nil {
				log.Errorf("tcr CreateNamespace %s error: %s", ns, err)
				failedList = append(failedList, ns)
			}
		}
	}
	return failedList, nil
}
func (c *Client) CreateTcrNs(tcrClient *tcrapis.TCRAPIClient, ccrNs, tcrNs []string,
	secret map[string]configs.Secret, region string, tcrID string) ([]string, error) {
	var failedList []string
	secretID, secretKey, err := tcrapis.GetTcrSecret(secret)
	if err != nil {
		log.Errorf("GetTcrSecret error: %s", err)
		return failedList, err
	}
	for _, ns := range ccrNs {
		if !utils.IsContain(tcrNs, ns) {
			log.Infof("create namespace %s", ns)
			_, err := tcrClient.CreateNamespace(secretID, secretKey, region, tcrID, ns)
			if err != nil {
				log.Errorf("tcr CreateNamespace %s error:  %s", ns, err)
				failedList = append(failedList, ns)
			}
		}
	}
	return failedList, nil
}
func (c *Client) NormalTransfer(imageList map[string]string, ccrClient *ccrapis.CCRAPIClient, tcrClient *tcrapis.TCRAPIClient, repoChan chan string) error {
	jobListChan := make(chan *transfer.Job, c.config.FlagConf.Config.RoutineNums)
	fmt.Println("Start to handle transfer jobs, please wait ...")
	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		c.jobsHandler(jobListChan)
	}()
	if c.config.FlagConf.Config.CCRToTCR {
		wg.Add(1)
		go func() {
			defer wg.Done()
			c.HandleCcrToTCrTags(repoChan)
			c.SetURLPairFinished()
		}()
	} else {
		for source, target := range imageList {
			c.urlPairList.PushBack(&URLPair{
				source: source,
				target: target,
			})
		}
		wg.Add(1)
		go func() {
			defer wg.Done()
			c.HandleURLPair()
			c.SetURLPairFinished()
		}()
	}
	c.rulesHandler(jobListChan)
	wg.Wait()
	log.Infof("Start to retry failed jobs...")
	for times := 0; times < c.config.FlagConf.Config.RetryNums; times++ {
		log.Debugf("failedjobList len %d, failedGenNormalURLPairList len %d, failedJobGenerateList len %d", c.failedJobList.Len(), c.failedGenNormalURLPairList.Len(), c.failedJobGenerateList.Len())
		c.Retry()
	}
	if c.failedJobList.Len() != 0 {
		log.Infof("################# %v failed transfer jobs: #################", c.failedJobList.Len())
		for e := c.failedJobList.Front(); e != nil; e = e.Next() {
			log.Infof(e.Value.(*transfer.Job).Source.GetRegistry() + "/" +
				e.Value.(*transfer.Job).Source.GetRepository() + ":" + e.Value.(*transfer.Job).Source.GetTag())
		}
	}
	if c.failedGenNormalURLPairList.Len() != 0 {
		log.Infof("################# %v failed generate Normal urlPair: #################", c.failedGenNormalURLPairList.Len())
		for e := c.failedGenNormalURLPairList.Front(); e != nil; e = e.Next() {
			log.Infof(e.Value.(*URLPair).source + ": " + e.Value.(*URLPair).target)
		}
	}
	if c.failedJobGenerateList.Len() != 0 {
		log.Infof("################# %v failed generate jobs: #################", c.failedJobGenerateList.Len())
		for e := c.failedJobGenerateList.Front(); e != nil; e = e.Next() {
			log.Infof(e.Value.(*URLPair).source + ": " + e.Value.(*URLPair).target)
		}
	}
	log.Infof("################# Finished, %v transfer jobs failed, %v normal urlPair generate failed, %v jobs generate failed #################",
		c.failedJobList.Len(), c.failedGenNormalURLPairList.Len(), c.failedJobGenerateList.Len())
	return nil
}
func (c *Client) Retry() {
	retryJobListChan := make(chan *transfer.Job, c.config.FlagConf.Config.RoutineNums)
	wg1 := sync.WaitGroup{}
	wg1.Add(1)
	go func() {
		defer func() {
			wg1.Done()
		}()
		c.jobsHandler(retryJobListChan)
	}()
	if c.failedJobList.Len() != 0 {
		failedJobListChan := make(chan *transfer.Job, c.failedJobList.Len())
		for {
			failedJob := c.failedJobList.Front()
			if failedJob == nil {
				break
			}
			log.Infof("put failed job to failedJobListChan %s/%s:%s %s/%s:%s", failedJob.Value.(*transfer.Job).Source.GetRegistry(), failedJob.Value.(*transfer.Job).Source.GetRepository(), failedJob.Value.(*transfer.Job).Source.GetTag(), failedJob.Value.(*transfer.Job).Target.GetRegistry(), failedJob.Value.(*transfer.Job).Target.GetRepository(), failedJob.Value.(*transfer.Job).Target.GetTag())
			failedJobListChan <- failedJob.Value.(*transfer.Job)
			c.failedJobList.Remove(failedJob)
		}
		close(failedJobListChan)
		wg1.Add(1)
		go func() {
			defer wg1.Done()
			c.jobsHandler(failedJobListChan)
		}()
	}
	if c.failedGenNormalURLPairList.Len() != 0 || c.failedJobGenerateList.Len() != 0 {
		if c.failedGenNormalURLPairList.Len() != 0 {
			c.urlPairList.PushBackList(c.failedGenNormalURLPairList)
			c.failedGenNormalURLPairList.Init()
			if !c.config.FlagConf.Config.CCRToTCR {
				c.HandleURLPair()
			} else {
				c.CcrtoTcrGenTagRetry()
			}
		}
		if c.failedJobGenerateList.Len() != 0 {
			c.normalURLPairList.PushBackList(c.failedJobGenerateList)
			c.failedJobGenerateList.Init()
		}
		c.rulesHandler(retryJobListChan)
	} else {
		close(retryJobListChan)
	}
	wg1.Wait()
}
func NewTransferClient(opts *options.ClientOptions) (*Client, error) {
	clientConfig, err := configs.InitConfigs(opts)
	if err != nil {
		return nil, err
	}
	return &Client{
		jobList:                         list.New(),
		urlPairList:                     list.New(),
		failedJobList:                   list.New(),
		failedJobGenerateList:           list.New(),
		normalURLPairList:               list.New(),
		failedGenNormalURLPairList:      list.New(),
		config:                          clientConfig,
		jobListMutex:                    sync.Mutex{},
		urlPairListMutex:                sync.Mutex{},
		failedJobListMutex:              sync.Mutex{},
		failedJobGenerateListMutex:      sync.Mutex{},
		urlPairFinishedMutex:            sync.Mutex{},
		failedGenNormalURLPairListMutex: sync.Mutex{},
	}, nil
}
func (c *Client) rulesHandler(jobListChan chan *transfer.Job) {
	defer func() {
		close(jobListChan)
	}()
	routineNum := c.config.FlagConf.Config.RoutineNums
	wg := sync.WaitGroup{}
	for i := 0; i < routineNum; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer log.Debugf("exit rule handler main loop")
			for {
				urlPair, empty := c.GetNormalURLPair()
				if empty && c.IsURLPairFinished() {
					log.Debugf("url pair is empty")
					break
				}
				if empty {
					log.Debugf("not finieshed but url pair is empty")
					time.Sleep(100 * time.Millisecond)
					continue
				}
				log.Infof("generate job source %s, target %s", urlPair.source, urlPair.target)
				err := c.GenerateTransferJob(jobListChan, urlPair.source, urlPair.target)
				if err != nil {
					log.Errorf("Generate transfer job %s to %s error: %v", urlPair.source, urlPair.target, err)
					c.PutAFailedURLPair(urlPair)
				}
			}
		}()
	}
	wg.Wait()
}
func (c *Client) jobsHandler(jobListChan chan *transfer.Job) {
	routineNum := c.config.FlagConf.Config.RoutineNums
	wg := sync.WaitGroup{}
	for i := 0; i < routineNum; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				job, ok := <-jobListChan
				if !ok {
					break
				}
				if err := job.Run(); err != nil {
					log.Errorf("handle job failed %s/%s:%s, %s", job.Source.GetRegistry(), job.Source.GetRepository(), job.Source.GetTag(), err)
					c.PutAFailedJob(job)
				}
			}
		}()
	}
	wg.Wait()
}
func (c *Client) GetURLPair() (*URLPair, bool) {
	c.urlPairListMutex.Lock()
	defer func() {
		c.urlPairListMutex.Unlock()
	}()
	urlPair := c.urlPairList.Front()
	if urlPair == nil {
		return nil, true
	}
	c.urlPairList.Remove(urlPair)
	return urlPair.Value.(*URLPair), false
}
func (c *Client) PutURLPair(urlPair *URLPair) {
	c.urlPairListMutex.Lock()
	defer c.urlPairListMutex.Unlock()
	c.urlPairList.PushBack(urlPair)
}
func (c *Client) GetNormalURLPair() (*URLPair, bool) {
	c.normalURLPairListMutex.Lock()
	defer func() {
		c.normalURLPairListMutex.Unlock()
	}()
	urlPair := c.normalURLPairList.Front()
	if urlPair == nil {
		return nil, true
	}
	c.normalURLPairList.Remove(urlPair)
	return urlPair.Value.(*URLPair), false
}
func (c *Client) PutNormalURLPair(urlPair *URLPair) {
	c.normalURLPairListMutex.Lock()
	defer c.normalURLPairListMutex.Unlock()
	c.normalURLPairList.PushBack(urlPair)
}
func (c *Client) GetJob() (*transfer.Job, bool) {
	c.jobListMutex.Lock()
	defer func() {
		c.jobListMutex.Unlock()
	}()
	job := c.jobList.Front()
	if job == nil {
		return nil, true
	}
	c.jobList.Remove(job)
	return job.Value.(*transfer.Job), false
}
func (c *Client) PutJob(job *transfer.Job) {
	c.jobListMutex.Lock()
	defer func() {
		c.jobListMutex.Unlock()
	}()
	if c.jobList != nil {
		c.jobList.PushBack(job)
	}
}
func (c *Client) GenerateTransferJob(jobListChan chan *transfer.Job, source string, target string) error {
	if source == "" {
		return fmt.Errorf("source url should not be empty")
	}
	sourceURL, err := utils.NewRepoURL(source)
	if err != nil {
		return fmt.Errorf("url %s format error: %v", source, err)
	}
	if target == "" {
		return fmt.Errorf("target url should not be empty")
	}
	targetURL, err := utils.NewRepoURL(target)
	if err != nil {
		return fmt.Errorf("url %s format error: %v", target, err)
	}
	if sourceURL.GetTag() == "" {
		return fmt.Errorf("source tag empty, source: %s", sourceURL.GetURL())
	}
	if targetURL.GetTag() == "" {
		return fmt.Errorf("target tag empty, target: %s", targetURL.GetURL())
	}
	var imageSource *transfer.ImageSource
	var imageTarget *transfer.ImageTarget
	sourceSecurity, exist := c.config.GetSecuritySpecific(sourceURL.GetRegistry(), sourceURL.GetNamespace())
	if exist {
		log.Infof("Find auth information for %v, username: %v", sourceURL.GetURL(), sourceSecurity.Username)
	} else {
		log.Infof("Cannot find auth information for %v, pull actions will be anonymous", sourceURL.GetURL())
	}
	targetSecurity, exist := c.config.GetSecuritySpecific(targetURL.GetRegistry(), targetURL.GetNamespace())
	if exist {
		log.Infof("Find auth information for %v, username: %v", targetURL.GetURL(), targetSecurity.Username)
	} else {
		log.Infof("Cannot find auth information for %v, push actions will be anonymous", targetURL.GetURL())
	}
	imageSource, err = transfer.NewImageSource(sourceURL.GetRegistry(), sourceURL.GetRepoWithNamespace(), sourceURL.GetTag(), sourceSecurity.Username, sourceSecurity.Password, sourceSecurity.Insecure)
	if err != nil {
		return fmt.Errorf("generate %s image source error: %v", sourceURL.GetURL(), err)
	}
	imageTarget, err = transfer.NewImageTarget(targetURL.GetRegistry(), targetURL.GetRepoWithNamespace(), targetURL.GetTag(), targetSecurity.Username, targetSecurity.Password, targetSecurity.Insecure)
	if err != nil {
		return fmt.Errorf("generate %s image target error: %v", sourceURL.GetURL(), err)
	}
	jobListChan <- transfer.NewJob(imageSource, imageTarget)
	log.Infof("Generate a job for %s to %s", sourceURL.GetURL(), targetURL.GetURL())
	return nil
}
func (c *Client) GetFailedJob() (*transfer.Job, bool) {
	c.failedJobListMutex.Lock()
	defer func() {
		c.failedJobListMutex.Unlock()
	}()
	failedJob := c.failedJobList.Front()
	if failedJob == nil {
		return nil, true
	}
	c.failedJobList.Remove(failedJob)
	return failedJob.Value.(*transfer.Job), false
}
func (c *Client) PutAFailedJob(failedJob *transfer.Job) {
	c.failedJobListMutex.Lock()
	defer func() {
		c.failedJobListMutex.Unlock()
	}()
	if c.failedJobList != nil {
		c.failedJobList.PushBack(failedJob)
	}
}
func (c *Client) GetAFailedURLPair() (*URLPair, bool) {
	c.failedJobGenerateListMutex.Lock()
	defer func() {
		c.failedJobGenerateListMutex.Unlock()
	}()
	failedURLPair := c.failedJobGenerateList.Front()
	if failedURLPair == nil {
		return nil, true
	}
	c.failedJobGenerateList.Remove(failedURLPair)
	return failedURLPair.Value.(*URLPair), false
}
func (c *Client) PutAFailedURLPair(failedURLPair *URLPair) {
	c.failedJobGenerateListMutex.Lock()
	defer func() {
		c.failedJobGenerateListMutex.Unlock()
	}()
	if c.failedJobGenerateList != nil {
		c.failedJobGenerateList.PushBack(failedURLPair)
	}
}
func (c *Client) GetAFailedGenNormalURLPair() (*URLPair, bool) {
	c.failedGenNormalURLPairListMutex.Lock()
	defer func() {
		c.failedGenNormalURLPairListMutex.Unlock()
	}()
	failedURLPair := c.failedGenNormalURLPairList.Front()
	if failedURLPair == nil {
		return nil, true
	}
	c.failedGenNormalURLPairList.Remove(failedURLPair)
	return failedURLPair.Value.(*URLPair), false
}
func (c *Client) PutAFailedGenNormalURLPair(failedURLPair *URLPair) {
	c.failedGenNormalURLPairListMutex.Lock()
	defer func() {
		c.failedGenNormalURLPairListMutex.Unlock()
	}()
	if c.failedGenNormalURLPairList != nil {
		c.failedGenNormalURLPairList.PushBack(failedURLPair)
	}
}