JCC-CSScheduler/executor/internal/task/cache_move_package.go

package task

import (
	"fmt"
	"time"

	"gitlink.org.cn/cloudream/common/pkgs/logger"
	"gitlink.org.cn/cloudream/common/pkgs/task"
	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
	schglb "gitlink.org.cn/cloudream/scheduler/common/globals"
	exectsk "gitlink.org.cn/cloudream/scheduler/common/pkgs/mq/executor/task"
)

type CacheMovePackage struct {
	*exectsk.CacheMovePackage
}

func NewCacheMovePackage(info *exectsk.CacheMovePackage) *CacheMovePackage {
	return &CacheMovePackage{
		CacheMovePackage: info,
	}
}

func (t *CacheMovePackage) Execute(task *task.Task[TaskContext], ctx TaskContext, complete CompleteFn) {
	log := logger.WithType[CacheMovePackage]("Task")
	log.Debugf("begin with %v", logger.FormatStruct(t.CacheMovePackage))
	defer log.Debugf("end")

	err := t.do(ctx)
	if err != nil {
		ctx.reporter.Report(task.ID(), exectsk.NewCacheMovePackageStatus(err.Error()))
	} else {
		ctx.reporter.Report(task.ID(), exectsk.NewCacheMovePackageStatus(""))
	}
	ctx.reporter.ReportNow()

	complete(err, CompleteOption{
		RemovingDelay: time.Minute,
	})
}

func (t *CacheMovePackage) do(ctx TaskContext) error {
	stgCli, err := schglb.CloudreamStoragePool.Acquire()
	if err != nil {
		return fmt.Errorf("new cloudream storage client: %w", err)
	}
	defer schglb.CloudreamStoragePool.Release(stgCli)

	_, err = stgCli.CacheMovePackage(cdssdk.CacheMovePackageReq{
		UserID:    t.UserID,
		PackageID: t.PackageID,
		NodeID:    t.StgNodeID,
	})
	if err != nil {
		return err
	}

	return err
}

func init() {
	Register(NewCacheMovePackage)
}