Show the detail error message when failed to fetch the artifacts during replication

Show the detail error message when failed to fetch the artifacts during replication

Signed-off-by: Wenkai Yin <yinw@vmware.com>

src/common/utils/passports.go

@@ -3,8 +3,6 @@ package utils
 import (
 	"context"
 	"sync"
-
-	"github.com/goharbor/harbor/src/lib/log"
 )
 
 // PassportsPool holds a given number of passports, they can be applied or be revoked. PassportsPool
@@ -60,16 +58,15 @@ func (p *passportsPool) Revoke() bool {
 type LimitedConcurrentRunner interface {
 	// AddTask adds a task to run
 	AddTask(task func() error)
-	// Wait waits all the tasks to be finished
-	Wait()
+	// Wait waits all the tasks to be finished, returns error if the any of the tasks gets error
+	Wait() (err error)
 	// Cancel cancels all tasks, tasks that already started will continue to run
-	Cancel()
-	// IsCancelled checks whether context is cancelled. This happens when some task encountered
-	// critical errors.
-	IsCancelled() bool
+	Cancel(err error)
 }
 
 type limitedConcurrentRunner struct {
+	sync.Mutex
+	err           error
 	wg            *sync.WaitGroup
 	ctx           context.Context
 	cancel        context.CancelFunc
@@ -106,23 +103,23 @@ func (r *limitedConcurrentRunner) AddTask(task func() error) {
 
 		err := task()
 		if err != nil {
-			log.Errorf("%v", err)
-			r.cancel()
+			r.Cancel(err)
 		}
 	}()
 }
 
 // Wait waits all the tasks to be finished
-func (r *limitedConcurrentRunner) Wait() {
+func (r *limitedConcurrentRunner) Wait() (err error) {
 	r.wg.Wait()
+	return r.err
 }
 
 // Cancel cancels all tasks, tasks that already started will continue to run
-func (r *limitedConcurrentRunner) Cancel() {
+func (r *limitedConcurrentRunner) Cancel(err error) {
+	if err != nil {
+		r.Lock()
+		defer r.Unlock()
+		r.err = err
+	}
 	r.cancel()
 }
-
-// IsCancelled checks whether context is cancelled. This happens when some task encountered critical errors.
-func (r *limitedConcurrentRunner) IsCancelled() bool {
-	return r.ctx.Err() != nil
-}

src/replication/adapter/aliacr/adapter.go

@@ -273,7 +273,6 @@ func (a *adapter) FetchArtifacts(filters []*model.Filter) (resources []*model.Re
 
 	var rawResources = make([]*model.Resource, len(repositories))
 	runner := utils.NewLimitedConcurrentRunner(adp.MaxConcurrency)
-	defer runner.Cancel()
 
 	for i, r := range repositories {
 		index := i
@@ -317,12 +316,9 @@ func (a *adapter) FetchArtifacts(filters []*model.Filter) (resources []*model.Re
 			return nil
 		})
 	}
-	runner.Wait()
-
-	if runner.IsCancelled() {
-		return nil, fmt.Errorf("FetchArtifacts error when collect tags for repos")
+	if err = runner.Wait(); err != nil {
+		return nil, fmt.Errorf("failed to fetch artifacts: %v", err)
 	}
-
 	for _, r := range rawResources {
 		if r != nil {
 			resources = append(resources, r)

src/replication/adapter/dockerhub/adapter.go

@@ -277,7 +277,6 @@ func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, er
 
 	var rawResources = make([]*model.Resource, len(repos))
 	runner := utils.NewLimitedConcurrentRunner(adp.MaxConcurrency)
-	defer runner.Cancel()
 	for i, r := range repos {
 		index := i
 		repo := r
@@ -341,12 +340,9 @@ func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, er
 			return nil
 		})
 	}
-	runner.Wait()
-
-	if runner.IsCancelled() {
-		return nil, fmt.Errorf("FetchArtifacts error when collect tags for repos")
+	if err = runner.Wait(); err != nil {
+		return nil, fmt.Errorf("failed to fetch artifacts: %v", err)
 	}
-
 	var resources []*model.Resource
 	for _, r := range rawResources {
 		if r != nil {

src/replication/adapter/harbor/v1/adapter.go

@@ -59,7 +59,6 @@ func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, er
 
 		var rawResources = make([]*model.Resource, len(repositories))
 		runner := utils.NewLimitedConcurrentRunner(adp.MaxConcurrency)
-		defer runner.Cancel()
 
 		for i, r := range repositories {
 			index := i
@@ -88,12 +87,9 @@ func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, er
 				return nil
 			})
 		}
-		runner.Wait()
-
-		if runner.IsCancelled() {
-			return nil, fmt.Errorf("FetchArtifacts error when collect tags for repos")
+		if err = runner.Wait(); err != nil {
+			return nil, fmt.Errorf("failed to fetch artifacts: %v", err)
 		}
-
 		for _, r := range rawResources {
 			if r != nil {
 				resources = append(resources, r)

src/replication/adapter/harbor/v2/adapter.go

@@ -68,7 +68,6 @@ func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, er
 
 		var rawResources = make([]*model.Resource, len(repositories))
 		runner := utils.NewLimitedConcurrentRunner(adp.MaxConcurrency)
-		defer runner.Cancel()
 
 		for i, r := range repositories {
 			index := i
@@ -97,10 +96,8 @@ func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, er
 				return nil
 			})
 		}
-		runner.Wait()
-
-		if runner.IsCancelled() {
-			return nil, fmt.Errorf("FetchArtifacts error when collect tags for repos")
+		if err = runner.Wait(); err != nil {
+			return nil, err
 		}
 
 		for _, r := range rawResources {

src/replication/adapter/jfrog/adapter.go

@@ -168,7 +168,6 @@ func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, er
 
 	var rawResources = make([]*model.Resource, len(repositories))
 	runner := utils.NewLimitedConcurrentRunner(adp.MaxConcurrency)
-	defer runner.Cancel()
 
 	for i, r := range repositories {
 		index := i
@@ -195,12 +194,9 @@ func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, er
 			return nil
 		})
 	}
-	runner.Wait()
-
-	if runner.IsCancelled() {
-		return nil, fmt.Errorf("FetchArtifacts error when collect tags for repos")
+	if err = runner.Wait(); err != nil {
+		return nil, fmt.Errorf("failed to fetch artifacts: %v", err)
 	}
-
 	var resources []*model.Resource
 	for _, r := range rawResources {
 		if r != nil {

src/replication/adapter/native/adapter.go

@@ -141,7 +141,6 @@ func (a *Adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, er
 
 	var rawResources = make([]*model.Resource, len(repositories))
 	runner := utils.NewLimitedConcurrentRunner(adp.MaxConcurrency)
-	defer runner.Cancel()
 
 	for i, r := range repositories {
 		index := i
@@ -168,10 +167,8 @@ func (a *Adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, er
 			return nil
 		})
 	}
-	runner.Wait()
-
-	if runner.IsCancelled() {
-		return nil, fmt.Errorf("FetchArtifacts error when collect tags for repos")
+	if err = runner.Wait(); err != nil {
+		return nil, fmt.Errorf("failed to fetch artifacts: %v", err)
 	}
 
 	var resources []*model.Resource