Merge pull request #8302 from ywk253100/190711_client

Implement the retention client

src/core/api/harborapi_test.go

@@ -555,7 +555,7 @@ func (a testapi) GetRepos(authInfo usrInfo, projectID, keyword string) (
 	return code, nil, nil
 }
 
-func (a testapi) GetTag(authInfo usrInfo, repository string, tag string) (int, *tagResp, error) {
+func (a testapi) GetTag(authInfo usrInfo, repository string, tag string) (int, *TagResp, error) {
 	_sling := sling.New().Get(a.basePath).Path(fmt.Sprintf("/api/repositories/%s/tags/%s", repository, tag))
 	code, data, err := request(_sling, jsonAcceptHeader, authInfo)
 	if err != nil {
@@ -567,7 +567,7 @@ func (a testapi) GetTag(authInfo usrInfo, repository string, tag string) (int, *
 		return code, nil, nil
 	}
 
-	result := tagResp{}
+	result := TagResp{}
 	if err := json.Unmarshal(data, &result); err != nil {
 		return 0, nil, err
 	}
@@ -591,7 +591,7 @@ func (a testapi) GetReposTags(authInfo usrInfo, repoName string) (int, interface
 		return httpStatusCode, body, nil
 	}
 
-	result := []tagResp{}
+	result := []TagResp{}
 	if err := json.Unmarshal(body, &result); err != nil {
 		return 0, nil, err
 	}

src/core/api/repository.go

@@ -17,7 +17,6 @@ package api
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/goharbor/harbor/src/pkg/scan"
 	"io/ioutil"
 	"net/http"
 	"sort"
@@ -25,6 +24,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/goharbor/harbor/src/pkg/scan"
+
 	"errors"
 
 	"github.com/docker/distribution/manifest/schema1"
@@ -96,7 +97,8 @@ type cfg struct {
 	Labels map[string]string `json:"labels"`
 }
 
-type tagResp struct {
+// TagResp holds the information of one image tag
+type TagResp struct {
 	tagDetail
 	Signature    *notary.Target          `json:"signature"`
 	ScanOverview *models.ImgScanOverview `json:"scan_overview,omitempty"`
@@ -608,7 +610,7 @@ func (ra *RepositoryAPI) GetTags() {
 // get config, signature and scan overview and assemble them into one
 // struct for each tag in tags
 func assembleTagsInParallel(client *registry.Repository, repository string,
-	tags []string, username string) []*tagResp {
+	tags []string, username string) []*TagResp {
 	var err error
 	signatures := map[string][]notary.Target{}
 	if config.WithNotary() {
@@ -619,13 +621,13 @@ func assembleTagsInParallel(client *registry.Repository, repository string,
 		}
 	}
 
-	c := make(chan *tagResp)
+	c := make(chan *TagResp)
 	for _, tag := range tags {
 		go assembleTag(c, client, repository, tag, config.WithClair(),
 			config.WithNotary(), signatures)
 	}
-	result := []*tagResp{}
-	var item *tagResp
+	result := []*TagResp{}
+	var item *TagResp
 	for i := 0; i < len(tags); i++ {
 		item = <-c
 		if item == nil {
@@ -636,10 +638,10 @@ func assembleTagsInParallel(client *registry.Repository, repository string,
 	return result
 }
 
-func assembleTag(c chan *tagResp, client *registry.Repository,
+func assembleTag(c chan *TagResp, client *registry.Repository,
 	repository, tag string, clairEnabled, notaryEnabled bool,
 	signatures map[string][]notary.Target) {
-	item := &tagResp{}
+	item := &TagResp{}
 	// labels
 	image := fmt.Sprintf("%s:%s", repository, tag)
 	labels, err := dao.GetLabelsOfResource(common.ResourceTypeImage, image)

src/core/api/repository_test.go

@@ -96,7 +96,7 @@ func TestGetReposTags(t *testing.T) {
 		t.Errorf("failed to get tags of repository %s: %v", repository, err)
 	} else {
 		assert.Equal(int(200), code, "httpStatusCode should be 200")
-		if tg, ok := tags.([]tagResp); ok {
+		if tg, ok := tags.([]TagResp); ok {
 			assert.Equal(1, len(tg), fmt.Sprintf("there should be only one tag, but now %v", tg))
 			assert.Equal(tg[0].Name, "latest", "the tag should be latest")
 		} else {

src/jobservice/job/known_jobs.go

@@ -30,4 +30,6 @@ const (
 	Replication = "REPLICATION"
 	// ReplicationScheduler : the name of the replication scheduler job in job service
 	ReplicationScheduler = "IMAGE_REPLICATE"
+	// Retention : the name of the retention job
+	Retention = "RETENTION"
 )

src/pkg/clients/core/chart.go

@@ -0,0 +1,35 @@
+// Copyright Project Harbor Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package core
+
+import (
+	"fmt"
+
+	"github.com/goharbor/harbor/src/chartserver"
+)
+
+func (c *client) ListAllCharts(project, repository string) ([]*chartserver.ChartVersion, error) {
+	url := c.buildURL(fmt.Sprintf("/api/chartrepo/%s/charts/%s", project, repository))
+	var charts []*chartserver.ChartVersion
+	if err := c.httpclient.Get(url, &charts); err != nil {
+		return nil, err
+	}
+	return charts, nil
+}
+
+func (c *client) DeleteChart(project, repository, version string) error {
+	url := c.buildURL(fmt.Sprintf("/api/chartrepo/%s/charts/%s/%s", project, repository, version))
+	return c.httpclient.Delete(url)
+}

src/pkg/clients/core/client.go

@@ -0,0 +1,62 @@
+// Copyright Project Harbor Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package core
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/goharbor/harbor/src/chartserver"
+	chttp "github.com/goharbor/harbor/src/common/http"
+	"github.com/goharbor/harbor/src/common/http/modifier"
+	"github.com/goharbor/harbor/src/core/api"
+)
+
+// Client defines the methods that a core client should implement
+// Currently, it contains only part of the whole method collection
+// and we should expand it when needed
+type Client interface {
+	ImageClient
+	ChartClient
+}
+
+// ImageClient defines the methods that an image client should implement
+type ImageClient interface {
+	ListAllImages(project, repository string) ([]*api.TagResp, error)
+	DeleteImage(project, repository, tag string) error
+}
+
+// ChartClient defines the methods that a chart client should implement
+type ChartClient interface {
+	ListAllCharts(project, repository string) ([]*chartserver.ChartVersion, error)
+	DeleteChart(project, repository, version string) error
+}
+
+// New returns an instance of the client which is a default implement for Client
+func New(url string, httpclient *http.Client, authorizer modifier.Modifier) Client {
+	return &client{
+		url:        url,
+		httpclient: chttp.NewClient(httpclient, authorizer),
+	}
+}
+
+type client struct {
+	url        string
+	httpclient *chttp.Client
+}
+
+func (c *client) buildURL(path string) string {
+	return fmt.Sprintf("%s/%s", c.url, path)
+}

src/pkg/clients/core/image.go

@@ -0,0 +1,35 @@
+// Copyright Project Harbor Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package core
+
+import (
+	"fmt"
+
+	"github.com/goharbor/harbor/src/core/api"
+)
+
+func (c *client) ListAllImages(project, repository string) ([]*api.TagResp, error) {
+	url := c.buildURL(fmt.Sprintf("/api/repositories/%s/%s/tags", project, repository))
+	var images []*api.TagResp
+	if err := c.httpclient.GetAndIteratePagination(url, &images); err != nil {
+		return nil, err
+	}
+	return images, nil
+}
+
+func (c *client) DeleteImage(project, repository, tag string) error {
+	url := c.buildURL(fmt.Sprintf("/api/repositories/%s/%s/tags/%s", project, repository, tag))
+	return c.httpclient.Delete(url)
+}

src/pkg/retention/client.go

@@ -15,6 +15,16 @@
 package retention
 
 import (
+	"errors"
+	"fmt"
+	"net/http"
+
+	"github.com/goharbor/harbor/src/common/http/modifier/auth"
+	cjob "github.com/goharbor/harbor/src/common/job"
+	"github.com/goharbor/harbor/src/common/job/models"
+	"github.com/goharbor/harbor/src/core/config"
+	"github.com/goharbor/harbor/src/jobservice/job"
+	"github.com/goharbor/harbor/src/pkg/clients/core"
 	"github.com/goharbor/harbor/src/pkg/retention/policy"
 	"github.com/goharbor/harbor/src/pkg/retention/res"
 )
@@ -43,36 +53,137 @@ type Client interface {
 	// SubmitTask to jobservice
 	//
 	//  Arguments:
-	//    repository: *res.Repository : repository info
+	//    taskID                      : the ID of task
+	//    repository *res.Repository  : repository info
 	//    meta *policy.LiteMeta       : policy lite metadata
 	//
 	//  Returns:
 	//    string : the job ID
 	//    error  : common error if any errors occurred
-	SubmitTask(repository *res.Repository, meta *policy.LiteMeta) (string, error)
+	SubmitTask(taskID int64, repository *res.Repository, meta *policy.LiteMeta) (string, error)
 }
 
 // New basic client
-func New() Client {
-	return &basicClient{}
+func New(client ...*http.Client) Client {
+	var c *http.Client
+	if len(client) > 0 {
+		c = client[0]
+	}
+	if c == nil {
+		c = http.DefaultClient
+	}
+
+	// init core client
+	internalCoreURL := config.InternalCoreURL()
+	jobserviceSecret := config.JobserviceSecret()
+	authorizer := auth.NewSecretAuthorizer(jobserviceSecret)
+	coreClient := core.New(internalCoreURL, c, authorizer)
+
+	// init jobservice client
+	internalJobserviceURL := config.InternalJobServiceURL()
+	coreSecret := config.CoreSecret()
+	jobserviceClient := cjob.NewDefaultClient(internalJobserviceURL, coreSecret)
+
+	return &basicClient{
+		internalCoreURL:  internalCoreURL,
+		coreClient:       coreClient,
+		jobserviceClient: jobserviceClient,
+	}
 }
 
 // basicClient is a default
-type basicClient struct{}
+type basicClient struct {
+	internalCoreURL  string
+	coreClient       core.Client
+	jobserviceClient cjob.Client
+}
 
 // GetCandidates gets the tag candidates under the repository
-func (bc *basicClient) GetCandidates(repo *res.Repository) ([]*res.Candidate, error) {
-	results := make([]*res.Candidate, 0)
-
-	return results, nil
+func (bc *basicClient) GetCandidates(repository *res.Repository) ([]*res.Candidate, error) {
+	if repository == nil {
+		return nil, errors.New("repository is nil")
+	}
+	candidates := make([]*res.Candidate, 0)
+	switch repository.Kind {
+	case CandidateKindImage:
+		images, err := bc.coreClient.ListAllImages(repository.Namespace, repository.Name)
+		if err != nil {
+			return nil, err
+		}
+		for _, image := range images {
+			labels := []string{}
+			for _, label := range image.Labels {
+				labels = append(labels, label.Name)
+			}
+			candidate := &res.Candidate{
+				Kind:         CandidateKindImage,
+				Namespace:    repository.Namespace,
+				Repository:   repository.Name,
+				Tag:          image.Name,
+				Labels:       labels,
+				CreationTime: image.Created.Unix(),
+				// TODO: populate the pull/push time
+				// PulledTime: ,
+				// PushedTime:,
+			}
+			candidates = append(candidates, candidate)
+		}
+	case CandidateKindChart:
+		charts, err := bc.coreClient.ListAllCharts(repository.Namespace, repository.Name)
+		if err != nil {
+			return nil, err
+		}
+		for _, chart := range charts {
+			labels := []string{}
+			for _, label := range chart.Labels {
+				labels = append(labels, label.Name)
+			}
+			candidate := &res.Candidate{
+				Kind:         CandidateKindChart,
+				Namespace:    repository.Namespace,
+				Repository:   repository.Name,
+				Tag:          chart.Name,
+				Labels:       labels,
+				CreationTime: chart.Created.Unix(),
+				// TODO: populate the pull/push time
+				// PulledTime: ,
+				// PushedTime:,
+			}
+			candidates = append(candidates, candidate)
+		}
+	default:
+		return nil, fmt.Errorf("unsupported repository kind: %s", repository.Kind)
+	}
+	return candidates, nil
 }
 
 // Deletes the specified candidate
 func (bc *basicClient) Delete(candidate *res.Candidate) error {
-	return nil
+	if candidate == nil {
+		return errors.New("candidate is nil")
+	}
+	switch candidate.Kind {
+	case CandidateKindImage:
+		return bc.coreClient.DeleteImage(candidate.Namespace, candidate.Repository, candidate.Tag)
+	case CandidateKindChart:
+		return bc.coreClient.DeleteChart(candidate.Namespace, candidate.Repository, candidate.Tag)
+	default:
+		return fmt.Errorf("unsupported candidate kind: %s", candidate.Kind)
+	}
 }
 
 // SubmitTask to jobservice
-func (bc *basicClient) SubmitTask(*res.Repository, *policy.LiteMeta) (string, error) {
-	return "", nil
+func (bc *basicClient) SubmitTask(taskID int64, repository *res.Repository, meta *policy.LiteMeta) (string, error) {
+	j := &models.JobData{
+		Metadata: &models.JobMetadata{
+			JobKind: job.KindGeneric,
+		},
+		StatusHook: fmt.Sprintf("%s/service/notifications/jobs/retention/tasks/%d", bc.internalCoreURL, taskID),
+	}
+	j.Name = job.Retention
+	j.Parameters = map[string]interface{}{
+		ParamRepo: repository,
+		ParamMeta: meta,
+	}
+	return bc.jobserviceClient.SubmitJob(j)
 }

src/pkg/retention/client_test.go

@@ -0,0 +1,142 @@
+// Copyright Project Harbor Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package retention
+
+import (
+	"testing"
+
+	"github.com/goharbor/harbor/src/chartserver"
+	"github.com/goharbor/harbor/src/common/job/models"
+	"github.com/goharbor/harbor/src/core/api"
+	"github.com/goharbor/harbor/src/jobservice/job"
+	"github.com/goharbor/harbor/src/pkg/retention/res"
+	"github.com/goharbor/harbor/src/testing/clients"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+	"k8s.io/helm/pkg/proto/hapi/chart"
+	"k8s.io/helm/pkg/repo"
+)
+
+type fakeCoreClient struct {
+	clients.DumbCoreClient
+}
+
+func (f *fakeCoreClient) ListAllImages(project, repository string) ([]*api.TagResp, error) {
+	image := &api.TagResp{}
+	image.Name = "latest"
+	return []*api.TagResp{image}, nil
+}
+
+func (f *fakeCoreClient) ListAllCharts(project, repository string) ([]*chartserver.ChartVersion, error) {
+	metadata := &chart.Metadata{
+		Name: "1.0",
+	}
+	chart := &chartserver.ChartVersion{}
+	chart.ChartVersion = repo.ChartVersion{
+		Metadata: metadata,
+	}
+	return []*chartserver.ChartVersion{chart}, nil
+}
+
+type fakeJobserviceClient struct{}
+
+func (f *fakeJobserviceClient) SubmitJob(*models.JobData) (string, error) {
+	return "1", nil
+}
+func (f *fakeJobserviceClient) GetJobLog(uuid string) ([]byte, error) {
+	return nil, nil
+}
+func (f *fakeJobserviceClient) PostAction(uuid, action string) error {
+	return nil
+}
+func (f *fakeJobserviceClient) GetExecutions(uuid string) ([]job.Stats, error) {
+	return nil, nil
+}
+
+type clientTestSuite struct {
+	suite.Suite
+}
+
+func (c *clientTestSuite) TestGetCandidates() {
+	client := &basicClient{}
+	client.coreClient = &fakeCoreClient{}
+	var repository *res.Repository
+	// nil repository
+	candidates, err := client.GetCandidates(repository)
+	require.NotNil(c.T(), err)
+
+	// image repository
+	repository = &res.Repository{}
+	repository.Kind = CandidateKindImage
+	repository.Namespace = "library"
+	repository.Name = "hello-world"
+	candidates, err = client.GetCandidates(repository)
+	require.Nil(c.T(), err)
+	assert.Equal(c.T(), 1, len(candidates))
+	assert.Equal(c.T(), CandidateKindImage, candidates[0].Kind)
+	assert.Equal(c.T(), "library", candidates[0].Namespace)
+	assert.Equal(c.T(), "hello-world", candidates[0].Repository)
+	assert.Equal(c.T(), "latest", candidates[0].Tag)
+
+	// chart repository
+	repository.Kind = CandidateKindChart
+	repository.Namespace = "goharbor"
+	repository.Name = "harbor"
+	candidates, err = client.GetCandidates(repository)
+	require.Nil(c.T(), err)
+	assert.Equal(c.T(), 1, len(candidates))
+	assert.Equal(c.T(), CandidateKindChart, candidates[0].Kind)
+	assert.Equal(c.T(), "goharbor", candidates[0].Namespace)
+	assert.Equal(c.T(), "1.0", candidates[0].Tag)
+}
+
+func (c *clientTestSuite) TestDelete() {
+	client := &basicClient{}
+	client.coreClient = &fakeCoreClient{}
+
+	var candidate *res.Candidate
+	// nil candidate
+	err := client.Delete(candidate)
+	require.NotNil(c.T(), err)
+
+	// image
+	candidate = &res.Candidate{}
+	candidate.Kind = CandidateKindImage
+	err = client.Delete(candidate)
+	require.Nil(c.T(), err)
+
+	// chart
+	candidate.Kind = CandidateKindChart
+	err = client.Delete(candidate)
+	require.Nil(c.T(), err)
+
+	// unsupported type
+	candidate.Kind = "unsupported"
+	err = client.Delete(candidate)
+	require.NotNil(c.T(), err)
+}
+
+func (c *clientTestSuite) TestSubmitTask() {
+	client := &basicClient{}
+	client.jobserviceClient = &fakeJobserviceClient{}
+	jobID, err := client.SubmitTask(1, nil, nil)
+	require.Nil(c.T(), err)
+	assert.Equal(c.T(), "1", jobID)
+}
+
+func TestClientTestSuite(t *testing.T) {
+	suite.Run(t, new(clientTestSuite))
+}

src/pkg/retention/launcher.go

@@ -28,7 +28,10 @@ import (
 )
 
 // TODO init the client
-var client Client
+var (
+	client Client
+	mgr    Manager
+)
 
 // Launcher provides function to launch the async jobs to run retentions based on the provided policy.
 type Launcher interface {
@@ -37,11 +40,12 @@ type Launcher interface {
 	//
 	//  Arguments:
 	//   policy *policy.Metadata: the policy info
+	//   executionID int64      : the execution ID
 	//
 	//  Returns:
-	//   []*TaskSubmitResult : the submit results of tasks
+	//   int64               : the count of tasks
 	//   error               : common error if any errors occurred
-	Launch(policy *policy.Metadata) ([]*TaskSubmitResult, error)
+	Launch(policy *policy.Metadata, executionID int64) (int64, error)
 }
 
 // NewLauncher returns an instance of Launcher
@@ -52,17 +56,23 @@ func NewLauncher() Launcher {
 type launcher struct {
 }
 
-func (l *launcher) Launch(ply *policy.Metadata) ([]*TaskSubmitResult, error) {
+type jobData struct {
+	repository *res.Repository
+	policy     *policy.LiteMeta
+	taskID     int64
+}
+
+func (l *launcher) Launch(ply *policy.Metadata, executionID int64) (int64, error) {
 	if ply == nil {
-		return nil, launcherError(fmt.Errorf("the policy is nil"))
+		return 0, launcherError(fmt.Errorf("the policy is nil"))
 	}
 	// no rules, return directly
 	if len(ply.Rules) == 0 {
-		return nil, nil
+		return 0, nil
 	}
 	scope := ply.Scope
 	if scope == nil {
-		return nil, launcherError(fmt.Errorf("the scope of policy is nil"))
+		return 0, launcherError(fmt.Errorf("the scope of policy is nil"))
 	}
 
 	repositoryRules := make(map[res.Repository]*policy.LiteMeta, 0)
@@ -73,7 +83,7 @@ func (l *launcher) Launch(ply *policy.Metadata) ([]*TaskSubmitResult, error) {
 		// get projects
 		projectCandidates, err = getProjects()
 		if err != nil {
-			return nil, launcherError(err)
+			return 0, launcherError(err)
 		}
 	}
 
@@ -85,11 +95,11 @@ func (l *launcher) Launch(ply *policy.Metadata) ([]*TaskSubmitResult, error) {
 				selector, err := selectors.Get(projectSelector.Kind, projectSelector.Decoration,
 					projectSelector.Pattern)
 				if err != nil {
-					return nil, launcherError(err)
+					return 0, launcherError(err)
 				}
 				projectCandidates, err = selector.Select(projectCandidates)
 				if err != nil {
-					return nil, launcherError(err)
+					return 0, launcherError(err)
 				}
 			}
 		case "project":
@@ -103,7 +113,7 @@ func (l *launcher) Launch(ply *policy.Metadata) ([]*TaskSubmitResult, error) {
 		for _, projectCandidate := range projectCandidates {
 			repositories, err := getRepositories(projectCandidate.NamespaceID)
 			if err != nil {
-				return nil, launcherError(err)
+				return 0, launcherError(err)
 			}
 			repositoryCandidates = append(repositoryCandidates, repositories...)
 		}
@@ -112,11 +122,11 @@ func (l *launcher) Launch(ply *policy.Metadata) ([]*TaskSubmitResult, error) {
 			selector, err := selectors.Get(repositorySelector.Kind, repositorySelector.Decoration,
 				repositorySelector.Pattern)
 			if err != nil {
-				return nil, launcherError(err)
+				return 0, launcherError(err)
 			}
 			repositoryCandidates, err = selector.Select(repositoryCandidates)
 			if err != nil {
-				return nil, launcherError(err)
+				return 0, launcherError(err)
 			}
 		}
 
@@ -134,19 +144,40 @@ func (l *launcher) Launch(ply *policy.Metadata) ([]*TaskSubmitResult, error) {
 			repositoryRules[repository].Rules = append(repositoryRules[repository].Rules, &rule)
 		}
 	}
+	// no tasks need to be submitted
+	if len(repositoryRules) == 0 {
+		return 0, nil
+	}
 
-	var result []*TaskSubmitResult
-	for repository, rule := range repositoryRules {
-		jobID, err := client.SubmitTask(&repository, rule)
-		result = append(result, &TaskSubmitResult{
-			JobID: jobID,
-			Error: err,
+	// create task records
+	jobDatas := []*jobData{}
+	for repository, policy := range repositoryRules {
+		taskID, err := mgr.CreateTask(&Task{
+			ExecutionID: executionID,
 		})
 		if err != nil {
-			log.Error(launcherError(fmt.Errorf("failed to submit task: %v", err)))
+			return 0, launcherError(err)
 		}
+		jobDatas = append(jobDatas, &jobData{
+			repository: &repository,
+			policy:     policy,
+			taskID:     taskID,
+		})
 	}
-	return result, nil
+
+	allFailed := true
+	for _, jobData := range jobDatas {
+		_, err := client.SubmitTask(jobData.taskID, jobData.repository, jobData.policy)
+		if err != nil {
+			log.Error(launcherError(fmt.Errorf("failed to submit task %d: %v", jobData.taskID, err)))
+			continue
+		}
+		allFailed = false
+	}
+	if allFailed {
+		return 0, launcherError(fmt.Errorf("all tasks failed"))
+	}
+	return int64(len(jobDatas)), nil
 }
 
 func launcherError(err error) error {

src/pkg/retention/launcher_test.go

@@ -19,18 +19,18 @@ import (
 	"strconv"
 	"testing"
 
-	"github.com/stretchr/testify/suite"
-
 	"github.com/goharbor/harbor/src/chartserver"
 	"github.com/goharbor/harbor/src/common/models"
 	"github.com/goharbor/harbor/src/pkg/project"
 	"github.com/goharbor/harbor/src/pkg/repository"
 	"github.com/goharbor/harbor/src/pkg/retention/policy"
 	"github.com/goharbor/harbor/src/pkg/retention/policy/rule"
+	"github.com/goharbor/harbor/src/pkg/retention/q"
 	"github.com/goharbor/harbor/src/pkg/retention/res"
 	_ "github.com/goharbor/harbor/src/pkg/retention/res/selectors/regexp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
 )
 
 type fakeProjectManager struct {
@@ -85,11 +85,56 @@ func (f *fakeClient) GetCandidates(repo *res.Repository) ([]*res.Candidate, erro
 func (f *fakeClient) Delete(candidate *res.Candidate) error {
 	return nil
 }
-func (f *fakeClient) SubmitTask(repository *res.Repository, meta *policy.LiteMeta) (string, error) {
+func (f *fakeClient) SubmitTask(taskID int64, repository *res.Repository, meta *policy.LiteMeta) (string, error) {
 	f.id++
 	return strconv.Itoa(f.id), nil
 }
 
+type fakeRetentionManager struct{}
+
+func (f *fakeRetentionManager) CreatePolicy(p *policy.Metadata) (int64, error) {
+	return 0, nil
+}
+func (f *fakeRetentionManager) UpdatePolicy(p *policy.Metadata) error {
+	return nil
+}
+func (f *fakeRetentionManager) DeletePolicy(ID int64) error {
+	return nil
+}
+func (f *fakeRetentionManager) GetPolicy(ID int64) (*policy.Metadata, error) {
+	return nil, nil
+}
+func (f *fakeRetentionManager) CreateExecution(execution *Execution) (int64, error) {
+	return 0, nil
+}
+func (f *fakeRetentionManager) UpdateExecution(execution *Execution) error {
+	return nil
+}
+func (f *fakeRetentionManager) GetExecution(eid int64) (*Execution, error) {
+	return nil, nil
+}
+func (f *fakeRetentionManager) ListTasks(query *q.Query) ([]*Task, error) {
+	return nil, nil
+}
+func (f *fakeRetentionManager) CreateTask(task *Task) (int64, error) {
+	return 0, nil
+}
+func (f *fakeRetentionManager) UpdateTask(task *Task) error {
+	return nil
+}
+func (f *fakeRetentionManager) GetTaskLog(taskID int64) ([]byte, error) {
+	return nil, nil
+}
+func (f *fakeRetentionManager) ListExecutions(query *q.Query) ([]*Execution, error) {
+	return nil, nil
+}
+func (f *fakeRetentionManager) AppendHistory(history *History) error {
+	return nil
+}
+func (f *fakeRetentionManager) ListHistories(executionID int64, query *q.Query) ([]*History, error) {
+	return nil, nil
+}
+
 type launchTestSuite struct {
 	suite.Suite
 }
@@ -116,6 +161,7 @@ func (l *launchTestSuite) SetupTest() {
 		},
 	}
 	client = &fakeClient{}
+	mgr = &fakeRetentionManager{}
 }
 
 func (l *launchTestSuite) TestGetProjects() {
@@ -142,14 +188,14 @@ func (l *launchTestSuite) TestLaunch() {
 	launcher := NewLauncher()
 	var ply *policy.Metadata
 	// nil policy
-	result, err := launcher.Launch(ply)
+	n, err := launcher.Launch(ply, 1)
 	require.NotNil(l.T(), err)
 
 	// nil rules
 	ply = &policy.Metadata{}
-	result, err = launcher.Launch(ply)
+	n, err = launcher.Launch(ply, 1)
 	require.Nil(l.T(), err)
-	assert.Equal(l.T(), 0, len(result))
+	assert.Equal(l.T(), int64(0), n)
 
 	// nil scope
 	ply = &policy.Metadata{
@@ -157,7 +203,7 @@ func (l *launchTestSuite) TestLaunch() {
 			{},
 		},
 	}
-	_, err = launcher.Launch(ply)
+	_, err = launcher.Launch(ply, 1)
 	require.NotNil(l.T(), err)
 
 	// system scope
@@ -186,14 +232,9 @@ func (l *launchTestSuite) TestLaunch() {
 			},
 		},
 	}
-
-	result, err = launcher.Launch(ply)
+	n, err = launcher.Launch(ply, 1)
 	require.Nil(l.T(), err)
-	assert.Equal(l.T(), 2, len(result))
-	assert.Equal(l.T(), "1", result[0].JobID)
-	assert.Nil(l.T(), result[0].Error)
-	assert.Equal(l.T(), "2", result[1].JobID)
-	assert.Nil(l.T(), result[1].Error)
+	assert.Equal(l.T(), int64(2), n)
 }
 
 func TestLaunchTestSuite(t *testing.T) {

src/pkg/retention/manager.go

@@ -16,16 +16,17 @@ package retention
 
 import (
 	"encoding/json"
+	"time"
+
 	"github.com/goharbor/harbor/src/pkg/retention/dao"
 	"github.com/goharbor/harbor/src/pkg/retention/dao/models"
 	"github.com/goharbor/harbor/src/pkg/retention/policy"
 	"github.com/goharbor/harbor/src/pkg/retention/q"
-	"time"
 )
 
 // Manager defines operations of managing policy
 type Manager interface {
-	// Create new policy and return uuid
+	// Create new policy and return ID
 	CreatePolicy(p *policy.Metadata) (int64, error)
 	// Update the existing policy
 	// Full update
@@ -41,6 +42,14 @@ type Manager interface {
 	UpdateExecution(execution *Execution) error
 	// Get the specified execution
 	GetExecution(eid int64) (*Execution, error)
+	// List tasks histories
+	ListTasks(query *q.Query) ([]*Task, error)
+	// Create a new retention task
+	CreateTask(task *Task) (int64, error)
+	// Update the specified task
+	UpdateTask(task *Task) error
+	// Get the log of the specified task
+	GetTaskLog(taskID int64) ([]byte, error)
 	// List execution histories
 	ListExecutions(query *q.Query) ([]*Execution, error)
 	// Add new history
@@ -150,6 +159,26 @@ func (d *DefaultManager) GetExecution(eid int64) (*Execution, error) {
 	return e1, nil
 }
 
+// CreateTask creates task record
+func (d *DefaultManager) CreateTask(task *Task) (int64, error) {
+	panic("implement me")
+}
+
+// ListTasks lists tasks according to the query
+func (d *DefaultManager) ListTasks(query *q.Query) ([]*Task, error) {
+	panic("implement me")
+}
+
+// UpdateTask updates the task
+func (d *DefaultManager) UpdateTask(task *Task) error {
+	panic("implement me")
+}
+
+// GetTaskLog gets the logs of task
+func (d *DefaultManager) GetTaskLog(taskID int64) ([]byte, error) {
+	panic("implement me")
+}
+
 // ListHistories List Histories
 func (d *DefaultManager) ListHistories(executionID int64, query *q.Query) ([]*History, error) {
 	his, err := dao.ListExecHistories(executionID, query)

src/pkg/retention/models.go

@@ -16,21 +16,39 @@ package retention
 
 import "time"
 
+// const definitions
+const (
+	ExecutionStatusInProgress string = "InProgress"
+	ExecutionStatusSucceed    string = "Succeed"
+	ExecutionStatusFailed     string = "Failed"
+	ExecutionStatusStopped    string = "Stopped"
+
+	TaskStatusPending    string = "Pending"
+	TaskStatusInProgress string = "InProgress"
+	TaskStatusSucceed    string = "Succeed"
+	TaskStatusFailed     string = "Failed"
+	TaskStatusStopped    string = "Stopped"
+
+	CandidateKindImage string = "image"
+	CandidateKindChart string = "chart"
+)
+
 // Execution of retention
 type Execution struct {
-	ID        int64     `json:"id,omitempty"`
+	ID        int64     `json:"id"`
 	PolicyID  int64     `json:"policy_id"`
 	StartTime time.Time `json:"start_time"`
 	EndTime   time.Time `json:"end_time,omitempty"`
 	Status    string    `json:"status"`
 }
 
-// TaskSubmitResult is the result of task submitting
-// If the task is submitted successfully, JobID will be set
-// and the Error is nil
-type TaskSubmitResult struct {
-	JobID string
-	Error error
+// Task of retention
+type Task struct {
+	ID          int64     `json:"id"`
+	ExecutionID int64     `json:"execution_id"`
+	Status      string    `json:"status"`
+	StartTime   time.Time `json:"start_time"`
+	EndTime     time.Time `json:"end_time"`
 }
 
 // History of retention

src/pkg/retention/policy/models.go

@@ -25,8 +25,8 @@ const (
 
 // Metadata of policy
 type Metadata struct {
-	// UUID of the policy
-	ID int64 `json:"id,omitempty"`
+	// ID of the policy
+	ID int64 `json:"id"`
 
 	// Algorithm applied to the rules
 	// "OR" / "AND"

src/testing/clients/dumb_core_client.go

@@ -0,0 +1,44 @@
+// Copyright Project Harbor Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package clients
+
+import (
+	"github.com/goharbor/harbor/src/chartserver"
+	"github.com/goharbor/harbor/src/core/api"
+)
+
+// DumbCoreClient provides an empty implement for pkg/clients/core.Client
+// it is only used for testing
+type DumbCoreClient struct{}
+
+// ListAllImages ...
+func (d *DumbCoreClient) ListAllImages(project, repository string) ([]*api.TagResp, error) {
+	return nil, nil
+}
+
+// DeleteImage ...
+func (d *DumbCoreClient) DeleteImage(project, repository, tag string) error {
+	return nil
+}
+
+// ListAllCharts ...
+func (d *DumbCoreClient) ListAllCharts(project, repository string) ([]*chartserver.ChartVersion, error) {
+	return nil, nil
+}
+
+// DeleteChart ...
+func (d *DumbCoreClient) DeleteChart(project, repository, version string) error {
+	return nil
+}