refactor: refacts the seed peer selection logic from grpc resolver to self pick

Signed-off-by: chlins <chlins.zhang@gmail.com>

Author: chlins

scheduler/job/job.go

@@ -39,12 +39,13 @@ import (
 	commonv1 "d7y.io/api/v2/pkg/apis/common/v1"
 	commonv2 "d7y.io/api/v2/pkg/apis/common/v2"
 	dfdaemonv2 "d7y.io/api/v2/pkg/apis/dfdaemon/v2"
-	managerv2 "d7y.io/api/v2/pkg/apis/manager/v2"
 
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	internaljob "d7y.io/dragonfly/v2/internal/job"
 	managertypes "d7y.io/dragonfly/v2/manager/types"
+	"d7y.io/dragonfly/v2/pkg/dfnet"
 	"d7y.io/dragonfly/v2/pkg/idgen"
+	cndsystemclient "d7y.io/dragonfly/v2/pkg/rpc/cdnsystem/client"
 	dfdaemonclient "d7y.io/dragonfly/v2/pkg/rpc/dfdaemon/client"
 	"d7y.io/dragonfly/v2/scheduler/config"
 	resource "d7y.io/dragonfly/v2/scheduler/resource/standard"
@@ -79,10 +80,11 @@ type job struct {
 	localJob     *internaljob.Job
 	resource     resource.Resource
 	config       *config.Config
+	dialOptions  []grpc.DialOption
 }
 
 // New creates a new Job.
-func New(cfg *config.Config, resource resource.Resource) (Job, error) {
+func New(cfg *config.Config, resource resource.Resource, dialOptions ...grpc.DialOption) (Job, error) {
 	redisConfig := &internaljob.Config{
 		Addrs:            cfg.Database.Redis.Addrs,
 		MasterName:       cfg.Database.Redis.MasterName,
@@ -127,6 +129,7 @@ func New(cfg *config.Config, resource resource.Resource) (Job, error) {
 		localJob:     localJob,
 		resource:     resource,
 		config:       cfg,
+		dialOptions:  dialOptions,
 	}
 
 	namedJobFuncs := map[string]any{
@@ -241,11 +244,6 @@ func (j *job) PreheatSingleSeedPeer(ctx context.Context, req *internaljob.Prehea
 		return nil, fmt.Errorf("cluster %d scheduler %s has disabled seed peer", j.config.Manager.SchedulerClusterID, j.config.Server.AdvertiseIP)
 	}
 
-	// If scheduler has no available seed peer, return error.
-	if len(j.resource.SeedPeer().Client().Addrs()) == 0 {
-		return nil, fmt.Errorf("cluster %d scheduler %s has no available seed peer", j.config.Manager.SchedulerClusterID, j.config.Server.AdvertiseIP)
-	}
-
 	// Preheat by v2 grpc protocol. If seed peer does not support
 	// v2 protocol, preheat by v1 grpc protocol.
 	resp, err := j.preheatV2SingleSeedPeer(ctx, req, log)
@@ -275,8 +273,22 @@ func (j *job) preheatV1SingleSeedPeer(ctx context.Context, req *internaljob.Preh
 		Priority:    commonv1.Priority(req.Priority),
 	}
 
+	selectedSeedPeer, err := j.resource.SeedPeer().SelectSeedPeer(ctx, taskID)
+	if err != nil {
+		return nil, err
+	}
+
+	log.Infof("[preheat]: selected seed peer %s", selectedSeedPeer)
+
+	// TODO(chlins): reuse the client if we encounter the performance issue in future.
+	client, err := cndsystemclient.GetClientByAddr(ctx, dfnet.NetAddr{Type: dfnet.TCP, Addr: selectedSeedPeer}, j.dialOptions...)
+	if err != nil {
+		return nil, err
+	}
+	defer client.Close()
+
 	// Trigger seed peer download seeds.
-	stream, err := j.resource.SeedPeer().Client().ObtainSeeds(ctx, &cdnsystemv1.SeedRequest{
+	stream, err := client.ObtainSeeds(ctx, &cdnsystemv1.SeedRequest{
 		TaskId:  taskID,
 		Url:     req.URL,
 		UrlMeta: urlMeta,
@@ -343,7 +355,22 @@ func (j *job) preheatV2SingleSeedPeerByURL(ctx context.Context, url string, req
 	filteredQueryParams := idgen.ParseFilteredQueryParams(req.FilteredQueryParams)
 	taskID := idgen.TaskIDV2ByURLBased(url, req.PieceLength, req.Tag, req.Application, filteredQueryParams)
 	advertiseIP := j.config.Server.AdvertiseIP.String()
-	stream, err := j.resource.SeedPeer().Client().DownloadTask(ctx, taskID, &dfdaemonv2.DownloadTaskRequest{
+
+	selectedSeedPeer, err := j.resource.SeedPeer().SelectSeedPeer(ctx, taskID)
+	if err != nil {
+		return nil, err
+	}
+
+	log.Infof("[preheat]: selected seed peer %s", selectedSeedPeer)
+
+	// TODO(chlins): reuse the client if we encounter the performance issue in future.
+	client, err := dfdaemonclient.GetV2ByAddr(ctx, selectedSeedPeer, j.dialOptions...)
+	if err != nil {
+		return nil, err
+	}
+	defer client.Close()
+
+	stream, err := client.DownloadTask(ctx, taskID, &dfdaemonv2.DownloadTaskRequest{
 		Download: &commonv2.Download{
 			Url:                 url,
 			PieceLength:         req.PieceLength,
@@ -410,7 +437,7 @@ func (j *job) PreheatAllSeedPeers(ctx context.Context, req *internaljob.PreheatR
 			for _, seedPeer := range seedPeers {
 				var (
 					hostname = seedPeer.Hostname
-					ip       = seedPeer.Ip
+					ip       = seedPeer.IP
 					port     = seedPeer.Port
 				)
 
@@ -554,20 +581,20 @@ func (j *job) PreheatAllSeedPeers(ctx context.Context, req *internaljob.PreheatR
 // 2. Count: If count is provided, selects up to the specified number of seed peers. If count exceeds the number of available seed peers, all seed peers are selected.
 // 3. Percentage: If percentage is provided, selects a proportional number of seed peers (rounded down). Ensures at least one seed peer is selected if percentage > 0.
 // Priority: IPs > Count > Percentage
-func (j *job) selectSeedPeers(ips []string, count *uint32, percentage *uint32, log *logger.SugaredLoggerOnWith) ([]*managerv2.SeedPeer, error) {
+func (j *job) selectSeedPeers(ips []string, count *uint32, percentage *uint32, log *logger.SugaredLoggerOnWith) ([]*resource.Host, error) {
 	if !j.config.SeedPeer.Enable {
 		return nil, fmt.Errorf("cluster %d scheduler %s has disabled seed peer", j.config.Manager.SchedulerClusterID, j.config.Server.AdvertiseIP)
 	}
 
-	seedPeers := j.resource.SeedPeer().Client().SeedPeers()
+	seedPeers := j.resource.HostManager().LoadAllSeeds()
 	if len(seedPeers) == 0 {
 		return nil, fmt.Errorf("cluster %d scheduler %s has no available seed peer", j.config.Manager.SchedulerClusterID, j.config.Server.AdvertiseIP)
 	}
 
 	if len(ips) > 0 {
-		selectedSeedPeers := make([]*managerv2.SeedPeer, 0, len(ips))
+		selectedSeedPeers := make([]*resource.Host, 0, len(ips))
 		for _, seedPeer := range seedPeers {
-			if slices.Contains(ips, seedPeer.Ip) {
+			if slices.Contains(ips, seedPeer.IP) {
 				selectedSeedPeers = append(selectedSeedPeers, seedPeer)
 				continue
 			}

scheduler/resource/standard/host_manager.go

@@ -63,6 +63,9 @@ type HostManager interface {
 	// LoadAllNormals loads all normal hosts through the Range of sync.Map.
 	LoadAllNormals() []*Host
 
+	// LoadAllSeeds loads all seed hosts through the Range of sync.Map.
+	LoadAllSeeds() []*Host
+
 	// Try to reclaim host.
 	RunGC(context.Context) error
 }
@@ -163,6 +166,27 @@ func (h *hostManager) LoadAllNormals() []*Host {
 	return hosts
 }
 
+// LoadAllSeeds loads all seed hosts through the Range of sync.Map.
+func (h *hostManager) LoadAllSeeds() []*Host {
+	hosts := make([]*Host, 0)
+	h.Map.Range(func(key, value any) bool {
+		host, ok := value.(*Host)
+		if !ok {
+			host.Log.Error("invalid host")
+			return true
+		}
+
+		if host.Type == types.HostTypeNormal {
+			return true
+		}
+
+		hosts = append(hosts, host)
+		return true
+	})
+
+	return hosts
+}
+
 // LoadRandom loads host randomly through the Range of sync.Map.
 func (h *hostManager) LoadRandom(n int, blocklist set.SafeSet[string]) []*Host {
 	hosts := make([]*Host, 0, n)

scheduler/resource/standard/host_manager_mock.go

@@ -41,6 +41,9 @@ type Resource interface {
 	// Task manager interface.
 	TaskManager() TaskManager
 
+	// Serve serves resource service.
+	Serve() error
+
 	// Stop resource service.
 	Stop() error
 }
@@ -64,7 +67,7 @@ type resource struct {
 }
 
 // New returns Resource interface.
-func New(cfg *config.Config, gc gc.GC, dynconfig config.DynconfigInterface, transportCredentials credentials.TransportCredentials) (Resource, error) {
+func New(cfg *config.Config, gc gc.GC, transportCredentials credentials.TransportCredentials) (Resource, error) {
 	resource := &resource{config: cfg}
 
 	// Initialize host manager interface.
@@ -91,12 +94,7 @@ func New(cfg *config.Config, gc gc.GC, dynconfig config.DynconfigInterface, tran
 	// Initialize seed peer interface.
 	if cfg.SeedPeer.Enable {
 		dialOptions := []grpc.DialOption{grpc.WithStatsHandler(otelgrpc.NewClientHandler()), grpc.WithTransportCredentials(transportCredentials)}
-		client, err := newSeedPeerClient(dynconfig, hostManager, dialOptions...)
-		if err != nil {
-			return nil, err
-		}
-
-		resource.seedPeer = newSeedPeer(client, peerManager, hostManager)
+		resource.seedPeer = newSeedPeer(peerManager, hostManager, dialOptions...)
 	}
 
 	return resource, nil
@@ -122,6 +120,15 @@ func (r *resource) TaskManager() TaskManager {
 	return r.taskManager
 }
 
+// Serve serves resource service.
+func (r *resource) Serve() error {
+	if r.config.SeedPeer.Enable {
+		return r.seedPeer.Serve()
+	}
+
+	return nil
+}
+
 // Stop resource service.
 func (r *resource) Stop() error {
 	if r.config.SeedPeer.Enable {

scheduler/resource/standard/resource.go

@@ -23,43 +23,25 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/mock/gomock"
-	"google.golang.org/grpc/resolver"
-
-	managerv2 "d7y.io/api/v2/pkg/apis/manager/v2"
 
 	"d7y.io/dragonfly/v2/pkg/gc"
 	"d7y.io/dragonfly/v2/pkg/rpc"
 	"d7y.io/dragonfly/v2/scheduler/config"
-	configmocks "d7y.io/dragonfly/v2/scheduler/config/mocks"
 )
 
 func TestResource_New(t *testing.T) {
 	tests := []struct {
 		name   string
 		config *config.Config
-		mock   func(mg *gc.MockGCMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder)
+		mock   func(mg *gc.MockGCMockRecorder)
 		expect func(t *testing.T, resource Resource, err error)
 	}{
 		{
 			name:   "new resource",
 			config: config.New(),
-			mock: func(mg *gc.MockGCMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
+			mock: func(mg *gc.MockGCMockRecorder) {
 				gomock.InOrder(
 					mg.Add(gomock.Any()).Return(nil).Times(3),
-					md.Get().Return(&config.DynconfigData{
-						Scheduler: &managerv2.Scheduler{
-							SeedPeers: []*managerv2.SeedPeer{
-								{
-									Id: 1,
-								},
-							},
-						},
-					}, nil).Times(1),
-					md.Register(gomock.Any()).Return().Times(1),
-					md.GetResolveSeedPeerAddrs().Return([]resolver.Address{}, nil).Times(1),
-					md.Register(gomock.Any()).Return().Times(1),
-					md.GetResolveSeedPeerAddrs().Return([]resolver.Address{}, nil).Times(1),
-					md.Register(gomock.Any()).Return().Times(1),
 				)
 			},
 			expect: func(t *testing.T, resource Resource, err error) {
@@ -71,7 +53,7 @@ func TestResource_New(t *testing.T) {
 		{
 			name:   "new resource failed because of host manager error",
 			config: config.New(),
-			mock: func(mg *gc.MockGCMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
+			mock: func(mg *gc.MockGCMockRecorder) {
 				mg.Add(gomock.Any()).Return(errors.New("foo")).Times(1)
 			},
 			expect: func(t *testing.T, resource Resource, err error) {
@@ -82,7 +64,7 @@ func TestResource_New(t *testing.T) {
 		{
 			name:   "new resource failed because of task manager error",
 			config: config.New(),
-			mock: func(mg *gc.MockGCMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
+			mock: func(mg *gc.MockGCMockRecorder) {
 				gomock.InOrder(
 					mg.Add(gomock.Any()).Return(nil).Times(1),
 					mg.Add(gomock.Any()).Return(errors.New("foo")).Times(1),
@@ -96,7 +78,7 @@ func TestResource_New(t *testing.T) {
 		{
 			name:   "new resource failed because of peer manager error",
 			config: config.New(),
-			mock: func(mg *gc.MockGCMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
+			mock: func(mg *gc.MockGCMockRecorder) {
 				gomock.InOrder(
 					mg.Add(gomock.Any()).Return(nil).Times(2),
 					mg.Add(gomock.Any()).Return(errors.New("foo")).Times(1),
@@ -107,36 +89,12 @@ func TestResource_New(t *testing.T) {
 				assert.EqualError(err, "foo")
 			},
 		},
-		{
-			name:   "new resource faild because of dynconfig get error",
-			config: config.New(),
-			mock: func(mg *gc.MockGCMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
-				gomock.InOrder(
-					mg.Add(gomock.Any()).Return(nil).Times(3),
-					md.Get().Return(&config.DynconfigData{}, errors.New("foo")).Times(1),
-				)
-			},
-			expect: func(t *testing.T, resource Resource, err error) {
-				assert := assert.New(t)
-				assert.EqualError(err, "foo")
-			},
-		},
 		{
 			name:   "new resource faild because of seed peer list is empty",
 			config: config.New(),
-			mock: func(mg *gc.MockGCMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
+			mock: func(mg *gc.MockGCMockRecorder) {
 				gomock.InOrder(
 					mg.Add(gomock.Any()).Return(nil).Times(3),
-					md.Get().Return(&config.DynconfigData{
-						Scheduler: &managerv2.Scheduler{
-							SeedPeers: []*managerv2.SeedPeer{},
-						},
-					}, nil).Times(1),
-					md.Register(gomock.Any()).Return().Times(1),
-					md.GetResolveSeedPeerAddrs().Return([]resolver.Address{}, nil).Times(1),
-					md.Register(gomock.Any()).Return().Times(1),
-					md.GetResolveSeedPeerAddrs().Return([]resolver.Address{}, nil).Times(1),
-					md.Register(gomock.Any()).Return().Times(1),
 				)
 			},
 			expect: func(t *testing.T, resource Resource, err error) {
@@ -159,7 +117,7 @@ func TestResource_New(t *testing.T) {
 					Enable: false,
 				},
 			},
-			mock: func(mg *gc.MockGCMockRecorder, md *configmocks.MockDynconfigInterfaceMockRecorder) {
+			mock: func(mg *gc.MockGCMockRecorder) {
 				mg.Add(gomock.Any()).Return(nil).Times(3)
 			},
 			expect: func(t *testing.T, resource Resource, err error) {
@@ -175,10 +133,9 @@ func TestResource_New(t *testing.T) {
 			ctl := gomock.NewController(t)
 			defer ctl.Finish()
 			gc := gc.NewMockGC(ctl)
-			dynconfig := configmocks.NewMockDynconfigInterface(ctl)
-			tc.mock(gc.EXPECT(), dynconfig.EXPECT())
+			tc.mock(gc.EXPECT())
 
-			resource, err := New(tc.config, gc, dynconfig, rpc.NewInsecureCredentials())
+			resource, err := New(tc.config, gc, rpc.NewInsecureCredentials())
 			tc.expect(t, resource, err)
 		})
 	}