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

package cmdline

import (
	"fmt"
	"os"
	"time"

	"github.com/spf13/cobra"
	"gitlink.org.cn/cloudream/common/pkgs/logger"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/accessstat"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/accesstoken"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/cluster"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/config"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/db"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader/strategy"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/http"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/metacache"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/mount"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/publock"
	myrpc "gitlink.org.cn/cloudream/jcs-pub/client/internal/rpc"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/services"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/spacesyncer"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/speedstats"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/ticktock"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/uploader"
	stgglb "gitlink.org.cn/cloudream/jcs-pub/common/globals"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/connectivity"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/rpc"
	clirpc "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/rpc/client"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/pool"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/sysevent"
	jcstypes "gitlink.org.cn/cloudream/jcs-pub/common/types"
	"gitlink.org.cn/cloudream/jcs-pub/common/types/datamap"
)

// 初始化函数，将ServeHTTP命令注册到命令列表中。
func init() {
	var configPath string
	var opt serveHTTPOptions
	cmd := cobra.Command{
		Use:   "serve",
		Short: "start serving storage service",
		Run: func(cmd *cobra.Command, args []string) {
			serveHTTP(configPath, opt)
		},
	}
	cmd.Flags().StringVarP(&configPath, "config", "c", "", "config file path")
	cmd.Flags().BoolVarP(&opt.DisableHTTP, "no-http", "", false, "disable http server")
	cmd.Flags().StringVarP(&opt.HTTPListenAddr, "listen", "", "", "http listen address, will override config file")
	cmd.Flags().BoolVarP(&opt.DisableMount, "no-mount", "", false, "disable mount")
	cmd.Flags().StringVarP(&opt.MountPoint, "mount", "", "", "mount point, will override config file")
	cmd.Flags().BoolVarP(&opt.Standalone, "standalone", "", false, "standalone mode")
	RootCmd.AddCommand(&cmd)
}

type serveHTTPOptions struct {
	DisableHTTP    bool
	HTTPListenAddr string
	DisableMount   bool
	MountPoint     string
	Standalone     bool
}

