355 lines
9.3 KiB
Go
355 lines
9.3 KiB
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/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/services"
|
||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/spacesyncer"
|
||
"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/models/datamap"
|
||
"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/connectivity"
|
||
"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/publock"
|
||
"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/pool"
|
||
"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/sysevent"
|
||
)
|
||
|
||
// 初始化函数，将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)
|
||
|
||
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)
|
||
}
|
||
|
||
stgglb.UserID = accToken.GetToken().UserID
|
||
stgglb.InitPools(hubRPCCfg, corRPCCfg)
|
||
} else {
|
||
stgglb.UserID = 0
|
||
accToken = accesstoken.NewDisabled()
|
||
}
|
||
accTokenChan := accToken.Start()
|
||
defer accToken.Stop()
|
||
|
||
// 数据库
|
||
db, err := db.NewDB(&config.Cfg().DB)
|
||
if err != nil {
|
||
logger.Fatalf("new db failed, err: %s", err.Error())
|
||
}
|
||
|
||
// 初始化系统事件发布器
|
||
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()
|
||
|
||
// 元数据缓存
|
||
metaCacheHost := metacache.NewHost(db)
|
||
go metaCacheHost.Serve()
|
||
spaceMeta := metaCacheHost.AddStorageMeta()
|
||
hubMeta := metaCacheHost.AddHubMeta()
|
||
conMeta := metaCacheHost.AddConnectivity()
|
||
|
||
// 公共锁
|
||
publock := publock.NewService()
|
||
|
||
// 访问统计
|
||
acStat := accessstat.NewAccessStat(accessstat.Config{
|
||
// TODO 考虑放到配置里
|
||
ReportInterval: time.Second * 10,
|
||
}, db)
|
||
acStatChan := acStat.Start()
|
||
defer acStat.Stop()
|
||
|
||
// 存储管理器
|
||
stgPool := pool.NewPool()
|
||
|
||
// 下载策略
|
||
strgSel := strategy.NewSelector(config.Cfg().DownloadStrategy, spaceMeta, hubMeta, conMeta)
|
||
|
||
// 下载器
|
||
dlder := downloader.NewDownloader(config.Cfg().Downloader, conCol, stgPool, strgSel, db)
|
||
|
||
// 上传器
|
||
uploader := uploader.NewUploader(publock, conCol, stgPool, spaceMeta, db)
|
||
|
||
// 定时任务
|
||
tktk := ticktock.New(config.Cfg().TickTock, db, spaceMeta, stgPool, evtPub, publock)
|
||
tktk.Start()
|
||
defer tktk.Stop()
|
||
|
||
// 用户空间同步功能
|
||
spaceSync := spacesyncer.New(db, stgPool, spaceMeta)
|
||
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)
|
||
|
||
// 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)
|
||
httpChan := httpSvr.Start()
|
||
defer httpSvr.Stop()
|
||
|
||
/// 开始监听各个模块的事件
|
||
|
||
accTokenEvt := accTokenChan.Receive()
|
||
evtPubEvt := evtPubChan.Receive()
|
||
conColEvt := conColChan.Receive()
|
||
acStatEvt := acStatChan.Receive()
|
||
spaceSyncEvt := spaceSyncChan.Receive()
|
||
// replEvt := replCh.Receive()
|
||
httpEvt := httpChan.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 := <-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)
|
||
}
|