JCS-pub/hub/internal/cmd/serve.go

package cmd

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/go-co-op/gocron/v2"
	"github.com/spf13/cobra"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/rpc"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/pool"
	"gitlink.org.cn/cloudream/jcs-pub/hub/internal/http"
	myrpc "gitlink.org.cn/cloudream/jcs-pub/hub/internal/rpc"

	"gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec"
	"gitlink.org.cn/cloudream/common/pkgs/logger"
	stgglb "gitlink.org.cn/cloudream/jcs-pub/common/globals"
	"gitlink.org.cn/cloudream/jcs-pub/common/models/datamap"
	hubrpc "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/rpc/hub"
	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/sysevent"
	cortypes "gitlink.org.cn/cloudream/jcs-pub/coordinator/types"
	"gitlink.org.cn/cloudream/jcs-pub/hub/internal/config"

	coormq "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/rpc/coordinator"
)

func init() {
	var configPath string
	var httpAddr string

	cmd := &cobra.Command{
		Use:   "serve",
		Short: "start jcs-pub hub service",
		Run: func(cmd *cobra.Command, args []string) {
			serve(configPath, httpAddr)
		},
	}
	cmd.Flags().StringVarP(&configPath, "config", "c", "", "path to config file")
	cmd.Flags().StringVarP(&httpAddr, "http", "", "127.0.0.1:8890", "http listen address")
	RootCmd.AddCommand(cmd)
}

func serve(configPath string, httpAddr 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.InitPools(&hubrpc.PoolConfig{}, &config.Cfg().CoordinatorRPC)
	// stgglb.Stats.SetupHubStorageTransfer(*config.Cfg().Local.HubID)
	// stgglb.Stats.SetupHubTransfer(*config.Cfg().Local.HubID)
	// 获取Hub配置
	hubCfg := downloadHubConfig()

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

	// 初始化执行器
	worker := exec.NewWorker()

	// 初始化HTTP服务
	httpSvr, err := http.NewServer(httpAddr, http.NewService(&worker, stgPool))
	if err != nil {
		logger.Fatalf("new http server failed, err: %s", err.Error())
	}
	go serveHTTP(httpSvr)

	// 启动网络连通性检测，并就地检测一次
	// conCol := connectivity.NewCollector(&config.Cfg().Connectivity, func(collector *connectivity.Collector) {
	// 	log := logger.WithField("Connectivity", "")

	// 	coorCli, err := stgglb.CoordinatorMQPool.Acquire()
	// 	if err != nil {
	// 		log.Warnf("acquire coordinator mq failed, err: %s", err.Error())
	// 		return
	// 	}
	// 	defer stgglb.CoordinatorMQPool.Release(coorCli)

	// 	cons := collector.GetAll()
	// 	hubCons := make([]cortypes.HubConnectivity, 0, len(cons))
	// 	for _, con := range cons {
	// 		var delay *float32
	// 		if con.Latency != nil {
	// 			v := float32(con.Latency.Microseconds()) / 1000
	// 			delay = &v
	// 		}

	// 		hubCons = append(hubCons, cortypes.HubConnectivity{
	// 			FromHubID: *stgglb.Local.HubID,
	// 			ToHubID:   con.ToHubID,
	// 			Latency:   delay,
	// 			TestTime:  con.TestTime,
	// 		})
	// 	}

	// 	_, err = coorCli.UpdateHubConnectivities(coormq.ReqUpdateHubConnectivities(hubCons))
	// 	if err != nil {
	// 		log.Warnf("update hub connectivities: %v", err)
	// 	}
	// })
	// conCol.CollectInPlace()

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

	// 启动访问统计服务
	// acStat := accessstat.NewAccessStat(accessstat.Config{
	// 	// TODO 考虑放到配置里
	// 	ReportInterval: time.Second * 10,
	// })
	// go serveAccessStat(acStat)

	// 初始化系统事件发布器
	evtPub, err := sysevent.NewPublisher(sysevent.ConfigFromMQConfig(config.Cfg().RabbitMQ), &datamap.SourceHub{
		HubID:   hubCfg.Hub.HubID,
		HubName: hubCfg.Hub.Name,
	})
	if err != nil {
		logger.Errorf("new sysevent publisher: %v", err)
		os.Exit(1)
	}
	go servePublisher(evtPub)

	// 初始化定时任务执行器
	sch := setupTickTask(stgPool, evtPub)
	sch.Start()
	defer sch.Shutdown()

	// RPC服务
	rpcSvr := hubrpc.NewServer(config.Cfg().RPC, myrpc.NewService(&worker, stgPool))
	rpcSvrChan := rpcSvr.Start()
	defer rpcSvr.Stop()

	/// 开始监听各个模块的事件
	rpcEvt := rpcSvrChan.Receive()

loop:
	for {
		select {
		case e := <-rpcEvt.Chan():
			if e.Err != nil {
				logger.Errorf("receive rpc event: %v", e.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 = rpcSvrChan.Receive()
		}
	}

}

func downloadHubConfig() coormq.GetHubConfigResp {
	coorCli := stgglb.CoordinatorRPCPool.Get()
	defer coorCli.Release()

	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	cfgResp, cerr := coorCli.GetHubConfig(ctx, coormq.ReqGetHubConfig(cortypes.HubID(config.Cfg().ID)))
	if cerr != nil {
		logger.Errorf("getting hub config: %v", cerr)
		os.Exit(1)
	}

	return *cfgResp
}

func servePublisher(evtPub *sysevent.Publisher) {
	logger.Info("start serving sysevent publisher")

	ch := evtPub.Start()

loop:
	for {
		val, err := ch.Receive().Wait(context.Background())
		if err != nil {
			logger.Errorf("sysevent publisher stopped with error: %s", err.Error())
			break
		}

		switch val := val.(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)
		}
	}
	logger.Info("sysevent publisher stopped")

	// TODO 仅简单结束了程序
	os.Exit(1)
}

func setupTickTask(hubPool *pool.Pool, evtPub *sysevent.Publisher) gocron.Scheduler {
	sch, err := gocron.NewScheduler()
	if err != nil {
		logger.Errorf("new cron scheduler: %s", err.Error())
		os.Exit(1)
	}

	// sch.NewJob(gocron.DailyJob(1, gocron.NewAtTimes(
	// 	gocron.NewAtTime(0, 0, 0),
	// )), gocron.NewTask(tickevent.ReportStorageStats, hubPool, evtPub))

	// sch.NewJob(gocron.DailyJob(1, gocron.NewAtTimes(
	// 	gocron.NewAtTime(0, 0, 1),
	// )), gocron.NewTask(tickevent.ReportHubTransferStats, evtPub))

	// sch.NewJob(gocron.DailyJob(1, gocron.NewAtTimes(
	// 	gocron.NewAtTime(0, 0, 2),
	// )), gocron.NewTask(tickevent.ReportHubStorageTransferStats, evtPub))

	return sch
}

func serveHTTP(server *http.Server) {
	logger.Info("start serving http")

	err := server.Serve()

	if err != nil {
		logger.Errorf("http stopped with error: %s", err.Error())
	}

	logger.Info("http stopped")

	// TODO 仅简单结束了程序
	os.Exit(1)
}