feat: change request header in response (#3038)

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

go.mod

@@ -3,7 +3,7 @@ module d7y.io/dragonfly/v2
 go 1.21
 
 require (
-	d7y.io/api/v2 v2.0.83
+	d7y.io/api/v2 v2.0.84
 	github.com/MysteriousPotato/go-lockable v1.0.0
 	github.com/RichardKnop/machinery v1.10.6
 	github.com/Showmax/go-fqdn v1.0.0

go.sum

@@ -49,8 +49,8 @@ cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0Zeo
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
-d7y.io/api/v2 v2.0.83 h1:+XIFKmVX1bywIZJyLVzfQC8riVe+gYcuJGxWMZDKAoI=
-d7y.io/api/v2 v2.0.83/go.mod h1:nBK3pWGNkbZTI49Rt9oV4KeGtELO2m2puDgltVpIRg4=
+d7y.io/api/v2 v2.0.84 h1:V4lpcsMccllHH4/yUgWYcEdrl8XgZoijX92Tp2as0e0=
+d7y.io/api/v2 v2.0.84/go.mod h1:nBK3pWGNkbZTI49Rt9oV4KeGtELO2m2puDgltVpIRg4=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20201218220906-28db891af037/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/azure-sdk-for-go v16.2.1+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=

scheduler/scheduling/scheduling.go

@@ -627,7 +627,7 @@ func ConstructSuccessNormalTaskResponse(candidateParents []*resource.Peer) *sche
 			Tag:           &candidateParent.Task.Tag,
 			Application:   &candidateParent.Task.Application,
 			Filters:       candidateParent.Task.Filters,
-			Header:        candidateParent.Task.Header,
+			RequestHeader: candidateParent.Task.Header,
 			PieceLength:   uint32(candidateParent.Task.PieceLength),
 			ContentLength: uint64(candidateParent.Task.ContentLength.Load()),
 			PieceCount:    uint32(candidateParent.Task.TotalPieceCount.Load()),

scheduler/scheduling/scheduling_test.go

@@ -1667,7 +1667,7 @@ func TestScheduling_ConstructSuccessNormalTaskResponse(t *testing.T) {
 									Tag:           &candidateParents[0].Task.Tag,
 									Application:   &candidateParents[0].Task.Application,
 									Filters:       candidateParents[0].Task.Filters,
-									Header:        candidateParents[0].Task.Header,
+									RequestHeader: candidateParents[0].Task.Header,
 									PieceLength:   uint32(candidateParents[0].Task.PieceLength),
 									ContentLength: uint64(candidateParents[0].Task.ContentLength.Load()),
 									PieceCount:    uint32(candidateParents[0].Task.TotalPieceCount.Load()),

scheduler/service/service_v2.go

@@ -113,7 +113,7 @@ func (v *V2) AnnouncePeer(stream schedulerv2.Scheduler_AnnouncePeerServer) error
 		case *schedulerv2.AnnouncePeerRequest_RegisterPeerRequest:
 			registerPeerRequest := announcePeerRequest.RegisterPeerRequest
 			log.Infof("receive RegisterPeerRequest, url: %s, range: %#v, header: %#v, need back-to-source: %t",
-				registerPeerRequest.Download.GetUrl(), registerPeerRequest.Download.GetRange(), registerPeerRequest.Download.GetHeader(), registerPeerRequest.Download.GetNeedBackToSource())
+				registerPeerRequest.Download.GetUrl(), registerPeerRequest.Download.GetRange(), registerPeerRequest.Download.GetRequestHeader(), registerPeerRequest.Download.GetNeedBackToSource())
 			if err := v.handleRegisterPeerRequest(ctx, stream, req.GetHostId(), req.GetTaskId(), req.GetPeerId(), registerPeerRequest); err != nil {
 				log.Error(err)
 				return err
@@ -261,7 +261,7 @@ func (v *V2) StatPeer(ctx context.Context, req *schedulerv2.StatPeerRequest) (*c
 		Tag:           &peer.Task.Tag,
 		Application:   &peer.Task.Application,
 		Filters:       peer.Task.Filters,
-		Header:        peer.Task.Header,
+		RequestHeader: peer.Task.Header,
 		PieceLength:   uint32(peer.Task.PieceLength),
 		ContentLength: uint64(peer.Task.ContentLength.Load()),
 		PieceCount:    uint32(peer.Task.TotalPieceCount.Load()),
@@ -416,7 +416,7 @@ func (v *V2) StatTask(ctx context.Context, req *schedulerv2.StatTaskRequest) (*c
 		Tag:           &task.Tag,
 		Application:   &task.Application,
 		Filters:       task.Filters,
-		Header:        task.Header,
+		RequestHeader: task.Header,
 		PieceLength:   uint32(task.PieceLength),
 		ContentLength: uint64(task.ContentLength.Load()),
 		PieceCount:    uint32(task.TotalPieceCount.Load()),
@@ -1288,12 +1288,12 @@ func (v *V2) handleResource(ctx context.Context, stream schedulerv2.Scheduler_An
 		}
 
 		task = resource.NewTask(taskID, download.GetUrl(), download.GetTag(), download.GetApplication(), download.GetType(),
-			download.GetFilters(), download.GetHeader(), int32(v.config.Scheduler.BackToSourceCount), options...)
+			download.GetFilters(), download.GetRequestHeader(), int32(v.config.Scheduler.BackToSourceCount), options...)
 		v.resource.TaskManager().Store(task)
 	} else {
 		task.URL = download.GetUrl()
 		task.Filters = download.GetFilters()
-		task.Header = download.GetHeader()
+		task.Header = download.GetRequestHeader()
 	}
 
 	// Store new peer or load peer.

scheduler/service/service_v2_test.go

@@ -146,7 +146,7 @@ func TestServiceV2_StatPeer(t *testing.T) {
 						Tag:           &peer.Task.Tag,
 						Application:   &peer.Task.Application,
 						Filters:       peer.Task.Filters,
-						Header:        peer.Task.Header,
+						RequestHeader: peer.Task.Header,
 						PieceLength:   uint32(peer.Task.PieceLength),
 						ContentLength: uint64(peer.Task.ContentLength.Load()),
 						PieceCount:    uint32(peer.Task.TotalPieceCount.Load()),
@@ -372,7 +372,7 @@ func TestServiceV2_StatTask(t *testing.T) {
 					Tag:           &task.Tag,
 					Application:   &task.Application,
 					Filters:       task.Filters,
-					Header:        task.Header,
+					RequestHeader: task.Header,
 					PieceLength:   uint32(task.PieceLength),
 					ContentLength: uint64(task.ContentLength.Load()),
 					PieceCount:    uint32(task.TotalPieceCount.Load()),
@@ -2938,9 +2938,9 @@ func TestServiceV2_handleResource(t *testing.T) {
 		{
 			name: "task can be loaded",
 			download: &commonv2.Download{
-				Url:     "foo",
-				Filters: []string{"bar"},
-				Header:  map[string]string{"baz": "bas"},
+				Url:           "foo",
+				Filters:       []string{"bar"},
+				RequestHeader: map[string]string{"baz": "bas"},
 			},
 			run: func(t *testing.T, svc *V2, download *commonv2.Download, stream schedulerv2.Scheduler_AnnouncePeerServer, mockHost *resource.Host, mockTask *resource.Task, mockPeer *resource.Peer,
 				hostManager resource.HostManager, taskManager resource.TaskManager, peerManager resource.PeerManager, mr *resource.MockResourceMockRecorder, mh *resource.MockHostManagerMockRecorder,
@@ -2961,16 +2961,16 @@ func TestServiceV2_handleResource(t *testing.T) {
 				assert.Equal(task.ID, mockTask.ID)
 				assert.Equal(task.URL, download.Url)
 				assert.EqualValues(task.Filters, download.Filters)
-				assert.EqualValues(task.Header, download.Header)
+				assert.EqualValues(task.Header, download.RequestHeader)
 			},
 		},
 		{
 			name: "task can not be loaded",
 			download: &commonv2.Download{
-				Url:     "foo",
-				Filters: []string{"bar"},
-				Header:  map[string]string{"baz": "bas"},
-				Digest:  &dgst,
+				Url:           "foo",
+				Filters:       []string{"bar"},
+				RequestHeader: map[string]string{"baz": "bas"},
+				Digest:        &dgst,
 			},
 			run: func(t *testing.T, svc *V2, download *commonv2.Download, stream schedulerv2.Scheduler_AnnouncePeerServer, mockHost *resource.Host, mockTask *resource.Task, mockPeer *resource.Peer,
 				hostManager resource.HostManager, taskManager resource.TaskManager, peerManager resource.PeerManager, mr *resource.MockResourceMockRecorder, mh *resource.MockHostManagerMockRecorder,
@@ -2994,7 +2994,7 @@ func TestServiceV2_handleResource(t *testing.T) {
 				assert.Equal(task.Digest.String(), download.GetDigest())
 				assert.Equal(task.URL, download.GetUrl())
 				assert.EqualValues(task.Filters, download.GetFilters())
-				assert.EqualValues(task.Header, download.Header)
+				assert.EqualValues(task.Header, download.RequestHeader)
 			},
 		},
 		{
@@ -3020,10 +3020,10 @@ func TestServiceV2_handleResource(t *testing.T) {
 		{
 			name: "peer can be loaded",
 			download: &commonv2.Download{
-				Url:     "foo",
-				Filters: []string{"bar"},
-				Header:  map[string]string{"baz": "bas"},
-				Digest:  &dgst,
+				Url:           "foo",
+				Filters:       []string{"bar"},
+				RequestHeader: map[string]string{"baz": "bas"},
+				Digest:        &dgst,
 			},
 			run: func(t *testing.T, svc *V2, download *commonv2.Download, stream schedulerv2.Scheduler_AnnouncePeerServer, mockHost *resource.Host, mockTask *resource.Task, mockPeer *resource.Peer,
 				hostManager resource.HostManager, taskManager resource.TaskManager, peerManager resource.PeerManager, mr *resource.MockResourceMockRecorder, mh *resource.MockHostManagerMockRecorder,
@@ -3045,18 +3045,18 @@ func TestServiceV2_handleResource(t *testing.T) {
 				assert.Equal(task.Digest.String(), download.GetDigest())
 				assert.Equal(task.URL, download.GetUrl())
 				assert.EqualValues(task.Filters, download.GetFilters())
-				assert.EqualValues(task.Header, download.Header)
+				assert.EqualValues(task.Header, download.RequestHeader)
 				assert.EqualValues(peer, mockPeer)
 			},
 		},
 		{
 			name: "peer can not be loaded",
 			download: &commonv2.Download{
-				Url:      "foo",
-				Filters:  []string{"bar"},
-				Header:   map[string]string{"baz": "bas"},
-				Digest:   &dgst,
-				Priority: commonv2.Priority_LEVEL1,
+				Url:           "foo",
+				Filters:       []string{"bar"},
+				RequestHeader: map[string]string{"baz": "bas"},
+				Digest:        &dgst,
+				Priority:      commonv2.Priority_LEVEL1,
 				Range: &commonv2.Range{
 					Start:  uint64(mockPeerRange.Start),
 					Length: uint64(mockPeerRange.Length),
@@ -3084,7 +3084,7 @@ func TestServiceV2_handleResource(t *testing.T) {
 				assert.Equal(task.Digest.String(), download.GetDigest())
 				assert.Equal(task.URL, download.GetUrl())
 				assert.EqualValues(task.Filters, download.GetFilters())
-				assert.EqualValues(task.Header, download.Header)
+				assert.EqualValues(task.Header, download.RequestHeader)
 				assert.Equal(peer.ID, mockPeer.ID)
 				assert.Equal(peer.Priority, download.Priority)
 				assert.Equal(peer.Range.Start, int64(download.Range.Start))