feat: extend the p2p preheat policy

Makefile

@@ -468,7 +468,7 @@ misspell:
 	@echo checking misspell...
 	@find . -type d \( -path ./tests \) -prune -o -name '*.go' -print | xargs misspell -error
 
-# golangci-lint binary installation or refer to https://golangci-lint.run/usage/install/#local-installation 
+# golangci-lint binary installation or refer to https://golangci-lint.run/usage/install/#local-installation
 # curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v1.55.2
 GOLANGCI_LINT := $(shell go env GOPATH)/bin/golangci-lint
 lint:

api/v2.0/swagger.yaml

@@ -1193,7 +1193,7 @@ paths:
         '404':
           $ref: '#/responses/404'
         '422':
-          $ref: '#/responses/422'  
+          $ref: '#/responses/422'
         '500':
           $ref: '#/responses/500'
   /projects/{project_name}/repositories/{repository_name}/artifacts/{reference}/scan/stop:
@@ -1226,7 +1226,7 @@ paths:
         '404':
           $ref: '#/responses/404'
         '422':
-          $ref: '#/responses/422'  
+          $ref: '#/responses/422'
         '500':
           $ref: '#/responses/500'
   /projects/{project_name}/repositories/{repository_name}/artifacts/{reference}/scan/{report_id}/log:
@@ -6553,7 +6553,7 @@ responses:
         description: The ID of the corresponding request for the response
         type: string
     schema:
-      $ref: '#/definitions/Errors'    
+      $ref: '#/definitions/Errors'
   '500':
     description: Internal server error
     headers:
@@ -7095,9 +7095,9 @@ definitions:
         type: boolean
         description: Whether the preheat policy enabled
         x-omitempty: false
-      scope:
+      extra_attrs:
         type: string
-        description: The scope of preheat policy
+        description: The extra attributes of preheat policy
       creation_time:
         type: string
         format: date-time
@@ -7937,7 +7937,7 @@ definitions:
     properties:
       resource:
         type: string
-        description: The resource of the access. Possible resources are listed here for system and project level https://github.com/goharbor/harbor/blob/main/src/common/rbac/const.go 
+        description: The resource of the access. Possible resources are listed here for system and project level https://github.com/goharbor/harbor/blob/main/src/common/rbac/const.go
       action:
         type: string
         description: The action of the access. Possible actions are *, pull, push, create, read, update, delete, list, operate, scanner-pull and stop.
@@ -10112,4 +10112,4 @@ definitions:
       scan_type:
         type: string
         description: 'The scan type for the scan request. Two options are currently supported, vulnerability and sbom'
-        enum: [ vulnerability, sbom ]
+        enum: [ vulnerability, sbom ]

make/migrations/postgresql/0160_2.13.0_schema.up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE p2p_preheat_policy DROP COLUMN IF EXISTS scope;
+ALTER TABLE p2p_preheat_policy ADD COLUMN IF NOT EXISTS extra_attrs text;

make/photon/Makefile

@@ -246,4 +246,3 @@ cleanimage:
 
 .PHONY: clean
 clean: cleanimage
-

src/controller/p2p/preheat/enforcer.go

