[fix][dingoscheduler] Add caching tasks concurrently

internal/dao/dao.go

@@ -17,4 +17,4 @@ package dao
 import "github.com/google/wire"
 
 var DaoProvider = wire.NewSet(NewDingospeedDao, NewModelFileRecordDao, NewModelFileProcessDao, NewCacheJobDao,
-	NewRepositoryDao, NewTagDao, NewRepositoryTagDao, NewOrganizationDao, NewHfTokenDao)
+	NewRepositoryDao, NewTagDao, NewRepositoryTagDao, NewOrganizationDao, NewHfTokenDao, NewLockDao)

internal/dao/lock_dao.go

@@ -0,0 +1,39 @@
+package dao
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"dingoscheduler/internal/data"
+)
+
+type LockDao struct {
+	baseData           *data.BaseData
+	cacheJobReqMu      sync.Mutex
+	cacheJobReqTimeout time.Duration
+}
+
+func NewLockDao(baseData *data.BaseData) *LockDao {
+	return &LockDao{baseData: baseData, cacheJobReqTimeout: 60 * time.Second}
+}
+
+func (f *LockDao) GetCacheJobReqLock(orgRepoKey string) *sync.RWMutex {
+	if val, ok := f.baseData.Cache.Get(orgRepoKey); ok {
+		f.baseData.Cache.Set(orgRepoKey, val, f.cacheJobReqTimeout)
+		return val.(*sync.RWMutex)
+	}
+	f.cacheJobReqMu.Lock()
+	defer f.cacheJobReqMu.Unlock()
+	if val, ok := f.baseData.Cache.Get(orgRepoKey); ok {
+		f.baseData.Cache.Set(orgRepoKey, val, f.cacheJobReqTimeout)
+		return val.(*sync.RWMutex)
+	}
+	newLock := &sync.RWMutex{}
+	f.baseData.Cache.Set(orgRepoKey, newLock, f.cacheJobReqTimeout)
+	return newLock
+}
+
+func GetCacheJobOrgRepoKey(orgRepo string) string {
+	return fmt.Sprintf("cacheJob/%s", orgRepo)
+}

internal/dao/repository_dao.go

@@ -55,7 +55,7 @@ func NewRepositoryDao(data *data.BaseData, repositoryTagDao *RepositoryTagDao, t
 }
 
 func (r *RepositoryDao) PersistRepo(persistRepoReq *query.PersistRepoReq) error {
-	zap.S().Debugf("PersistRepo instanceId:%s， org:%s, repo:%s", persistRepoReq.InstanceIds, persistRepoReq.Org, persistRepoReq.Repo)
+	zap.S().Debugf("PersistRepo start instanceId:%s， org:%s, repo:%s", persistRepoReq.InstanceIds, persistRepoReq.Org, persistRepoReq.Repo)
 	var (
 		pipelineMap map[string]string
 		err         error
@@ -93,6 +93,7 @@ func (r *RepositoryDao) PersistRepo(persistRepoReq *query.PersistRepoReq) error
 			}
 		}
 	}
+	zap.S().Debugf("PersistRepo end instanceId:%s， org:%s, repo:%s", persistRepoReq.InstanceIds, persistRepoReq.Org, persistRepoReq.Repo)
 	return nil
 }
 
@@ -117,7 +118,8 @@ func (r *RepositoryDao) singleRepositoryPersist(repository *model.Repository, in
 			return err
 		}
 		if !isComplete {
-			return myerr.New(fmt.Sprintf("repo file unComplete.%s", orgRepo))
+			zap.S().Infof("repo file unComplete.%s", orgRepo)
+			return nil
 		}
 	}
 	// 保存组织图片

internal/service/cache_job_service.go

