feat: add RemoteIp for calling client (#4231)

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

Author: gaius-qi

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.44
+	d7y.io/api/v2 v2.1.47
 	github.com/MysteriousPotato/go-lockable v1.0.0
 	github.com/Showmax/go-fqdn v1.0.0
 	github.com/VividCortex/mysqlerr v1.0.0

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.44 h1:h762BncKXXKBd+LgU58wzFgNwLpqPxW7wrgcZJNcV+4=
-d7y.io/api/v2 v2.1.44/go.mod h1:IbhylQWRkqRka+oUl73Fzz331fHFIAwS2m4cMNpFWdk=
+d7y.io/api/v2 v2.1.47 h1:HpH8iPpaVmIP5c01yaJzMM1HytkRttij91SU8/+b4u0=
+d7y.io/api/v2 v2.1.47/go.mod h1:IbhylQWRkqRka+oUl73Fzz331fHFIAwS2m4cMNpFWdk=
 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=

scheduler/job/job.go

@@ -329,6 +329,7 @@ func (j *job) preheatV2SingleSeedPeer(ctx context.Context, req *internaljob.Preh
 func (j *job) preheatV2SingleSeedPeerByURL(ctx context.Context, url string, req *internaljob.PreheatRequest, log *logger.SugaredLoggerOnWith) (*internaljob.PreheatResponse, error) {
 	taskID := idgen.TaskIDV2ByURLBased(url, req.PieceLength, req.Tag, req.Application, strings.Split(req.FilteredQueryParams, idgen.FilteredQueryParamsSeparator))
 	filteredQueryParams := strings.Split(req.FilteredQueryParams, idgen.FilteredQueryParamsSeparator)
+	advertiseIP := j.config.Server.AdvertiseIP.String()
 	stream, err := j.resource.SeedPeer().Client().DownloadTask(ctx, taskID, &dfdaemonv2.DownloadTaskRequest{
 		Download: &commonv2.Download{
 			Url:                 url,
@@ -341,6 +342,7 @@ func (j *job) preheatV2SingleSeedPeerByURL(ctx context.Context, url string, req
 			RequestHeader:       req.Headers,
 			CertificateChain:    req.CertificateChain,
 			LoadToCache:         req.LoadToCache,
+			RemoteIp:            &advertiseIP,
 		}})
 	if err != nil {
 		log.Errorf("[preheat]: preheat failed: %s", err.Error())
@@ -421,6 +423,7 @@ func (j *job) preheatAllSeedPeers(ctx context.Context, req *internaljob.PreheatR
 					return err
 				}
 
+				advertiseIP := j.config.Server.AdvertiseIP.String()
 				stream, err := dfdaemonClient.DownloadTask(
 					ctx,
 					taskID,
@@ -436,6 +439,7 @@ func (j *job) preheatAllSeedPeers(ctx context.Context, req *internaljob.PreheatR
 						Timeout:             durationpb.New(req.Timeout),
 						CertificateChain:    req.CertificateChain,
 						LoadToCache:         req.LoadToCache,
+						RemoteIp:            &advertiseIP,
 					}})
 				if err != nil {
 					log.Errorf("[preheat]: preheat failed: %s", err.Error())
@@ -626,6 +630,7 @@ func (j *job) preheatAllPeers(ctx context.Context, req *internaljob.PreheatReque
 					return err
 				}
 
+				advertiseIP := j.config.Server.AdvertiseIP.String()
 				stream, err := dfdaemonClient.DownloadTask(
 					ctx,
 					taskID,
@@ -641,6 +646,7 @@ func (j *job) preheatAllPeers(ctx context.Context, req *internaljob.PreheatReque
 						Timeout:             durationpb.New(req.Timeout),
 						CertificateChain:    req.CertificateChain,
 						LoadToCache:         req.LoadToCache,
+						RemoteIp:            &advertiseIP,
 					}})
 				if err != nil {
 					log.Errorf("[preheat]: preheat failed: %s", err.Error())
@@ -890,8 +896,10 @@ func (j *job) deleteTask(ctx context.Context, data string) (string, error) {
 			continue
 		}
 
+		advertiseIP := j.config.Server.AdvertiseIP.String()
 		if err = dfdaemonClient.DeleteTask(ctx, &dfdaemonv2.DeleteTaskRequest{
-			TaskId: req.TaskID,
+			TaskId:   req.TaskID,
+			RemoteIp: &advertiseIP,
 		}); err != nil {
 			log.Errorf("delete task failed: %s", err.Error())
 			failureTasks = append(failureTasks, &internaljob.DeleteFailureTask{

scheduler/service/service_v2.go

@@ -2447,7 +2447,8 @@ func (v *V2) DeletePersistentCachePeer(_ctx context.Context, req *schedulerv2.De
 		return err
 	}
 
-	if err := dfdaemonClient.DeletePersistentCacheTask(ctx, &dfdaemonv2.DeletePersistentCacheTaskRequest{TaskId: peer.Task.ID}); err != nil {
+	advertiseIP := v.config.Server.AdvertiseIP.String()
+	if err := dfdaemonClient.DeletePersistentCacheTask(ctx, &dfdaemonv2.DeletePersistentCacheTaskRequest{TaskId: peer.Task.ID, RemoteIp: &advertiseIP}); err != nil {
 		peer.Log.Errorf("delete persistent cache task %s from peer %s failed %s", peer.Task.ID, peer.ID, err)
 	}
 
@@ -2653,11 +2654,13 @@ func (v *V2) downloadPersistentCacheTaskByPeer(ctx context.Context, task *persis
 		return err
 	}
 
+	advertiseIP := v.config.Server.AdvertiseIP.String()
 	stream, err := dfdaemonClient.DownloadPersistentCacheTask(ctx, &dfdaemonv2.DownloadPersistentCacheTaskRequest{
 		TaskId:      task.ID,
 		Persistent:  true,
 		Tag:         &task.Tag,
 		Application: &task.Application,
+		RemoteIp:    &advertiseIP,
 	})
 	if err != nil {
 		task.Log.Errorf("download persistent cache task failed %s", err)
@@ -2689,9 +2692,11 @@ func (v *V2) persistPersistentCacheTaskByPeer(ctx context.Context, peer *persist
 		return err
 	}
 
+	advertiseIP := v.config.Server.AdvertiseIP.String()
 	if err := dfdaemonClient.UpdatePersistentCacheTask(ctx, &dfdaemonv2.UpdatePersistentCacheTaskRequest{
 		TaskId:     peer.Task.ID,
 		Persistent: true,
+		RemoteIp:   &advertiseIP,
 	}); err != nil {
 		peer.Log.Errorf("update persistent cache task failed %s", err)
 		return err
@@ -2826,7 +2831,8 @@ func (v *V2) DeletePersistentCacheTask(_ctx context.Context, req *schedulerv2.De
 			continue
 		}
 
-		if err := dfdaemonClient.DeletePersistentCacheTask(ctx, &dfdaemonv2.DeletePersistentCacheTaskRequest{TaskId: peer.Task.ID}); err != nil {
+		advertiseIP := v.config.Server.AdvertiseIP.String()
+		if err := dfdaemonClient.DeletePersistentCacheTask(ctx, &dfdaemonv2.DeletePersistentCacheTaskRequest{TaskId: peer.Task.ID, RemoteIp: &advertiseIP}); err != nil {
 			peer.Log.Errorf("delete persistent cache task %s from peer %s failed %s", peer.Task.ID, peer.ID, err)
 			continue
 		}