JCS-pub/client/internal/cmdline/mount.go

package cmdline

import (
	"fmt"
	"os"
	"time"

	"github.com/spf13/cobra"
	"gitlink.org.cn/cloudream/common/pkgs/logger"
	"gitlink.org.cn/cloudream/storage2/client/internal/config"
	db2 "gitlink.org.cn/cloudream/storage2/client/internal/db"
	"gitlink.org.cn/cloudream/storage2/client/internal/downloader"
	"gitlink.org.cn/cloudream/storage2/client/internal/downloader/strategy"
	"gitlink.org.cn/cloudream/storage2/client/internal/metacache"
	"gitlink.org.cn/cloudream/storage2/client/internal/mount"
	mntcfg "gitlink.org.cn/cloudream/storage2/client/internal/mount/config"
	"gitlink.org.cn/cloudream/storage2/client/internal/uploader"
	stgglb "gitlink.org.cn/cloudream/storage2/common/globals"
	"gitlink.org.cn/cloudream/storage2/common/pkgs/connectivity"
	"gitlink.org.cn/cloudream/storage2/common/pkgs/distlock"
	hubrpc "gitlink.org.cn/cloudream/storage2/common/pkgs/grpc/hub"
	hubpool "gitlink.org.cn/cloudream/storage2/common/pkgs/storage/pool"
)

func init() {
	var configPath string

	cmd := &cobra.Command{
		Use:  "mount",
		Args: cobra.ExactArgs(1),
		Run: func(cmd *cobra.Command, args []string) {
			mountCmd(args[0], configPath)
		},
	}
	cmd.Flags().StringVarP(&configPath, "config", "c", "", "path to config file")

	RootCmd.AddCommand(cmd)
}

func mountCmd(mountPoint string, configPath string) {
	err := config.Init(configPath)
	if err != nil {
		fmt.Printf("init config failed, err: %s", err.Error())
		os.Exit(1)
	}

	err = logger.Init(&config.Cfg().Logger)
	if err != nil {
		fmt.Printf("init logger failed, err: %s", err.Error())
		os.Exit(1)
	}

	stgglb.InitLocal(config.Cfg().Local)
	stgglb.InitMQPool(config.Cfg().RabbitMQ)
	stgglb.InitHubRPCPool(&hubrpc.PoolConfig{})
	// stgglb.Stats.SetupHubStorageTransfer(*config.Cfg().Local.HubID)
	// stgglb.Stats.SetupHubTransfer(*config.Cfg().Local.HubID)

	// 初始化存储服务管理器
	stgPool := hubpool.NewPool()

	db, err := db2.NewDB(&config.Cfg().DB)
	if err != nil {
		logger.Fatalf("new db failed, err: %s", err.Error())
	}

	// 启动网络连通性检测，并就地检测一次
	conCol := connectivity.NewCollector(&config.Cfg().Connectivity, nil)
	// conCol.CollectInPlace()

	// 初始化元数据缓存服务
	metacacheHost := metacache.NewHost(db)
	go metacacheHost.Serve()
	stgMeta := metacacheHost.AddStorageMeta()
	hubMeta := metacacheHost.AddHubMeta()
	conMeta := metacacheHost.AddConnectivity()

	// 分布式锁
	distlockSvc, err := distlock.NewService(&config.Cfg().DistLock)
	if err != nil {
		logger.Warnf("new distlock service failed, err: %s", err.Error())
		os.Exit(1)
	}
	go serveDistLock(distlockSvc)

	// 初始化下载策略选择器
	strgSel := strategy.NewSelector(config.Cfg().DownloadStrategy, stgMeta, hubMeta, conMeta)

	// 初始化下载器
	dlder := downloader.NewDownloader(config.Cfg().Downloader, &conCol, stgPool, strgSel, db)

	// 上传器
	uploader := uploader.NewUploader(distlockSvc, &conCol, stgPool, stgMeta, db)

	mnt := mount.NewMount(&mntcfg.Config{
		DataDir:             "./cache/data",
		MetaDir:             "./cache/meta",
		MountPoint:          mountPoint,
		AttrTimeout:         time.Second * 5,
		UploadPendingTime:   time.Second * 10,
		CacheActiveTime:     time.Second * 10,
		CacheExpireTime:     time.Second * 60,
		ScanDataDirInterval: 30 * time.Second,
	}, db, uploader, &dlder)

	ch := mnt.Start()
	for {
		evt, err := ch.Receive()
		if err != nil {
			break
		}

		switch e := evt.(type) {
		case mount.MountingFailedEvent:
			fmt.Println("mounting failed:", e.Err)
			return

		case mount.MountExitEvent:
			fmt.Printf("mount exit\n")
			return
		}
	}
}