feat: update delete task job and list tasks with concurrency and filter valid tasks.

Signed-off-by: Asklv <boironic@gmail.com>

Author: IRONICBo

scheduler/job/job.go

@@ -23,11 +23,18 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/RichardKnop/machinery/v1"
 	"github.com/go-playground/validator/v10"
+	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
+	grpc_zap "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap"
+	grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry"
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -46,10 +53,14 @@ import (
 const (
 	// preheatTimeout is timeout of preheating.
 	preheatTimeout = 20 * time.Minute
-	// listTasksTimeout is timeout of listing tasks.
-	listTasksTimeout = 10 * time.Minute
 	// deleteTaskTimeout is timeout of deleting task.
 	deleteTaskTimeout = 20 * time.Minute
+	// deleteTaskConcurrency is the number of concurrent delete tasks.
+	deleteTaskConcurrency = 10
+	// deleteTaskMaxRetries is the maximum number of retries for delete tasks.
+	deleteTaskMaxRetries = 3
+	// deleteTaskBackoffWaitBetween is waiting for a fixed period of time between calls in backoff linear.
+	deleteTaskBackoffWaitBetween = 500 * time.Millisecond
 )
 
 // Job is an interface for job.
@@ -306,12 +317,6 @@ func (j *job) syncPeers() (string, error) {
 
 // listTasks is a job to list tasks.
 func (j *job) listTasks(ctx context.Context, data string) (string, error) {
-	// TODO:
-	// 1. query all peers with task id
-	// 2. delete current task by task id and host id
-	ctx, cancel := context.WithTimeout(ctx, listTasksTimeout)
-	defer cancel()
-
 	req := &internaljob.ListTasksRequest{}
 	if err := internaljob.UnmarshalRequest(data, req); err != nil {
 		logger.Errorf("unmarshal request err: %s, request body: %s", err.Error(), data)
@@ -324,26 +329,21 @@ func (j *job) listTasks(ctx context.Context, data string) (string, error) {
 	}
 
 	// Get all peers by task id
-	peers, err := j.getPeers(req.TaskID)
+	peers, err := j.getValidPeers(req.TaskID)
 	if err != nil {
 		logger.Errorf("get peers by task id %s failed: %s", req.TaskID, err.Error())
 		return "", err
 	}
 
-	// Return peers by page
 	listTaskResponse := &internaljob.ListTasksResponse{
-		Total: len(peers),
-		Page:  req.Page,
-		Peers: peers[req.Page*req.PerPage : (req.Page+1)*req.PerPage],
+		Peers: peers,
 	}
 
 	return internaljob.MarshalResponse(listTaskResponse)
 }
 
 // deleteTask is a job to delete task.
 func (j *job) deleteTask(ctx context.Context, data string) (string, error) {
-	// TODO:
-	// 1. query all peers with task id
 	ctx, cancel := context.WithTimeout(ctx, deleteTaskTimeout)
 	defer cancel()
 
@@ -359,43 +359,82 @@ func (j *job) deleteTask(ctx context.Context, data string) (string, error) {
 	}
 
 	// Get all peers by task id
-	peers, err := j.getPeers(req.TaskID)
+	peers, err := j.getValidPeers(req.TaskID)
 	if err != nil {
 		logger.Errorf("get peers by task id %s failed: %s", req.TaskID, err.Error())
 		return "", err
 	}
 
 	// Delete task by task id and host id
-	successTasks := make([]*internaljob.TaskInfo, 0)
-	failureTasks := make([]*internaljob.TaskInfo, 0)
+	successTasks := make([]*internaljob.Task, 0)
+	failureTasks := make([]*internaljob.Task, 0)
 
+	// Create a wait group to limit delete rpc concurrency
+	// and avoid too many rpc requests to the host.
+	wg := sync.WaitGroup{}
+	deleteTaskLimit := make(chan struct{}, deleteTaskConcurrency)
 	for _, peer := range peers {
-		// hostID := peer.Host.ID
-		// get task info by task id
-		task, ok := j.resource.TaskManager().Load(req.TaskID)
-		if !ok {
-			logger.Errorf("task %s not found", req.TaskID)
-			failureTasks = append(failureTasks, &internaljob.TaskInfo{
-				Task: nil,
-				Peer: peer,
-				Desc: "task not found",
+		wg.Add(1)
+		deleteTaskLimit <- struct{}{}
+		go func(peer *resource.Peer) {
+			defer func() {
+				wg.Done()
+				<-deleteTaskLimit
+			}()
+
+			// Get dfdaemon client from host
+			target := fmt.Sprintf("%s:%d", peer.Host.IP, peer.Host.Port)
+			conn, err := grpc.DialContext(
+				ctx,
+				target,
+				grpc.WithIdleTimeout(0),
+				grpc.WithDefaultCallOptions(
+					grpc.MaxCallRecvMsgSize(math.MaxInt32),
+					grpc.MaxCallSendMsgSize(math.MaxInt32),
+				),
+				grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(
+					grpc_prometheus.UnaryClientInterceptor,
+					grpc_zap.UnaryClientInterceptor(logger.GrpcLogger.Desugar()),
+					grpc_retry.UnaryClientInterceptor(
+						grpc_retry.WithMax(deleteTaskMaxRetries),
+						grpc_retry.WithBackoff(grpc_retry.BackoffLinear(deleteTaskBackoffWaitBetween)),
+					),
+				)),
+			)
+			if err != nil {
+				logger.Errorf("create grpc client to %s failed: %s", target, err.Error())
+				failureTasks = append(failureTasks, &internaljob.Task{
+					Task:        peer.Task,
+					Peer:        peer,
+					Description: err.Error(),
+				})
+				return
+			}
+
+			dfdaemonUploadClient := dfdaemonv2.NewDfdaemonUploadClient(conn)
+			_, err = dfdaemonUploadClient.DeleteCacheTask(ctx, &dfdaemonv2.DeleteCacheTaskRequest{
+				TaskId: req.TaskID,
 			})
-			continue
-		}
+			if err != nil {
+				logger.Errorf("delete task %s from %s failed: %s", req.TaskID, target, err.Error())
+				failureTasks = append(failureTasks, &internaljob.Task{
+					Task:        peer.Task,
+					Peer:        peer,
+					Description: err.Error(),
+				})
+				return
+			}
 
-		// TODO: change to scheduler delete task grpc function
-		// and add batch delete
-		j.resource.SeedPeer().Client().DeleteCacheTask(ctx, &dfdaemonv2.DeleteCacheTaskRequest{
-			TaskId: req.TaskID,
-		})
-
-		successTasks = append(successTasks, &internaljob.TaskInfo{
-			Task: task,
-			Peer: peer,
-			Desc: "success",
-		})
+			successTasks = append(successTasks, &internaljob.Task{
+				Task:        peer.Task,
+				Peer:        peer,
+				Description: fmt.Sprintf("delete task %s from %s success", req.TaskID, target),
+			})
+		}(peer)
 	}
 
+	wg.Wait()
+
 	deleteTaskResponse := &internaljob.DeleteTaskResponse{
 		SuccessTasks: successTasks,
 		FailureTasks: failureTasks,
@@ -404,8 +443,8 @@ func (j *job) deleteTask(ctx context.Context, data string) (string, error) {
 	return internaljob.MarshalResponse(deleteTaskResponse)
 }
 
-// getPeers try to get peers by task id
-func (j *job) getPeers(taskID string) ([]*resource.Peer, error) {
+// getValidPeers try to get valid peers by task id
+func (j *job) getValidPeers(taskID string) ([]*resource.Peer, error) {
 	// get task info by task id
 	task, ok := j.resource.TaskManager().Load(taskID)
 	if !ok {
@@ -424,5 +463,14 @@ func (j *job) getPeers(taskID string) ([]*resource.Peer, error) {
 		peers = append(peers, peer)
 	}
 
-	return peers, nil
+	// Choose finished peers as list tasks result
+	finishedPeers := make([]*resource.Peer, len(peers))
+	for _, peer := range peers {
+		currentState := peer.FSM.Current()
+		if currentState == resource.PeerStateSucceeded || currentState == resource.PeerStateFailed {
+			finishedPeers = append(finishedPeers, peer)
+		}
+	}
+
+	return finishedPeers, nil
 }