feat: implement v2 version of scheduler service

Implement StatTask, AnnounceHost and LeaveHost api in scheduler service v2. Signed-off-by: Gaius <[email protected]>
dragonflyoss · Feb 28, 2023 · c0a11ef · c0a11ef
1 parent de2a7c9
commit c0a11ef
Show file tree

Hide file tree

Showing 11 changed files with 1,373 additions and 59 deletions.
diff --git a/client/daemon/peer/peertask_manager_mock.go b/client/daemon/peer/peertask_manager_mock.go
diff --git a/client/daemon/peer/piece_manager_mock.go b/client/daemon/peer/piece_manager_mock.go
diff --git a/pkg/types/types.go b/pkg/types/types.go
@@ -109,8 +109,8 @@ const (
 )
 
 // Name returns the name of host type.
-func (h *HostType) Name() string {
-	switch *h {
+func (h HostType) Name() string {
+	switch h {
 	case HostTypeSuperSeed:
 		return HostTypeSuperSeedName
 	case HostTypeStrongSeed:

diff --git a/scheduler/metrics/metrics.go b/scheduler/metrics/metrics.go
@@ -88,6 +88,20 @@ var (
 		Help:      "Counter of the number of failed of the leaving peer.",
 	})
 
+	ExchangePeerCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: types.MetricsNamespace,
+		Subsystem: types.SchedulerMetricsName,
+		Name:      "exchange_peer_total",
+		Help:      "Counter of the number of the leaving peer.",
+	})
+
+	ExchangePeerFailureCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: types.MetricsNamespace,
+		Subsystem: types.SchedulerMetricsName,
+		Name:      "exchange_peer_failure_total",
+		Help:      "Counter of the number of failed of the leaving peer.",
+	})
+
 	RegisterTaskCount = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: types.MetricsNamespace,
 		Subsystem: types.SchedulerMetricsName,
@@ -152,12 +166,13 @@ var (
 	}, []string{"os", "platform", "platform_family", "platform_version",
 		"kernel_version", "git_version", "git_commit", "go_version", "build_platform"})
 
-	AnnounceHostFailureCount = promauto.NewCounter(prometheus.CounterOpts{
+	AnnounceHostFailureCount = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: types.MetricsNamespace,
 		Subsystem: types.SchedulerMetricsName,
 		Name:      "announce_host_failure_total",
 		Help:      "Counter of the number of failed of the announce host.",
-	})
+	}, []string{"os", "platform", "platform_family", "platform_version",
+		"kernel_version", "git_version", "git_commit", "go_version", "build_platform"})
 
 	LeaveTaskCount = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: types.MetricsNamespace,

