chore: update client version (#3267)

* chore: update client version

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

* feat: update client verison

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

---------

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

Author: gaius-qi

README.md

@@ -71,12 +71,6 @@ Join the conversation and help the community.
 - **Twitter**: [@dragonfly_oss](https://twitter.com/dragonfly_oss)
 - **DingTalk**: [22880028764](https://qr.dingtalk.com/action/joingroup?code=v1,k1,pkV9IbsSyDusFQdByPSK3HfCG61ZCLeb8b/lpQ3uUqI=&_dt_no_comment=1&origin=11)
 
-<!-- markdownlint-disable -->
-<div align="center">
-  <img src="docs/images/community/dingtalk-group.jpeg" width="300" title="dingtalk">
-</div>
-<!-- markdownlint-restore -->
-
 ## Contributing
 
 You should check out our

client-rs

@@ -3,7 +3,7 @@ module d7y.io/dragonfly/v2
 go 1.21
 
 require (
-	d7y.io/api/v2 v2.0.112
+	d7y.io/api/v2 v2.0.113
 	github.com/MysteriousPotato/go-lockable v1.0.0
 	github.com/RichardKnop/machinery v1.10.8
 	github.com/Showmax/go-fqdn v1.0.0

go.mod

@@ -51,8 +51,8 @@ cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0Zeo
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
-d7y.io/api/v2 v2.0.112 h1:I3HCB3GH1tWbp3pFSPIlx7r+XpPcl8UBcNz0A3Ftink=
-d7y.io/api/v2 v2.0.112/go.mod h1:L35KjMvZmedXQ6UGHieIddUT1ZZjUAJUqJ9lf52r4fw=
+d7y.io/api/v2 v2.0.113 h1:09Zo115B6wQI3Jn+RSUpLfzAmIA+0xgY3a03Ti6X+sw=
+d7y.io/api/v2 v2.0.113/go.mod h1:NMBc/M0ekCHgfn4wCJfLDXdDm/2wKMB0HOxpyUREGAk=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/azure-sdk-for-go v16.2.1+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v1.0.0/go.mod h1:uGG2W01BaETf0Ozp+QxxKJdMBNRWPdstHG0Fmdwn1/U=

go.sum

@@ -171,6 +171,12 @@ func WithTaskAndPeerID(taskID, peerID string) *SugaredLoggerOnWith {
 	}
 }
 
+func WithHostAndTaskID(hostID, taskID string) *SugaredLoggerOnWith {
+	return &SugaredLoggerOnWith{
+		withArgs: []any{"hostID", hostID, "taskID", taskID},
+	}
+}
+
 func WithHostnameAndIP(hostname, ip string) *SugaredLoggerOnWith {
 	return &SugaredLoggerOnWith{
 		withArgs: []any{"hostname", hostname, "ip", ip},

internal/dflog/logger.go

@@ -134,6 +134,9 @@ type V2 interface {
 	// Checks information of task.
 	StatTask(context.Context, *schedulerv2.StatTaskRequest, ...grpc.CallOption) (*commonv2.Task, error)
 
+	// LeaveTask releases task in scheduler.
+	LeaveTask(context.Context, *schedulerv2.LeaveTaskRequest, ...grpc.CallOption) error
+
 	// AnnounceHost announces host to scheduler.
 	AnnounceHost(context.Context, *schedulerv2.AnnounceHostRequest, ...grpc.CallOption) error
 
@@ -215,6 +218,20 @@ func (v *v2) StatTask(ctx context.Context, req *schedulerv2.StatTaskRequest, opt
 	)
 }
 
+// LeaveTask releases task in scheduler.
+func (v *v2) LeaveTask(ctx context.Context, req *schedulerv2.LeaveTaskRequest, opts ...grpc.CallOption) error {
+	ctx, cancel := context.WithTimeout(ctx, contextTimeout)
+	defer cancel()
+
+	_, err := v.SchedulerClient.LeaveTask(
+		context.WithValue(ctx, pkgbalancer.ContextKey, req.TaskId),
+		req,
+		opts...,
+	)
+
+	return err
+}
+
 // AnnounceHost announces host to scheduler.
 func (v *v2) AnnounceHost(ctx context.Context, req *schedulerv2.AnnounceHostRequest, opts ...grpc.CallOption) error {
 	ctx, cancel := context.WithTimeout(ctx, contextTimeout)

pkg/rpc/scheduler/client/client_v2.go

@@ -194,6 +194,20 @@ var (
 		Help:      "Counter of the number of failed of the stat task.",
 	})
 
+	LeaveTaskCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: types.MetricsNamespace,
+		Subsystem: types.SchedulerMetricsName,
+		Name:      "leave_task_total",
+		Help:      "Counter of the number of the leaving task.",
+	})
+
+	LeaveTaskFailureCount = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: types.MetricsNamespace,
+		Subsystem: types.SchedulerMetricsName,
+		Name:      "leave_task_failure_total",
+		Help:      "Counter of the number of failed of the leaving task.",
+	})
+
 	AnnounceHostCount = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: types.MetricsNamespace,
 		Subsystem: types.SchedulerMetricsName,

pkg/rpc/scheduler/client/mocks/client_v2_mock.go

@@ -124,6 +124,19 @@ func (s *schedulerServerV2) StatTask(ctx context.Context, req *schedulerv2.StatT
 	return resp, nil
 }
 
+// LeaveTask releases task in scheduler.
+func (s *schedulerServerV2) LeaveTask(ctx context.Context, req *schedulerv2.LeaveTaskRequest) (*emptypb.Empty, error) {
+	// Collect LeaseTaskCount metrics.
+	metrics.LeaveTaskCount.Inc()
+	if err := s.service.LeaveTask(ctx, req); err != nil {
+		// Collect LeaseTaskFailureCount metrics.
+		metrics.LeaveTaskFailureCount.Inc()
+		return nil, err
+	}
+
+	return new(emptypb.Empty), nil
+}
+
 // AnnounceHost announces host to scheduler.
 func (s *schedulerServerV2) AnnounceHost(ctx context.Context, req *schedulerv2.AnnounceHostRequest) (*emptypb.Empty, error) {
 	// Collect AnnounceHostCount metrics.

scheduler/metrics/metrics.go

@@ -462,6 +462,39 @@ func (v *V2) StatTask(ctx context.Context, req *schedulerv2.StatTaskRequest) (*c
 	return resp, nil
 }
 
+// LeaveTask releases task in scheduler.
+func (v *V2) LeaveTask(ctx context.Context, req *schedulerv2.LeaveTaskRequest) error {
+	log := logger.WithHostAndTaskID(req.GetHostId(), req.GetTaskId())
+	log.Infof("leave task request: %#v", req)
+
+	host, loaded := v.resource.HostManager().Load(req.GetHostId())
+	if !loaded {
+		msg := fmt.Sprintf("host %s not found", req.GetHostId())
+		log.Error(msg)
+		return status.Error(codes.NotFound, msg)
+	}
+
+	host.Peers.Range(func(key, value any) bool {
+		peer, ok := value.(*resource.Peer)
+		if !ok {
+			host.Log.Errorf("invalid peer %s %#v", key, value)
+			return true
+		}
+
+		if peer.Task.ID == req.GetTaskId() {
+			if err := peer.FSM.Event(ctx, resource.PeerEventLeave); err != nil {
+				msg := fmt.Sprintf("peer fsm event failed: %s", err.Error())
+				peer.Log.Error(msg)
+				return true
+			}
+		}
+
+		return true
+	})
+
+	return nil
+}
+
 // AnnounceHost announces host to scheduler.
 func (v *V2) AnnounceHost(ctx context.Context, req *schedulerv2.AnnounceHostRequest) error {
 	logger.WithHostID(req.Host.GetId()).Infof("announce host request: %#v", req.GetHost())