Merge pull request #12378 from steven-zou/feat/preheat_policy_enforcement

feat(preheat):implement policy enforcer

src/controller/p2p/preheat/enforcer.go

@@ -15,21 +15,483 @@
 package preheat
 
 import (
-	"github.com/goharbor/harbor/src/pkg/artifact"
-	"github.com/goharbor/harbor/src/pkg/p2p/preheat/models/policy"
+	"context"
+	"fmt"
+	"strings"
+
+	tk "github.com/docker/distribution/registry/auth/token"
+	"github.com/goharbor/harbor/src/common/models"
+	"github.com/goharbor/harbor/src/controller/artifact"
+	"github.com/goharbor/harbor/src/controller/project"
+	"github.com/goharbor/harbor/src/controller/scan"
+	"github.com/goharbor/harbor/src/controller/tag"
+	"github.com/goharbor/harbor/src/core/config"
+	"github.com/goharbor/harbor/src/core/service/token"
+	"github.com/goharbor/harbor/src/jobservice/job"
+	"github.com/goharbor/harbor/src/jobservice/logger"
+	"github.com/goharbor/harbor/src/lib/errors"
+	"github.com/goharbor/harbor/src/lib/q"
+	"github.com/goharbor/harbor/src/lib/selector"
+	"github.com/goharbor/harbor/src/pkg/p2p/preheat"
+	pol "github.com/goharbor/harbor/src/pkg/p2p/preheat/models/policy"
+	"github.com/goharbor/harbor/src/pkg/p2p/preheat/models/provider"
+	"github.com/goharbor/harbor/src/pkg/p2p/preheat/policy"
+	pr "github.com/goharbor/harbor/src/pkg/p2p/preheat/provider"
+	"github.com/goharbor/harbor/src/pkg/scan/report"
+	v1 "github.com/goharbor/harbor/src/pkg/scan/rest/v1"
+	"github.com/goharbor/harbor/src/pkg/scan/vuln"
+	"github.com/goharbor/harbor/src/pkg/task"
 )
 