@@ -34,15 +34,17 @@ type CacheJobService struct {
 	modelFileProcessDao *dao.ModelFileProcessDao
 	cacheJobDao         *dao.CacheJobDao
 	hfTokenDao          *dao.HfTokenDao
+	lockDao             *dao.LockDao
 }
 
 func NewCacheJobService(dingospeedDao *dao.DingospeedDao, modelFileProcessDao *dao.ModelFileProcessDao,
-	cacheJobDao *dao.CacheJobDao, hfTokenDao *dao.HfTokenDao) *CacheJobService {
+	cacheJobDao *dao.CacheJobDao, hfTokenDao *dao.HfTokenDao, lockDao *dao.LockDao) *CacheJobService {
 	return &CacheJobService{
 		dingospeedDao:       dingospeedDao,
 		cacheJobDao:         cacheJobDao,
 		modelFileProcessDao: modelFileProcessDao,
 		hfTokenDao:          hfTokenDao,
+		lockDao:             lockDao,
 	}
 }
 
@@ -62,6 +64,9 @@ func (c *CacheJobService) ListCacheJob(instanceId, datatype string, page, pageSi
 
 func (c *CacheJobService) CreateCacheJob(createCacheJobReq *query.CreateCacheJobReq) (*common.Response, error) {
 	zap.S().Debugf("Cache instanceId:%s, %s/%s", createCacheJobReq.InstanceId, createCacheJobReq.Org, createCacheJobReq.Repo)
+	lock := c.lockDao.GetCacheJobReqLock(createCacheJobReq.OrgRepo)
+	lock.Lock()
+	defer lock.Unlock()
 	cacheJob, err := c.cacheJobDao.GetCacheJob(&query.CacheJobQuery{InstanceId: createCacheJobReq.InstanceId, Type: createCacheJobReq.Type,
 		Org: createCacheJobReq.Org, Repo: createCacheJobReq.Repo, Datatype: createCacheJobReq.Datatype})
 	if err != nil {
@@ -86,6 +91,9 @@ func (c *CacheJobService) CreateCacheJob(createCacheJobReq *query.CreateCacheJob
 }
 
 func (c *CacheJobService) StopCacheJob(jobStatusReq *query.JobStatusReq) error {
+	lock := c.lockDao.GetCacheJobReqLock(util.Itoa(jobStatusReq.Id))
+	lock.Lock()
+	defer lock.Unlock()
 	cacheJob, err := c.cacheJobDao.GetCacheJob(&query.CacheJobQuery{Id: jobStatusReq.Id})
 	if err != nil {
 		return err
@@ -120,6 +128,9 @@ func (c *CacheJobService) StopCacheJob(jobStatusReq *query.JobStatusReq) error {
 }
 
 func (c *CacheJobService) ResumeCacheJob(resumeCacheJobReq *query.ResumeCacheJobReq) error {
+	lock := c.lockDao.GetCacheJobReqLock(util.Itoa(resumeCacheJobReq.Id))
+	lock.Lock()
+	defer lock.Unlock()
 	cacheJob, err := c.cacheJobDao.GetCacheJob(&query.CacheJobQuery{Id: resumeCacheJobReq.Id})
 	if err != nil {
 		return err
@@ -158,6 +169,9 @@ func (c *CacheJobService) ResumeCacheJob(resumeCacheJobReq *query.ResumeCacheJob
 }
 
 func (c *CacheJobService) DeleteCacheJob(id int64) error {
+	lock := c.lockDao.GetCacheJobReqLock(util.Itoa(id))
+	lock.Lock()
+	defer lock.Unlock()
 	cacheJob, err := c.cacheJobDao.GetCacheJob(&query.CacheJobQuery{Id: id})
 	if err != nil {
 		return err

pkg/util/avatar_util.go

@@ -96,7 +96,6 @@ func FetchAvatarURL(orgName string) (string, error) {
 	}
 	findAvatar(doc)
 	if avatarURL == "" {
-		zap.S().Errorf("在组织页面（%s）中未找到符合特征的头像节点", orgUri)
 		return "", fmt.Errorf("未在组织页面（%s）中找到头像元素", orgUri)
 	}
 	return avatarURL, nil