func serveHTTP(configPath string, opts serveHTTPOptions) {
	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)

	// 集群模式
	clster := cluster.New(config.Cfg().Cluster)
	err = clster.Start()
	if err != nil {
		logger.Errorf("start cluster failed, err: %v", err)
		os.Exit(1)
	}

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

	// 元数据缓存
	metaCacheHost := metacache.NewHost(db)
	go metaCacheHost.Serve()
	spaceMeta := metaCacheHost.AddStorageMeta()
	hubMeta := metaCacheHost.AddHubMeta()
	conMeta := metaCacheHost.AddConnectivity()

	stgglb.StandaloneMode = opts.Standalone || config.Cfg().AccessToken == nil

	// 访问令牌
	var accToken *accesstoken.Keeper
	if !stgglb.StandaloneMode {
		tempCli, err := config.Cfg().CoordinatorRPC.BuildTempClient()
		if err != nil {
			logger.Warnf("build coordinator rpc temp client: %v", err)
			os.Exit(1)
		}

		accToken, err = accesstoken.New(*config.Cfg().AccessToken, tempCli)
		tempCli.Release()
		if err != nil {
			logger.Warnf("new access token keeper: %v", err)
			os.Exit(1)
		}

		hubRPCCfg, err := config.Cfg().HubRPC.Build(accToken)
		if err != nil {
			logger.Warnf("build hub rpc pool config: %v", err)
			os.Exit(1)
		}

		corRPCCfg, err := config.Cfg().CoordinatorRPC.Build(accToken)
		if err != nil {
			logger.Warnf("build coordinator rpc pool config: %v", err)
			os.Exit(1)
		}

		// 通过HubID获取Hub的地址
		hubRPCCfg.AddrProvider = func(id jcstypes.HubID) string {
			hub := hubMeta.Get(id)
			if hub == nil {
				return ""
			}

			addr, ok := hub.Address.(*jcstypes.GRPCAddressInfo)
			if !ok {
				return ""
			}

			ip, port := stgglb.SelectGRPCAddress(hub, addr)
			return fmt.Sprintf("%v:%v", ip, port)
		}

		stgglb.UserID = accToken.GetToken().UserID
		stgglb.InitPools(hubRPCCfg, corRPCCfg)
	} else {
		stgglb.UserID = 0
		accToken = accesstoken.NewDisabled()
	}
	accTokenChan := accToken.Start()
	defer accToken.Stop()

	// 初始化系统事件发布器
	evtPub, err := sysevent.NewPublisher(config.Cfg().SysEvent, &datamap.SourceClient{
		UserID: stgglb.UserID,
	})
	if err != nil {
		logger.Errorf("new sysevent publisher: %v", err)
		os.Exit(1)
	}
	evtPubChan := evtPub.Start()
	defer evtPub.Stop()

	// 连接性信息收集
	var conCol *connectivity.Collector
	if stgglb.StandaloneMode {
		conCol = connectivity.NewDisabled()
	} else {
		conCol = connectivity.NewEnabled(config.Cfg().Connectivity)
	}
	conColChan := conCol.Start()
	defer conCol.Stop()
	conCol.CollectInPlace()

	// 公共锁
	publock := publock.NewMaster()
	publock.Start()
	defer publock.Stop()

	// 访问统计
	acStat := accessstat.NewAccessStat(accessstat.Config{
		// TODO 考虑放到配置里
		ReportInterval: time.Second * 10,
	}, db)
	acStatChan := acStat.Start()
	defer acStat.Stop()

	// 存储管理器
	stgPool := pool.NewPool()

	// 传输速度统计
	spdStats := speedstats.New()

	// 下载策略
	strgSel := strategy.NewSelector(config.Cfg().DownloadStrategy, spaceMeta, hubMeta, conMeta)

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

	// 上传器
	uploader := uploader.NewUploader(publock, conCol, stgPool, spaceMeta, db)

	// 定时任务
	tktk := ticktock.New(config.Cfg().TickTock, db, spaceMeta, stgPool, evtPub, publock, spdStats, clster)
	tktk.Start()
	defer tktk.Stop()

	// 用户空间同步功能
	spaceSync := spacesyncer.New(db, stgPool, spaceMeta, clster)
	spaceSyncChan := spaceSync.Start()
	defer spaceSync.Stop()

	// 交互式命令行
	// rep := repl.New(db, tktk)
	// replCh := rep.Start()

	// 挂载
	mntCfg := config.Cfg().Mount
	if !opts.DisableMount && mntCfg != nil && mntCfg.Enabled {
		if opts.MountPoint != "" {
			mntCfg.MountPoint = opts.MountPoint
		}
	} else {
		mntCfg = nil
	}
	mnt := mount.NewMount(mntCfg, db, uploader, dlder)
	mntChan := mnt.Start()
	defer mnt.Stop()

	svc := services.NewService(publock, dlder, acStat, uploader, strgSel, spaceMeta, db, evtPub, mnt, stgPool, spaceSync, tktk, spdStats, accToken)

	// HTTP接口
	httpCfgJSON := config.Cfg().HTTP
	if !opts.DisableHTTP && httpCfgJSON != nil && httpCfgJSON.Enabled {
		if opts.HTTPListenAddr != "" {
			httpCfgJSON.Listen = opts.HTTPListenAddr
		}
	} else {
		httpCfgJSON = &http.ConfigJSON{
			Enabled: false,
		}
	}
	httpCfg, err := httpCfgJSON.Build()
	if err != nil {
		logger.Errorf("build http config: %v", err)
		os.Exit(1)
	}
	httpSvr := http.NewServer(httpCfg, svc, clster)
	httpChan := httpSvr.Start()
	defer httpSvr.Stop()

	// RPC接口
	rpcSvr := clirpc.NewServer(config.Cfg().RPC, myrpc.NewService(publock, clster, httpSvr), nil)
	rpcChan := rpcSvr.Start()
	defer rpcSvr.Stop()

	/// 开始监听各个模块的事件

	accTokenEvt := accTokenChan.Receive()
	evtPubEvt := evtPubChan.Receive()
	conColEvt := conColChan.Receive()
	acStatEvt := acStatChan.Receive()
	spaceSyncEvt := spaceSyncChan.Receive()
	// replEvt := replCh.Receive()
	httpEvt := httpChan.Receive()
	rpcEvt := rpcChan.Receive()
	mntEvt := mntChan.Receive()