@@ -402,7 +402,7 @@ func (de *defaultEnforcer) launchExecutions(ctx context.Context, candidates []*s
 	// Start tasks
 	count := 0
 	for _, c := range candidates {
-		if _, err = de.startTask(ctx, eid, c, insData, pl.Scope); err != nil {
+		if _, err = de.startTask(ctx, eid, c, insData, pl.ExtraAttrs); err != nil {
 			// Just log the error and skip
 			log.Errorf("start task error for preheating image: %s/%s:%s@%s", c.Namespace, c.Repository, c.Tags[0], c.Digest)
 			continue
@@ -421,7 +421,7 @@ func (de *defaultEnforcer) launchExecutions(ctx context.Context, candidates []*s
 }
 
 // startTask starts the preheat task(job) for the given candidate
-func (de *defaultEnforcer) startTask(ctx context.Context, executionID int64, candidate *selector.Candidate, instance, scope string) (int64, error) {
+func (de *defaultEnforcer) startTask(ctx context.Context, executionID int64, candidate *selector.Candidate, instance string, extraAttrs map[string]interface{}) (int64, error) {
 	u, err := de.fullURLGetter(candidate)
 	if err != nil {
 		return -1, err
@@ -438,10 +438,10 @@ func (de *defaultEnforcer) startTask(ctx context.Context, executionID int64, can
 		Headers: map[string]interface{}{
 			accessCredHeaderKey: cred,
 		},
-		ImageName: fmt.Sprintf("%s/%s", candidate.Namespace, candidate.Repository),
-		Tag:       candidate.Tags[0],
-		Digest:    candidate.Digest,
-		Scope:     scope,
+		ImageName:  fmt.Sprintf("%s/%s", candidate.Namespace, candidate.Repository),
+		Tag:        candidate.Tags[0],
+		Digest:     candidate.Digest,
+		ExtraAttrs: extraAttrs,
 	}
 
 	piData, err := pi.ToJSON()

src/controller/p2p/preheat/enforcer_test.go

@@ -210,7 +210,6 @@ func mockPolicies() []*po.Schema {
 				Type: po.TriggerTypeManual,
 			},
 			Enabled:     true,
-			Scope:       "single_peer",
 			CreatedAt:   time.Now().UTC(),
 			UpdatedTime: time.Now().UTC(),
 		}, {
@@ -236,7 +235,6 @@ func mockPolicies() []*po.Schema {
 			Trigger: &po.Trigger{
 				Type: po.TriggerTypeEventBased,
 			},
-			Scope:       "all_peers",
 			Enabled:     true,
 			CreatedAt:   time.Now().UTC(),
 			UpdatedTime: time.Now().UTC(),

src/pkg/p2p/preheat/models/policy/policy.go

@@ -30,9 +30,6 @@ func init() {
 	beego_orm.RegisterModel(&Schema{})
 }
 
-// ScopeType represents the preheat scope type.
-type ScopeType = string
-
 const (
 	// Filters:
 	//  Repository : type=Repository value=name text (double star pattern used)
@@ -58,11 +55,6 @@ const (
 	TriggerTypeScheduled TriggerType = "scheduled"
 	// TriggerTypeEventBased represents the event_based trigger type
 	TriggerTypeEventBased TriggerType = "event_based"
-
-	// ScopeTypeSinglePeer represents preheat image to single peer in p2p cluster.
-	ScopeTypeSinglePeer ScopeType = "single_peer"
-	// ScopeTypeAllPeers represents preheat image to all peers in p2p cluster.
-	ScopeTypeAllPeers ScopeType = "all_peers"
 )
 
 // Schema defines p2p preheat policy schema
@@ -80,10 +72,11 @@ type Schema struct {
 	// Use JSON data format (query by trigger type should be supported)
 	TriggerStr string `orm:"column(trigger)" json:"-"`
 	Enabled    bool   `orm:"column(enabled)" json:"enabled"`
-	// Scope decides the preheat scope.
-	Scope       string    `orm:"column(scope)" json:"scope"`
-	CreatedAt   time.Time `orm:"column(creation_time)" json:"creation_time"`
-	UpdatedTime time.Time `orm:"column(update_time)" json:"update_time"`
+	// ExtraAttrs is used to store extra attributes provided by vendor.
+	ExtraAttrsStr string                 `orm:"column(extra_attrs)" json:"-"`
+	ExtraAttrs    map[string]interface{} `orm:"-" json:"extra_attrs"`
+	CreatedAt     time.Time              `orm:"column(creation_time)" json:"creation_time"`
+	UpdatedTime   time.Time              `orm:"column(update_time)" json:"update_time"`
 }
 
 // TableName specifies the policy schema table name.
@@ -136,11 +129,6 @@ func (s *Schema) ValidatePreheatPolicy() error {
 		}
 	}
 
-	// validate preheat scope
-	if s.Scope != "" && s.Scope != ScopeTypeSinglePeer && s.Scope != ScopeTypeAllPeers {
-		return errors.New(nil).WithCode(errors.BadRequestCode).WithMessagef("invalid scope for preheat policy: %s", s.Scope)
-	}
-
 	return nil
 }
 
@@ -162,6 +150,14 @@ func (s *Schema) Encode() error {
 		s.TriggerStr = string(triggerStr)
 	}
 
+	if s.ExtraAttrs != nil {
+		extraAttrsStr, err := json.Marshal(s.ExtraAttrs)
+		if err != nil {
+			return err
+		}
+		s.ExtraAttrsStr = string(extraAttrsStr)
+	}
+
 	return nil
 }
 
@@ -181,6 +177,13 @@ func (s *Schema) Decode() error {
 	}
 	s.Trigger = trigger
 
+	// parse extra attributes
+	extraAttrs, err := decodeExtraAttrs(s.ExtraAttrsStr)
+	if err != nil {
+		return err
+	}
+	s.ExtraAttrs = extraAttrs
+
 	return nil
 }
 
@@ -230,3 +233,17 @@ func decodeTrigger(triggerStr string) (*Trigger, error) {
 
 	return trigger, nil
 }
+
+// decodeExtraAttrs parse extraAttrsStr to extraAttrs.
+func decodeExtraAttrs(extraAttrsStr string) (map[string]interface{}, error) {
+	if len(extraAttrsStr) == 0 {
+		return nil, nil
+	}
+
+	extraAttrs := make(map[string]interface{})
+	if err := json.Unmarshal([]byte(extraAttrsStr), &extraAttrs); err != nil {
+		return nil, err
+	}
+
+	return extraAttrs, nil
+}

src/pkg/p2p/preheat/models/policy/policy_test.go

@@ -64,35 +64,28 @@ func (p *PolicyTestSuite) TestValidatePreheatPolicy() {
 	// valid cron string
 	p.schema.Trigger.Settings.Cron = "0 0 0 1 1 *"
 	p.NoError(p.schema.ValidatePreheatPolicy())
-
-	// invalid preheat scope
-	p.schema.Scope = "invalid scope"
-	p.Error(p.schema.ValidatePreheatPolicy())
-	// valid preheat scope
-	p.schema.Scope = "single_peer"
-	p.NoError(p.schema.ValidatePreheatPolicy())
 }
 
 // TestDecode tests decode.
 func (p *PolicyTestSuite) TestDecode() {
 	s := &Schema{
-		ID:          100,
-		Name:        "test-for-decode",
-		Description: "",
-		ProjectID:   1,
-		ProviderID:  1,
-		Filters:     nil,
-		FiltersStr:  "[{\"type\":\"repository\",\"value\":\"**\"},{\"type\":\"tag\",\"value\":\"**\"},{\"type\":\"label\",\"value\":\"test\"}]",
-		Trigger:     nil,
-		TriggerStr:  "{\"type\":\"event_based\",\"trigger_setting\":{\"cron\":\"\"}}",
-		Enabled:     false,
-		Scope:       "all_peers",
+		ID:            100,
+		Name:          "test-for-decode",
+		Description:   "",
+		ProjectID:     1,
+		ProviderID:    1,
+		Filters:       nil,
+		FiltersStr:    "[{\"type\":\"repository\",\"value\":\"**\"},{\"type\":\"tag\",\"value\":\"**\"},{\"type\":\"label\",\"value\":\"test\"}]",
+		Trigger:       nil,
+		TriggerStr:    "{\"type\":\"event_based\",\"trigger_setting\":{\"cron\":\"\"}}",
+		Enabled:       false,
+		ExtraAttrsStr: "{\"key\":\"value\"}",
 	}
 	p.NoError(s.Decode())
 	p.Len(s.Filters, 3)
 	p.NotNil(s.Trigger)
 
-	p.Equal(ScopeTypeAllPeers, s.Scope)
+	p.Equal(map[string]interface{}{"key": "value"}, s.ExtraAttrs)
 
 	// invalid filter or trigger
 	s.FiltersStr = ""
@@ -132,10 +125,12 @@ func (p *PolicyTestSuite) TestEncode() {
 		},
 		TriggerStr: "",
 		Enabled:    false,
-		Scope:      "single_peer",
+		ExtraAttrs: map[string]interface{}{
+			"key": "value",
+		},
 	}
 	p.NoError(s.Encode())
 	p.Equal(`[{"type":"repository","value":"**"},{"type":"tag","value":"**"},{"type":"label","value":"test"}]`, s.FiltersStr)
 	p.Equal(`{"type":"event_based","trigger_setting":{}}`, s.TriggerStr)
-	p.Equal(ScopeTypeSinglePeer, s.Scope)
+	p.Equal(`{"key":"value"}`, s.ExtraAttrsStr)
 }

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

@@ -36,6 +36,15 @@ const (
 
 	// dragonflyJobPath is the job path for dragonfly openapi.
 	dragonflyJobPath = "/oapi/v1/jobs"
+
+	// scopeTypeSingleSeedPeer represents preheat image to single seed peer in p2p cluster.
+	scopeTypeSingleSeedPeer = "single_seed_peer"
+
+	// scopeTypeAllSeedPeers represents preheat image to all seed peers in p2p cluster.
+	scopeTypeAllSeedPeers = "all_seed_peers"
+
+	// scopeTypeAllPeers represents preheat image to all peers in p2p cluster.
+	scopeTypeAllPeers = "all_peers"
 )
 
 const (
@@ -54,13 +63,13 @@ const (
 
 type dragonflyCreateJobRequest struct {
 	// Type is the job type, support preheat.
-	Type string `json:"type" binding:"required"`
+	Type string `json:"type"`
 
 	// Args is the preheating args.
-	Args dragonflyCreateJobRequestArgs `json:"args" binding:"omitempty"`
+	Args dragonflyCreateJobRequestArgs `json:"args"`
 
 	// SchedulerClusterIDs is the scheduler cluster ids for preheating.
-	SchedulerClusterIDs []uint `json:"scheduler_cluster_ids" binding:"omitempty"`
+	SchedulerClusterIDs []uint `json:"scheduler_cluster_ids"`
 }
 
 type dragonflyCreateJobRequestArgs struct {
@@ -150,6 +159,14 @@ type dragonflyJobResponse struct {
 	} `json:"result"`
 }
 
+// dragonflyExtraAttrs is the extra attributes model definition for dragonfly provider.
+type dragonflyExtraAttrs struct {
+	// Scope is the scope for preheating, default behavior is single_peer.
+	Scope string `json:"scope"`
+	// ClusterIDs is the cluster ids for dragonfly provider.
+	ClusterIDs []uint `json:"cluster_ids"`
+}
+
 // DragonflyDriver implements the provider driver interface for Alibaba dragonfly.
 // More details, please refer to https://github.com/alibaba/Dragonfly
 type DragonflyDriver struct {
@@ -201,6 +218,18 @@ func (dd *DragonflyDriver) Preheat(preheatingImage *PreheatImage) (*PreheatingSt
 		return nil, errors.New("no image specified")
 	}
 
+	var extraAttrs dragonflyExtraAttrs
+	if len(preheatingImage.ExtraAttrs) > 0 {
+		extraAttrsStr, err := json.Marshal(preheatingImage.ExtraAttrs)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to marshal extra attributes")
+		}
+
+		if err := json.Unmarshal(extraAttrsStr, &extraAttrs); err != nil {
+			return nil, errors.Wrap(err, "failed to unmarshal extra attributes")
+		}
+	}
+
 	// Construct the preheat job request by the given parameters of the preheating image .
 	req := &dragonflyCreateJobRequest{
 		Type: "preheat",
@@ -209,10 +238,23 @@ func (dd *DragonflyDriver) Preheat(preheatingImage *PreheatImage) (*PreheatingSt
 			Type:    preheatingImage.Type,
 			URL:     preheatingImage.URL,
 			Headers: headerToMapString(preheatingImage.Headers),
-			Scope:   preheatingImage.Scope,
 		},
 	}
 
+	// Set the scope if it is specified.
+	if extraAttrs.Scope != "" {
+		if err := isScopeValid(extraAttrs.Scope); err != nil {
+			return nil, errors.Wrap(err, "specify the invalid scope")
+		}
+
+		req.Args.Scope = extraAttrs.Scope
+	}
+
+	// Set the cluster ids if it is specified.
+	if len(extraAttrs.ClusterIDs) > 0 {
+		req.SchedulerClusterIDs = extraAttrs.ClusterIDs
+	}
+
 	url := fmt.Sprintf("%s%s", strings.TrimSuffix(dd.instance.Endpoint, "/"), dragonflyJobPath)
 	data, err := client.GetHTTPClient(dd.instance.Insecure).Post(url, dd.getCred(), req, nil)
 	if err != nil {
@@ -326,3 +368,15 @@ func headerToMapString(header map[string]interface{}) map[string]string {
 
 	return m
 }
+
+// isScopeValid checks whether the scope is valid.
+func isScopeValid(scope string) error {
+	switch scope {
+	case scopeTypeSingleSeedPeer,
+		scopeTypeAllSeedPeers,
+		scopeTypeAllPeers:
+		return nil
+	default:
+		return errors.Errorf("invalid scope: %s", scope)
+	}
+}