Repository: dragonflyoss/Dragonfly2
Pull request #1528 (2022-08-04 06:16:01)

hotfix: peer with same host and manager embed assets
<!--- Provide a general summary of your changes in the Title above -->

## Description

<!--- Describe your changes in detail -->

## Related Issue

<!--- This project only accepts pull requests related to open issues -->
<!--- If suggesting a new feature or change, please discuss it in an issue first -->
<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
<!--- Please link to the issue here: -->

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## Screenshots (if appropriate)

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [ ] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.



Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index e1ee970f..00c0f4fc 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -26,6 +26,12 @@ jobs:
 
       - name: Checkout code
         uses: actions/checkout@v2
+        with:
+          submodules: recursive
+
+      - name: Build manager console assets
+        run: |-
+          make build-manager-console
 
       - name: Run Unit tests
         run: |-
@@ -55,6 +61,12 @@ jobs:
 
       - name: Check out code
         uses: actions/checkout@v2
+        with:
+          submodules: recursive
+
+      - name: Build manager console assets
+        run: |-
+          make build-manager-console
 
       - name: Build
         run: make build
diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml
index a2221202..5c00a161 100644
--- a/.github/workflows/lint.yml
+++ b/.github/workflows/lint.yml
@@ -17,6 +17,12 @@ jobs:
     steps:
       - name: Checkout code
         uses: actions/checkout@v2
+        with:
+          submodules: recursive
+
+      - name: Build manager console assets
+        run: |-
+          make build-manager-console
 
       - name: Golangci lint
         uses: golangci/golangci-lint-action@v2
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 72b45a6e..8dcf7bcd 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -21,6 +21,10 @@ jobs:
         with:
           go-version: 1.18
 
+      - name: Build manager console assets
+        run: |-
+          make build-manager-console
+
       - name: Check GoReleaser config
         uses: goreleaser/goreleaser-action@v2
         with:
diff --git a/.gitignore b/.gitignore
index a0d99905..b5d201f9 100644
--- a/.gitignore
+++ b/.gitignore
@@ -20,6 +20,8 @@ vendor/
 *.DS_Store
 target/
 scheduler/test/misc
