Skip to content

Commit abb5cca

Browse files
dask-58dask-58
committed
feat: support enableTaskIDBasedBlobDigest for preheat job in Manager
Signed-off-by: dask-58 <140686560+dask-58@users.noreply.github.com>
1 parent ce173de commit abb5cca

6 files changed

Lines changed: 191 additions & 111 deletions

File tree

‎internal/job/image.go‎

Lines changed: 19 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,9 @@ type ManifestRequest struct {
165165

166166
// InsecureSkipVerify indicates whether to skip TLS verification.
167167
InsecureSkipVerify bool
168+
169+
// EnableTaskIDBasedBlobDigest indicates whether to use the blob digest for task ID calculation.
170+
EnableTaskIDBasedBlobDigest *bool
168171
}
169172

170173
// Image implements the interface for handling container images.
@@ -353,21 +356,22 @@ func buildPreheatRequestFromManifests(manifests []distribution.Manifest, req *Ma
353356
}
354357

355358
layers := &PreheatRequest{
356-
URLs: layerURLs,
357-
PieceLength: req.PieceLength,
358-
Tag: req.Tag,
359-
Application: req.Application,
360-
FilteredQueryParams: req.FilteredQueryParams,
361-
Headers: nethttp.HeaderToMap(header),
362-
Scope: req.Scope,
363-
IPs: req.IPs,
364-
Percentage: req.Percentage,
365-
Count: req.Count,
366-
ConcurrentTaskCount: req.ConcurrentTaskCount,
367-
ConcurrentPeerCount: req.ConcurrentPeerCount,
368-
CertificateChain: certificateChain,
369-
InsecureSkipVerify: req.InsecureSkipVerify,
370-
Timeout: req.Timeout,
359+
URLs: layerURLs,
360+
PieceLength: req.PieceLength,
361+
Tag: req.Tag,
362+
Application: req.Application,
363+
FilteredQueryParams: req.FilteredQueryParams,
364+
Headers: nethttp.HeaderToMap(header),
365+
Scope: req.Scope,
366+
IPs: req.IPs,
367+
Percentage: req.Percentage,
368+
Count: req.Count,
369+
ConcurrentTaskCount: req.ConcurrentTaskCount,
370+
ConcurrentPeerCount: req.ConcurrentPeerCount,
371+
CertificateChain: certificateChain,
372+
InsecureSkipVerify: req.InsecureSkipVerify,
373+
Timeout: req.Timeout,
374+
EnableTaskIDBasedBlobDigest: req.EnableTaskIDBasedBlobDigest,
371375
}
372376

373377
return []*PreheatRequest{layers}, nil

‎internal/job/types.go‎

Lines changed: 22 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -31,27 +31,28 @@ type PreheatRequest struct {
3131
URL string `json:"url" validate:"omitempty,url"`
3232

3333
// Use for V2 preheating to support multiple URLs.
34-
URLs []string `json:"urls" validate:"omitempty"`
35-
PieceLength *uint64 `json:"pieceLength" binding:"omitempty,gte=4194304"`
36-
Tag string `json:"tag" validate:"omitempty"`
37-
FilteredQueryParams string `json:"filtered_query_params" validate:"omitempty"`
38-
Headers map[string]string `json:"headers" validate:"omitempty"`
39-
Application string `json:"application" validate:"omitempty"`
40-
Priority int32 `json:"priority" validate:"omitempty"`
41-
Scope string `json:"scope" validate:"omitempty"`
42-
IPs []string `json:"ips" validate:"omitempty"`
43-
Percentage *uint32 `json:"percentage" validate:"omitempty,gte=1,lte=100"`
44-
Count *uint32 `json:"count" validate:"omitempty,gte=1,lte=200"`
45-
ConcurrentTaskCount int64 `json:"concurrent_task_count" validate:"omitempty"`
46-
ConcurrentPeerCount int64 `json:"concurrent_peer_count" validate:"omitempty"`
47-
CertificateChain [][]byte `json:"certificate_chain" validate:"omitempty"`
48-
InsecureSkipVerify bool `json:"insecure_skip_verify" validate:"omitempty"`
49-
Timeout time.Duration `json:"timeout" validate:"omitempty"`
50-
GroupUUID string `json:"group_uuid" validate:"omitempty"`
51-
TaskUUID string `json:"task_uuid" validate:"omitempty"`
52-
ObjectStorage *v2.ObjectStorage `json:"object_storage" validate:"omitempty"`
53-
Hdfs *v2.HDFS `json:"hdfs" validate:"omitempty"`
54-
OutputPath *string `json:"output_path" validate:"omitempty"`
34+
URLs []string `json:"urls" validate:"omitempty"`
35+
PieceLength *uint64 `json:"pieceLength" binding:"omitempty,gte=4194304"`
36+
Tag string `json:"tag" validate:"omitempty"`
37+
FilteredQueryParams string `json:"filtered_query_params" validate:"omitempty"`
38+
Headers map[string]string `json:"headers" validate:"omitempty"`
39+
Application string `json:"application" validate:"omitempty"`
40+
Priority int32 `json:"priority" validate:"omitempty"`
41+
Scope string `json:"scope" validate:"omitempty"`
42+
IPs []string `json:"ips" validate:"omitempty"`
43+
Percentage *uint32 `json:"percentage" validate:"omitempty,gte=1,lte=100"`
44+
Count *uint32 `json:"count" validate:"omitempty,gte=1,lte=200"`
45+
ConcurrentTaskCount int64 `json:"concurrent_task_count" validate:"omitempty"`
46+
ConcurrentPeerCount int64 `json:"concurrent_peer_count" validate:"omitempty"`
47+
CertificateChain [][]byte `json:"certificate_chain" validate:"omitempty"`
48+
InsecureSkipVerify bool `json:"insecure_skip_verify" validate:"omitempty"`
49+
Timeout time.Duration `json:"timeout" validate:"omitempty"`
50+
GroupUUID string `json:"group_uuid" validate:"omitempty"`
51+
TaskUUID string `json:"task_uuid" validate:"omitempty"`
52+
ObjectStorage *v2.ObjectStorage `json:"object_storage" validate:"omitempty"`
53+
Hdfs *v2.HDFS `json:"hdfs" validate:"omitempty"`
54+
OutputPath *string `json:"output_path" validate:"omitempty"`
55+
EnableTaskIDBasedBlobDigest *bool `json:"enable_task_id_based_blob_digest" validate:"omitempty"`
5556
}
5657