-// Enforcer defines policy enforcement operations.
+const (
+	defaultSeverityCode     = 99
+	extraAttrTotal          = "totalCount"
+	extraAttrTrigger        = "trigger"
+	extraAttrTriggerSetting = "triggerSetting"
+	extraAttrArtifact       = "artifact"
+	extraAttrDigest         = "digest"
+	extraAttrKind           = "kind"
+	resourceActionType      = "repository"
+	resourcePullAction      = "pull"
+	manifestAPIPattern      = "%s/v2/%s/manifests/%s"
+	accessCredHeaderKey     = "Authorization"
+)
+
+// Enforcer defines preheat policy enforcement operations.
 type Enforcer interface {
-	// Enforce the specified policy.
+	// Enforce preheating action by the given policy.
+	// For manual and scheduled preheating scenarios.
 	//
 	// Arguments:
-	//   p *policy.Schema : the being enforced policy
-	//   art ...*artifact.Artifact （optional）: the relevant artifact referred by the happening events
-	//   that defined in the event-based policy p.
+	//   ctx context.Context : system context
+	//   policyID int64 : ID of the being enforced policy
 	//
 	// Returns:
 	//   - ID of the execution
 	//   - non-nil error if any error occurred during the enforcement
-	Enforce(p *policy.Schema, art ...*artifact.Artifact) (int64, error)
+	EnforcePolicy(ctx context.Context, policyID int64) (int64, error)
+
+	// Enforce preheating action by the given artifact.
+	// For event-based cases.
+	// Using the given artifact to located the matched preheat policy and bound this action
+	// with the located preheat policy.
+	//
+	// Arguments:
+	//   ctx context.Context : system context
+	//   art *artifact.Artifact: Artifact contained in the occurred events.
+	//
+	// Returns:
+	//   - IDs of the executions
+	//   - non-nil error if any error occurred during the enforcement
+	//
+	// Notes:
+	//  The current design is artifact central mode (identified by digest). All the tags of
+	//  the artifact are kept together. However, the preheating action is based on the specified
+	//  tag and we need to split the all-tags-in-one artifact to one-tag artifacts here.
+	PreheatArtifact(ctx context.Context, art *artifact.Artifact) ([]int64, error)
+}
+
+// extURLGetter is a func template to get the external access endpoint
+// The purpose of defining such a func template is decoupling code
+type extURLGetter func(c *selector.Candidate) (string, error)
+
+// accessCredMaker is a func template to generate the required credential header value
+// The purpose of defining such a func template is decoupling code
+type accessCredMaker func(c *selector.Candidate) (string, error)
+
+// defaultEnforcer is default implementation of Enforcer.
+type defaultEnforcer struct {
+	// for policy management
+	policyMgr policy.Manager
+	// for talking to job service to launch tasks
+	executionMgr task.ExecutionManager
+	taskMgr      task.Manager
+	// for retrieving the artifact candidates (including tags and labels)
+	artCtl artifact.Controller
+	// for getting vulnerability severity of the specified artifact
+	scanCtl scan.Controller
+	// for getting project related info
+	proCtl project.Controller
+	// TODO: Need preheat provider manager
+	//
+	// for getting the access endpoint of registry V2
+	fullURLGetter extURLGetter
+	// for creating the access credential
+	credMaker accessCredMaker
+}
+
+// NewEnforcer creat a new enforcer
+func NewEnforcer() Enforcer {
+	return &defaultEnforcer{
+		policyMgr:    policy.New(),
+		executionMgr: task.NewExecutionManager(),
+		taskMgr:      task.NewManager(),
+		artCtl:       artifact.NewController(),
+		scanCtl:      scan.DefaultController,
+		proCtl:       project.NewController(),
+		fullURLGetter: func(c *selector.Candidate) (s string, e error) {
+			edp, err := config.ExtEndpoint()
+			if err != nil {
+				return "", err
+			}
+
+			r := fmt.Sprintf("%s/%s", c.Namespace, c.Repository)
+			return fmt.Sprintf(manifestAPIPattern, edp, r, c.Tags[0]), nil
+		},
+		credMaker: func(c *selector.Candidate) (s string, e error) {
+			r := fmt.Sprintf("%s/%s", c.Namespace, c.Repository)
+
+			ac := []*tk.ResourceActions{
+				{
+					Type: resourceActionType,
+					Name: r,
+					// Only pull action is enough
+					Actions: []string{resourcePullAction},
+				},
+			}
+			t, err := token.MakeToken("distributor", token.Registry, ac)
+			if err != nil {
+				return "", err
+			}
+
+			return fmt.Sprintf("Bearer %s", t.Token), nil
+		},
+	}
+}
+
+// EnforcePolicy enforces preheating action by the given policy
+func (de *defaultEnforcer) EnforcePolicy(ctx context.Context, policyID int64) (int64, error) {
+	// Get the the given policy data
+	pl, err := de.policyMgr.Get(ctx, policyID)
+	if err != nil {
+		return -1, enforceError(err)
+	}
+
+	// Check if policy is enabled
+	if !pl.Enabled {
+		return -1, enforceError(errors.Errorf("policy %d:%s is not enabled", pl.ID, pl.Name))
+	}
+
+	// Retrieve the initial candidates
+	candidates, err := de.getCandidates(ctx, pl)
+	if err != nil {
+		return -1, enforceError(err)
+	}
+
+	// Do filters
+	filtered, err := policy.NewFilter().
+		BuildFrom(pl).
+		Filter(candidates)
+	if err != nil {
+		return -1, enforceError(err)
+	}
+
+	// Launch execution
+	eid, err := de.launchExecutions(ctx, filtered, pl)
+	if err != nil {
+		// NOTES: Please pay attention here, even the non-nil error returned, it does not mean
+		// the relevant execution is not available. The execution ID should also be checked(>0)
+		// at any time.
+		return eid, enforceError(err)
+	}
+
+	return eid, nil
+}
+
+// PreheatArtifact enforces preheating action by the given artifact.
+func (de *defaultEnforcer) PreheatArtifact(ctx context.Context, art *artifact.Artifact) ([]int64, error) {
+	if art == nil {
+		return nil, errors.New("nil artifact")
+	}
+
+	// Get project info
+	p, err := de.proCtl.Get(ctx, art.ProjectID, project.CVEAllowlist(true))
+	if err != nil {
+		return nil, enforceErrorExt(err, art)
+	}
+
+	// Convert to candidates
+	candidates, err := de.toCandidates(ctx, p, []*artifact.Artifact{art})
+	if err != nil {
+		return nil, enforceErrorExt(err, art)
+	}
+
+	// Find all the policies that match the given artifact
+	_, l, err := de.policyMgr.ListPoliciesByProject(ctx, art.ProjectID, nil)
+	if err != nil {
+		return nil, enforceErrorExt(err, art)
+	}
+
+	matched := 0
+	ids := make([]int64, 0)
+	for _, pl := range l {
+		// Skip disabled policies
+		if !pl.Enabled {
+			continue
+		}
+
+		// Only look for the event-based policies
+		if pl.Trigger == nil ||
+			pl.Trigger.Type != pol.TriggerTypeEventBased {
+			// Skip
+			continue
+		}
+
+		filtered, err := policy.NewFilter().BuildFrom(pl).Filter(candidates)
+		if err != nil {
+			// Log error and continue
+			logger.Errorf("Failed to do filter for policy %d:%s with error: %s", pl.ID, pl.Name, err.Error())
+			continue
+		}
+
+		matched++
+
+		if len(filtered) > 0 {
+			// Matched
+			eid, err := de.launchExecutions(ctx, filtered, pl)
+			if err != nil {
+				// Log error and continue
+				logger.Errorf("Failed to launch execution for policy %d:%s with error: %s", pl.ID, pl.Name, err.Error())
+			} else {
+				// Success and then append the execution id to list
+				ids = append(ids, eid)
+			}
+		}
+	}
+
+	if matched != len(ids) {
+		// Some policy enforcement are failed
+		// Treat it as an error case
+		return ids, enforceErrorExt(errors.Errorf("%d policies matched but only %d successfully enforced", matched, len(ids)), art)
+	}
+
+	return ids, nil
+}
+
+// getCandidates get the initial candidates by evaluating the policy
+func (de *defaultEnforcer) getCandidates(ctx context.Context, ps *pol.Schema) ([]*selector.Candidate, error) {
+	// Get project info
+	p, err := de.proCtl.Get(ctx, ps.ProjectID, project.CVEAllowlist(true))
+	if err != nil {
+		return nil, err
+	}
+
+	// Get the initial candidates
+	// Here we have a hidden filter, the artifact type filter.
+	// Only get the image type at this moment.
+	arts, err := de.artCtl.List(ctx, &q.Query{
+		Keywords: map[string]interface{}{
+			"project_id": ps.ProjectID,
+			"type":       pr.SupportedType,
+		},
+	}, &artifact.Option{
+		WithLabel: true,
+		WithTag:   true,
+		TagOption: &tag.Option{
+			WithSignature: true,
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return de.toCandidates(ctx, p, arts)
+}
+
+// launchExecutions create execution record and launch tasks to preheat the filtered artifacts.
+func (de *defaultEnforcer) launchExecutions(ctx context.Context, candidates []*selector.Candidate, pl *pol.Schema) (int64, error) {
+	// Create execution first anyway
+	attrs := map[string]interface{}{
+		extraAttrTotal:          len(candidates),
+		extraAttrTrigger:        pl.Trigger.Type,
+		extraAttrTriggerSetting: pl.Trigger.Settings.Cron,
+	}
+	if pl.Trigger.Type != pol.TriggerTypeScheduled {
+		attrs[extraAttrTriggerSetting] = "-"
+	}
+
+	eid, err := de.executionMgr.Create(ctx, job.P2PPreheat, pl.ID, pl.Trigger.Type, attrs)
+	if err != nil {
+		return -1, err
+	}
+
+	// Handle empty candidate list case
+	if len(candidates) == 0 {
+		// Return earlier
+		if err := de.executionMgr.MarkDone(ctx, eid, "no artifacts to preheat"); err != nil {
+			return eid, err
+		}
+
+		return eid, nil
+	}
+
+	// TODO: Get provider instance by the provider ID
+	// Placeholder
+	ins := &provider.Instance{}
+	insData, err := ins.ToJSON()
+	if err != nil {
+		// In case
+		if er := de.executionMgr.MarkError(ctx, eid, err.Error()); er != nil {
+			return eid, errors.Wrap(er, err.Error())
+		}
+
+		return eid, nil
+	}
+
+	// Start tasks
+	count := 0
+	for _, c := range candidates {
+		if _, err = de.startTask(ctx, eid, c, insData); err != nil {
+			// Just log the error and skip
+			logger.Errorf("start task error for preheating image: %s/%s:%s@%s", c.Namespace, c.Repository, c.Tags[0], c.Digest)
+			continue
+		}
+
+		count++
+	}
+
+	if count != len(candidates) {
+		// Obviously, failed to start some tasks
+		// Return as an error but the execution can still be queried.
+		return eid, errors.Errorf("some errors occurred when enforcing policy '%s(%d)' but execution '%d' is still available", pl.Name, pl.ID, eid)
+	}
+
+	return eid, nil
+}
+
+// startTask starts the preheat task(job) for the given candidate
+func (de *defaultEnforcer) startTask(ctx context.Context, executionID int64, candidate *selector.Candidate, instance string) (int64, error) {
+	u, err := de.fullURLGetter(candidate)
+	if err != nil {
+		return -1, err
+	}
+
+	cred, err := de.credMaker(candidate)
+	if err != nil {
+		return -1, err
+	}
+
+	pi := &pr.PreheatImage{
+		Type: pr.SupportedType,
+		URL:  u,
+		Headers: map[string]interface{}{
+			accessCredHeaderKey: cred,
+		},
+		ImageName: fmt.Sprintf("%s/%s", candidate.Namespace, candidate.Repository),
+		Tag:       candidate.Tags[0],
+	}
+
+	piData, err := pi.ToJSON()
+	if err != nil {
+		return -1, err
+	}
+
+	j := &task.Job{
+		Name: job.P2PPreheat,
+		Parameters: job.Parameters{
+			preheat.PreheatParamProvider: instance,
+			preheat.PreheatParamImage:    piData,
+		},
+		Metadata: &job.Metadata{
+			JobKind:  job.KindGeneric,
+			IsUnique: true,
+		},
+	}
+
+	tid, err := de.taskMgr.Create(ctx, executionID, j, map[string]interface{}{
+		extraAttrArtifact: fmt.Sprintf("%s:%s", pi.ImageName, pi.Tag),
+		extraAttrDigest:   candidate.Digest,
+		extraAttrKind:     pi.Type,
+	})
+	if err != nil {
+		return -1, err
+	}
+
+	return tid, nil
+}
+
+// getVulnerabilitySev gets the severity code value for the given artifact with allowlist option set
+func (de *defaultEnforcer) getVulnerabilitySev(ctx context.Context, p *models.Project, art *artifact.Artifact) (uint, error) {
+	al := report.CVESet(p.CVEAllowlist.CVESet())
+	r, err := de.scanCtl.GetSummary(ctx, art, []string{v1.MimeTypeNativeReport}, report.WithCVEAllowlist(&al))
+	if err != nil {
+		if errors.IsNotFoundErr(err) {
+			// no vulnerability report
+			return defaultSeverityCode, nil
+		}
+
+		return defaultSeverityCode, errors.Wrap(err, "get vulnerability severity")
+	}
+
+	// Severity is based on the native report format.
+	// In case no supported report format, treat as same to the no report scenario
+	sum, ok := r[v1.MimeTypeNativeReport]
+	if !ok {
+		return defaultSeverityCode, nil
+	}
+
+	sm, ok := sum.(*vuln.NativeReportSummary)
+	if !ok {
+		return defaultSeverityCode, errors.New("malformed native summary report")
+	}
+
+	return (uint)(sm.Severity.Code()), nil
+}
+
+// toCandidates converts the artifacts to filtering candidates
+func (de *defaultEnforcer) toCandidates(ctx context.Context, p *models.Project, arts []*artifact.Artifact) ([]*selector.Candidate, error) {
+	// Convert to filtering candidates first
+	candidates := make([]*selector.Candidate, 0)
+
+	for _, a := range arts {
+		// Vulnerability severity is property of artifact
+		sev, err := de.getVulnerabilitySev(ctx, p, a)
+		if err != nil {
+			return nil, err
+		}
+
+		// If artifact has more than one tag, then split them into separate candidate for easy filtering.
+		for _, t := range a.Tags {
+			candidates = append(candidates, &selector.Candidate{
+				NamespaceID: p.ProjectID,
+				Namespace:   p.Name,
+				Repository:  pureRepository(p.Name, a.RepositoryName),
+				Kind:        pr.SupportedType,
+				Digest:      a.Digest,
+				Tags:        []string{t.Name},
+				Labels:      getLabels(a.Labels),
+				Signatures: map[string]bool{
+					t.Name: t.Signed,
+				},
+				VulnerabilitySeverity: sev,
+			})
+		}
+	}
+
+	return candidates, nil
+}
+
+// enforceError is a wrap error
+func enforceError(e error) error {
+	return errors.Wrap(e, "enforce policy error")
+}
+
+// enforceErrorExt is an enhanced wrap error
+func enforceErrorExt(e error, art *artifact.Artifact) error {
+	return errors.Wrap(e, fmt.Sprintf("enforce policy for given artifact error: %s@%s", art.RepositoryName, art.Digest))
+}
+
+// pureRepository removes project name from the repository
+func pureRepository(ns, r string) string {
+	return strings.TrimPrefix(r, fmt.Sprintf("%s/", ns))
+}
+
+// getLabels gets label texts from the label objects
+func getLabels(labels []*models.Label) []string {
+	lt := make([]string, 0)
+	for _, l := range labels {
+		lt = append(lt, l.Name)
+	}
+
+	return lt
 }

src/controller/p2p/preheat/enforcer_test.go

@@ -0,0 +1,284 @@
+// 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 preheat
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/goharbor/harbor/src/common/models"
+	car "github.com/goharbor/harbor/src/controller/artifact"
+	"github.com/goharbor/harbor/src/controller/tag"
+	"github.com/goharbor/harbor/src/lib/selector"
+	ar "github.com/goharbor/harbor/src/pkg/artifact"
+	po "github.com/goharbor/harbor/src/pkg/p2p/preheat/models/policy"
+	v1 "github.com/goharbor/harbor/src/pkg/scan/rest/v1"
+	"github.com/goharbor/harbor/src/pkg/scan/vuln"
+	ta "github.com/goharbor/harbor/src/pkg/tag/model/tag"
+	"github.com/goharbor/harbor/src/testing/controller/artifact"
+	"github.com/goharbor/harbor/src/testing/controller/project"
+	"github.com/goharbor/harbor/src/testing/controller/scan"
+	"github.com/goharbor/harbor/src/testing/mock"
+	"github.com/goharbor/harbor/src/testing/pkg/p2p/preheat/policy"
+	"github.com/goharbor/harbor/src/testing/pkg/task"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+)
+
+// EnforcerTestSuite is a test suite of testing preheat enforcer
+type EnforcerTestSuite struct {
+	suite.Suite
+
+	enforcer *defaultEnforcer
+}
+
+// TestEnforcer is an entry method of running EnforcerTestSuite
+func TestEnforcer(t *testing.T) {
+	suite.Run(t, &EnforcerTestSuite{})
+}
+
+// SetupSuite prepares env for running EnforcerTestSuite
+func (suite *EnforcerTestSuite) SetupSuite() {
+	fakePolicies := mockPolicies()
+	fakePolicyManager := &policy.FakeManager{}
+	fakePolicyManager.On("Get",
+		context.TODO(),
+		mock.AnythingOfType("int64")).
+		Return(fakePolicies[0], nil)
+	fakePolicyManager.On("ListPoliciesByProject",
+		context.TODO(),
+		mock.AnythingOfType("int64"),
+		mock.AnythingOfType("*q.Query"),
+	).Return((int64)(2), fakePolicies, nil)
+
+	fakeExecManager := &task.FakeExecutionManager{}
+	fakeExecManager.On("Create",
+		context.TODO(),
+		mock.AnythingOfType("string"),
+		mock.AnythingOfType("int64"),
+		mock.AnythingOfType("string"),
+		mock.AnythingOfType("map[string]interface {}"),
+	).Return(time.Now().Unix(), nil)
+
+	fakeTaskManager := &task.FakeManager{}
+	fakeTaskManager.On("Create",
+		context.TODO(),
+		mock.AnythingOfType("int64"),
+		mock.AnythingOfType("*task.Job"),
+		mock.AnythingOfType("map[string]interface {}"),
+	).Return(time.Now().Unix(), nil)
+
+	fakeArtCtl := &artifact.Controller{}
+	fakeArtCtl.On("List",
+		context.TODO(),
+		mock.AnythingOfType("*q.Query"),
+		mock.AnythingOfType("*artifact.Option"),
+	).Return(mockArtifacts(), nil)
+
+	fakeScanCtl := &scan.Controller{}
+	fakeScanCtl.On("GetSummary",
+		context.TODO(),
+		mock.AnythingOfType("*artifact.Artifact"),
+		[]string{v1.MimeTypeNativeReport},
+		mock.AnythingOfType("report.Option"),
+	).Return(mockVulnerabilitySummary(), nil)
+
+	fakeProCtl := &project.Controller{}
+	fakeProCtl.On("Get",
+		context.TODO(),
+		(int64)(1),
+		mock.Anything,
+	).Return(&models.Project{
+		ProjectID:    1,
+		Name:         "library",
+		CVEAllowlist: models.CVEAllowlist{},
+	}, nil)
+
+	suite.enforcer = &defaultEnforcer{
+		policyMgr:    fakePolicyManager,
+		executionMgr: fakeExecManager,
+		taskMgr:      fakeTaskManager,
+		artCtl:       fakeArtCtl,
+		scanCtl:      fakeScanCtl,
+		proCtl:       fakeProCtl,
+		fullURLGetter: func(c *selector.Candidate) (s string, e error) {
+			r := fmt.Sprintf("%s/%s", c.Namespace, c.Repository)
+			return fmt.Sprintf(manifestAPIPattern, "https://testing.harbor.com", r, c.Tags[0]), nil
+		},
+		credMaker: func(c *selector.Candidate) (s string, e error) {
+			return "fake-token", nil
+		},
+	}
+
+}
+
+// TestEnforcePolicy tests the policy enforcement case.
+func (suite *EnforcerTestSuite) TestEnforcePolicy() {
+	eid, err := suite.enforcer.EnforcePolicy(context.TODO(), 1)
+	require.NoError(suite.T(), err, "enforce policy")
+	suite.Condition(func() (success bool) {
+		return eid > 0
+	}, "execution created")
+}
+
+// TestPreheatArtifact tests the artifact preheating case
+func (suite *EnforcerTestSuite) TestPreheatArtifact() {
+	ids, err := suite.enforcer.PreheatArtifact(context.TODO(), mockArtifacts()[1])
+	require.NoError(suite.T(), err, "preheat given artifact")
+	suite.Equal(1, len(ids), "executions created")
+}
+
+// mock policies for reusing
+func mockPolicies() []*po.Schema {
+	return []*po.Schema{
+		{
+			ID:          1,
+			Name:        "manual_policy",
+			Description: "for testing",
+			ProjectID:   1,
+			ProviderID:  1,
+			Filters: []*po.Filter{
+				{
+					Type:  po.FilterTypeRepository,
+					Value: "sub/**",
+				},
+				{
+					Type:  po.FilterTypeTag,
+					Value: "prod*",
+				},
+				{
+					Type:  po.FilterTypeLabel,
+					Value: "approved,ready",
+				},
+				{
+					Type:  po.FilterTypeSignature,
+					Value: true,
+				},
+				{
+					Type:  po.FilterTypeVulnerability,
+					Value: 3, // medium
+				},
+			},
+			Trigger: &po.Trigger{
+				Type: po.TriggerTypeManual,
+			},
+			Enabled:     true,
+			CreatedAt:   time.Now().UTC(),
+			UpdatedTime: time.Now().UTC(),
+		}, {
+			ID:          2,
+			Name:        "event_based_policy",
+			Description: "for testing",
+			ProjectID:   1,
+			ProviderID:  1,
+			Filters: []*po.Filter{
+				{
+					Type:  po.FilterTypeRepository,
+					Value: "busy*",
+				},
+				{
+					Type:  po.FilterTypeTag,
+					Value: "stage*",
+				},
+				{
+					Type:  po.FilterTypeLabel,
+					Value: "staged",
+				},
+			},
+			Trigger: &po.Trigger{
+				Type: po.TriggerTypeEventBased,
+			},
+			Enabled:     true,
+			CreatedAt:   time.Now().UTC(),
+			UpdatedTime: time.Now().UTC(),
+		},
+	}
+}
+
+// mock artifacts
+func mockArtifacts() []*car.Artifact {
+	// Skip all the unused properties
+	return []*car.Artifact{
+		{
+			Artifact: ar.Artifact{
+				ID:             1,
+				Type:           "image",
+				ProjectID:      1,
+				RepositoryName: "library/sub/busybox",
+				Digest:         "sha256@fake1",
+			},
+			Tags: []*tag.Tag{
+				{
+					Tag: ta.Tag{
+						Name: "prod",
+					},
+					Signed: true,
+				}, {
+					Tag: ta.Tag{
+						Name: "stage",
+					},
+					Signed: false,
+				},
+			},
+			Labels: []*models.Label{
+				{
+					Name: "approved",
+				}, {
+					Name: "ready",
+				},
+			},
+		}, {
+			Artifact: ar.Artifact{
+				ID:             2,
+				Type:           "image",
+				ProjectID:      1,
+				RepositoryName: "library/busybox",
+				Digest:         "sha256@fake2",
+			},
+			Tags: []*tag.Tag{
+				{
+					Tag: ta.Tag{
+						Name: "latest",
+					},
+					Signed: true,
+				}, {
+					Tag: ta.Tag{
+						Name: "stage",
+					},
+					Signed: false,
+				},
+			},
+			Labels: []*models.Label{
+				{
+					Name: "approved",
+				}, {
+					Name: "staged",
+				},
+			},
+		},
+	}
+}
+
+// mock vulnerability summary
+func mockVulnerabilitySummary() map[string]interface{} {
+	// skip all unused properties
+	return map[string]interface{}{
+		v1.MimeTypeNativeReport: &vuln.NativeReportSummary{
+			Severity: vuln.Low,
+		},
+	}
+}

src/jobservice/job/known_jobs.go

@@ -36,4 +36,6 @@ const (
 	SlackJob = "SLACK"
 	// Retention : the name of the retention job
 	Retention = "RETENTION"
+	// P2PPreheat : the name of the P2P preheat job
+	P2PPreheat = "P2P_PREHEAT"
 )

src/jobservice/runtime/bootstrap.go

@@ -23,6 +23,8 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/goharbor/harbor/src/pkg/p2p/preheat"
+
 	"github.com/goharbor/harbor/src/jobservice/api"
 	"github.com/goharbor/harbor/src/jobservice/common/utils"
 	"github.com/goharbor/harbor/src/jobservice/config"
@@ -262,6 +264,7 @@ func (bs *Bootstrap) loadAndRunRedisWorkerPool(
 			scheduler.JobNameScheduler: (*scheduler.PeriodicJob)(nil),
 			job.WebhookJob:             (*notification.WebhookJob)(nil),
 			job.SlackJob:               (*notification.SlackJob)(nil),
+			job.P2PPreheat:             (*preheat.Job)(nil),
 		}); err != nil {
 		// exit
 		return nil, err

src/pkg/p2p/preheat/job.go

@@ -25,10 +25,10 @@ import (
 )
 
 const (
-	// parameter keeps the preheating provider instance info.
-	preheatParamProvider = "provider"
-	// parameter keeps the preheating artifact (image) info.
-	preheatParamImage = "image"
+	// PreheatParamProvider is a parameter keeping the preheating provider instance info.
+	PreheatParamProvider = "provider"
+	// PreheatParamImage is a parameter keeping the preheating artifact (image) info.
+	PreheatParamImage = "image"
 	// checkInterval indicates the interval of loop check.
 	checkInterval = 10 * time.Second
 	// checkTimeout indicates the overall timeout of the loop check.
@@ -181,7 +181,7 @@ func preheatJobRunningError(err error) error {
 
 // parseParamProvider parses the provider param.
 func parseParamProvider(params job.Parameters) (*provider.Instance, error) {
-	data, err := parseStrValue(params, preheatParamProvider)
+	data, err := parseStrValue(params, PreheatParamProvider)
 	if err != nil {
 		return nil, err
 	}
@@ -209,7 +209,7 @@ func parseParamProvider(params job.Parameters) (*provider.Instance, error) {
 
 // parseParamImage parses the preheating image param.
 func parseParamImage(params job.Parameters) (*pr.PreheatImage, error) {
-	data, err := parseStrValue(params, preheatParamImage)
+	data, err := parseStrValue(params, PreheatParamImage)
 	if err != nil {
 		return nil, err
 	}

src/pkg/p2p/preheat/job_test.go

@@ -101,12 +101,12 @@ func (suite *JobTestSuite) validateJob(j job.Interface, params job.Parameters) {
 func (suite *JobTestSuite) runJob(ins *p.Instance) {
 	params := make(job.Parameters)
 	data, err := ins.ToJSON()
-	require.NoError(suite.T(), err, "encode parameter", preheatParamProvider)
-	params[preheatParamProvider] = data
+	require.NoError(suite.T(), err, "encode parameter", PreheatParamProvider)
+	params[PreheatParamProvider] = data
 
 	data, err = suite.preheatingImage.ToJSON()
-	require.NoError(suite.T(), err, "encode parameter", preheatParamImage)
-	params[preheatParamImage] = data
+	require.NoError(suite.T(), err, "encode parameter", PreheatParamImage)
+	params[PreheatParamImage] = data
 
 	j := &Job{}
 	suite.validateJob(j, params)

src/pkg/p2p/preheat/provider/preheat_image.go

@@ -42,6 +42,9 @@ type PreheatImage struct {
 
 	// The tag
 	Tag string `json:"tag,omitempty"`
+
+	// Digest of the preheating image
+	Digest string `json:"digest"`
 }
 
 // FromJSON build preheating image from the given data.

src/testing/pkg/p2p/preheat/policy/manager.go

@@ -0,0 +1,156 @@
+// Code generated by mockery v2.0.3. DO NOT EDIT.
+
+package policy
+
+import (
+	context "context"
+
+	policy "github.com/goharbor/harbor/src/pkg/p2p/preheat/models/policy"
+	mock "github.com/stretchr/testify/mock"
+
+	q "github.com/goharbor/harbor/src/lib/q"
+)
+
+// FakeManager is an autogenerated mock type for the Manager type
+type FakeManager struct {
+	mock.Mock
+}
+
+// Create provides a mock function with given fields: ctx, schema
+func (_m *FakeManager) Create(ctx context.Context, schema *policy.Schema) (int64, error) {
+	ret := _m.Called(ctx, schema)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, *policy.Schema) int64); ok {
+		r0 = rf(ctx, schema)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *policy.Schema) error); ok {
+		r1 = rf(ctx, schema)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Delete provides a mock function with given fields: ctx, id
+func (_m *FakeManager) Delete(ctx context.Context, id int64) error {
+	ret := _m.Called(ctx, id)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
+		r0 = rf(ctx, id)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Get provides a mock function with given fields: ctx, id
+func (_m *FakeManager) Get(ctx context.Context, id int64) (*policy.Schema, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 *policy.Schema
+	if rf, ok := ret.Get(0).(func(context.Context, int64) *policy.Schema); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*policy.Schema)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// ListPolicies provides a mock function with given fields: ctx, query
+func (_m *FakeManager) ListPolicies(ctx context.Context, query *q.Query) (int64, []*policy.Schema, error) {
+	ret := _m.Called(ctx, query)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) int64); ok {
+		r0 = rf(ctx, query)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 []*policy.Schema
+	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) []*policy.Schema); ok {
+		r1 = rf(ctx, query)
+	} else {
+		if ret.Get(1) != nil {
+			r1 = ret.Get(1).([]*policy.Schema)
+		}
+	}
+
+	var r2 error
+	if rf, ok := ret.Get(2).(func(context.Context, *q.Query) error); ok {
+		r2 = rf(ctx, query)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
+}
+
+// ListPoliciesByProject provides a mock function with given fields: ctx, project, query
+func (_m *FakeManager) ListPoliciesByProject(ctx context.Context, project int64, query *q.Query) (int64, []*policy.Schema, error) {
+	ret := _m.Called(ctx, project, query)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, int64, *q.Query) int64); ok {
+		r0 = rf(ctx, project, query)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 []*policy.Schema
+	if rf, ok := ret.Get(1).(func(context.Context, int64, *q.Query) []*policy.Schema); ok {
+		r1 = rf(ctx, project, query)
+	} else {
+		if ret.Get(1) != nil {
+			r1 = ret.Get(1).([]*policy.Schema)
+		}
+	}
+
+	var r2 error
+	if rf, ok := ret.Get(2).(func(context.Context, int64, *q.Query) error); ok {
+		r2 = rf(ctx, project, query)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
+}
+
+// Update provides a mock function with given fields: ctx, schema, props
+func (_m *FakeManager) Update(ctx context.Context, schema *policy.Schema, props ...string) error {
+	_va := make([]interface{}, len(props))
+	for _i := range props {
+		_va[_i] = props[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, ctx, schema)
+	_ca = append(_ca, _va...)
+	ret := _m.Called(_ca...)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, *policy.Schema, ...string) error); ok {
+		r0 = rf(ctx, schema, props...)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}

src/testing/pkg/task/execution_manager.go

@@ -0,0 +1,147 @@
+// Code generated by mockery v2.0.3. DO NOT EDIT.
+
+package task
+
+import (
+	context "context"
+
+	q "github.com/goharbor/harbor/src/lib/q"
+	mock "github.com/stretchr/testify/mock"
+
+	task "github.com/goharbor/harbor/src/pkg/task"
+)
+
+// FakeExecutionManager is an autogenerated mock type for the ExecutionManager type
+type FakeExecutionManager struct {
+	mock.Mock
+}
+
+// Create provides a mock function with given fields: ctx, vendorType, vendorID, trigger, extraAttrs
+func (_m *FakeExecutionManager) Create(ctx context.Context, vendorType string, vendorID int64, trigger string, extraAttrs ...map[string]interface{}) (int64, error) {
+	_va := make([]interface{}, len(extraAttrs))
+	for _i := range extraAttrs {
+		_va[_i] = extraAttrs[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, ctx, vendorType, vendorID, trigger)
+	_ca = append(_ca, _va...)
+	ret := _m.Called(_ca...)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, string, int64, string, ...map[string]interface{}) int64); ok {
+		r0 = rf(ctx, vendorType, vendorID, trigger, extraAttrs...)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string, int64, string, ...map[string]interface{}) error); ok {
+		r1 = rf(ctx, vendorType, vendorID, trigger, extraAttrs...)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Delete provides a mock function with given fields: ctx, id
+func (_m *FakeExecutionManager) Delete(ctx context.Context, id int64) error {
+	ret := _m.Called(ctx, id)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
+		r0 = rf(ctx, id)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Get provides a mock function with given fields: ctx, id
+func (_m *FakeExecutionManager) Get(ctx context.Context, id int64) (*task.Execution, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 *task.Execution
+	if rf, ok := ret.Get(0).(func(context.Context, int64) *task.Execution); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*task.Execution)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// List provides a mock function with given fields: ctx, query
+func (_m *FakeExecutionManager) List(ctx context.Context, query *q.Query) ([]*task.Execution, error) {
+	ret := _m.Called(ctx, query)
+
+	var r0 []*task.Execution
+	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) []*task.Execution); ok {
+		r0 = rf(ctx, query)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*task.Execution)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) error); ok {
+		r1 = rf(ctx, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// MarkDone provides a mock function with given fields: ctx, id, message
+func (_m *FakeExecutionManager) MarkDone(ctx context.Context, id int64, message string) error {
+	ret := _m.Called(ctx, id, message)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, int64, string) error); ok {
+		r0 = rf(ctx, id, message)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// MarkError provides a mock function with given fields: ctx, id, message
+func (_m *FakeExecutionManager) MarkError(ctx context.Context, id int64, message string) error {
+	ret := _m.Called(ctx, id, message)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, int64, string) error); ok {
+		r0 = rf(ctx, id, message)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Stop provides a mock function with given fields: ctx, id
+func (_m *FakeExecutionManager) Stop(ctx context.Context, id int64) error {
+	ret := _m.Called(ctx, id)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
+		r0 = rf(ctx, id)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}

src/testing/pkg/task/manager.go

@@ -0,0 +1,128 @@
+// Code generated by mockery v2.0.3. DO NOT EDIT.
+
+package task
+
+import (
+	context "context"
+
+	q "github.com/goharbor/harbor/src/lib/q"
+	mock "github.com/stretchr/testify/mock"
+
+	task "github.com/goharbor/harbor/src/pkg/task"
+)
+
+// FakeManager is an autogenerated mock type for the Manager type
+type FakeManager struct {
+	mock.Mock
+}
+
+// Create provides a mock function with given fields: ctx, executionID, job, extraAttrs
+func (_m *FakeManager) Create(ctx context.Context, executionID int64, job *task.Job, extraAttrs ...map[string]interface{}) (int64, error) {
+	_va := make([]interface{}, len(extraAttrs))
+	for _i := range extraAttrs {
+		_va[_i] = extraAttrs[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, ctx, executionID, job)
+	_ca = append(_ca, _va...)
+	ret := _m.Called(_ca...)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, int64, *task.Job, ...map[string]interface{}) int64); ok {
+		r0 = rf(ctx, executionID, job, extraAttrs...)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64, *task.Job, ...map[string]interface{}) error); ok {
+		r1 = rf(ctx, executionID, job, extraAttrs...)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Get provides a mock function with given fields: ctx, id
+func (_m *FakeManager) Get(ctx context.Context, id int64) (*task.Task, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 *task.Task
+	if rf, ok := ret.Get(0).(func(context.Context, int64) *task.Task); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*task.Task)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetLog provides a mock function with given fields: ctx, id
+func (_m *FakeManager) GetLog(ctx context.Context, id int64) ([]byte, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 []byte
+	if rf, ok := ret.Get(0).(func(context.Context, int64) []byte); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]byte)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// List provides a mock function with given fields: ctx, query
+func (_m *FakeManager) List(ctx context.Context, query *q.Query) ([]*task.Task, error) {
+	ret := _m.Called(ctx, query)
+
+	var r0 []*task.Task
+	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) []*task.Task); ok {
+		r0 = rf(ctx, query)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*task.Task)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) error); ok {
+		r1 = rf(ctx, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Stop provides a mock function with given fields: ctx, id
+func (_m *FakeManager) Stop(ctx context.Context, id int64) error {
+	ret := _m.Called(ctx, id)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
+		r0 = rf(ctx, id)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}