Merge pull request #8383 from ywk253100/190724_retention_client

Create retention client with the configurations got from jobservice config

src/jobservice/config/config.go

@@ -24,7 +24,7 @@ import (
 	"strings"
 
 	"github.com/goharbor/harbor/src/jobservice/common/utils"
-	"gopkg.in/yaml.v2"
+	yaml "gopkg.in/yaml.v2"
 )
 
 const (
@@ -37,6 +37,7 @@ const (
 	jobServiceRedisURL          = "JOB_SERVICE_POOL_REDIS_URL"
 	jobServiceRedisNamespace    = "JOB_SERVICE_POOL_REDIS_NAMESPACE"
 	jobServiceAuthSecret        = "JOBSERVICE_SECRET"
+	coreURL                     = "CORE_URL"
 
 	// JobServiceProtocolHTTPS points to the 'https' protocol
 	JobServiceProtocolHTTPS = "https"
@@ -163,6 +164,11 @@ func GetAuthSecret() string {
 	return utils.ReadEnv(jobServiceAuthSecret)
 }
 
+// GetCoreURL get the core url from the env
+func GetCoreURL() string {
+	return utils.ReadEnv(coreURL)
+}
+
 // GetUIAuthSecret get the auth secret of UI side
 func GetUIAuthSecret() string {
 	return utils.ReadEnv(uiAuthSecret)

src/jobservice/config/config_test.go

@@ -14,11 +14,12 @@
 package config
 
 import (
+	"os"
+	"testing"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
-	"os"
-	"testing"
 )
 
 // ConfigurationTestSuite tests the configuration loading
@@ -84,6 +85,7 @@ func (suite *ConfigurationTestSuite) TestConfigLoadingWithEnv() {
 	)
 	assert.Equal(suite.T(), "js_secret", GetAuthSecret(), "expect auth secret 'js_secret' but got '%s'", GetAuthSecret())
 	assert.Equal(suite.T(), "core_secret", GetUIAuthSecret(), "expect auth secret 'core_secret' but got '%s'", GetUIAuthSecret())
+	assert.Equal(suite.T(), "core_url", GetCoreURL(), "expect core url 'core_url' but got '%s'", GetCoreURL())
 }
 
 // TestDefaultConfig ...
@@ -134,6 +136,7 @@ func setENV() error {
 	err = os.Setenv("JOB_SERVICE_POOL_REDIS_NAMESPACE", "ut_namespace")
 	err = os.Setenv("JOBSERVICE_SECRET", "js_secret")
 	err = os.Setenv("CORE_SECRET", "core_secret")
+	err = os.Setenv("CORE_URL", "core_url")
 
 	return err
 }

src/jobservice/main.go

@@ -19,7 +19,6 @@ import (
 	"errors"
 	"flag"
 	"fmt"
-	"os"
 
 	"github.com/goharbor/harbor/src/common"
 	comcfg "github.com/goharbor/harbor/src/common/config"
@@ -64,7 +63,7 @@ func main() {
 		if utils.IsEmptyStr(secret) {
 			return nil, errors.New("empty auth secret")
 		}
-		coreURL := os.Getenv("CORE_URL")
+		coreURL := config.GetCoreURL()
 		configURL := coreURL + common.CoreConfigPath
 		cfgMgr := comcfg.NewRESTCfgManager(configURL, secret)
 		jobCtx := impl.NewContext(ctx, cfgMgr)

src/pkg/retention/boot.go

@@ -14,14 +14,10 @@
 
 package retention
 
-import "github.com/goharbor/harbor/src/pkg/retention/dep"
-
 // TODO: Move to api.Init()
 
 // Init the retention components
 func Init() error {
-	// New default retention client
-	dep.DefaultClient = dep.NewClient()
 
 	return nil
 }

src/pkg/retention/dep/client.go

@@ -17,29 +17,16 @@ package dep
 import (
 	"errors"
 	"fmt"
-	"github.com/goharbor/harbor/src/pkg/retention/policy/lwp"
 	"net/http"
 
 	"github.com/goharbor/harbor/src/common/http/modifier/auth"
-	cjob "github.com/goharbor/harbor/src/common/job"
+	"github.com/goharbor/harbor/src/jobservice/config"
-	"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/res"
 )
 
-const (
-	// ParamRepo ...
-	ParamRepo = "repository"
-	// ParamMeta ...
-	ParamMeta = "liteMeta"
-)
-
-// TODO: Move to api.Base
-
 // DefaultClient for the retention
-var DefaultClient Client
+var DefaultClient = NewClient()
 
 // Client is designed to access core service to get required infos
 type Client interface {
@@ -61,18 +48,6 @@ type Client interface {
 	//  Returns:
 	//    error : common error if any errors occurred
 	Delete(candidate *res.Candidate) error
-
-	// SubmitTask to jobservice
-	//
-	//  Arguments:
-	//    taskID                      : the ID of task
-	//    repository *res.Repository  : repository info
-	//    meta *lwp.Metadata          : policy lightweight metadata
-	//
-	//  Returns:
-	//    string : the job ID
-	//    error  : common error if any errors occurred
-	SubmitTask(taskID int64, repository *res.Repository, meta *lwp.Metadata) (string, error)
 }
 
 // NewClient new a basic client
@@ -86,28 +61,21 @@ func NewClient(client ...*http.Client) Client {
 	}
 
 	// init core client
-	internalCoreURL := config.InternalCoreURL()
+	internalCoreURL := config.GetCoreURL()
-	jobserviceSecret := config.JobserviceSecret()
+	jobserviceSecret := config.GetAuthSecret()
 	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,
+		internalCoreURL: internalCoreURL,
-		coreClient:       coreClient,
+		coreClient:      coreClient,
-		jobserviceClient: jobserviceClient,
 	}
 }
 
 // basicClient is a default
 type basicClient struct {
-	internalCoreURL  string
+	internalCoreURL string
-	coreClient       core.Client
+	coreClient      core.Client
-	jobserviceClient cjob.Client
 }
 
 // GetCandidates gets the tag candidates under the repository
@@ -183,19 +151,3 @@ func (bc *basicClient) Delete(candidate *res.Candidate) error {
 		return fmt.Errorf("unsupported candidate kind: %s", candidate.Kind)
 	}
 }
-
-// SubmitTask to jobservice
-func (bc *basicClient) SubmitTask(taskID int64, repository *res.Repository, meta *lwp.Metadata) (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/dep/client_test.go

@@ -129,14 +129,6 @@ func (c *clientTestSuite) TestDelete() {
 	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/job.go

@@ -205,28 +205,28 @@ func logError(logger logger.Interface, err error) error {
 }
 
 func getParamRepo(params job.Parameters) (*res.Repository, error) {
-	v, ok := params[dep.ParamRepo]
+	v, ok := params[ParamRepo]
 	if !ok {
-		return nil, errors.Errorf("missing parameter: %s", dep.ParamRepo)
+		return nil, errors.Errorf("missing parameter: %s", ParamRepo)
 	}
 
 	repo, ok := v.(*res.Repository)
 	if !ok {
-		return nil, errors.Errorf("invalid parameter: %s", dep.ParamRepo)
+		return nil, errors.Errorf("invalid parameter: %s", ParamRepo)
 	}
 
 	return repo, nil
 }
 
 func getParamMeta(params job.Parameters) (*lwp.Metadata, error) {
-	v, ok := params[dep.ParamMeta]
+	v, ok := params[ParamMeta]
 	if !ok {
-		return nil, errors.Errorf("missing parameter: %s", dep.ParamMeta)
+		return nil, errors.Errorf("missing parameter: %s", ParamMeta)
 	}
 
 	meta, ok := v.(*lwp.Metadata)
 	if !ok {
-		return nil, errors.Errorf("invalid parameter: %s", dep.ParamMeta)
+		return nil, errors.Errorf("invalid parameter: %s", ParamMeta)
 	}
 
 	return meta, nil

src/pkg/retention/job_test.go

@@ -21,32 +21,21 @@ import (
 	"testing"
 	"time"
 
+	"github.com/goharbor/harbor/src/jobservice/job"
+	"github.com/goharbor/harbor/src/jobservice/logger"
+	"github.com/goharbor/harbor/src/pkg/retention/dep"
+	"github.com/goharbor/harbor/src/pkg/retention/policy"
+	"github.com/goharbor/harbor/src/pkg/retention/policy/action"
 	"github.com/goharbor/harbor/src/pkg/retention/policy/alg"
 	"github.com/goharbor/harbor/src/pkg/retention/policy/alg/or"
-	"github.com/goharbor/harbor/src/pkg/retention/res/selectors"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/goharbor/harbor/src/pkg/retention/res/selectors/doublestar"
-
-	"github.com/goharbor/harbor/src/pkg/retention/res/selectors/label"
-
-	"github.com/goharbor/harbor/src/pkg/retention/policy/action"
-	"github.com/goharbor/harbor/src/pkg/retention/policy/rule/latestk"
-
-	"github.com/goharbor/harbor/src/pkg/retention/policy"
-	"github.com/goharbor/harbor/src/pkg/retention/policy/rule"
-
-	"github.com/goharbor/harbor/src/jobservice/logger"
-
-	"github.com/goharbor/harbor/src/jobservice/job"
-
 	"github.com/goharbor/harbor/src/pkg/retention/policy/lwp"
-
+	"github.com/goharbor/harbor/src/pkg/retention/policy/rule"
+	"github.com/goharbor/harbor/src/pkg/retention/policy/rule/latestk"
 	"github.com/goharbor/harbor/src/pkg/retention/res"
-
+	"github.com/goharbor/harbor/src/pkg/retention/res/selectors"
-	"github.com/goharbor/harbor/src/pkg/retention/dep"
+	"github.com/goharbor/harbor/src/pkg/retention/res/selectors/doublestar"
-
+	"github.com/goharbor/harbor/src/pkg/retention/res/selectors/label"
+	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -87,7 +76,7 @@ func (suite *JobTestSuite) TearDownSuite() {
 
 func (suite *JobTestSuite) TestRunSuccess() {
 	params := make(job.Parameters)
-	params[dep.ParamRepo] = &res.Repository{
+	params[ParamRepo] = &res.Repository{
 		Namespace: "library",
 		Name:      "harbor",
 		Kind:      res.Image,
@@ -103,7 +92,7 @@ func (suite *JobTestSuite) TestRunSuccess() {
 	ruleParams := make(rule.Parameters)
 	ruleParams[latestk.ParameterK] = 10
 
-	params[dep.ParamMeta] = &lwp.Metadata{
+	params[ParamMeta] = &lwp.Metadata{
 		Algorithm: policy.AlgorithmOR,
 		Rules: []*rule.Metadata{
 			{

src/pkg/retention/launcher.go

@@ -17,19 +17,28 @@ package retention
 import (
 	"fmt"
 
-	"github.com/goharbor/harbor/src/pkg/retention/policy/lwp"
+	cjob "github.com/goharbor/harbor/src/common/job"
-
+	"github.com/goharbor/harbor/src/common/job/models"
 	"github.com/goharbor/harbor/src/common/utils"
 	"github.com/goharbor/harbor/src/common/utils/log"
+	"github.com/goharbor/harbor/src/core/config"
+	"github.com/goharbor/harbor/src/jobservice/job"
 	"github.com/goharbor/harbor/src/pkg/project"
 	"github.com/goharbor/harbor/src/pkg/repository"
-	"github.com/goharbor/harbor/src/pkg/retention/dep"
 	"github.com/goharbor/harbor/src/pkg/retention/policy"
+	"github.com/goharbor/harbor/src/pkg/retention/policy/lwp"
 	"github.com/goharbor/harbor/src/pkg/retention/res"
 	"github.com/goharbor/harbor/src/pkg/retention/res/selectors"
 	"github.com/pkg/errors"
 )
 
+const (
+	// ParamRepo ...
+	ParamRepo = "repository"
+	// ParamMeta ...
+	ParamMeta = "liteMeta"
+)
+
 // Launcher provides function to launch the async jobs to run retentions based on the provided policy.
 type Launcher interface {
 	// Launch async jobs for the retention policy
@@ -47,20 +56,22 @@ type Launcher interface {
 
 // NewLauncher returns an instance of Launcher
 func NewLauncher(projectMgr project.Manager, repositoryMgr repository.Manager,
-	retentionMgr Manager, retentionClient dep.Client) Launcher {
+	retentionMgr Manager) Launcher {
 	return &launcher{
-		projectMgr:      projectMgr,
+		projectMgr:       projectMgr,
-		repositoryMgr:   repositoryMgr,
+		repositoryMgr:    repositoryMgr,
-		retentionMgr:    retentionMgr,
+		retentionMgr:     retentionMgr,
-		retentionClient: retentionClient,
+		jobserviceClient: cjob.GlobalClient,
+		internalCoreURL:  config.InternalCoreURL(),
 	}
 }
 
 type launcher struct {
-	retentionMgr    Manager
+	retentionMgr     Manager
-	retentionClient dep.Client
+	projectMgr       project.Manager
-	projectMgr      project.Manager
+	repositoryMgr    repository.Manager
-	repositoryMgr   repository.Manager
+	jobserviceClient cjob.Client
+	internalCoreURL  string
 }
 
 type jobData struct {
@@ -174,7 +185,18 @@ func (l *launcher) Launch(ply *policy.Metadata, executionID int64) (int64, error
 
 	allFailed := true
 	for _, jobData := range jobDatas {
-		_, err := l.retentionClient.SubmitTask(jobData.taskID, jobData.repository, jobData.policy)
+		j := &models.JobData{
+			Metadata: &models.JobMetadata{
+				JobKind: job.KindGeneric,
+			},
+			StatusHook: fmt.Sprintf("%s/service/notifications/jobs/retention/tasks/%d", l.internalCoreURL, jobData.taskID),
+		}
+		j.Name = job.Retention
+		j.Parameters = map[string]interface{}{
+			ParamRepo: jobData.repository,
+			ParamMeta: jobData.policy,
+		}
+		_, err := l.jobserviceClient.SubmitJob(j)
 		if err != nil {
 			log.Error(launcherError(fmt.Errorf("failed to submit task %d: %v", jobData.taskID, err)))
 			continue

src/pkg/retention/launcher_test.go

@@ -16,20 +16,18 @@ package retention
 
 import (
 	"fmt"
-	"strconv"
 	"testing"
 
 	"github.com/goharbor/harbor/src/chartserver"
+	"github.com/goharbor/harbor/src/common/job"
 	"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/dep"
 	"github.com/goharbor/harbor/src/pkg/retention/policy"
-	"github.com/goharbor/harbor/src/pkg/retention/policy/lwp"
 	"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/doublestar"
+	hjob "github.com/goharbor/harbor/src/testing/job"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
@@ -76,22 +74,6 @@ func (f *fakeRepositoryManager) ListChartRepositories(projectID int64) ([]*chart
 	return f.chartRepositories, nil
 }
 
-type fakeClient struct {
-	id int
-}
-
-func (f *fakeClient) GetCandidates(repo *res.Repository) ([]*res.Candidate, error) {
-	return nil, nil
-}
-
-func (f *fakeClient) Delete(candidate *res.Candidate) error {
-	return nil
-}
-func (f *fakeClient) SubmitTask(taskID int64, repository *res.Repository, meta *lwp.Metadata) (string, error) {
-	f.id++
-	return strconv.Itoa(f.id), nil
-}
-
 type fakeRetentionManager struct{}
 
 func (f *fakeRetentionManager) CreatePolicy(p *policy.Metadata) (int64, error) {
@@ -139,10 +121,10 @@ func (f *fakeRetentionManager) ListHistories(executionID int64, query *q.Query)
 
 type launchTestSuite struct {
 	suite.Suite
-	projectMgr      project.Manager
+	projectMgr       project.Manager
-	repositoryMgr   repository.Manager
+	repositoryMgr    repository.Manager
-	retentionMgr    Manager
+	retentionMgr     Manager
-	retentionClient dep.Client
+	jobserviceClient job.Client
 }
 
 func (l *launchTestSuite) SetupTest() {
@@ -167,7 +149,7 @@ func (l *launchTestSuite) SetupTest() {
 		},
 	}
 	l.retentionMgr = &fakeRetentionManager{}
-	l.retentionClient = &fakeClient{}
+	l.jobserviceClient = &hjob.MockJobClient{}
 }
 
 func (l *launchTestSuite) TestGetProjects() {
@@ -191,7 +173,13 @@ func (l *launchTestSuite) TestGetRepositories() {
 }
 
 func (l *launchTestSuite) TestLaunch() {
-	launcher := NewLauncher(l.projectMgr, l.repositoryMgr, l.retentionMgr, l.retentionClient)
+	launcher := &launcher{
+		projectMgr:       l.projectMgr,
+		repositoryMgr:    l.repositoryMgr,
+		retentionMgr:     l.retentionMgr,
+		jobserviceClient: l.jobserviceClient,
+	}
+
 	var ply *policy.Metadata
 	// nil policy
 	n, err := launcher.Launch(ply, 1)