5758
// PreheatResponse defines the response parameters for preheating.

‎manager/job/preheat.go‎

Lines changed: 35 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -95,24 +95,25 @@ func (p *preheat) CreatePreheat(ctx context.Context, schedulers []models.Schedul
9595
}
9696

9797
files, err = p.internalJobImage.CreatePreheatRequestsByManifestURL(ctx, &internaljob.ManifestRequest{
98-
URL: json.URL,
99-
PieceLength: json.PieceLength,
100-
Tag: json.Tag,
101-
Application: json.Application,
102-
FilteredQueryParams: json.FilteredQueryParams,
103-
Headers: json.Headers,
104-
Username: json.Username,
105-
Password: json.Password,
106-
Platform: json.Platform,
107-
Scope: json.Scope,
108-
IPs: json.IPs,
109-
Percentage: json.Percentage,
110-
Count: json.Count,
111-
ConcurrentTaskCount: json.ConcurrentTaskCount,
112-
ConcurrentPeerCount: json.ConcurrentPeerCount,
113-
Timeout: json.Timeout,
114-
RootCAs: p.rootCAs,
115-
InsecureSkipVerify: p.insecureSkipVerify,
98+
URL: json.URL,
99+
PieceLength: json.PieceLength,
100+
Tag: json.Tag,
101+
Application: json.Application,
102+
FilteredQueryParams: json.FilteredQueryParams,
103+
Headers: json.Headers,
104+
Username: json.Username,
105+
Password: json.Password,
106+
Platform: json.Platform,
107+
Scope: json.Scope,
108+
IPs: json.IPs,
109+
Percentage: json.Percentage,
110+
Count: json.Count,
111+
ConcurrentTaskCount: json.ConcurrentTaskCount,
112+
ConcurrentPeerCount: json.ConcurrentPeerCount,
113+
Timeout: json.Timeout,
114+
RootCAs: p.rootCAs,
115+
InsecureSkipVerify: p.insecureSkipVerify,
116+
EnableTaskIDBasedBlobDigest: json.EnableTaskIDBasedBlobDigest,
116117
})
117118
if err != nil {
118119
return nil, err
@@ -133,21 +134,22 @@ func (p *preheat) CreatePreheat(ctx context.Context, schedulers []models.Schedul
133134
}
134135

135136
files = append(files, &internaljob.PreheatRequest{
136-
URLs: urls,
137-
PieceLength: json.PieceLength,
138-
Tag: json.Tag,
139-
Application: json.Application,
140-
FilteredQueryParams: json.FilteredQueryParams,
141-
Headers: json.Headers,
142-
Scope: json.Scope,
143-
IPs: json.IPs,
144-
Percentage: json.Percentage,
145-
Count: json.Count,
146-
ConcurrentTaskCount: json.ConcurrentTaskCount,
147-
ConcurrentPeerCount: json.ConcurrentPeerCount,
148-
CertificateChain: certificateChain,
149-
InsecureSkipVerify: p.insecureSkipVerify,
150-
Timeout: json.Timeout,
137+
URLs: urls,
138+
PieceLength: json.PieceLength,
139+
Tag: json.Tag,
140+
Application: json.Application,
141+
FilteredQueryParams: json.FilteredQueryParams,
142+
Headers: json.Headers,
143+
Scope: json.Scope,
144+
IPs: json.IPs,
145+
Percentage: json.Percentage,
146+
Count: json.Count,
147+
ConcurrentTaskCount: json.ConcurrentTaskCount,
148+
ConcurrentPeerCount: json.ConcurrentPeerCount,
149+
CertificateChain: certificateChain,
150+
InsecureSkipVerify: p.insecureSkipVerify,
151+
Timeout: json.Timeout,
152+
EnableTaskIDBasedBlobDigest: json.EnableTaskIDBasedBlobDigest,
151153
})
152154

153155
default:

‎manager/types/job.go‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -183,6 +183,13 @@ type PreheatArgs struct {
183183

184184
// Timeout is the timeout for preheating, default is 60 minutes.
185185
Timeout time.Duration `json:"timeout" binding:"omitempty"`
186+
187+
// EnableTaskIDBasedBlobDigest indicates whether to use the blob digest for task ID calculation
188+
// when downloading from OCI registries. When enabled for OCI blob URLs (e.g., /v2/<name>/blobs/sha256:<digest>),
189+
// the task ID is derived from the blob digest rather than the full URL. This enables deduplication across
190+
// registries - the same blob from different registries shares one task ID, eliminating redundant downloads
191+
// and storage. Default is true.
192+
EnableTaskIDBasedBlobDigest *bool `json:"enable_task_id_based_blob_digest" binding:"omitempty"`
186193
}
187194

188195
type CreateSyncPeersJobRequest struct {

‎scheduler/job/enable_task_id_based_blob_digest_test.go‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
/*
2+
* Copyright 2025 The Dragonfly Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package job
18+
19+
import (
20+
"testing"
21+
22+
"github.com/stretchr/testify/assert"
23+
)
24+
25+
func TestGetEnableTaskIDBasedBlobDigest(t *testing.T) {
26+
tests := []struct {
27+
name string
28+
enable *bool
29+
expect bool
30+
}{
31+
{
32+
name: "nil returns true (default)",
33+
enable: nil,
34+
expect: true,
35+
},
36+
{
37+
name: "true returns true",
38+
enable: func() *bool { b := true; return &b }(),
39+
expect: true,
40+
},
41+
{
42+
name: "false returns false",
43+
enable: func() *bool { b := false; return &b }(),
44+
expect: false,
45+
},
46+
}
47+
48+
for _, tc := range tests {
49+
t.Run(tc.name, func(t *testing.T) {
50+
result := getEnableTaskIDBasedBlobDigest(tc.enable)
51+
assert.Equal(t, tc.expect, result)
52+
})
53+
}
54+
}

‎scheduler/job/job.go‎

Lines changed: 54 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -375,20 +375,21 @@ func (j *job) preheatV2SingleSeedPeerByURL(ctx context.Context, url string, req
375375

376376
stream, err := client.DownloadTask(ctx, taskID, &dfdaemonv2.DownloadTaskRequest{
377377
Download: &commonv2.Download{
378-
Url: url,
379-
PieceLength: req.PieceLength,
380-
Type: commonv2.TaskType_STANDARD,
381-
Tag: &req.Tag,
382-
Application: &req.Application,
383-
Priority: commonv2.Priority(req.Priority),
384-
FilteredQueryParams: filteredQueryParams,
385-
RequestHeader: req.Headers,
386-
CertificateChain: req.CertificateChain,
387-
RemoteIp: &advertiseIP,
388-
Timeout: durationpb.New(req.Timeout),
389-
ObjectStorage: req.ObjectStorage,
390-
Hdfs: req.Hdfs,
391-
OutputPath: req.OutputPath,
378+
Url: url,
379+
PieceLength: req.PieceLength,
380+
Type: commonv2.TaskType_STANDARD,
381+
Tag: &req.Tag,
382+
Application: &req.Application,
383+
Priority: commonv2.Priority(req.Priority),
384+
FilteredQueryParams: filteredQueryParams,
385+
RequestHeader: req.Headers,
386+
CertificateChain: req.CertificateChain,
387+
RemoteIp: &advertiseIP,
388+
Timeout: durationpb.New(req.Timeout),
389+
ObjectStorage: req.ObjectStorage,
390+
Hdfs: req.Hdfs,
391+
OutputPath: req.OutputPath,
392+
EnableTaskIdBasedBlobDigest: getEnableTaskIDBasedBlobDigest(req.EnableTaskIDBasedBlobDigest),
392393
}})
393394
if err != nil {
394395
log.Errorf("[preheat]: preheat failed: %s", err.Error())
@@ -476,20 +477,21 @@ func (j *job) PreheatAllSeedPeers(ctx context.Context, req *internaljob.PreheatR
476477
ctx,
477478
taskID,
478479
&dfdaemonv2.DownloadTaskRequest{Download: &commonv2.Download{
479-
Url: url,
480-
PieceLength: req.PieceLength,
481-
Type: commonv2.TaskType_STANDARD,
482-
Tag: &req.Tag,
483-
Application: &req.Application,
484-
Priority: commonv2.Priority(req.Priority),
485-
FilteredQueryParams: filteredQueryParams,
486-
RequestHeader: req.Headers,
487-
Timeout: durationpb.New(req.Timeout),
488-
CertificateChain: req.CertificateChain,
489-
RemoteIp: &advertiseIP,
490-
ObjectStorage: req.ObjectStorage,
491-
Hdfs: req.Hdfs,
492-
OutputPath: req.OutputPath,
480+
Url: url,
481+
PieceLength: req.PieceLength,
482+
Type: commonv2.TaskType_STANDARD,
483+
Tag: &req.Tag,
484+
Application: &req.Application,
485+
Priority: commonv2.Priority(req.Priority),
486+
FilteredQueryParams: filteredQueryParams,
487+
RequestHeader: req.Headers,
488+
Timeout: durationpb.New(req.Timeout),
489+
CertificateChain: req.CertificateChain,
490+
RemoteIp: &advertiseIP,
491+
ObjectStorage: req.ObjectStorage,
492+
Hdfs: req.Hdfs,
493+
OutputPath: req.OutputPath,
494+
EnableTaskIdBasedBlobDigest: getEnableTaskIDBasedBlobDigest(req.EnableTaskIDBasedBlobDigest),
493495
}})
494496
if err != nil {
495497
log.Errorf("[preheat]: preheat failed: %s", err.Error())
@@ -696,20 +698,21 @@ func (j *job) PreheatAllPeers(ctx context.Context, req *internaljob.PreheatReque
696698
ctx,
697699
taskID,
698700
&dfdaemonv2.DownloadTaskRequest{Download: &commonv2.Download{
699-
Url: url,
700-
PieceLength: req.PieceLength,
701-
Type: commonv2.TaskType_STANDARD,
702-
Tag: &req.Tag,
703-
Application: &req.Application,
704-
Priority: commonv2.Priority(req.Priority),
705-
FilteredQueryParams: filteredQueryParams,
706-
RequestHeader: req.Headers,
707-
Timeout: durationpb.New(req.Timeout),
708-
CertificateChain: req.CertificateChain,
709-
RemoteIp: &advertiseIP,
710-
ObjectStorage: req.ObjectStorage,
711-
Hdfs: req.Hdfs,
712-
OutputPath: req.OutputPath,
701+
Url: url,
702+
PieceLength: req.PieceLength,
703+
Type: commonv2.TaskType_STANDARD,
704+
Tag: &req.Tag,
705+
Application: &req.Application,
706+
Priority: commonv2.Priority(req.Priority),
707+
FilteredQueryParams: filteredQueryParams,
708+
RequestHeader: req.Headers,
709+
Timeout: durationpb.New(req.Timeout),
710+
CertificateChain: req.CertificateChain,
711+
RemoteIp: &advertiseIP,
712+
ObjectStorage: req.ObjectStorage,
713+
Hdfs: req.Hdfs,
714+
OutputPath: req.OutputPath,
715+
EnableTaskIdBasedBlobDigest: getEnableTaskIDBasedBlobDigest(req.EnableTaskIDBasedBlobDigest),
713716
}})
714717
if err != nil {
715718
log.Errorf("[preheat]: preheat failed: %s", err.Error())
@@ -1095,3 +1098,12 @@ func (j *job) ListTaskEntries(ctx context.Context, req *internaljob.ListTaskEntr
10951098
SchedulerID: j.config.Manager.SchedulerClusterID,
10961099
}, nil
10971100
}
1101+
1102+
// getEnableTaskIDBasedBlobDigest returns the value of EnableTaskIDBasedBlobDigest.
1103+
// If nil, returns true as the default value.
1104+
func getEnableTaskIDBasedBlobDigest(enable *bool) bool {
1105+
if enable == nil {
1106+
return true
1107+
}
1108+
return *enable
1109+
}

0 commit comments

Comments
 (0)