+manager/dist/*
+!manager/dist/.gitkeep
 .chglog
 
 # mysql
diff --git a/.goreleaser.yml b/.goreleaser.yml
index 70a6e63b..247aa3c9 100644
--- a/.goreleaser.yml
+++ b/.goreleaser.yml
@@ -1,6 +1,7 @@
 before:
   hooks:
     - hack/update-version-gorelease.sh
+    - make build-manager-console
     - go mod download
 
 builds:
@@ -12,6 +13,7 @@ builds:
       - darwin
     goarch:
       - amd64
+      - arm64
     ldflags:
       - -X d7y.io/dragonfly/v2/version.Major={{ .Major }}
       - -X d7y.io/dragonfly/v2/version.Minor={{ .Minor }}
@@ -29,6 +31,7 @@ builds:
       - darwin
     goarch:
       - amd64
+      - arm64
     ldflags:
       - -X d7y.io/dragonfly/v2/version.Major={{ .Major }}
       - -X d7y.io/dragonfly/v2/version.Minor={{ .Minor }}
@@ -46,6 +49,7 @@ builds:
       - darwin
     goarch:
       - amd64
+      - arm64
     ldflags:
       - -X d7y.io/dragonfly/v2/version.Major={{ .Major }}
       - -X d7y.io/dragonfly/v2/version.Minor={{ .Minor }}
@@ -63,6 +67,7 @@ builds:
       - darwin
     goarch:
       - amd64
+      - arm64
     ldflags:
       - -X d7y.io/dragonfly/v2/version.Major={{ .Major }}
       - -X d7y.io/dragonfly/v2/version.Minor={{ .Minor }}
@@ -80,6 +85,7 @@ builds:
       - darwin
     goarch:
       - amd64
+      - arm64
     ldflags:
       - -X d7y.io/dragonfly/v2/version.Major={{ .Major }}
       - -X d7y.io/dragonfly/v2/version.Minor={{ .Minor }}
diff --git a/Makefile b/Makefile
index c4ad9b14..3839e590 100644
--- a/Makefile
+++ b/Makefile
@@ -132,14 +132,20 @@ build-scheduler: build-dirs
 .PHONY: build-scheduler
 
 # Build manager
-build-manager: build-dirs
+build-manager: build-dirs build-manager-console
 	@echo "Begin to build manager."
+	make build-manager-server
+.PHONY: build-manager
+
+# Build manager server
+build-manager-server: build-dirs
+	@echo "Begin to build manager server."
 	./hack/build.sh manager
 .PHONY: build-manager
 
 # Build manager console
 build-manager-console: build-dirs
-	@echo "Begin to build manager."
+	@echo "Begin to build manager console."
 	./hack/build.sh manager-console
 .PHONY: build-manager-console
 
@@ -401,8 +407,10 @@ help:
 	@echo "make docker-push                    push dragonfly image"
 	@echo "make docker-build-dfdaemon          build dfdaemon image"
 	@echo "make docker-build-scheduler         build scheduler image"
+	@echo "make docker-build-manager           build manager image"
 	@echo "make docker-push-dfdaemon           push dfdaemon image"
 	@echo "make docker-push-scheduler          push scheduler image"
+	@echo "make docker-push-manager            push manager image"
 	@echo "make build                          build dragonfly"
 	@echo "make build-dfget                    build dfget"
 	@echo "make build-linux-dfget              build linux dfget"
@@ -412,6 +420,7 @@ help:
 	@echo "make build-linux-dfstore            build linux dfstore"
 	@echo "make build-scheduler                build scheduler"
 	@echo "make build-manager                  build manager"
+	@echo "make build-manager-server           build manager server"
 	@echo "make build-manager-console          build manager console"
 	@echo "make build-e2e-sha256sum            build sha256sum test tool"
 	@echo "make install-dfget                  install dfget"
diff --git a/build/images/manager/Dockerfile b/build/images/manager/Dockerfile
index 33ff3876..d01dacd6 100644
--- a/build/images/manager/Dockerfile
+++ b/build/images/manager/Dockerfile
@@ -1,5 +1,17 @@
 ARG BASE_IMAGE=alpine:3.16
 
+FROM node:12-alpine as console-builder
+
+WORKDIR /build
+
+COPY ./manager/console/package.json /build
+
+RUN npm install --loglevel warn --progress false
+
+COPY ./manager/console /build
+
+RUN npm run build
+
 FROM golang:1.18.3-alpine3.16 as server-builder
 
 ARG GOPROXY
@@ -12,19 +24,9 @@ RUN apk --no-cache add bash make gcc libc-dev git
 
 COPY . /go/src/d7y.io/dragonfly/v2
 
-RUN make build-manager && make install-manager
+COPY --from=console-builder /build/dist /go/src/d7y.io/dragonfly/v2/manager/dist
 
-FROM node:12-alpine as console-builder
-
-WORKDIR /build
-
-COPY ./manager/console/package.json /build
-
-RUN npm install --loglevel warn --progress false
-
-COPY ./manager/console /build
-
-RUN npm run build
+RUN make build-manager-server && make install-manager
 
 FROM ${BASE_IMAGE} as health
 
@@ -43,7 +45,6 @@ RUN mkdir -p /opt/dragonfly/bin/manager/console \
     && echo "hosts: files dns" > /etc/nsswitch.conf
 
 COPY --from=server-builder /opt/dragonfly/bin/manager /opt/dragonfly/bin/server
-COPY --from=console-builder /build/dist /opt/dragonfly/manager/console/dist
 COPY --from=health /bin/grpc_health_probe /bin/grpc_health_probe
 
 EXPOSE 8080 65003
diff --git a/build/plugin-builder/images/manager/Dockerfile b/build/plugin-builder/images/manager/Dockerfile
index 25f0904e..136c730e 100644
--- a/build/plugin-builder/images/manager/Dockerfile
+++ b/build/plugin-builder/images/manager/Dockerfile
@@ -1,20 +1,17 @@
 ARG BASE_IMAGE=alpine:3.16
 
-FROM node:12-alpine as console-builder
-WORKDIR /build
-COPY ./manager/console/package.json /build
-RUN npm install --loglevel warn --progress false
-COPY ./manager/console /build
-RUN npm run build
-
 FROM ${BASE_IMAGE}
+
 WORKDIR /opt/dragonfly
+
 ENV PATH=/opt/dragonfly/bin:$PATH
-RUN mkdir -p /opt/dragonfly/bin/manager/console \
-    && echo "hosts: files dns" > /etc/nsswitch.conf  && \
+
+RUN echo "hosts: files dns" > /etc/nsswitch.conf  && \
     mkdir -p /usr/local/dragonfly/plugins/
+
 COPY ./artifacts/binaries/manager /opt/dragonfly/bin/server
 COPY ./artifacts/plugins/d7y-manager-plugin-* /usr/local/dragonfly/plugins/
-COPY --from=console-builder /build/dist /opt/dragonfly/manager/console/dist
+
 EXPOSE 8080 65003
+
 ENTRYPOINT ["/opt/dragonfly/bin/server"]
diff --git a/client/daemon/daemon.go b/client/daemon/daemon.go
index 0f3f01a1..793406a4 100644
--- a/client/daemon/daemon.go
+++ b/client/daemon/daemon.go
@@ -167,9 +167,9 @@ func New(opt *config.DaemonOption, d dfpath.Dfpath) (Daemon, error) {
 			PeerId: request.PeerID,
 		})
 		if er != nil {
-			logger.Errorf("step 4:leave task %s/%s, error: %v", request.TaskID, request.PeerID, er)
+			logger.Errorf("step 4: leave task %s/%s, error: %v", request.TaskID, request.PeerID, er)
 		} else {
-			logger.Infof("step 4:leave task %s/%s state ok", request.TaskID, request.PeerID)
+			logger.Infof("step 4: leave task %s/%s state ok", request.TaskID, request.PeerID)
 		}
 	}
 	storageManager, err := storage.NewStorageManager(opt.Storage.StoreStrategy, &opt.Storage,
diff --git a/client/daemon/rpcserver/rpcserver.go b/client/daemon/rpcserver/rpcserver.go
index c5b09265..022f7867 100644
--- a/client/daemon/rpcserver/rpcserver.go
+++ b/client/daemon/rpcserver/rpcserver.go
@@ -123,13 +123,21 @@ func (s *server) GetPieceTasks(ctx context.Context, request *base.PieceTaskReque
 				err, request.TaskId, request.SrcPid, request.DstPid, request.StartNum, request.Limit)
 			return nil, dferrors.New(code, err.Error())
 		}
-		// dst peer is not running
+
+		// check whether dst peer is not running
 		task, ok := s.peerTaskManager.IsPeerTaskRunning(request.TaskId)
 		if !ok {
 			code = base.Code_PeerTaskNotFound
 			logger.Errorf("get piece tasks error: target peer task not found, task id: %s, src peer: %s, dst peer: %s, piece num: %d, limit: %d",
 				request.TaskId, request.SrcPid, request.DstPid, request.StartNum, request.Limit)
 			return nil, dferrors.New(code, err.Error())
+		} else if task.GetPeerID() == request.GetSrcPid() {
+			// 2. scheduler schedules same peer host, and the running peer is the source peer (schedulers misses the old peer leaved event)
+			code = base.Code_PeerTaskNotFound
+			logger.Errorf("get piece tasks error: target peer task not found, task id: %s, the src peer is same with running peer,"+
+				" src peer: %s, dst peer: %s, piece num: %d, limit: %d",
+				request.TaskId, request.SrcPid, request.DstPid, request.StartNum, request.Limit)
+			return nil, dferrors.New(code, err.Error())
 		}
 
 		if task.GetPeerID() != request.GetDstPid() {
diff --git a/deploy/docker-compose/template/manager.template.yaml b/deploy/docker-compose/template/manager.template.yaml
index 7e895b1f..560f5013 100644
--- a/deploy/docker-compose/template/manager.template.yaml
+++ b/deploy/docker-compose/template/manager.template.yaml
@@ -13,8 +13,6 @@ server:
   rest:
     # stand address
     addr: :8080
-  # front-end console resource path
-  # publicPath: /dist
 
 # database info used for server
 database:
diff --git a/hack/build.sh b/hack/build.sh
index e085e5e6..c990f321 100755
--- a/hack/build.sh
+++ b/hack/build.sh
@@ -26,8 +26,8 @@ GOLDFLAGS="${GOLDFLAGS} -X \"d7y.io/dragonfly/v2/version.Gotags=${GOTAGS:-none}\
 GOLDFLAGS="${GOLDFLAGS} -X \"d7y.io/dragonfly/v2/version.GoVersion=$(go version | grep -o 'go[^ ].*')\""
 GOLDFLAGS="${GOLDFLAGS} -X \"d7y.io/dragonfly/v2/version.Gogcflags=${GOGCFLAGS:-none}\""
 
-curDir=$(cd "$(dirname "$0")" && pwd)
-cd "${curDir}" || return
+CUR_DIR=$(cd "$(dirname "$0")" && pwd)
+cd "${CUR_DIR}" || return
 BUILD_SOURCE_HOME=$(cd ".." && pwd)
 
 . ./env.sh
@@ -114,10 +114,14 @@ build-manager-docker() {
 
 build-manager-console() {
     set -x
-    consoleDir=$(echo $curDir | sed 's#hack#manager/console#')
+    CONSOLE_DIR=$(echo $CUR_DIR | sed 's#hack#manager/console#')
+    MANAGER_DIR=$(echo $CUR_DIR | sed 's#hack#manager#')
+    CONSOLE_ASSETS_DIR=$CONSOLE_DIR/dist/
+    MANAGER_ASSETS_DIR=$MANAGER_DIR/dist/
     docker run --workdir=/build \
-        --rm -v ${consoleDir}:/build node:12-alpine \
+        --rm -v ${CONSOLE_DIR}:/build node:12-alpine \
         sh -c "npm install --loglevel warn --progress false && npm run build"
+    cp -r $CONSOLE_ASSETS_DIR $MANAGER_ASSETS_DIR
 }
 
 main() {
diff --git a/hack/gen-buildx.sh b/hack/gen-buildx.sh
index de4d4f3d..b546fdb3 100755
--- a/hack/gen-buildx.sh
+++ b/hack/gen-buildx.sh
@@ -1,6 +1,6 @@
 #!/bin/bash
 
-components="scheduler manager"
+components="scheduler manager-server"
 
 set -x
 
diff --git a/manager/config/config.go b/manager/config/config.go
index 78b68bcd..897fb17b 100644
--- a/manager/config/config.go
+++ b/manager/config/config.go
@@ -54,9 +54,6 @@ type ServerConfig struct {
 	// Server log directory.
 	LogDir string `yaml:"logDir" mapstructure:"logDir"`
 
-	// Console resource path.
-	PublicPath string `yaml:"publicPath" mapstructure:"publicPath"`
-
 	// GRPC server configuration.
 	GRPC *TCPListenConfig `yaml:"grpc" mapstructure:"grpc"`
 
@@ -249,8 +246,7 @@ type ObjectStorageConfig struct {
 func New() *Config {
 	return &Config{
 		Server: &ServerConfig{
-			Name:       DefaultServerName,
-			PublicPath: DefaultPublicPath,
+			Name: DefaultServerName,
 			GRPC: &TCPListenConfig{
 				PortRange: TCPListenPortRange{
 					Start: DefaultGRPCPort,
diff --git a/manager/config/constants.go b/manager/config/constants.go
index ea7dce81..8f43ee49 100644
--- a/manager/config/constants.go
+++ b/manager/config/constants.go
@@ -33,9 +33,6 @@ const (
 	// DefaultServerName is default server name.
 	DefaultServerName = "d7y/manager"
 
-	// DefaultPublicPath is default path for frontend assets.
-	DefaultPublicPath = "manager/console/dist"
-
 	// DefaultGRPCPort is default port for grpc server.
 	DefaultGRPCPort = 65003
 
diff --git a/manager/dist/.gitkeep b/manager/dist/.gitkeep
new file mode 100644
index 00000000..e69de29b
diff --git a/manager/manager.go b/manager/manager.go
index 0955bec3..4c55fece 100644
--- a/manager/manager.go
+++ b/manager/manager.go
@@ -18,9 +18,12 @@ package manager
 
 import (
 	"context"
+	"embed"
+	"io/fs"
 	"net/http"
 	"time"
 
+	"github.com/gin-contrib/static"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"google.golang.org/grpc"
 
@@ -42,10 +45,39 @@ import (
 
 const (
 	// gracefulStopTimeout specifies a time limit for
-	// grpc server to complete a graceful shutdown
+	// grpc server to complete a graceful shutdown.
 	gracefulStopTimeout = 10 * time.Second
+
+	// assetsTargetPath is target path of embed assets.
+	assetsTargetPath = "dist"
 )
 
+//go:embed dist/*
+var assets embed.FS
+
+type embedFileSystem struct {
+	http.FileSystem
+}
+
+func (e embedFileSystem) Exists(prefix string, path string) bool {
+	_, err := e.Open(path)
+	if err != nil {
+		return false
+	}
+	return true
+}
+
+func EmbedFolder(fsEmbed embed.FS, targetPath string) static.ServeFileSystem {
+	fsys, err := fs.Sub(fsEmbed, targetPath)
+	if err != nil {
+		panic(err)
+	}
+
+	return embedFileSystem{
+		FileSystem: http.FS(fsys),
+	}
+}
+
 type Server struct {
 	// Server configuration
 	config *config.Config
@@ -107,7 +139,7 @@ func New(cfg *config.Config, d dfpath.Dfpath) (*Server, error) {
 
 	// Initialize REST server
 	restService := service.New(db, cache, job, enforcer, objectStorage)
-	router, err := router.Init(cfg, d.LogDir(), restService, enforcer)
+	router, err := router.Init(cfg, d.LogDir(), restService, enforcer, EmbedFolder(assets, assetsTargetPath))
 	if err != nil {
 		return nil, err
 	}
diff --git a/manager/router/router.go b/manager/router/router.go
index 56428b6d..9938fc9e 100644
--- a/manager/router/router.go
+++ b/manager/router/router.go
@@ -18,6 +18,7 @@ package router
 
 import (
 	"io"
+	"net/http"
 	"os"
 	"path/filepath"
 
@@ -43,13 +44,13 @@ const (
 
 var GinLogFileName = "gin.log"
 
-func Init(cfg *config.Config, logDir string, service service.Service, enforcer *casbin.Enforcer) (*gin.Engine, error) {
-	// Set mode
+func Init(cfg *config.Config, logDir string, service service.Service, enforcer *casbin.Enforcer, assets static.ServeFileSystem) (*gin.Engine, error) {
+	// Set mode.
 	if !cfg.Verbose {
 		gin.SetMode(gin.ReleaseMode)
 	}
 
-	// Logging to a file
+	// Logging to a file.
 	if !cfg.Console {
 		gin.DisableConsoleColor()
 		logDir := filepath.Join(logDir, "manager")
@@ -60,7 +61,7 @@ func Init(cfg *config.Config, logDir string, service service.Service, enforcer *
 	r := gin.New()
 	h := handlers.New(service)
 
-	// Prometheus metrics
+	// Prometheus metrics.
 	p := ginprometheus.NewPrometheus(PrometheusSubsystemName)
 	// URL removes query string.
 	// Prometheus metrics need to reduce label,
@@ -91,8 +92,8 @@ func Init(cfg *config.Config, logDir string, service service.Service, enforcer *
 		return nil, err
 	}
 
-	// Manager View
-	r.Use(static.Serve("/", static.LocalFile(cfg.Server.PublicPath, true)))
+	// Manager view.
+	r.Use(static.Serve("/", assets))
 
 	// Router
 	apiv1 := r.Group("/api/v1")
@@ -237,9 +238,9 @@ func Init(cfg *config.Config, logDir string, service service.Service, enforcer *
 	apiSeagger := ginSwagger.URL("/swagger/doc.json")
 	r.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler, apiSeagger))
 
-	// Fallback To Manager View
+	// Fallback to manager view.
 	r.NoRoute(func(c *gin.Context) {
-		c.File(filepath.Join(cfg.Server.PublicPath, "index.html"))
+		c.Redirect(http.StatusMovedPermanently, "/")
 	})
 
 	return r, nil
diff --git a/scheduler/scheduler/scheduler.go b/scheduler/scheduler/scheduler.go
index 7e577dff..963791a2 100644
--- a/scheduler/scheduler/scheduler.go
+++ b/scheduler/scheduler/scheduler.go
@@ -266,6 +266,14 @@ func (s *scheduler) filterCandidateParents(peer *resource.Peer, blocklist set.Sa
 			continue
 		}
 
+		// Candidate parent host is not allowed to be the same as the peer host,
+		// because dfdaemon cannot handle the situation
+		// where two tasks are downloading and downloading each other.
+		if peer.Host.ID == candidateParent.Host.ID {
+			peer.Log.Debugf("candidate parent %s host %s is the same as peer host", candidateParent.ID, candidateParent.Host.ID)
+			continue
+		}
+
 		// Candidate parent is bad node.
 		if s.evaluator.IsBadNode(candidateParent) {
 			peer.Log.Debugf("candidate parent %s is not selected because it is bad node", candidateParent.ID)


Test patch:
diff --git a/manager/config/config_test.go b/manager/config/config_test.go
index cfa0c301..0c592eff 100644
--- a/manager/config/config_test.go
+++ b/manager/config/config_test.go
@@ -32,9 +32,8 @@ func TestManagerConfig_Load(t *testing.T) {
 
 	config := &Config{
 		Server: &ServerConfig{
-			Name:       "foo",
-			LogDir:     "foo",
-			PublicPath: "foo",
+			Name:   "foo",
+			LogDir: "foo",
 			GRPC: &TCPListenConfig{
 				Listen: "127.0.0.1",
 				PortRange: TCPListenPortRange{
diff --git a/manager/config/testdata/manager.yaml b/manager/config/testdata/manager.yaml
index 21b20024..2752003f 100644
--- a/manager/config/testdata/manager.yaml
+++ b/manager/config/testdata/manager.yaml
@@ -1,7 +1,6 @@
 server:
   name: foo
   logDir: foo
-  publicPath: foo
   grpc:
     listen: 127.0.0.1
     port:
diff --git a/scheduler/scheduler/scheduler_test.go b/scheduler/scheduler/scheduler_test.go
index 5d04e537..76d94be9 100644
--- a/scheduler/scheduler/scheduler_test.go
+++ b/scheduler/scheduler/scheduler_test.go
@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/golang/mock/gomock"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 
 	"d7y.io/dragonfly/v2/manager/types"
@@ -548,7 +549,17 @@ func TestScheduler_NotifyAndFindParent(t *testing.T) {
 			mock: func(peer *resource.Peer, mockHost *resource.Host, mockTask *resource.Task, mockPeer *resource.Peer, blocklist set.SafeSet[string], stream rpcscheduler.Scheduler_ReportPieceResultServer, dynconfig config.DynconfigInterface, ms *rpcschedulermocks.MockScheduler_ReportPieceResultServerMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
 				peer.FSM.SetState(resource.PeerStateRunning)
 				mockPeer.FSM.SetState(resource.PeerStateRunning)
-				candidatePeer := resource.NewPeer(idgen.PeerID("127.0.0.1"), mockTask, mockHost)
+				candidatePeer := resource.NewPeer(idgen.PeerID("127.0.0.1"), mockTask, resource.NewHost(&rpcscheduler.PeerHost{
+					Id:             idgen.HostID(uuid.New().String(), 8003),
+					Ip:             "127.0.0.1",
+					RpcPort:        8003,
+					DownPort:       8001,
+					HostName:       "hostname",
+					SecurityDomain: "security_domain",
+					Location:       "location",
+					Idc:            "idc",
+					NetTopology:    "net_topology",
+				}))
 				candidatePeer.FSM.SetState(resource.PeerStateRunning)
 				peer.Task.StorePeer(peer)
 				peer.Task.StorePeer(mockPeer)
@@ -584,7 +595,17 @@ func TestScheduler_NotifyAndFindParent(t *testing.T) {
 			mockHost := resource.NewHost(mockRawHost)
 			mockTask := resource.NewTask(mockTaskID, mockTaskURL, base.TaskType_Normal, mockTaskURLMeta, resource.WithBackToSourceLimit(mockTaskBackToSourceLimit))
 			peer := resource.NewPeer(mockPeerID, mockTask, mockHost)
-			mockPeer := resource.NewPeer(idgen.PeerID("127.0.0.1"), mockTask, mockHost)
+			mockPeer := resource.NewPeer(idgen.PeerID("127.0.0.1"), mockTask, resource.NewHost(&rpcscheduler.PeerHost{
+				Id:             idgen.HostID(uuid.New().String(), 8003),
+				Ip:             "127.0.0.1",
+				RpcPort:        8003,
+				DownPort:       8001,
+				HostName:       "hostname",
+				SecurityDomain: "security_domain",
+				Location:       "location",
+				Idc:            "idc",
+				NetTopology:    "net_topology",
+			}))
 			blocklist := set.NewSafeSet[string]()
 
 			tc.mock(peer, mockHost, mockTask, mockPeer, blocklist, stream, dynconfig, stream.EXPECT(), dynconfig.EXPECT())
@@ -794,6 +815,25 @@ func TestScheduler_FindParent(t *testing.T) {
 				assert.Equal(mockPeers[1].ID, parent.ID)
 			},
 		},
+		{
+			name: "find parent with same host",
+			mock: func(peer *resource.Peer, mockPeers []*resource.Peer, blocklist set.SafeSet[string], md *configmocks.MockDynconfigInterfaceMockRecorder) {
+				peer.FSM.SetState(resource.PeerStateRunning)
+				mockPeers[0].FSM.SetState(resource.PeerStateRunning)
+				mockPeers[1].FSM.SetState(resource.PeerStateRunning)
+				mockPeers[0].IsBackToSource.Store(true)
+				mockPeers[1].Host = peer.Host
+				peer.Task.StorePeer(peer)
+				peer.Task.StorePeer(mockPeers[0])
+				peer.Task.StorePeer(mockPeers[1])
+				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1)
+			},
+			expect: func(t *testing.T, peer *resource.Peer, mockPeers []*resource.Peer, parent *resource.Peer, ok bool) {
+				assert := assert.New(t)
+				assert.True(ok)
+				assert.Equal(mockPeers[0].ID, parent.ID)
+			},
+		},
 		{
 			name: "find parent and fetch filterParentLimit from manager dynconfig",
 			mock: func(peer *resource.Peer, mockPeers []*resource.Peer, blocklist set.SafeSet[string], md *configmocks.MockDynconfigInterfaceMockRecorder) {
@@ -835,6 +875,17 @@ func TestScheduler_FindParent(t *testing.T) {
 
 			var mockPeers []*resource.Peer
 			for i := 0; i < 11; i++ {
+				mockHost := resource.NewHost(&rpcscheduler.PeerHost{
+					Id:             idgen.HostID(uuid.New().String(), 8003),
+					Ip:             "127.0.0.1",
+					RpcPort:        8003,
+					DownPort:       8001,
+					HostName:       "hostname",
+					SecurityDomain: "security_domain",
+					Location:       "location",
+					Idc:            "idc",
+					NetTopology:    "net_topology",
+				})
 				peer := resource.NewPeer(idgen.PeerID(fmt.Sprintf("127.0.0.%d", i)), mockTask, mockHost)
 				mockPeers = append(mockPeers, peer)
 			}