diff --git a/scheduler/resource/seed_peer_mock.go b/scheduler/resource/seed_peer_mock.go
diff --git a/scheduler/rpcserver/scheduler_server_v1.go b/scheduler/rpcserver/scheduler_server_v1.go
@@ -86,7 +86,7 @@ func (s *schedulerServerV1) AnnounceTask(ctx context.Context, req *schedulerv1.A
 	metrics.AnnounceTaskCount.Inc()
 	if err := s.service.AnnounceTask(ctx, req); err != nil {
 		metrics.AnnounceTaskFailureCount.Inc()
-		return new(emptypb.Empty), err
+		return nil, err
 	}
 
 	return new(emptypb.Empty), nil
@@ -95,13 +95,13 @@ func (s *schedulerServerV1) AnnounceTask(ctx context.Context, req *schedulerv1.A
 // StatTask checks if the given task exists.
 func (s *schedulerServerV1) StatTask(ctx context.Context, req *schedulerv1.StatTaskRequest) (*schedulerv1.Task, error) {
 	metrics.StatTaskCount.Inc()
-	task, err := s.service.StatTask(ctx, req)
+	resp, err := s.service.StatTask(ctx, req)
 	if err != nil {
 		metrics.StatTaskFailureCount.Inc()
 		return nil, err
 	}
 
-	return task, nil
+	return resp, nil
 }
 
 // LeaveTask makes the peer unschedulable.
@@ -114,8 +114,9 @@ func (s *schedulerServerV1) AnnounceHost(ctx context.Context, req *schedulerv1.A
 	metrics.AnnounceHostCount.WithLabelValues(req.Os, req.Platform, req.PlatformFamily, req.PlatformVersion,
 		req.KernelVersion, req.Build.GitVersion, req.Build.GitCommit, req.Build.GoVersion, req.Build.Platform).Inc()
 	if err := s.service.AnnounceHost(ctx, req); err != nil {
-		metrics.AnnounceHostFailureCount.Inc()
-		return new(emptypb.Empty), err
+		metrics.AnnounceHostFailureCount.WithLabelValues(req.Os, req.Platform, req.PlatformFamily, req.PlatformVersion,
+			req.KernelVersion, req.Build.GitVersion, req.Build.GitCommit, req.Build.GoVersion, req.Build.Platform).Inc()
+		return nil, err
 	}
 
 	return new(emptypb.Empty), nil
@@ -126,7 +127,7 @@ func (s *schedulerServerV1) LeaveHost(ctx context.Context, req *schedulerv1.Leav
 	metrics.LeaveHostCount.Inc()
 	if err := s.service.LeaveHost(ctx, req); err != nil {
 		metrics.LeaveHostFailureCount.Inc()
-		return new(emptypb.Empty), err
+		return nil, err
 	}
 
 	return new(emptypb.Empty), nil

diff --git a/scheduler/rpcserver/scheduler_server_v2.go b/scheduler/rpcserver/scheduler_server_v2.go
@@ -58,45 +58,72 @@ func (s *schedulerServerV2) AnnouncePeer(stream schedulerv2.Scheduler_AnnouncePe
 // StatPeer checks information of peer.
 func (s *schedulerServerV2) StatPeer(ctx context.Context, req *schedulerv2.StatPeerRequest) (*commonv2.Peer, error) {
 	metrics.StatPeerCount.Inc()
-	peer, err := s.service.StatPeer(ctx, req)
+	resp, err := s.service.StatPeer(ctx, req)
 	if err != nil {
 		metrics.StatPeerFailureCount.Inc()
 		return nil, err
 	}
 
-	return peer, nil
+	return resp, nil
 }
 
 // LeavePeer releases peer in scheduler.
 func (s *schedulerServerV2) LeavePeer(ctx context.Context, req *schedulerv2.LeavePeerRequest) (*emptypb.Empty, error) {
 	metrics.LeavePeerCount.Inc()
 	if err := s.service.LeavePeer(ctx, req); err != nil {
 		metrics.LeavePeerFailureCount.Inc()
-		return new(emptypb.Empty), err
+		return nil, err
 	}
 
 	return new(emptypb.Empty), nil
 }
 
-// TODO exchange peer api definition.
 // ExchangePeer exchanges peer information.
 func (s *schedulerServerV2) ExchangePeer(ctx context.Context, req *schedulerv2.ExchangePeerRequest) (*schedulerv2.ExchangePeerResponse, error) {
-	return nil, nil
+	metrics.ExchangePeerCount.Inc()
+	resp, err := s.service.ExchangePeer(ctx, req)
+	if err != nil {
+		metrics.ExchangePeerFailureCount.Inc()
+		return nil, err
+	}
+
+	return resp, nil
 }
 
-// Checks information of task.
+// StatTask checks information of task.
 func (s *schedulerServerV2) StatTask(ctx context.Context, req *schedulerv2.StatTaskRequest) (*commonv2.Task, error) {
-	return nil, nil
+	metrics.StatTaskCount.Inc()
+	resp, err := s.service.StatTask(ctx, req)
+	if err != nil {
+		metrics.StatTaskFailureCount.Inc()
+		return nil, err
+	}
+
+	return resp, nil
 }
 
 // AnnounceHost announces host to scheduler.
 func (s *schedulerServerV2) AnnounceHost(ctx context.Context, req *schedulerv2.AnnounceHostRequest) (*emptypb.Empty, error) {
-	return nil, nil
+	metrics.AnnounceHostCount.WithLabelValues(req.Host.Os, req.Host.Platform, req.Host.PlatformFamily, req.Host.PlatformVersion,
+		req.Host.KernelVersion, req.Host.Build.GitVersion, req.Host.Build.GitCommit, req.Host.Build.GoVersion, req.Host.Build.Platform).Inc()
+	if err := s.service.AnnounceHost(ctx, req); err != nil {
+		metrics.AnnounceHostFailureCount.WithLabelValues(req.Host.Os, req.Host.Platform, req.Host.PlatformFamily, req.Host.PlatformVersion,
+			req.Host.KernelVersion, req.Host.Build.GitVersion, req.Host.Build.GitCommit, req.Host.Build.GoVersion, req.Host.Build.Platform).Inc()
+		return nil, err
+	}
+
+	return new(emptypb.Empty), nil
 }
 
 // LeaveHost releases host in scheduler.
 func (s *schedulerServerV2) LeaveHost(ctx context.Context, req *schedulerv2.LeaveHostRequest) (*emptypb.Empty, error) {
-	return nil, nil
+	metrics.LeaveHostCount.Inc()
+	if err := s.service.LeaveHost(ctx, req); err != nil {
+		metrics.LeaveHostFailureCount.Inc()
+		return nil, err
+	}
+
+	return new(emptypb.Empty), nil
 }
 
 // SyncProbes sync probes of the host.

diff --git a/scheduler/service/service_v1.go b/scheduler/service/service_v1.go
@@ -538,6 +538,7 @@ func (v *V1) AnnounceHost(ctx context.Context, req *schedulerv1.AnnounceHostRequ
 		return nil
 	}
 
+	// Host already exists and updates properties.
 	host.Port = req.Port
 	host.DownloadPort = req.DownloadPort
 	host.Type = types.ParseHostType(req.Type)
@@ -548,6 +549,10 @@ func (v *V1) AnnounceHost(ctx context.Context, req *schedulerv1.AnnounceHostRequ
 	host.KernelVersion = req.KernelVersion
 	host.UpdatedAt.Store(time.Now())
 
+	if concurrentUploadLimit > 0 {
+		host.ConcurrentUploadLimit.Store(concurrentUploadLimit)
+	}
+
 	if req.Cpu != nil {
 		host.CPU = resource.CPU{
 			LogicalCount:   req.Cpu.LogicalCount,
@@ -612,10 +617,6 @@ func (v *V1) AnnounceHost(ctx context.Context, req *schedulerv1.AnnounceHostRequ
 		}
 	}
 
-	if concurrentUploadLimit > 0 {
-		host.ConcurrentUploadLimit.Store(concurrentUploadLimit)
-	}
-
 	return nil
 }