初步实现跨云调度功能

client/internal/http/server.go

@@ -145,6 +145,7 @@ func (s *Server) routeV1(eg *gin.Engine, rt gin.IRoutes) {
 	v1.POST(cliapi.UserSpaceLoadPackagePath, awsAuth.Auth, s.UserSpace().LoadPackage)
 	v1.POST(cliapi.UserSpaceCreatePackagePath, awsAuth.Auth, s.UserSpace().CreatePackage)
 	v1.GET(cliapi.UserSpaceGetPath, awsAuth.Auth, s.UserSpace().Get)
+	rt.POST(cliapi.UserSpaceSpaceToSpacePath, s.UserSpace().SpaceToSpace)
 
 	// v1.POST(cdsapi.CacheMovePackagePath, awsAuth.Auth, s.Cache().MovePackage)
 

client/internal/http/user_space.go

@@ -83,3 +83,25 @@ func (s *UserSpaceService) Get(ctx *gin.Context) {
 		UserSpace: info,
 	}))
 }
+
+func (s *UserSpaceService) SpaceToSpace(ctx *gin.Context) {
+	log := logger.WithField("HTTP", "UserSpace.SpaceToSpace")
+
+	var req cliapi.UserSpaceSpaceToSpace
+	if err := ctx.ShouldBindJSON(&req); err != nil {
+		log.Warnf("binding body: %s", err.Error())
+		ctx.JSON(http.StatusBadRequest, Failed(errorcode.BadArgument, "missing argument or invalid argument"))
+		return
+	}
+
+	ret, err := s.svc.UserSpaceSvc().SpaceToSpace(req.SrcUserSpaceID, req.SrcPath, req.DstUserSpaceID, req.DstPath)
+	if err != nil {
+		log.Warnf("space2space: %s", err.Error())
+		ctx.JSON(http.StatusOK, Failed(errorcode.OperationFailed, "space2space failed"))
+		return
+	}
+
+	ctx.JSON(http.StatusOK, OK(cliapi.UserSpaceSpaceToSpaceResp{
+		SpaceToSpaceResult: ret,
+	}))
+}

client/internal/services/user_space.go