loop:
	for {
		select {
		case e := <-accTokenEvt.Chan():
			if e.Err != nil {
				logger.Errorf("receive access token event: %v", err)
				break loop
			}

			switch e := e.Value.(type) {
			case accesstoken.ExitEvent:
				if e.Err != nil {
					logger.Errorf("access token keeper exit with error: %v", err)
				} else {
					logger.Info("access token keeper exited")
				}
				break loop
			}
			accTokenEvt = accTokenChan.Receive()

		case e := <-evtPubEvt.Chan():
			if e.Err != nil {
				logger.Errorf("receive publisher event: %v", err)
				break loop
			}

			switch val := e.Value.(type) {
			case sysevent.PublishError:
				logger.Errorf("publishing event: %v", val)

			case sysevent.PublisherExited:
				if val.Err != nil {
					logger.Errorf("publisher exited with error: %v", val.Err)
				} else {
					logger.Info("publisher exited")
				}
				break loop

			case sysevent.OtherError:
				logger.Errorf("sysevent: %v", val)
			}
			evtPubEvt = evtPubChan.Receive()

		case e := <-conColEvt.Chan():
			if e.Err != nil {
				logger.Errorf("receive connectivity event: %v", err)
				break loop
			}

			switch e := e.Value.(type) {
			case connectivity.ExitEvent:
				if e.Err != nil {
					logger.Errorf("connectivity collector exited with error: %v", e.Err)
				} else {
					logger.Info("connectivity collector exited")
				}
				break loop

			case connectivity.CollectedEvent:
			}
			conColEvt = conColChan.Receive()

		case e := <-acStatEvt.Chan():
			if e.Err != nil {
				logger.Errorf("receive access stat event: %v", err)
				break loop
			}
			switch e := e.Value.(type) {
			case accessstat.ExitEvent:
				logger.Infof("access stat exited, err: %v", e.Err)
				break loop
			}
			acStatEvt = acStatChan.Receive()

		case e := <-spaceSyncEvt.Chan():
			if e.Err != nil {
				logger.Errorf("receive space sync event: %v", err)
				break loop
			}
			spaceSyncEvt = spaceSyncChan.Receive()

		// case e := <-replEvt.Chan():
		// 	if e.Err != nil {
		// 		logger.Errorf("receive repl event: %v", err)
		// 		break loop
		// 	}

		// 	switch e.Value.(type) {
		// 	case repl.ExitEvent:
		// 		logger.Info("exit by repl")
		// 		break loop
		// 	}
		// 	replEvt = replCh.Receive()

		case e := <-httpEvt.Chan():
			if e.Err != nil {
				logger.Errorf("receive http event: %v", err)
				break loop
			}

			switch e := e.Value.(type) {
			case http.ExitEvent:
				logger.Infof("http server exited, err: %v", e.Err)
				break loop
			}
			httpEvt = httpChan.Receive()

		case e := <-rpcEvt.Chan():
			if e.Err != nil {
				logger.Errorf("receive rpc event: %v", err)
				break loop
			}

			switch e := e.Value.(type) {
			case rpc.ExitEvent:
				if e.Err != nil {
					logger.Errorf("rpc server exited with error: %v", e.Err)
				} else {
					logger.Infof("rpc server exited")
				}
				break loop
			}
			rpcEvt = rpcChan.Receive()

		case e := <-mntEvt.Chan():
			if e.Err != nil {
				logger.Errorf("receive mount event: %v", e.Err)
				break loop
			}

			switch e := e.Value.(type) {
			case mount.ExitEvent:
				logger.Infof("mount exited, err: %v", e.Err)
				break loop
			}
			mntEvt = mntChan.Receive()
		}
	}

	// TODO 优雅退出
	os.Exit(0)
}