Repository: dragonflyoss/Dragonfly2
Pull request #1038 (2022-01-25 03:23:45)

feat: If cdn only updates IP, set cdn peers state to PeerStateLeave
<!--- Provide a general summary of your changes in the Title above -->

## Description
- If cdn only updates IP, set cdn peers state to PeerStateLeave.
<!--- 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)
- [x] 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! -->

- [x] Code compiles correctly.
- [x] My code follows the code style of this project.
- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [x] I have read the **CONTRIBUTING** document.
- [x] I have added tests to cover my changes.
- [x] All new and existing tests passed.



Patch:
diff --git a/Makefile b/Makefile
index 1d1509b4..9ed76e3f 100644
--- a/Makefile
+++ b/Makefile
@@ -203,8 +203,8 @@ test-coverage:
 
 # Run github actions E2E tests with coverage
 actions-e2e-test-coverage:
-	@ginkgo -v -r --race --failFast -cover test/e2e --trace --progress
-	@cat test/e2e/*.coverprofile >> coverage.txt
+	@ginkgo -v -r --race --fail-fast -cover --trace --progress test/e2e
+	@cat coverprofile.out >> coverage.txt
 .PHONY: actions-e2e-test-coverage
 
 # Install E2E tests environment
@@ -214,13 +214,13 @@ install-e2e-test:
 
 # Run E2E tests
 e2e-test: install-e2e-test
-	@ginkgo -v -r --race --failFast test/e2e --trace --progress
+	@ginkgo -v -r --race --fail-fast --cover --trace --progress test/e2e
 .PHONY: e2e-test
 
 # Run E2E tests with coverage
 e2e-test-coverage: install-e2e-test
-	@ginkgo -v -r --race --failFast -cover test/e2e --trace --progress
-	@cat test/e2e/*.coverprofile >> coverage.txt
+	@ginkgo -v -r --race --fail-fast --cover --trace --progress test/e2e
+	@cat coverprofile.out >> coverage.txt
 .PHONY: e2e-test-coverage
 
 # Clean E2E tests
diff --git a/go.mod b/go.mod
index fe570543..75529c04 100644
--- a/go.mod
+++ b/go.mod
@@ -42,8 +42,7 @@ require (
 	github.com/mcuadros/go-gin-prometheus v0.1.0
 	github.com/mitchellh/mapstructure v1.4.1
 	github.com/montanaflynn/stats v0.6.6
-	github.com/onsi/ginkgo v1.16.5
-	github.com/onsi/gomega v1.14.0
+	github.com/onsi/gomega v1.18.0
 	github.com/opencontainers/go-digest v1.0.0
 	github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
 	github.com/pkg/errors v0.9.1
@@ -70,7 +69,7 @@ require (
 	golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2
 	golang.org/x/oauth2 v0.0.0-20210201163806-010130855d6c
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
-	golang.org/x/sys v0.0.0-20211124211545-fe61309f8881
+	golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e
 	golang.org/x/time v0.0.0-20201208040808-7e3f01d25324
 	google.golang.org/api v0.39.0
 	google.golang.org/grpc v1.39.0
@@ -155,6 +154,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect
+	github.com/onsi/ginkgo/v2 v2.1.0 // indirect
 	github.com/opencontainers/image-spec v1.0.2 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pelletier/go-toml v1.8.1 // indirect
diff --git a/go.sum b/go.sum
index ffd95775..82dcc0cd 100644
--- a/go.sum
+++ b/go.sum
@@ -377,6 +377,7 @@ github.com/google/pprof v0.0.0-20201023163331-3e6fc7fc9c4c/go.mod h1:kpwsk12EmLe
 github.com/google/pprof v0.0.0-20201203190320-1bf35d6f28c2/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/pprof v0.0.0-20201218002935-b9804c9f04c2/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/pprof v0.0.0-20210122040257-d980be63207e/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
+github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
@@ -649,6 +650,9 @@ github.com/onsi/ginkgo v1.15.0/go.mod h1:hF8qUzuuC8DJGygJH3726JnCZX4MYbRB8yFfISq
 github.com/onsi/ginkgo v1.16.4/go.mod h1:dX+/inL/fNMqNlz0e9LfyB9TswhZpCVdJM/Z6Vvnwo0=
 github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
 github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
+github.com/onsi/ginkgo/v2 v2.0.0/go.mod h1:vw5CSIxN1JObi/U8gcbwft7ZxR2dgaR70JSE3/PpL4c=
+github.com/onsi/ginkgo/v2 v2.1.0 h1:Rj+Was8Gt/ICSZY/CihVKK2cGy3rDqBYKSxVnmOXZzI=
+github.com/onsi/ginkgo/v2 v2.1.0/go.mod h1:vw5CSIxN1JObi/U8gcbwft7ZxR2dgaR70JSE3/PpL4c=
 github.com/onsi/gomega v0.0.0-20170829124025-dcabb60a477c/go.mod h1:C1qb7wdrVGGVU+Z6iS04AVkA3Q65CEZX59MT0QO5uiA=
 github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
@@ -658,6 +662,10 @@ github.com/onsi/gomega v1.10.4/go.mod h1:g/HbgYopi++010VEqkFgJHKC09uJiW9UkXvMUuK
 github.com/onsi/gomega v1.10.5/go.mod h1:gza4q3jKQJijlu05nKWRCW/GavJumGt8aNRxWg7mt48=
 github.com/onsi/gomega v1.14.0 h1:ep6kpPVwmr/nTbklSx2nrLNSIO62DoYAhnPNIMhK8gI=
 github.com/onsi/gomega v1.14.0/go.mod h1:cIuvLEne0aoVhAgh/O6ac0Op8WWw9H6eYCriF+tEHG0=
+github.com/onsi/gomega v1.17.0 h1:9Luw4uT5HTjHTN8+aNcSThgH1vdXnmdJ8xIfZ4wyTRE=
+github.com/onsi/gomega v1.17.0/go.mod h1:HnhC7FXeEQY45zxNK3PPoIUhzk/80Xly9PcubAlGdZY=
+github.com/onsi/gomega v1.18.0 h1:ngbYoRctxjl8SiF7XgP0NxBFbfHcg3wfHMMaFHWwMTM=
+github.com/onsi/gomega v1.18.0/go.mod h1:0q+aL8jAiMXy9hbwj2mr5GziHiwhAIQpFmmtT5hitRs=
 github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.0.1/go.mod h1:BtxoFyWECRxE4U/7sNtV5W15zMzWCbyJoFRP3s7yZA0=
@@ -1119,6 +1127,8 @@ golang.org/x/sys v0.0.0-20210816074244-15123e1e1f71/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20211013075003-97ac67df715c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211124211545-fe61309f8881 h1:TyHqChC80pFkXWraUUf6RuB5IqFdQieMLwwCJokV2pc=
 golang.org/x/sys v0.0.0-20211124211545-fe61309f8881/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b h1:9zKuko04nR4gjZ4+DNjHqRlAJqbJETHwiNKDqTfOjfE=
diff --git a/scheduler/resource/cdn.go b/scheduler/resource/cdn.go
index eac5b68c..1c19ed2d 100644
--- a/scheduler/resource/cdn.go
+++ b/scheduler/resource/cdn.go
@@ -222,14 +222,24 @@ func (c *cdnClient) OnNotify(data *config.DynconfigData) {
 		return
 	}
 
-	// Update dynamic data
-	c.data = data
+	// If only the ip of the cdn host is changed,
+	// the cdn peer needs to be cleared.
+	diff := diffCDNs(c.data.CDNs, data.CDNs)
+	for _, v := range diff {
+		id := idgen.CDNHostID(v.Hostname, v.Port)
+		if host, ok := c.hostManager.Load(id); ok {
+			host.LeavePeers()
+		}
+	}
 
 	// Update host manager
 	for _, host := range cdnsToHosts(data.CDNs) {
 		c.hostManager.Store(host)
 	}
 
+	// Update dynamic data
+	c.data = data
+
 	// Update grpc cdn addresses
 	c.UpdateState(cdnsToNetAddrs(data.CDNs))
 	logger.Infof("cdn addresses have been updated: %v", ips)
@@ -274,6 +284,44 @@ func cdnsToNetAddrs(cdns []*config.CDN) []dfnet.NetAddr {
 	return netAddrs
 }
 
+// diffCDNs get cdns with the same HostID but different IP
+func diffCDNs(cx []*config.CDN, cy []*config.CDN) []*config.CDN {
+	var diff []*config.CDN
+	for _, x := range cx {
+		for _, y := range cy {
+			if x.Hostname != y.Hostname {
+				continue
+			}
+
+			if x.Port != y.Port {
+				continue
+			}
+
+			if x.IP == y.IP {
+				continue
+			}
+
+			diff = append(diff, x)
+		}
+	}
+
+	for _, x := range cx {
+		found := false
+		for _, y := range cy {
+			if idgen.CDNHostID(x.Hostname, x.Port) == idgen.CDNHostID(y.Hostname, y.Port) {
+				found = true
+				break
+			}
+		}
+
+		if !found {
+			diff = append(diff, x)
+		}
+	}
+
+	return diff
+}
+
 // getCDNIPs get ips by []*config.CDN.
 func getCDNIPs(cdns []*config.CDN) []string {
 	ips := []string{}
diff --git a/scheduler/resource/host.go b/scheduler/resource/host.go
index 6e80c9fa..d80f60a5 100644
--- a/scheduler/resource/host.go
+++ b/scheduler/resource/host.go
@@ -165,6 +165,27 @@ func (h *Host) LenPeers() int {
 	return len
 }
 
+// LeavePeers set peer state to PeerStateLeave
+func (h *Host) LeavePeers() {
+	h.Peers.Range(func(key, value interface{}) bool {
+		if peer, ok := value.(*Peer); ok {
+			if err := peer.FSM.Event(PeerEventDownloadFailed); err != nil {
+				peer.Log.Errorf("peer fsm event failed: %v", err)
+				return true
+			}
+
+			if err := peer.FSM.Event(PeerEventLeave); err != nil {
+				peer.Log.Errorf("peer fsm event failed: %v", err)
+				return true
+			}
+
+			peer.Log.Info("peer has been left")
+		}
+
+		return true
+	})
+}
+
 // FreeUploadLoad return free upload load of host
 func (h *Host) FreeUploadLoad() int32 {
 	return h.UploadLoadLimit.Load() - int32(h.LenPeers())
diff --git a/scheduler/resource/task.go b/scheduler/resource/task.go
index 651dc8c3..e14d8392 100644
--- a/scheduler/resource/task.go
+++ b/scheduler/resource/task.go
@@ -124,7 +124,7 @@ func NewTask(id, url string, backToSourceLimit int, meta *base.UrlMeta) *Task {
 	t.FSM = fsm.NewFSM(
 		TaskStatePending,
 		fsm.Events{
-			{Name: TaskEventDownload, Src: []string{TaskStatePending, TaskStateFailed}, Dst: TaskStateRunning},
+			{Name: TaskEventDownload, Src: []string{TaskStatePending, TaskStateSucceeded, TaskStateFailed}, Dst: TaskStateRunning},
 			{Name: TaskEventDownloadSucceeded, Src: []string{TaskStateRunning, TaskStateFailed}, Dst: TaskStateSucceeded},
 			{Name: TaskEventDownloadFailed, Src: []string{TaskStateRunning}, Dst: TaskStateFailed},
 		},
@@ -186,6 +186,26 @@ func (t *Task) LenPeers() int {
 	return len
 }
 
+// LenAvailablePeers return length of peers without state is PeerStateLeave
+func (t *Task) LenAvailablePeers() int {
+	var len int
+	t.Peers.Range(func(_, v interface{}) bool {
+		peer, ok := v.(*Peer)
+		if !ok {
+			return true
+		}
+
+		if peer.FSM.Is(PeerStateLeave) {
+			return true
+		}
+
+		len++
+		return true
+	})
+
+	return len
+}
+
 // LoadCDNPeer return latest cdn peer in peers sync map
 func (t *Task) LoadCDNPeer() (*Peer, bool) {
 	var peers []*Peer
diff --git a/scheduler/service/callback.go b/scheduler/service/callback.go
index aee32134..cff5c4fb 100644
--- a/scheduler/service/callback.go
+++ b/scheduler/service/callback.go
@@ -221,26 +221,18 @@ func (c *callback) PieceFail(ctx context.Context, peer *resource.Peer, piece *rp
 		peer.Log.Infof("parent %s is cdn", piece.DstPid)
 		fallthrough
 	case base.Code_CDNTaskNotFound:
-		// CDN peer exists in the scheduler, but the peer information in the cdn service has been recycled.
-		// Scheduler need to redownload cdn peer.
-		if err := parent.FSM.Event(resource.PeerEventRestart); err != nil {
-			peer.Log.Errorf("peer fsm event failed: %v", err)
-			break
-		}
-
+		c.PeerFail(ctx, parent)
 		go func() {
 			parent.Log.Info("cdn restart seed task")
-			parent, endOfPiece, err := c.resource.CDN().TriggerTask(context.Background(), parent.Task)
+			cdnPeer, endOfPiece, err := c.resource.CDN().TriggerTask(context.Background(), parent.Task)
 			if err != nil {
 				peer.Log.Errorf("retrigger task failed: %v", err)
-
 				c.TaskFail(ctx, parent.Task)
-				c.PeerFail(ctx, parent)
 				return
 			}
 
-			c.TaskSuccess(ctx, parent.Task, endOfPiece)
-			c.PeerSuccess(ctx, parent)
+			c.TaskSuccess(ctx, cdnPeer.Task, endOfPiece)
+			c.PeerSuccess(ctx, cdnPeer)
 		}()
 	default:
 	}
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index 6f9170c7..cf8170b5 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -88,7 +88,7 @@ func (s *service) CDN() resource.CDN {
 func (s *service) RegisterTask(ctx context.Context, req *rpcscheduler.PeerTaskRequest) (*resource.Task, error) {
 	task := resource.NewTask(idgen.TaskID(req.Url, req.UrlMeta), req.Url, s.config.Scheduler.BackSourceCount, req.UrlMeta)
 	task, ok := s.resource.TaskManager().LoadOrStore(task)
-	if ok && (task.FSM.Is(resource.TaskStateRunning) || task.FSM.Is(resource.TaskStateSucceeded)) {
+	if ok && (task.FSM.Is(resource.TaskStateRunning) || task.FSM.Is(resource.TaskStateSucceeded)) && task.LenAvailablePeers() != 0 {
 		// Task is healthy and can be reused
 		task.UpdateAt.Store(time.Now())
 		task.Log.Infof("reuse task and status is %s", task.FSM.Current())


Test patch:
diff --git a/.github/workflows/compatibility-e2e.yml b/.github/workflows/compatibility-e2e.yml
index ee965e29..6d819c4b 100644
--- a/.github/workflows/compatibility-e2e.yml
+++ b/.github/workflows/compatibility-e2e.yml
@@ -41,7 +41,7 @@ jobs:
       - name: Get dependencies
         run: |
           go mod download github.com/go-task/slim-sprig
-          go install github.com/onsi/ginkgo/ginkgo
+          go install github.com/onsi/ginkgo/v2/ginkgo@v2.1.0
           mkdir -p /tmp/artifact
           go mod vendor
 
diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml
index 2c4e5011..a9f84b88 100644
--- a/.github/workflows/e2e.yml
+++ b/.github/workflows/e2e.yml
@@ -36,7 +36,7 @@ jobs:
       - name: Get dependencies
         run: |
           go mod download github.com/go-task/slim-sprig
-          go install github.com/onsi/ginkgo/ginkgo
+          go install github.com/onsi/ginkgo/v2/ginkgo@v2.1.0
           mkdir -p /tmp/artifact
           go mod vendor
 
diff --git a/hack/install-e2e-test.sh b/hack/install-e2e-test.sh
index 629eaa27..13469be7 100755
--- a/hack/install-e2e-test.sh
+++ b/hack/install-e2e-test.sh
@@ -65,7 +65,7 @@ install-ginkgo() {
       print_step_info "ginkgo has been installed"
   else
       go mod download github.com/go-task/slim-sprig
-      go install github.com/onsi/ginkgo/ginkgo
+      go install github.com/onsi/ginkgo/v2/ginkgo@v2.1.0
   fi
 }
 
diff --git a/scheduler/resource/cdn_test.go b/scheduler/resource/cdn_test.go
index f9954fdb..7fb3d961 100644
--- a/scheduler/resource/cdn_test.go
+++ b/scheduler/resource/cdn_test.go
@@ -165,23 +165,77 @@ func TestCDNClient_OnNotify(t *testing.T) {
 		mock func(dynconfig *configmocks.MockDynconfigInterfaceMockRecorder, hostManager *MockHostManagerMockRecorder)
 	}{
 		{
-			name: "notify client",
+			name: "notify client without different cdns",
 			data: &config.DynconfigData{
 				CDNs: []*config.CDN{{
-					ID: 1,
-					IP: "0.0.0.0",
+					ID:       1,
+					Hostname: "foo",
+					IP:       "0.0.0.0",
+					Port:     8080,
 				}},
 			},
 			mock: func(dynconfig *configmocks.MockDynconfigInterfaceMockRecorder, hostManager *MockHostManagerMockRecorder) {
 				gomock.InOrder(
 					dynconfig.Get().Return(&config.DynconfigData{
 						CDNs: []*config.CDN{{
-							ID: 1,
-							IP: "127.0.0.1",
+							ID:       1,
+							Hostname: "foo",
+							IP:       "0.0.0.0",
+							Port:     8080,
+						}},
+					}, nil).Times(1),
+					hostManager.Store(gomock.Any()).Return().Times(1),
+					dynconfig.Register(gomock.Any()).Return().Times(1),
+				)
+			},
+		},
+		{
+			name: "notify client with different cdns",
+			data: &config.DynconfigData{
+				CDNs: []*config.CDN{{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "0.0.0.0",
+				}},
+			},
+			mock: func(dynconfig *configmocks.MockDynconfigInterfaceMockRecorder, hostManager *MockHostManagerMockRecorder) {
+				mockHost := NewHost(mockRawHost)
+				gomock.InOrder(
+					dynconfig.Get().Return(&config.DynconfigData{
+						CDNs: []*config.CDN{{
+							ID:       1,
+							Hostname: "foo",
+							IP:       "127.0.0.1",
+						}},
+					}, nil).Times(1),
+					hostManager.Store(gomock.Any()).Return().Times(1),
+					dynconfig.Register(gomock.Any()).Return().Times(1),
+					hostManager.Load(gomock.Any()).Return(mockHost, true).Times(1),
+					hostManager.Store(gomock.Any()).Return().Times(1),
+				)
+			},
+		},
+		{
+			name: "notify client with different cdns and load host failed",
+			data: &config.DynconfigData{
+				CDNs: []*config.CDN{{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "0.0.0.0",
+				}},
+			},
+			mock: func(dynconfig *configmocks.MockDynconfigInterfaceMockRecorder, hostManager *MockHostManagerMockRecorder) {
+				gomock.InOrder(
+					dynconfig.Get().Return(&config.DynconfigData{
+						CDNs: []*config.CDN{{
+							ID:       1,
+							Hostname: "foo",
+							IP:       "127.0.0.1",
 						}},
 					}, nil).Times(1),
 					hostManager.Store(gomock.Any()).Return().Times(1),
 					dynconfig.Register(gomock.Any()).Return().Times(1),
+					hostManager.Load(gomock.Any()).Return(nil, false).Times(1),
 					hostManager.Store(gomock.Any()).Return().Times(1),
 				)
 			},
@@ -364,6 +418,207 @@ func TestCDNClient_cdnsToNetAddrs(t *testing.T) {
 	}
 }
 
+func TestCDNClient_diffCDNs(t *testing.T) {
+	tests := []struct {
+		name   string
+		cx     []*config.CDN
+		cy     []*config.CDN
+		expect func(t *testing.T, diff []*config.CDN)
+	}{
+		{
+			name: "same cdn list",
+			cx: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+			},
+			cy: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+			},
+			expect: func(t *testing.T, diff []*config.CDN) {
+				assert := assert.New(t)
+				assert.EqualValues(diff, []*config.CDN(nil))
+			},
+		},
+		{
+			name: "different hostname",
+			cx: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "bar",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+			},
+			cy: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+			},
+			expect: func(t *testing.T, diff []*config.CDN) {
+				assert := assert.New(t)
+				assert.EqualValues(diff, []*config.CDN{
+					{
+						ID:       1,
+						Hostname: "bar",
+						IP:       "127.0.0.1",
+						Port:     8080,
+					},
+				})
+			},
+		},
+		{
+			name: "different port",
+			cx: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "127.0.0.1",
+					Port:     8081,
+				},
+			},
+			cy: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+			},
+			expect: func(t *testing.T, diff []*config.CDN) {
+				assert := assert.New(t)
+				assert.EqualValues(diff, []*config.CDN{
+					{
+						ID:       1,
+						Hostname: "foo",
+						IP:       "127.0.0.1",
+						Port:     8081,
+					},
+				})
+			},
+		},
+		{
+			name: "different ip",
+			cx: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "0.0.0.0",
+					Port:     8080,
+				},
+			},
+			cy: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+			},
+			expect: func(t *testing.T, diff []*config.CDN) {
+				assert := assert.New(t)
+				assert.EqualValues(diff, []*config.CDN{
+					{
+						ID:       1,
+						Hostname: "foo",
+						IP:       "0.0.0.0",
+						Port:     8080,
+					},
+				})
+			},
+		},
+		{
+			name: "remove y cdn",
+			cx: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+				{
+					ID:       2,
+					Hostname: "bar",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+			},
+			cy: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+			},
+			expect: func(t *testing.T, diff []*config.CDN) {
+				assert := assert.New(t)
+				assert.EqualValues(diff, []*config.CDN{
+					{
+						ID:       2,
+						Hostname: "bar",
+						IP:       "127.0.0.1",
+						Port:     8080,
+					},
+				})
+			},
+		},
+		{
+			name: "remove x cdn",
+			cx: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "foo",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+			},
+			cy: []*config.CDN{
+				{
+					ID:       1,
+					Hostname: "baz",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+				{
+					ID:       2,
+					Hostname: "bar",
+					IP:       "127.0.0.1",
+					Port:     8080,
+				},
+			},
+			expect: func(t *testing.T, diff []*config.CDN) {
+				assert := assert.New(t)
+				assert.EqualValues(diff, []*config.CDN{
+					{
+						ID:       1,
+						Hostname: "foo",
+						IP:       "127.0.0.1",
+						Port:     8080,
+					},
+				})
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			tc.expect(t, diffCDNs(tc.cx, tc.cy))
+		})
+	}
+}
+
 func TestCDNClient_getCDNIPs(t *testing.T) {
 	tests := []struct {
 		name   string
diff --git a/scheduler/resource/host_test.go b/scheduler/resource/host_test.go
index 614ccd09..c68a6922 100644
--- a/scheduler/resource/host_test.go
+++ b/scheduler/resource/host_test.go
@@ -361,6 +361,54 @@ func TestHost_LenPeers(t *testing.T) {
 	}
 }
 
+func TestHost_LeavePeers(t *testing.T) {
+	tests := []struct {
+		name    string
+		rawHost *scheduler.PeerHost
+		options []HostOption
+		expect  func(t *testing.T, host *Host, mockPeer *Peer)
+	}{
+		{
+			name:    "leave peers",
+			rawHost: mockRawHost,
+			expect: func(t *testing.T, host *Host, mockPeer *Peer) {
+				assert := assert.New(t)
+				host.StorePeer(mockPeer)
+				assert.Equal(host.LenPeers(), 1)
+				host.LeavePeers()
+				host.Peers.Range(func(_, value interface{}) bool {
+					peer := value.(*Peer)
+					assert.True(peer.FSM.Is(PeerStateLeave))
+					return true
+				})
+			},
+		},
+		{
+			name:    "peers is empty ",
+			rawHost: mockRawHost,
+			expect: func(t *testing.T, host *Host, mockPeer *Peer) {
+				assert := assert.New(t)
+				assert.Equal(host.LenPeers(), 0)
+				host.LeavePeers()
+				host.Peers.Range(func(_, value interface{}) bool {
+					assert.Fail("host peers is not empty")
+					return true
+				})
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			host := NewHost(tc.rawHost, tc.options...)
+			mockTask := NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
+			mockPeer := NewPeer(mockPeerID, mockTask, host)
+
+			tc.expect(t, host, mockPeer)
+		})
+	}
+}
+
 func TestHost_FreeUploadLoad(t *testing.T) {
 	tests := []struct {
 		name    string
diff --git a/scheduler/resource/task_test.go b/scheduler/resource/task_test.go
index d0e125a4..a684f6ba 100644
--- a/scheduler/resource/task_test.go
+++ b/scheduler/resource/task_test.go
@@ -359,6 +359,58 @@ func TestTask_LenPeers(t *testing.T) {
 	}
 }
 
+func TestTask_LenAvailablePeers(t *testing.T) {
+	tests := []struct {
+		name              string
+		id                string
+		urlMeta           *base.UrlMeta
+		url               string
+		backToSourceLimit int
+		expect            func(t *testing.T, task *Task, mockPeer *Peer)
+	}{
+		{
+			name:              "len available peers",
+			id:                mockTaskID,
+			urlMeta:           mockTaskURLMeta,
+			url:               mockTaskURL,
+			backToSourceLimit: mockTaskBackToSourceLimit,
+			expect: func(t *testing.T, task *Task, mockPeer *Peer) {
+				assert := assert.New(t)
+				task.StorePeer(mockPeer)
+				mockPeer.ID = idgen.PeerID("0.0.0.0")
+				task.StorePeer(mockPeer)
+				assert.Equal(task.LenAvailablePeers(), 2)
+				task.StorePeer(mockPeer)
+				assert.Equal(task.LenAvailablePeers(), 2)
+				mockPeer.FSM.SetState(PeerStateLeave)
+				task.StorePeer(mockPeer)
+				assert.Equal(task.LenAvailablePeers(), 0)
+			},
+		},
+		{
+			name:              "peer does not exist",
+			id:                mockTaskID,
+			urlMeta:           mockTaskURLMeta,
+			url:               mockTaskURL,
+			backToSourceLimit: mockTaskBackToSourceLimit,
+			expect: func(t *testing.T, task *Task, mockPeer *Peer) {
+				assert := assert.New(t)
+				assert.Equal(task.LenAvailablePeers(), 0)
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			mockHost := NewHost(mockRawHost)
+			task := NewTask(mockTaskID, mockTaskURL, mockTaskBackToSourceLimit, mockTaskURLMeta)
+			mockPeer := NewPeer(mockPeerID, task, mockHost)
+
+			tc.expect(t, task, mockPeer)
+		})
+	}
+}
+
 func TestTask_CDNPeers(t *testing.T) {
 	tests := []struct {
 		name   string
diff --git a/scheduler/service/service_test.go b/scheduler/service/service_test.go
index f2960af1..569dc736 100644
--- a/scheduler/service/service_test.go
+++ b/scheduler/service/service_test.go
@@ -138,6 +138,8 @@ func TestService_RegisterTask(t *testing.T) {
 			},
 			run: func(svc Service, req *rpcscheduler.PeerTaskRequest, mockTask *resource.Task, mockPeer *resource.Peer, taskManager resource.TaskManager, cdn resource.CDN, mr *resource.MockResourceMockRecorder, mt *resource.MockTaskManagerMockRecorder, mc *resource.MockCDNMockRecorder) {
 				mockTask.FSM.SetState(resource.TaskStateRunning)
+				mockTask.StorePeer(mockPeer)
+				mockPeer.FSM.SetState(resource.PeerStateRunning)
 				gomock.InOrder(
 					mr.TaskManager().Return(taskManager).Times(1),
 					mt.LoadOrStore(gomock.Any()).Return(mockTask, true).Times(1),
@@ -157,6 +159,8 @@ func TestService_RegisterTask(t *testing.T) {
 			},
 			run: func(svc Service, req *rpcscheduler.PeerTaskRequest, mockTask *resource.Task, mockPeer *resource.Peer, taskManager resource.TaskManager, cdn resource.CDN, mr *resource.MockResourceMockRecorder, mt *resource.MockTaskManagerMockRecorder, mc *resource.MockCDNMockRecorder) {
 				mockTask.FSM.SetState(resource.TaskStateSucceeded)
+				mockTask.StorePeer(mockPeer)
+				mockPeer.FSM.SetState(resource.PeerStateRunning)
 				gomock.InOrder(
 					mr.TaskManager().Return(taskManager).Times(1),
 					mt.LoadOrStore(gomock.Any()).Return(mockTask, true).Times(1),
diff --git a/test/e2e/concurrency_test.go b/test/e2e/concurrency_test.go
index b97fad01..597143a8 100644
--- a/test/e2e/concurrency_test.go
+++ b/test/e2e/concurrency_test.go
@@ -19,8 +19,8 @@ package e2e
 import (
 	"fmt"
 
-	. "github.com/onsi/ginkgo" //nolint
-	. "github.com/onsi/gomega" //nolint
+	. "github.com/onsi/ginkgo/v2" //nolint
+	. "github.com/onsi/gomega"    //nolint
 
 	"d7y.io/dragonfly/v2/test/e2e/e2eutil"
 )
diff --git a/test/e2e/containerd_test.go b/test/e2e/containerd_test.go
index 3c48afd6..7c7e8c5b 100644
--- a/test/e2e/containerd_test.go
+++ b/test/e2e/containerd_test.go
@@ -19,8 +19,8 @@ package e2e
 import (
 	"fmt"
 
-	. "github.com/onsi/ginkgo" //nolint
-	. "github.com/onsi/gomega" //nolint
+	. "github.com/onsi/ginkgo/v2" //nolint
+	. "github.com/onsi/gomega"    //nolint
 
 	"d7y.io/dragonfly/v2/test/e2e/e2eutil"
 )
diff --git a/test/e2e/dfget_test.go b/test/e2e/dfget_test.go
index 6ff78836..10b0f91d 100644
--- a/test/e2e/dfget_test.go
+++ b/test/e2e/dfget_test.go
@@ -21,8 +21,8 @@ import (
 	"strings"
 	"time"
 
-	. "github.com/onsi/ginkgo" //nolint
-	. "github.com/onsi/gomega" //nolint
+	. "github.com/onsi/ginkgo/v2" //nolint
+	. "github.com/onsi/gomega"    //nolint
 
 	"d7y.io/dragonfly/v2/test/e2e/e2eutil"
 )
diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go
index 9bc23222..e8546262 100644
--- a/test/e2e/e2e_test.go
+++ b/test/e2e/e2e_test.go
@@ -23,8 +23,8 @@ import (
 	"strings"
 	"testing"
 
-	. "github.com/onsi/ginkgo" //nolint
-	. "github.com/onsi/gomega" //nolint
+	. "github.com/onsi/ginkgo/v2" //nolint
+	. "github.com/onsi/gomega"    //nolint
 
 	"d7y.io/dragonfly/v2/test/e2e/e2eutil"
 	_ "d7y.io/dragonfly/v2/test/e2e/manager"
diff --git a/test/e2e/manager/preheat.go b/test/e2e/manager/preheat.go
index c0f7fbc7..e6fcf486 100644
--- a/test/e2e/manager/preheat.go
+++ b/test/e2e/manager/preheat.go
@@ -25,8 +25,8 @@ import (
 	"time"
 
 	machineryv1tasks "github.com/RichardKnop/machinery/v1/tasks"
-	. "github.com/onsi/ginkgo" //nolint
-	. "github.com/onsi/gomega" //nolint
+	. "github.com/onsi/ginkgo/v2" //nolint
+	. "github.com/onsi/gomega"    //nolint
 
 	internaljob "d7y.io/dragonfly/v2/internal/job"
 	"d7y.io/dragonfly/v2/manager/model"