@@ -4,16 +4,21 @@ import (
 	"context"
 	"fmt"
 	"path"
+	"strings"
 
 	"gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec"
+	"gitlink.org.cn/cloudream/common/pkgs/logger"
+	"gitlink.org.cn/cloudream/common/pkgs/trie"
+	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
 	clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types"
 
 	"gitlink.org.cn/cloudream/jcs-pub/client/internal/db"
 	"gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader/strategy"
-	"gitlink.org.cn/cloudream/jcs-pub/client/types"
 	stgglb "gitlink.org.cn/cloudream/jcs-pub/common/globals"
 	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2"
 	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/parser"
+	hubmq "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/mq/hub"
+	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/types"
 )
 
 type UserSpaceService struct {
@@ -24,11 +29,11 @@ func (svc *Service) UserSpaceSvc() *UserSpaceService {
 	return &UserSpaceService{Service: svc}
 }
 
-func (svc *UserSpaceService) Get(userspaceID clitypes.UserSpaceID) (types.UserSpace, error) {
+func (svc *UserSpaceService) Get(userspaceID clitypes.UserSpaceID) (clitypes.UserSpace, error) {
 	return svc.DB.UserSpace().GetByID(svc.DB.DefCtx(), userspaceID)
 }
 
-func (svc *UserSpaceService) GetByName(name string) (types.UserSpace, error) {
+func (svc *UserSpaceService) GetByName(name string) (clitypes.UserSpace, error) {
 	return svc.DB.UserSpace().GetByName(svc.DB.DefCtx(), name)
 }
 
@@ -77,7 +82,7 @@ func (svc *UserSpaceService) LoadPackage(packageID clitypes.PackageID, userspace
 			return fmt.Errorf("unsupported download strategy: %T", strg)
 		}
 
-		ft.AddTo(ioswitch2.NewLoadToPublic(*destStg.MasterHub, *destStg, path.Join(rootPath, obj.Object.Path)))
+		ft.AddTo(ioswitch2.NewToPublicStore(*destStg.MasterHub, *destStg, path.Join(rootPath, obj.Object.Path)))
 		// 顺便保存到同存储服务的分片存储中
 		if destStg.UserSpace.ShardStore != nil {
 			ft.AddTo(ioswitch2.NewToShardStore(*destStg.MasterHub, *destStg, ioswitch2.RawStream(), ""))
@@ -115,3 +120,152 @@ func (svc *UserSpaceService) LoadPackage(packageID clitypes.PackageID, userspace
 
 	return nil
 }
+
+func (svc *UserSpaceService) SpaceToSpace(srcSpaceID clitypes.UserSpaceID, srcPath string, dstSpaceID clitypes.UserSpaceID, dstPath string) (clitypes.SpaceToSpaceResult, error) {
+	srcSpace := svc.UserSpaceMeta.Get(srcSpaceID)
+	if srcSpace == nil {
+		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("source userspace not found: %d", srcSpaceID)
+	}
+	if srcSpace.MasterHub == nil {
+		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("source userspace %v has no master hub", srcSpaceID)
+	}
+	srcSpaceCli, err := stgglb.HubMQPool.Acquire(srcSpace.MasterHub.HubID)
+	if err != nil {
+		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("new source userspace client: %w", err)
+	}
+	defer stgglb.HubMQPool.Release(srcSpaceCli)
+
+	dstSpace := svc.UserSpaceMeta.Get(dstSpaceID)
+	if dstSpace == nil {
+		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("destination userspace not found: %d", dstSpaceID)
+	}
+	if dstSpace.MasterHub == nil {
+		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("destination userspace %v has no master hub", dstSpaceID)
+	}
+	dstSpaceCli, err := stgglb.HubMQPool.Acquire(dstSpace.MasterHub.HubID)
+	if err != nil {
+		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("new destination userspace client: %w", err)
+	}
+	defer stgglb.HubMQPool.Release(dstSpaceCli)
+
+	srcPath = strings.Trim(srcPath, cdssdk.ObjectPathSeparator)
+	dstPath = strings.Trim(dstPath, cdssdk.ObjectPathSeparator)
+
+	if srcPath == "" {
+		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("source path is empty")
+	}
+
+	if dstPath == "" {
+		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("destination path is empty")
+	}
+
+	listAllResp, err := srcSpaceCli.PublicStoreListAll(&hubmq.PublicStoreListAll{
+		UserSpace: *srcSpace,
+		Path:      srcPath,
+	})
+	if err != nil {
+		return clitypes.SpaceToSpaceResult{}, fmt.Errorf("list all from source userspace: %w", err)
+	}
+
+	srcPathComps := clitypes.SplitObjectPath(srcPath)
+	srcDirCompLen := len(srcPathComps) - 1
+
+	entryTree := trie.NewTrie[*types.PublicStoreEntry]()
+	for _, e := range listAllResp.Entries {
+		pa, ok := strings.CutSuffix(e.Path, clitypes.ObjectPathSeparator)
+		comps := clitypes.SplitObjectPath(pa)
+		e.Path = pa
+
+		e2 := e
+		entryTree.CreateWords(comps[srcDirCompLen:]).Value = &e2
+		e2.IsDir = e2.IsDir || ok
+	}
+
+	entryTree.Iterate(func(path []string, node *trie.Node[*types.PublicStoreEntry], isWordNode bool) trie.VisitCtrl {
+		if node.Value == nil {
+			return trie.VisitContinue
+		}
+
+		if node.Value.IsDir && len(node.WordNexts) > 0 {
+			node.Value = nil
+			return trie.VisitContinue
+		}
+
+		if !node.Value.IsDir && len(node.WordNexts) == 0 {
+			node.WordNexts = nil
+		}
+
+		return trie.VisitContinue
+	})
+
+	var filePathes []string
+	var dirPathes []string
+	entryTree.Iterate(func(path []string, node *trie.Node[*types.PublicStoreEntry], isWordNode bool) trie.VisitCtrl {
+		if node.Value == nil {
+			return trie.VisitContinue
+		}
+
+		if node.Value.IsDir {
+			dirPathes = append(dirPathes, node.Value.Path)
+		} else {
+			filePathes = append(filePathes, node.Value.Path)
+		}
+
+		return trie.VisitContinue
+	})
+
+	var success []string
+	var failed []string
+
+	for _, f := range filePathes {
+		newPath := strings.Replace(f, srcPath, dstPath, 1)
+
+		ft := ioswitch2.NewFromTo()
+		ft.AddFrom(ioswitch2.NewFromPublicStore(*srcSpace.MasterHub, *srcSpace, f))
+		ft.AddTo(ioswitch2.NewToPublicStore(*dstSpace.MasterHub, *dstSpace, newPath))
+
+		plans := exec.NewPlanBuilder()
+		err = parser.Parse(ft, plans)
+		if err != nil {
+			failed = append(failed, f)
+			logger.Warnf("s2s: parse plan of file %v: %v", f, err)
+			continue
+		}
+
+		_, err = plans.Execute(exec.NewExecContext()).Wait(context.Background())
+		if err != nil {
+			failed = append(failed, f)
+			logger.Warnf("s2s: execute plan of file %v: %v", f, err)
+			continue
+		}
+
+		success = append(success, f)
+	}
+
+	newDirPathes := make([]string, 0, len(dirPathes))
+	for i := range dirPathes {
+		newDirPathes = append(newDirPathes, strings.Replace(dirPathes[i], srcPath, dstPath, 1))
+	}
+
+	mkdirResp, err := dstSpaceCli.PublicStoreMkdirs(&hubmq.PublicStoreMkdirs{
+		UserSpace: *dstSpace,
+		Pathes:    newDirPathes,
+	})
+	if err != nil {
+		failed = append(failed, dirPathes...)
+		logger.Warnf("s2s: mkdirs to destination userspace: %v", err)
+	} else {
+		for i := range dirPathes {
+			if mkdirResp.Successes[i] {
+				success = append(success, dirPathes[i])
+			} else {
+				failed = append(failed, dirPathes[i])
+			}
+		}
+	}
+
+	return clitypes.SpaceToSpaceResult{
+		Success: success,
+		Failed:  failed,
+	}, nil
+}

client/internal/uploader/create_load.go

@@ -49,7 +49,7 @@ func (u *CreateLoadUploader) Upload(pa string, stream io.Reader, opts ...UploadO
 	ft.AddFrom(fromExec)
 	for i, space := range u.targetSpaces {
 		ft.AddTo(ioswitch2.NewToShardStore(*space.MasterHub, space, ioswitch2.RawStream(), "shardInfo"))
-		ft.AddTo(ioswitch2.NewLoadToPublic(*space.MasterHub, space, path.Join(u.loadRoots[i], pa)))
+		ft.AddTo(ioswitch2.NewToPublicStore(*space.MasterHub, space, path.Join(u.loadRoots[i], pa)))
 		spaceIDs = append(spaceIDs, space.UserSpace.UserSpaceID)
 	}
 

client/internal/uploader/update.go

@@ -60,7 +60,7 @@ func (w *UpdateUploader) Upload(pat string, stream io.Reader, opts ...UploadOpti
 		AddTo(ioswitch2.NewToShardStore(*w.targetSpace.MasterHub, w.targetSpace, ioswitch2.RawStream(), "shardInfo"))
 
 	for i, space := range w.loadToSpaces {
-		ft.AddTo(ioswitch2.NewLoadToPublic(*space.MasterHub, space, path.Join(w.loadToPath[i], pat)))
+		ft.AddTo(ioswitch2.NewToPublicStore(*space.MasterHub, space, path.Join(w.loadToPath[i], pat)))
 	}
 
 	plans := exec.NewPlanBuilder()

client/sdk/api/userspace.go

@@ -76,3 +76,28 @@ func (r *UserSpaceGetResp) ParseResponse(resp *http.Response) error {
 func (c *Client) UserSpaceGet(req UserSpaceGet) (*UserSpaceGetResp, error) {
 	return JSONAPI(c.cfg, http.DefaultClient, &req, &UserSpaceGetResp{})
 }
+
+const UserSpaceSpaceToSpacePath = "/v1/userspace/spaceToSpace"
+
+type UserSpaceSpaceToSpace struct {
+	SrcUserSpaceID clitypes.UserSpaceID `json:"srcUserSpaceID" binding:"required"`
+	DstUserSpaceID clitypes.UserSpaceID `json:"dstUserSpaceID" binding:"required"`
+	SrcPath        string               `json:"srcPath" binding:"required"`
+	DstPath        string               `json:"dstPath" binding:"required"`
+}
+
+func (r *UserSpaceSpaceToSpace) MakeParam() *sdks.RequestParam {
+	return sdks.MakeJSONParam(http.MethodPost, UserSpaceSpaceToSpacePath, r)
+}
+
+type UserSpaceSpaceToSpaceResp struct {
+	clitypes.SpaceToSpaceResult
+}
+
+func (r *UserSpaceSpaceToSpaceResp) ParseResponse(resp *http.Response) error {
+	return sdks.ParseCodeDataJSONResponse(resp, r)
+}
+
+func (c *Client) UserSpaceSpaceToSpace(req UserSpaceSpaceToSpace) (*UserSpaceSpaceToSpaceResp, error) {
+	return JSONAPI(c.cfg, http.DefaultClient, &req, &UserSpaceSpaceToSpaceResp{})
+}

client/types/types.go

@@ -229,3 +229,8 @@ type PackageDetail struct {
 	ObjectCount int64
 	TotalSize   int64
 }
+
+type SpaceToSpaceResult struct {
+	Success []string `json:"success"`
+	Failed  []string `json:"failed"`
+}

common/pkgs/ioswitch2/fromto.go

@@ -221,7 +221,7 @@ type LoadToPublic struct {
 	ObjectPath string
 }
 
-func NewLoadToPublic(hub cortypes.Hub, space clitypes.UserSpaceDetail, objectPath string) *LoadToPublic {
+func NewToPublicStore(hub cortypes.Hub, space clitypes.UserSpaceDetail, objectPath string) *LoadToPublic {
 	return &LoadToPublic{
 		Hub:        hub,
 		Space:      space,

common/pkgs/mq/hub/user_space.go

@@ -8,6 +8,8 @@ import (
 
 type UserSpaceService interface {
 	PublicStoreListAll(msg *PublicStoreListAll) (*PublicStoreListAllResp, *mq.CodeMessage)
+
+	PublicStoreMkdirs(msg *PublicStoreMkdirs) (*PublicStoreMkdirsResp, *mq.CodeMessage)
 }
 
 // 启动从UserSpace上传Package的任务
@@ -26,3 +28,20 @@ type PublicStoreListAllResp struct {
 func (client *Client) PublicStoreListAll(msg *PublicStoreListAll, opts ...mq.RequestOption) (*PublicStoreListAllResp, error) {
 	return mq.Request(Service.PublicStoreListAll, client.rabbitCli, msg, opts...)
 }
+
+var _ = Register(Service.PublicStoreMkdirs)
+
+type PublicStoreMkdirs struct {
+	mq.MessageBodyBase
+	UserSpace clitypes.UserSpaceDetail
+	Pathes    []string
+}
+
+type PublicStoreMkdirsResp struct {
+	mq.MessageBodyBase
+	Successes []bool
+}
+
+func (client *Client) PublicStoreMkdirs(msg *PublicStoreMkdirs, opts ...mq.RequestOption) (*PublicStoreMkdirsResp, error) {
+	return mq.Request(Service.PublicStoreMkdirs, client.rabbitCli, msg, opts...)
+}

common/pkgs/storage/local/public_store.go

@@ -55,6 +55,16 @@ func (s *PublicStore) Read(objPath string) (io.ReadCloser, error) {
 	return f, nil
 }
 
+func (s *PublicStore) Mkdir(path string) error {
+	absObjPath := filepath.Join(s.root, path)
+	err := os.MkdirAll(absObjPath, 0755)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (s *PublicStore) ListAll(path string) ([]types.PublicStoreEntry, error) {
 	absObjPath := filepath.Join(s.root, path)
 

common/pkgs/storage/s3/public_store.go

@@ -1,6 +1,7 @@
 package s3
 
 import (
+	"bytes"
 	"context"
 	"io"
 
@@ -52,6 +53,15 @@ func (s *PublicStore) Read(objPath string) (io.ReadCloser, error) {
 	return resp.Body, nil
 }
 
+func (s *PublicStore) Mkdir(path string) error {
+	_, err := s.cli.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(s.Bucket),
+		Key:    aws.String(path + "/"),
+		Body:   bytes.NewReader([]byte{}),
+	})
+	return err
+}
+
 func (s *PublicStore) ListAll(path string) ([]types.PublicStoreEntry, error) {
 	key := path
 	// TODO 待测试

common/pkgs/storage/types/public_store.go

@@ -13,6 +13,8 @@ type PublicStoreEntry struct {
 type PublicStore interface {
 	Write(path string, stream io.Reader) error
 	Read(path string) (io.ReadCloser, error)
+	// 创建指定路径的文件夹。对于不支持空文件夹的存储系统来说，可以采用创建以/结尾的对象的方式来模拟文件夹。
+	Mkdir(path string) error
 	// 返回指定路径下的所有文件，文件路径是包含path在内的完整路径。返回结果的第一条一定是路径本身，可能是文件，也可能是目录。
 	// 如果路径不存在，那么不会返回错误，而是返回一个空列表。
 	// 返回的内容严格按照存储系统的原始结果来，比如当存储系统是一个对象存储时，那么就可能不会包含目录，或者包含用于模拟的以“/”结尾的对象。

hub/internal/mq/user_space.go

@@ -2,6 +2,7 @@ package mq
 
 import (
 	"gitlink.org.cn/cloudream/common/consts/errorcode"
+	"gitlink.org.cn/cloudream/common/pkgs/logger"
 	"gitlink.org.cn/cloudream/common/pkgs/mq"
 	hubmq "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/mq/hub"
 )
@@ -21,3 +22,24 @@ func (svc *Service) PublicStoreListAll(msg *hubmq.PublicStoreListAll) (*hubmq.Pu
 		Entries: es,
 	})
 }
+
+func (svc *Service) PublicStoreMkdirs(msg *hubmq.PublicStoreMkdirs) (*hubmq.PublicStoreMkdirsResp, *mq.CodeMessage) {
+	pub, err := svc.stgPool.GetPublicStore(&msg.UserSpace)
+	if err != nil {
+		return nil, mq.Failed(errorcode.OperationFailed, err.Error())
+	}
+
+	var suc []bool
+	for _, p := range msg.Pathes {
+		if err := pub.Mkdir(p); err != nil {
+			suc = append(suc, false)
+			logger.Warnf("userspace %v mkdir %s: %v", msg.UserSpace, p, err)
+		} else {
+			suc = append(suc, true)
+		}
+	}
+
+	return mq.ReplyOK(&hubmq.PublicStoreMkdirsResp{
+		Successes: suc,
+	})
+}