Upstream/harbor
1
0
Fork 0
mirror of https://github.com/goharbor/harbor.git synced 2024-11-23 10:45:45 +01:00
Code Issues Projects Releases Wiki Activity

fix #2382 (#2422)

Browse Source 
* fix #2382
This commit is contained in:
Daniel Jiang 2017-06-05 21:51:50 +08:00 committed by GitHub
parent 5512478593
commit c099ccf02e
2 changed files with 14 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
src/jobservice/job/statemachine.go
View File

@ -54,7 +54,7 @@ func (sm *SM) EnterState(s string) (string, error) {
return "", err return "", err
} }
} else { } else {
log.Debugf("Job: %d, no handler found for state:%s, skip", sm.CurrentJob, sm.CurrentState) log.Debugf("Job: %v, no exit handler found for state:%s, skip", sm.CurrentJob, sm.CurrentState)
} }
enterHandler, ok := sm.Handlers[s] enterHandler, ok := sm.Handlers[s]
var next = models.JobContinue var next = models.JobContinue
@ -101,7 +101,7 @@ func (sm *SM) Start(s string) {
log.Debugf("Job: %v, next state from handler: %s", sm.CurrentJob, n) log.Debugf("Job: %v, next state from handler: %s", sm.CurrentJob, n)
} }
if err != nil { if err != nil {
log.Warningf("Job: %v, the statemachin will enter error state due to error: %v", sm.CurrentJob, err) log.Warningf("Job: %v, the statemachine will enter error state due to error: %v", sm.CurrentJob, err)
sm.EnterState(models.JobError) sm.EnterState(models.JobError)
} }
} }
@ -187,6 +187,7 @@ func (sm *SM) Reset(j Job) error {
sm.AddTransition(models.JobRetrying, models.JobRunning, StatusUpdater{sm.CurrentJob, models.JobRunning}) sm.AddTransition(models.JobRetrying, models.JobRunning, StatusUpdater{sm.CurrentJob, models.JobRunning})
sm.Handlers[models.JobError] = StatusUpdater{sm.CurrentJob, models.JobError} sm.Handlers[models.JobError] = StatusUpdater{sm.CurrentJob, models.JobError}
sm.Handlers[models.JobStopped] = StatusUpdater{sm.CurrentJob, models.JobStopped} sm.Handlers[models.JobStopped] = StatusUpdater{sm.CurrentJob, models.JobStopped}
sm.Handlers[models.JobCanceled] = StatusUpdater{sm.CurrentJob, models.JobCanceled}
sm.Handlers[models.JobRetrying] = Retry{sm.CurrentJob} sm.Handlers[models.JobRetrying] = Retry{sm.CurrentJob}
if err := sm.CurrentJob.Init(); err != nil { if err := sm.CurrentJob.Init(); err != nil {
return err return err
@ -201,12 +202,14 @@ func (sm *SM) kickOff() error {
if repJob, ok := sm.CurrentJob.(*RepJob); ok { if repJob, ok := sm.CurrentJob.(*RepJob); ok {
if repJob.parm.Enabled == 0 { if repJob.parm.Enabled == 0 {
log.Debugf("The policy of job:%v is disabled, will cancel the job", repJob) log.Debugf("The policy of job:%v is disabled, will cancel the job", repJob)
if err := repJob.UpdateStatus(models.JobCanceled); err != nil { _, err := sm.EnterState(models.JobCanceled)
log.Warningf("Failed to update status of job: %v to 'canceled', error: %v", repJob, err) if err != nil {
log.Warningf("For job: %v, failed to update state to 'canceled', error: %v", repJob, err)
} }
return err
} }
} }
log.Debugf("In kickOff: will start job: %v", sm.CurrentJob)
sm.Start(models.JobRunning) sm.Start(models.JobRunning)
return nil return nil
} }

11
src/jobservice/job/workerpool.go
View File

@ -103,9 +103,10 @@ func (w *Worker) handle(job Job) {
} }
// NewWorker returns a pointer to new instance of worker // NewWorker returns a pointer to new instance of worker
func NewWorker(id int, wp *workerPool) *Worker { func NewWorker(id int, t Type, wp *workerPool) *Worker {
w := &Worker{ w := &Worker{
ID: id, ID: id,
Type: t,
Jobs: make(chan Job), Jobs: make(chan Job),
quit: make(chan bool), quit: make(chan bool),
queue: wp.workerChan, queue: wp.workerChan,
@ -125,19 +126,19 @@ func InitWorkerPools() error {
return err return err
} }
WorkerPools = make(map[Type]*workerPool) WorkerPools = make(map[Type]*workerPool)
WorkerPools[ReplicationType] = createWorkerPool(maxRepWorker) WorkerPools[ReplicationType] = createWorkerPool(maxRepWorker, ReplicationType)
WorkerPools[ScanType] = createWorkerPool(maxScanWorker) WorkerPools[ScanType] = createWorkerPool(maxScanWorker, ScanType)
return nil return nil
} }
//createWorkerPool create workers according to parm //createWorkerPool create workers according to parm
func createWorkerPool(n int) *workerPool { func createWorkerPool(n int, t Type) *workerPool {
wp := &workerPool{ wp := &workerPool{
workerChan: make(chan *Worker, n), workerChan: make(chan *Worker, n),
workerList: make([]*Worker, 0, n), workerList: make([]*Worker, 0, n),
} }
for i := 0; i < n; i++ { for i := 0; i < n; i++ {
worker := NewWorker(i, wp) worker := NewWorker(i, t, wp)
wp.workerList = append(wp.workerList, worker) wp.workerList = append(wp.workerList, worker)
worker.Start() worker.Start()
log.Debugf("worker %v started", worker) log.Debugf("worker %v started", worker)