feat: enhance preheat job concurrency control (#4233)

Signed-off-by: Gaius <gaius.qi@gmail.com>

Author: gaius-qi

.github/workflows/compatibility-e2e.yml

@@ -31,22 +31,22 @@ jobs:
         include:
           - module: manager
             image: manager
-            image-tag: v2.3.1-beta.0
+            image-tag: v2.3.1-beta.1
             chart-name: manager
             skip: "Rate Limit | preheat files in cache"
           - module: scheduler
             image: scheduler
-            image-tag: v2.3.1-beta.0
+            image-tag: v2.3.1-beta.1
             chart-name: scheduler
             skip: "Rate Limit | preheat files in cache"
           - module: client
             image: client
-            image-tag: v1.0.1
+            image-tag: v1.0.5
             chart-name: client
             skip: "Rate Limit"
           - module: seed-client
             image: client
-            image-tag: v1.0.1
+            image-tag: v1.0.5
             chart-name: seed-client
             skip: "Rate Limit"
 

internal/job/types.go

@@ -37,7 +37,8 @@ type PreheatRequest struct {
 	IPs                 []string          `json:"ips" validate:"omitempty"`
 	Percentage          *uint8            `json:"percentage" validate:"omitempty,gte=1,lte=100"`
 	Count               *uint32           `json:"count" validate:"omitempty,gte=1,lte=200"`
-	ConcurrentCount     int64             `json:"concurrent_count" validate:"omitempty"`
+	ConcurrentTaskCount int64             `json:"concurrent_task_count" validate:"omitempty"`
+	ConcurrentPeerCount int64             `json:"concurrent_peer_count" validate:"omitempty"`
 	CertificateChain    [][]byte          `json:"certificate_chain" validate:"omitempty"`
 	InsecureSkipVerify  bool              `json:"insecure_skip_verify" validate:"omitempty"`
 	Timeout             time.Duration     `json:"timeout" validate:"omitempty"`

manager/job/preheat.go

@@ -149,7 +149,8 @@ func (p *preheat) CreatePreheat(ctx context.Context, schedulers []models.Schedul
 			IPs:                 json.IPs,
 			Percentage:          json.Percentage,
 			Count:               json.Count,
-			ConcurrentCount:     json.ConcurrentCount,
+			ConcurrentTaskCount: json.ConcurrentTaskCount,
+			ConcurrentPeerCount: json.ConcurrentPeerCount,
 			CertificateChain:    certificateChain,
 			InsecureSkipVerify:  p.insecureSkipVerify,
 			Timeout:             json.Timeout,
@@ -390,7 +391,8 @@ func parseLayers(manifests []distribution.Manifest, args types.PreheatArgs, head
 		IPs:                 args.IPs,
 		Percentage:          args.Percentage,
 		Count:               args.Count,
-		ConcurrentCount:     args.ConcurrentCount,
+		ConcurrentTaskCount: args.ConcurrentTaskCount,
+		ConcurrentPeerCount: args.ConcurrentPeerCount,
 		CertificateChain:    certificateChain,
 		InsecureSkipVerify:  insecureSkipVerify,
 		Timeout:             args.Timeout,

manager/service/job.go

@@ -50,9 +50,6 @@ const (
 
 	// DefaultGCJobPollingInterval is the default interval for polling GC job.
 	DefaultGCJobPollingInterval = 5 * time.Second
-
-	// DefaultGetImageDistributionJobConcurrentCount is the default concurrent count for getting image distribution job.
-	DefaultGetImageDistributionJobConcurrentCount = 12
 )
 
 func (s *service) CreateGCJob(ctx context.Context, json types.CreateGCJobRequest) (*models.Job, error) {
@@ -119,8 +116,12 @@ func (s *service) CreatePreheatJob(ctx context.Context, json types.CreatePreheat
 		json.Args.Scope = types.SingleSeedPeerScope
 	}
 
-	if json.Args.ConcurrentCount == 0 {
-		json.Args.ConcurrentCount = types.DefaultPreheatConcurrentCount
+	if json.Args.ConcurrentTaskCount == 0 {
+		json.Args.ConcurrentTaskCount = types.DefaultPreheatConcurrentTaskCount
+	}
+
+	if json.Args.ConcurrentPeerCount == 0 {
+		json.Args.ConcurrentPeerCount = types.DefaultPreheatConcurrentPeerCount
 	}
 
 	if json.Args.Timeout == 0 {
@@ -222,6 +223,10 @@ func (s *service) CreateGetTaskJob(ctx context.Context, json types.CreateGetTask
 }
 
 func (s *service) CreateGetImageDistributionJob(ctx context.Context, json types.CreateGetImageDistributionJobRequest) (*types.CreateGetImageDistributionJobResponse, error) {
+	if json.Args.ConcurrentLayerCount == 0 {
+		json.Args.ConcurrentLayerCount = types.DefaultPreheatConcurrentLayerCount
+	}
+
 	imageLayers, err := s.getImageLayers(ctx, json)
 	if err != nil {
 		err = fmt.Errorf("get image layers failed: %w", err)
@@ -312,7 +317,7 @@ func (s *service) createGetTaskJobsSync(ctx context.Context, layers []internaljo
 	var mu sync.Mutex
 	jobs := make([]*models.Job, 0, len(layers))
 	eg, ctx := errgroup.WithContext(ctx)
-	eg.SetLimit(DefaultGetImageDistributionJobConcurrentCount)
+	eg.SetLimit(int(json.Args.ConcurrentLayerCount))
 	for _, file := range layers {
 		eg.Go(func() error {
 			job, err := s.createGetTaskJobSync(ctx, types.CreateGetTaskJobRequest{

manager/types/job.go

@@ -30,8 +30,14 @@ const (
 )
 
 const (
-	// DefaultPreheatConcurrentCount is the default concurrent count for preheating all peers.
-	DefaultPreheatConcurrentCount = 1000
+	// DefaultPreheatConcurrentPeerCount is the default concurrent peer count for preheating all peers.
+	DefaultPreheatConcurrentPeerCount = 500
+
+	// DefaultPreheatConcurrentTaskCount is the default concurrent task count for preheating all peers.
+	DefaultPreheatConcurrentTaskCount = 8
+
+	// DefaultPreheatConcurrentLayerCount is the default concurrent layer count for getting image distribution.
+	DefaultPreheatConcurrentLayerCount = 8
 
 	// DefaultJobTimeout is the default timeout for executing job.
 	DefaultJobTimeout = 60 * time.Minute
@@ -161,8 +167,16 @@ type PreheatArgs struct {
 	// Applies to 'all_peers' and 'all_seed_peers' scopes.
 	Count *uint32 `json:"count" binding:"omitempty,gte=1,lte=200"`
 
-	// BatchSize is the batch size for preheating all peers, default is 50.
-	ConcurrentCount int64 `json:"concurrent_count" binding:"omitempty,gte=1,lte=500"`
+	// ConcurrentTaskCount specifies the maximum number of tasks (e.g., image layers) to preheat concurrently.
+	// For example, if preheating 100 layers with ConcurrentTaskCount set to 10, up to 10 layers are processed simultaneously.
+	// If ConcurrentPeerCount is 10 for 1000 peers, each layer is preheated by 10 peers concurrently.
+	// Default is 8, maximum is 100.
+	ConcurrentTaskCount int64 `json:"concurrent_task_count" binding:"omitempty,gte=1,lte=100"`
+
+	// ConcurrentPeerCount specifies the maximum number of peers to preheat concurrently for a single task (e.g., an image layer).
+	// For example, if preheating a layer with ConcurrentPeerCount set to 10, up to 10 peers process that layer simultaneously.
+	// Default is 500, maximum is 1000.
+	ConcurrentPeerCount int64 `json:"concurrent_peer_count" binding:"omitempty,gte=1,lte=1000"`
 
 	// Timeout is the timeout for preheating, default is 30 minutes.
 	Timeout time.Duration `json:"timeout" binding:"omitempty"`
@@ -277,6 +291,9 @@ type GetImageDistributionArgs struct {
 
 	// The image type preheating task can specify the image architecture type. eg: linux/amd64.
 	Platform string `json:"platform" binding:"omitempty"`
+
+	// ConcurrentLayerCount specifies the maximum number of layers to get concurrently.
+	ConcurrentLayerCount int64 `json:"concurrent_layer_count" binding:"omitempty,gte=1,lte=100"`
 }
 
 // CreateGetImageDistributionJobResponse is the response for creating a get image job.