feat: add content for calculating task id when preheat file and handle task

api/manager/docs.go

@@ -4199,18 +4199,26 @@ const docTemplate = `{
             "type": "object",
             "properties": {
                 "actor_name": {
+                    "description": "ActorName represents the actor name, it can be the username or token name.",
                     "type": "string"
                 },
                 "actor_type": {
+                    "description": "ActorType represents the actor type, which can be one of the following:\n- UNKNOWN: Represents an unknown actor such as not authenticated.\n- USER: Represents a user.\n- PAT: Represents a personal access token.",
                     "type": "string"
                 },
                 "created_at": {
                     "type": "string"
                 },
                 "detail": {
-                    "$ref": "#/definitions/d7y_io_dragonfly_v2_manager_models.JSONMap"
+                    "description": "Detail represents the detail, leave for extension for future use.",
+                    "allOf": [
+                        {
+                            "$ref": "#/definitions/d7y_io_dragonfly_v2_manager_models.JSONMap"
+                        }
+                    ]
                 },
                 "event_type": {
+                    "description": "EventType represents the event type, indicates the type of event, API for http request,\ncan expand to other types such as SYSTEM for internal system events in future.",
                     "type": "string"
                 },
                 "id": {
@@ -4220,18 +4228,23 @@ const docTemplate = `{
                     "type": "integer"
                 },
                 "operated_at": {
+                    "description": "OperatedAt represents the operation time.",
                     "type": "string"
                 },
                 "operation": {
+                    "description": "Operation represents the operation, it will be the HTTP method for API events.",
                     "type": "string"
                 },
                 "path": {
+                    "description": "Path represents the request path, it will be the URL path for API events.",
                     "type": "string"
                 },
                 "state": {
+                    "description": "State represents the state, it indicates the state of the operation, e.g SUCCESS for API status code \u003e= 200 \u0026 \u003c 300.",
                     "type": "string"
                 },
                 "status_code": {
+                    "description": "StatusCode represents the status code, can be ignored for non-API events.",
                     "type": "integer"
                 },
                 "updated_at": {

api/manager/swagger.json

@@ -4193,18 +4193,26 @@
             "type": "object",
             "properties": {
                 "actor_name": {
+                    "description": "ActorName represents the actor name, it can be the username or token name.",
                     "type": "string"
                 },
                 "actor_type": {
+                    "description": "ActorType represents the actor type, which can be one of the following:\n- UNKNOWN: Represents an unknown actor such as not authenticated.\n- USER: Represents a user.\n- PAT: Represents a personal access token.",
                     "type": "string"
                 },
                 "created_at": {
                     "type": "string"
                 },
                 "detail": {
-                    "$ref": "#/definitions/d7y_io_dragonfly_v2_manager_models.JSONMap"
+                    "description": "Detail represents the detail, leave for extension for future use.",
+                    "allOf": [
+                        {
+                            "$ref": "#/definitions/d7y_io_dragonfly_v2_manager_models.JSONMap"
+                        }
+                    ]
                 },
                 "event_type": {
+                    "description": "EventType represents the event type, indicates the type of event, API for http request,\ncan expand to other types such as SYSTEM for internal system events in future.",
                     "type": "string"
                 },
                 "id": {
@@ -4214,18 +4222,23 @@
                     "type": "integer"
                 },
                 "operated_at": {
+                    "description": "OperatedAt represents the operation time.",
                     "type": "string"
                 },
                 "operation": {
+                    "description": "Operation represents the operation, it will be the HTTP method for API events.",
                     "type": "string"
                 },
                 "path": {
+                    "description": "Path represents the request path, it will be the URL path for API events.",
                     "type": "string"
                 },
                 "state": {
+                    "description": "State represents the state, it indicates the state of the operation, e.g SUCCESS for API status code \u003e= 200 \u0026 \u003c 300.",
                     "type": "string"
                 },
                 "status_code": {
+                    "description": "StatusCode represents the status code, can be ignored for non-API events.",
                     "type": "integer"
                 },
                 "updated_at": {

api/manager/swagger.yaml

@@ -28,28 +28,50 @@ definitions:
   d7y_io_dragonfly_v2_manager_models.Audit:
     properties:
       actor_name:
+        description: ActorName represents the actor name, it can be the username or
+          token name.
         type: string
       actor_type:
+        description: |-
+          ActorType represents the actor type, which can be one of the following:
+          - UNKNOWN: Represents an unknown actor such as not authenticated.
+          - USER: Represents a user.
+          - PAT: Represents a personal access token.
         type: string
       created_at:
         type: string
       detail:
-        $ref: '#/definitions/d7y_io_dragonfly_v2_manager_models.JSONMap'
+        allOf:
+        - $ref: '#/definitions/d7y_io_dragonfly_v2_manager_models.JSONMap'
+        description: Detail represents the detail, leave for extension for future
+          use.
       event_type:
+        description: |-
+          EventType represents the event type, indicates the type of event, API for http request,
+          can expand to other types such as SYSTEM for internal system events in future.
         type: string
       id:
         type: integer
       is_del:
         type: integer
       operated_at:
+        description: OperatedAt represents the operation time.
         type: string
       operation:
+        description: Operation represents the operation, it will be the HTTP method
+          for API events.
         type: string
       path:
+        description: Path represents the request path, it will be the URL path for
+          API events.
         type: string
       state:
+        description: State represents the state, it indicates the state of the operation,
+          e.g SUCCESS for API status code >= 200 & < 300.
         type: string
       status_code:
+        description: StatusCode represents the status code, can be ignored for non-API
+          events.
         type: integer
       updated_at:
         type: string

go.mod

@@ -4,7 +4,7 @@ go 1.23.8
 
 require (
 	cloud.google.com/go/storage v1.50.0
-	d7y.io/api/v2 v2.1.36
+	d7y.io/api/v2 v2.1.39
 	github.com/MysteriousPotato/go-lockable v1.0.0
 	github.com/RichardKnop/machinery v1.10.8
 	github.com/Showmax/go-fqdn v1.0.0
@@ -85,7 +85,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.35.0
 	go.opentelemetry.io/otel/trace v1.35.0
 	go.uber.org/atomic v1.11.0
-	go.uber.org/mock v0.5.0
+	go.uber.org/mock v0.5.1
 	go.uber.org/zap v1.27.0
 	golang.org/x/crypto v0.37.0
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56

go.sum

@@ -63,8 +63,8 @@ cloud.google.com/go/storage v1.50.0 h1:3TbVkzTooBvnZsk7WaAQfOsNrdoM8QHusXA1cpk6Q
 cloud.google.com/go/storage v1.50.0/go.mod h1:l7XeiD//vx5lfqE3RavfmU9yvk5Pp0Zhcv482poyafY=
 cloud.google.com/go/trace v1.11.2 h1:4ZmaBdL8Ng/ajrgKqY5jfvzqMXbrDcBsUGXOT9aqTtI=
 cloud.google.com/go/trace v1.11.2/go.mod h1:bn7OwXd4pd5rFuAnTrzBuoZ4ax2XQeG3qNgYmfCy0Io=
-d7y.io/api/v2 v2.1.36 h1:Gs9NNC9U9ou5QdIxrZT6hpmNwuIYVU66t1Af5RhzgXk=
-d7y.io/api/v2 v2.1.36/go.mod h1:VVOSCRsCW+r8/GnGbqAKQDC1sWLltRTT80ANHi3t1ZE=
+d7y.io/api/v2 v2.1.39 h1:BxbccbEUooXuCr8aqRr5hNQONFdL7DOWo5pz2CnBg/0=
+d7y.io/api/v2 v2.1.39/go.mod h1:5DPjID9MNDgw0mOqf49KF6Mho4x2v+C3uKmWWQYCWUs=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA=
 filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4=
@@ -1714,8 +1714,8 @@ go.uber.org/goleak v1.1.11/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ
 go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-go.uber.org/mock v0.5.0 h1:KAMbZvZPyBPWgD14IrIQ38QCyjwpvVVV6K/bHl1IwQU=
-go.uber.org/mock v0.5.0/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM=
+go.uber.org/mock v0.5.1 h1:ASgazW/qBmR+A32MYFDB6E2POoTgOwT509VP0CT/fjs=
+go.uber.org/mock v0.5.1/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=
 go.uber.org/multierr v1.5.0/go.mod h1:FeouvMocqHpRaaGuG9EjoKcStLC43Zu/fmqdUMPcKYU=

internal/job/types.go

@@ -22,19 +22,20 @@ import (
 
 // PreheatRequest defines the request parameters for preheating.
 type PreheatRequest struct {
-	URL                 string            `json:"url" validate:"required,url"`
-	PieceLength         *uint64           `json:"pieceLength" binding:"omitempty,gte=4194304"`
-	Tag                 string            `json:"tag" validate:"omitempty"`
-	FilteredQueryParams string            `json:"filtered_query_params" validate:"omitempty"`
-	Headers             map[string]string `json:"headers" validate:"omitempty"`
-	Application         string            `json:"application" validate:"omitempty"`
-	Priority            int32             `json:"priority" validate:"omitempty"`
-	Scope               string            `json:"scope" validate:"omitempty"`
-	ConcurrentCount     int64             `json:"concurrent_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"`
-	LoadToCache         bool              `json:"load_to_cache" validate:"omitempty"`
+	URL                         string            `json:"url" validate:"required,url"`
+	PieceLength                 *uint64           `json:"pieceLength" binding:"omitempty,gte=4194304"`
+	Tag                         string            `json:"tag" validate:"omitempty"`
+	FilteredQueryParams         string            `json:"filtered_query_params" validate:"omitempty"`
+	Headers                     map[string]string `json:"headers" validate:"omitempty"`
+	Application                 string            `json:"application" validate:"omitempty"`
+	Priority                    int32             `json:"priority" validate:"omitempty"`
+	Scope                       string            `json:"scope" validate:"omitempty"`
+	ConcurrentCount             int64             `json:"concurrent_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"`
+	LoadToCache                 bool              `json:"load_to_cache" validate:"omitempty"`
+	ContentForCalculatingTaskID *string           `json:"content_for_calculating_task_id" validate:"omitempty"`
 }
 
 // PreheatResponse defines the response parameters for preheating.

manager/job/preheat.go

@@ -123,17 +123,19 @@ func (p *preheat) CreatePreheat(ctx context.Context, schedulers []models.Schedul
 	case PreheatFileType:
 		files = []internaljob.PreheatRequest{
 			{
-				URL:                 json.URL,
-				PieceLength:         json.PieceLength,
-				Tag:                 json.Tag,
-				FilteredQueryParams: json.FilteredQueryParams,
-				Headers:             json.Headers,
-				Scope:               json.Scope,
-				ConcurrentCount:     json.ConcurrentCount,
-				CertificateChain:    p.certificateChain,
-				InsecureSkipVerify:  p.insecureSkipVerify,
-				Timeout:             json.Timeout,
-				LoadToCache:         json.LoadToCache,
+				URL:                         json.URL,
+				PieceLength:                 json.PieceLength,
+				Tag:                         json.Tag,
+				Application:                 json.Application,
+				FilteredQueryParams:         json.FilteredQueryParams,
+				Headers:                     json.Headers,
+				Scope:                       json.Scope,
+				ConcurrentCount:             json.ConcurrentCount,
+				CertificateChain:            p.certificateChain,
+				InsecureSkipVerify:          p.insecureSkipVerify,
+				Timeout:                     json.Timeout,
+				LoadToCache:                 json.LoadToCache,
+				ContentForCalculatingTaskID: json.ContentForCalculatingTaskID,
 			},
 		}
 	default:
@@ -349,6 +351,7 @@ func (p *preheat) parseLayers(manifests []distribution.Manifest, args types.Preh
 				URL:                 image.blobsURL(v.Digest.String()),
 				PieceLength:         args.PieceLength,
 				Tag:                 args.Tag,
+				Application:         args.Application,
 				FilteredQueryParams: args.FilteredQueryParams,
 				Headers:             nethttp.HeaderToMap(header),
 				Scope:               args.Scope,

manager/job/task.go

@@ -63,7 +63,9 @@ func (t *task) CreateGetTask(ctx context.Context, schedulers []models.Scheduler,
 
 	taskID := json.TaskID
 	if json.URL != "" {
-		taskID = idgen.TaskIDV2(json.URL, json.PieceLength, json.Tag, json.Application, idgen.ParseFilteredQueryParams(json.FilteredQueryParams))
+		taskID = idgen.TaskIDV2ByURLBased(json.URL, json.PieceLength, json.Tag, json.Application, idgen.ParseFilteredQueryParams(json.FilteredQueryParams))
+	} else if json.ContentForCalculatingTaskID != nil {
+		taskID = idgen.TaskIDV2ByContent(*json.ContentForCalculatingTaskID)
 	}
 
 	args, err := internaljob.MarshalRequest(internaljob.GetTaskRequest{
@@ -121,7 +123,9 @@ func (t *task) CreateDeleteTask(ctx context.Context, schedulers []models.Schedul
 
 	taskID := json.TaskID
 	if json.URL != "" {
-		taskID = idgen.TaskIDV2(json.URL, json.PieceLength, json.Tag, json.Application, idgen.ParseFilteredQueryParams(json.FilteredQueryParams))
+		taskID = idgen.TaskIDV2ByURLBased(json.URL, json.PieceLength, json.Tag, json.Application, idgen.ParseFilteredQueryParams(json.FilteredQueryParams))
+	} else if json.ContentForCalculatingTaskID != nil {
+		taskID = idgen.TaskIDV2ByContent(*json.ContentForCalculatingTaskID)
 	}
 
 	args, err := internaljob.MarshalRequest(internaljob.DeleteTaskRequest{

manager/types/job.go

@@ -109,7 +109,7 @@ type PreheatArgs struct {
 	Type string `json:"type" binding:"required,oneof=image file"`
 
 	// URL is the image url for preheating.
-	URL string `json:"url" binding:"required"`
+	URL string `json:"url" binding:"omitempty"`
 
 	// PieceLength is the piece length(bytes) for downloading file. The value needs to
 	// be greater than or equal to 4194304, for example: 4194304(4mib), 8388608(8mib).
@@ -120,6 +120,9 @@ type PreheatArgs struct {
 	// Tag is the tag for preheating.
 	Tag string `json:"tag" binding:"omitempty"`
 
+	// Application is the application string for preheating.
+	Application string `json:"application" binding:"omitempty"`
+
 	// FilteredQueryParams is the filtered query params for preheating.
 	FilteredQueryParams string `json:"filtered_query_params" binding:"omitempty"`
 
@@ -146,6 +149,12 @@ type PreheatArgs struct {
 
 	// LoadToCache is the flag for preheating content in cache storage, default is false.
 	LoadToCache bool `json:"load_to_cache" binding:"omitempty"`
+
+	// ContentForCalculatingTaskID is the content used to calculate the task id.
+	// If ContentForCalculatingTaskID is set, use its value to calculate the task ID.
+	// Otherwise, calculate the task ID based on url, piece_length, tag, application,
+	// and filtered_query_params. It is only used for file preheating task.
+	ContentForCalculatingTaskID *string `json:"content_for_calculating_task_id" binding:"omitempty"`
 }
 
 type CreateSyncPeersJobRequest struct {
@@ -200,6 +209,11 @@ type GetTaskArgs struct {
 
 	// FilteredQueryParams is the filtered query params of the task.
 	FilteredQueryParams string `json:"filtered_query_params" binding:"omitempty"`
+
+	// ContentForCalculatingTaskID is the content used to calculate the task id.
+	// If ContentForCalculatingTaskID is set, use its value to calculate the task ID.
+	// Otherwise, calculate the task ID based on url, piece_length, tag, application, and filtered_query_params.
+	ContentForCalculatingTaskID *string `json:"content_for_calculating_task_id" binding:"omitempty"`
 }
 
 type CreateDeleteTaskJobRequest struct {
@@ -243,4 +257,9 @@ type DeleteTaskArgs struct {
 
 	// Timeout is the timeout for deleting, default is 30 minutes.
 	Timeout time.Duration `json:"timeout" binding:"omitempty"`
+
+	// ContentForCalculatingTaskID is the content used to calculate the task id.
+	// If ContentForCalculatingTaskID is set, use its value to calculate the task ID.
+	// Otherwise, calculate the task ID based on url, piece_length, tag, application, and filtered_query_params.
+	ContentForCalculatingTaskID *string `json:"content_for_calculating_task_id" binding:"omitempty"`
 }

pkg/idgen/task_id.go

@@ -91,8 +91,8 @@ func ParseFilteredQueryParams(rawFilteredQueryParams string) []string {
 	return strings.Split(rawFilteredQueryParams, FilteredQueryParamsSeparator)
 }
 
-// TaskIDV2 generates v2 version of task id.
-func TaskIDV2(url string, pieceLength *uint64, tag, application string, filteredQueryParams []string) string {
+// TaskIDV2ByURLBased generates v2 version of task id by url based.
+func TaskIDV2ByURLBased(url string, pieceLength *uint64, tag, application string, filteredQueryParams []string) string {
 	url, err := neturl.FilterQueryParams(url, filteredQueryParams)
 	if err != nil {
 		url = ""
@@ -105,3 +105,8 @@ func TaskIDV2(url string, pieceLength *uint64, tag, application string, filtered
 
 	return pkgdigest.SHA256FromStrings(params...)
 }
+
+// TaskIDV2ByContent generates v2 version of task id by content.
+func TaskIDV2ByContent(content string) string {
+	return pkgdigest.SHA256FromStrings(content)
+}