移植scanner代码；增加交互式命令行功能
This commit is contained in:
Sydonian
parent
bb0fbeb7d6
commit
938d3b69b8
|
|
@ -15,7 +15,9 @@ import (
|
|||
"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/repl"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/services"
|
||||
"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"
|
||||
|
|
@ -124,6 +126,15 @@ func serveHTTP(configPath string, opts serveHTTPOptions) {
|
|||
// 上传器
|
||||
uploader := uploader.NewUploader(distlockSvc, &conCol, stgPool, stgMeta, db)
|
||||
|
||||
// 定时任务
|
||||
tktk := ticktock.New(config.Cfg().TickTock, db, stgMeta, stgPool, evtPub)
|
||||
tktk.Start()
|
||||
defer tktk.Stop()
|
||||
|
||||
// 交互式命令行
|
||||
rep := repl.New(db, tktk)
|
||||
replCh := rep.Start()
|
||||
|
||||
// 挂载
|
||||
mntCfg := config.Cfg().Mount
|
||||
if !opts.DisableMount && mntCfg != nil && mntCfg.Enabled {
|
||||
|
|
@ -156,6 +167,7 @@ func serveHTTP(configPath string, opts serveHTTPOptions) {
|
|||
|
||||
evtPubEvt := evtPubChan.Receive()
|
||||
acStatEvt := acStatChan.Receive()
|
||||
replEvt := replCh.Receive()
|
||||
httpEvt := httpChan.Receive()
|
||||
mntEvt := mntChan.Receive()
|
||||
|
||||
|
|
@ -197,6 +209,19 @@ loop:
|
|||
}
|
||||
acStatEvt = acStatChan.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)
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ import (
|
|||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader/strategy"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/http"
|
||||
mntcfg "gitlink.org.cn/cloudream/jcs-pub/client/internal/mount/config"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/ticktock"
|
||||
stgglb "gitlink.org.cn/cloudream/jcs-pub/common/globals"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/connectivity"
|
||||
hubrpc "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/grpc/hub"
|
||||
|
|
@ -25,6 +26,7 @@ type Config struct {
|
|||
Connectivity connectivity.Config `json:"connectivity"`
|
||||
Downloader downloader.Config `json:"downloader"`
|
||||
DownloadStrategy strategy.Config `json:"downloadStrategy"`
|
||||
TickTock ticktock.Config `json:"tickTock"`
|
||||
HTTP *http.Config `json:"http"`
|
||||
Mount *mntcfg.Config `json:"mount"`
|
||||
}
|
||||
|
|
|
|||
|
|
@ -253,6 +253,45 @@ func (db *ObjectDB) BatchGetDetails(ctx SQLContext, objectIDs []types.ObjectID)
|
|||
return details, nil
|
||||
}
|
||||
|
||||
func (db *ObjectDB) BatchGetDetailsPaged(ctx SQLContext, pkgID types.PackageID, lastObjID types.ObjectID, maxCnt int) ([]types.ObjectDetail, error) {
|
||||
var objs []types.Object
|
||||
|
||||
err := ctx.Table("Object").Where("ObjectID > ? ORDER BY ObjectID ASC LIMIT ?", lastObjID, maxCnt).Find(&objs).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
objIDs := make([]types.ObjectID, len(objs))
|
||||
for i, obj := range objs {
|
||||
objIDs[i] = obj.ObjectID
|
||||
}
|
||||
|
||||
// 获取所有的 ObjectBlock
|
||||
var allBlocks []types.ObjectBlock
|
||||
err = ctx.Table("ObjectBlock").Where("ObjectID IN ?", objIDs).Order("ObjectID, `Index` ASC").Find(&allBlocks).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// 获取所有的 PinnedObject
|
||||
var allPinnedObjs []types.PinnedObject
|
||||
err = ctx.Table("PinnedObject").Where("ObjectID IN ?", objIDs).Order("ObjectID ASC").Find(&allPinnedObjs).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
details := make([]types.ObjectDetail, len(objs))
|
||||
for i, obj := range objs {
|
||||
details[i] = types.ObjectDetail{
|
||||
Object: obj,
|
||||
}
|
||||
}
|
||||
|
||||
types.DetailsFillObjectBlocks(details, allBlocks)
|
||||
types.DetailsFillPinnedAt(details, allPinnedObjs)
|
||||
return details, nil
|
||||
}
|
||||
|
||||
func (db *ObjectDB) Create(ctx SQLContext, obj types.Object) (types.ObjectID, error) {
|
||||
err := ctx.Table("Object").Create(&obj).Error
|
||||
if err != nil {
|
||||
|
|
|
|||
|
|
@ -28,6 +28,82 @@ func (db *PackageDB) GetByName(ctx SQLContext, bucketID types.BucketID, name str
|
|||
return ret, err
|
||||
}
|
||||
|
||||
func (db *PackageDB) GetDetail(ctx SQLContext, packageID types.PackageID) (types.PackageDetail, error) {
|
||||
var pkg types.Package
|
||||
err := ctx.Table("Package").Where("PackageID = ?", packageID).First(&pkg).Error
|
||||
if err != nil {
|
||||
return types.PackageDetail{}, err
|
||||
}
|
||||
|
||||
var ret struct {
|
||||
ObjectCount int64
|
||||
TotalSize int64
|
||||
}
|
||||
|
||||
err = ctx.Table("Object").
|
||||
Select("COUNT(*) as ObjectCount, SUM(Size) as TotalSize").
|
||||
Where("PackageID = ?", packageID).
|
||||
First(&ret).
|
||||
Error
|
||||
if err != nil {
|
||||
return types.PackageDetail{}, err
|
||||
}
|
||||
|
||||
return types.PackageDetail{
|
||||
Package: pkg,
|
||||
ObjectCount: ret.ObjectCount,
|
||||
TotalSize: ret.TotalSize,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (db *PackageDB) BatchGetDetailPaged(ctx SQLContext, lastPkgID types.PackageID, count int) ([]types.PackageDetail, error) {
|
||||
var pkgs []types.Package
|
||||
err := ctx.Table("Package").
|
||||
Where("PackageID > ?", lastPkgID).
|
||||
Order("PackageID ASC").
|
||||
Limit(count).
|
||||
Find(&pkgs).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if len(pkgs) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
minPkgID := pkgs[0].PackageID
|
||||
maxPkgID := pkgs[len(pkgs)-1].PackageID
|
||||
|
||||
type detail struct {
|
||||
ObjectCount int64
|
||||
TotalSize int64
|
||||
}
|
||||
|
||||
var details []detail
|
||||
err = ctx.
|
||||
Table("Package").
|
||||
Select("COUNT(ObjectID) as ObjectCount, SUM(Size) as TotalSize").
|
||||
Joins("left join Object o on Package.PackageID = o.PackageID").
|
||||
Where("Package.PackageID >= ? AND Package.PackageID <= ?", minPkgID, maxPkgID).
|
||||
Group("Package.PackageID").
|
||||
Order("Package.PackageID ASC").
|
||||
Find(&details).Error
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ret := make([]types.PackageDetail, len(pkgs))
|
||||
for i := range pkgs {
|
||||
ret[i] = types.PackageDetail{
|
||||
Package: pkgs[i],
|
||||
ObjectCount: details[i].ObjectCount,
|
||||
TotalSize: details[i].TotalSize,
|
||||
}
|
||||
}
|
||||
|
||||
return ret, nil
|
||||
}
|
||||
|
||||
func (db *PackageDB) BatchTestPackageID(ctx SQLContext, pkgIDs []types.PackageID) (map[types.PackageID]bool, error) {
|
||||
if len(pkgIDs) == 0 {
|
||||
return make(map[types.PackageID]bool), nil
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
package cmdline
|
||||
package repl
|
||||
|
||||
/*
|
||||
import (
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
package cmdline
|
||||
package repl
|
||||
|
||||
/*
|
||||
import (
|
||||
|
|
@ -1,6 +1,5 @@
|
|||
package cmdline
|
||||
package repl
|
||||
|
||||
/*
|
||||
import (
|
||||
"fmt"
|
||||
"strconv"
|
||||
|
|
@ -8,7 +7,7 @@ import (
|
|||
|
||||
"github.com/jedib0t/go-pretty/v6/table"
|
||||
"github.com/spf13/cobra"
|
||||
cdssdk "gitlink.org.cn/cloudream/jcs-pub/client/types"
|
||||
clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types"
|
||||
)
|
||||
|
||||
func init() {
|
||||
|
|
@ -27,7 +26,7 @@ func init() {
|
|||
return
|
||||
}
|
||||
|
||||
lspOneByID(cmdCtx, cdssdk.PackageID(id))
|
||||
lspOneByID(cmdCtx, clitypes.PackageID(id))
|
||||
} else {
|
||||
lspByPath(cmdCtx, args[0])
|
||||
}
|
||||
|
|
@ -39,13 +38,15 @@ func init() {
|
|||
}
|
||||
|
||||
func lspByPath(cmdCtx *CommandContext, path string) {
|
||||
comps := strings.Split(strings.Trim(path, cdssdk.ObjectPathSeparator), cdssdk.ObjectPathSeparator)
|
||||
db2 := cmdCtx.repl.db
|
||||
|
||||
comps := strings.Split(strings.Trim(path, clitypes.ObjectPathSeparator), clitypes.ObjectPathSeparator)
|
||||
if len(comps) != 2 {
|
||||
fmt.Printf("Package path must be in format of <bucket>/<package>")
|
||||
return
|
||||
}
|
||||
|
||||
pkg, err := cmdCtx.Cmdline.Svc.PackageSvc().GetByFullName(comps[0], comps[1])
|
||||
pkg, err := db2.Package().GetByFullName(db2.DefCtx(), comps[0], comps[1])
|
||||
if err != nil {
|
||||
fmt.Println(err)
|
||||
return
|
||||
|
|
@ -57,8 +58,10 @@ func lspByPath(cmdCtx *CommandContext, path string) {
|
|||
fmt.Println(wr.Render())
|
||||
}
|
||||
|
||||
func lspOneByID(cmdCtx *CommandContext, id cdssdk.PackageID) {
|
||||
pkg, err := cmdCtx.Cmdline.Svc.PackageSvc().Get(id)
|
||||
func lspOneByID(cmdCtx *CommandContext, id clitypes.PackageID) {
|
||||
db2 := cmdCtx.repl.db
|
||||
|
||||
pkg, err := db2.Package().GetByID(db2.DefCtx(), id)
|
||||
if err != nil {
|
||||
fmt.Println(err)
|
||||
return
|
||||
|
|
@ -69,4 +72,3 @@ func lspOneByID(cmdCtx *CommandContext, id cdssdk.PackageID) {
|
|||
wr.AppendRow(table.Row{pkg.PackageID, pkg.Name})
|
||||
fmt.Println(wr.Render())
|
||||
}
|
||||
*/
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
package cmdline
|
||||
package repl
|
||||
|
||||
/*
|
||||
import (
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
package cmdline
|
||||
package repl
|
||||
|
||||
/*
|
||||
import (
|
||||
|
|
@ -0,0 +1,91 @@
|
|||
package repl
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
|
||||
"github.com/c-bata/go-prompt"
|
||||
"github.com/spf13/cobra"
|
||||
"gitlink.org.cn/cloudream/common/pkgs/async"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/db"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/ticktock"
|
||||
)
|
||||
|
||||
var (
|
||||
cmdCtxKey = &CommandContext{}
|
||||
)
|
||||
|
||||
type ReplEventChan = async.UnboundChannel[ReplEvent]
|
||||
|
||||
type ReplEvent interface {
|
||||
IsReplEvent() bool
|
||||
}
|
||||
|
||||
type ExitEvent struct {
|
||||
ReplEvent
|
||||
}
|
||||
|
||||
type Repl struct {
|
||||
prompt *prompt.Prompt
|
||||
evtCh *ReplEventChan
|
||||
db *db.DB
|
||||
tktk *ticktock.TickTock
|
||||
}
|
||||
|
||||
func New(db *db.DB, ticktock *ticktock.TickTock) *Repl {
|
||||
r := &Repl{
|
||||
evtCh: async.NewUnboundChannel[ReplEvent](),
|
||||
db: db,
|
||||
tktk: ticktock,
|
||||
}
|
||||
r.prompt = prompt.New(
|
||||
r.executor,
|
||||
r.completer,
|
||||
prompt.OptionPrefix(">>> "),
|
||||
prompt.OptionTitle("JCS"),
|
||||
prompt.OptionSetExitCheckerOnInput(r.exitChecker),
|
||||
)
|
||||
return r
|
||||
}
|
||||
|
||||
func (r *Repl) Start() *ReplEventChan {
|
||||
go func() {
|
||||
r.prompt.Run()
|
||||
r.evtCh.Send(ExitEvent{})
|
||||
}()
|
||||
return r.evtCh
|
||||
}
|
||||
|
||||
func (r *Repl) completer(d prompt.Document) []prompt.Suggest {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Repl) executor(input string) {
|
||||
fields := strings.Fields(input)
|
||||
if len(fields) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
RootCmd.SetArgs(fields)
|
||||
RootCmd.ExecuteContext(context.WithValue(context.Background(), cmdCtxKey, &CommandContext{
|
||||
repl: r,
|
||||
}))
|
||||
}
|
||||
|
||||
func (r *Repl) exitChecker(input string, breakline bool) bool {
|
||||
if breakline && input == "exit" {
|
||||
return true
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
var RootCmd = &cobra.Command{}
|
||||
|
||||
type CommandContext struct {
|
||||
repl *Repl
|
||||
}
|
||||
|
||||
func GetCmdCtx(cmd *cobra.Command) *CommandContext {
|
||||
return cmd.Context().Value(cmdCtxKey).(*CommandContext)
|
||||
}
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
package cmdline
|
||||
package repl
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
package repl
|
||||
|
||||
import "github.com/spf13/cobra"
|
||||
|
||||
func init() {
|
||||
ttCmd := &cobra.Command{
|
||||
Use: "ticktock",
|
||||
Short: "ticktock command",
|
||||
}
|
||||
RootCmd.AddCommand(ttCmd)
|
||||
|
||||
runCmd := &cobra.Command{
|
||||
Use: "run [jobName]",
|
||||
Short: "run job now",
|
||||
Args: cobra.ExactArgs(1),
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
tickTockRun(GetCmdCtx(cmd), args[0])
|
||||
},
|
||||
}
|
||||
ttCmd.AddCommand(runCmd)
|
||||
}
|
||||
|
||||
func tickTockRun(ctx *CommandContext, jobName string) {
|
||||
ctx.repl.tktk.RunNow(jobName)
|
||||
}
|
||||
|
|
@ -0,0 +1,186 @@
|
|||
package ticktock
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"gitlink.org.cn/cloudream/common/pkgs/logger"
|
||||
"gitlink.org.cn/cloudream/common/utils/reflect2"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/db"
|
||||
clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/common/models/datamap"
|
||||
)
|
||||
|
||||
const (
|
||||
BatchGetPackageDetailCount = 100
|
||||
BatchGetObjectDetailCount = 1000
|
||||
)
|
||||
|
||||
type ChangeRedundancy struct {
|
||||
}
|
||||
|
||||
func (j *ChangeRedundancy) Name() string {
|
||||
return reflect2.TypeNameOf[ChangeRedundancy]()
|
||||
}
|
||||
|
||||
func (j *ChangeRedundancy) Execute(t *TickTock) {
|
||||
log := logger.WithType[ChangeRedundancy]("TickTock")
|
||||
startTime := time.Now()
|
||||
log.Debugf("job start")
|
||||
defer func() {
|
||||
log.Debugf("job end, time: %v", time.Since(startTime))
|
||||
}()
|
||||
|
||||
ctx := &changeRedundancyContext{
|
||||
ticktock: t,
|
||||
allUserSpaces: make(map[clitypes.UserSpaceID]*userSpaceLoadInfo),
|
||||
}
|
||||
|
||||
spaceIDs, err := t.db.UserSpace().GetAllIDs(t.db.DefCtx())
|
||||
if err != nil {
|
||||
log.Warnf("get user space ids: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
spaces := t.spaceMeta.GetMany(spaceIDs)
|
||||
for _, space := range spaces {
|
||||
if space == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
ctx.allUserSpaces[space.UserSpace.UserSpaceID] = &userSpaceLoadInfo{
|
||||
UserSpace: space,
|
||||
}
|
||||
}
|
||||
|
||||
lastPkgID := clitypes.PackageID(0)
|
||||
for {
|
||||
pkgs, err := db.DoTx21(t.db, t.db.Package().BatchGetDetailPaged, lastPkgID, BatchGetPackageDetailCount)
|
||||
if err != nil {
|
||||
log.Warnf("get package details: %v", err)
|
||||
return
|
||||
}
|
||||
if len(pkgs) == 0 {
|
||||
break
|
||||
}
|
||||
lastPkgID = pkgs[len(pkgs)-1].Package.PackageID
|
||||
|
||||
for _, p := range pkgs {
|
||||
err := j.changeOne(ctx, p)
|
||||
if err != nil {
|
||||
log.Warnf("change redundancy: %v", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type changeRedundancyContext struct {
|
||||
ticktock *TickTock
|
||||
allUserSpaces map[clitypes.UserSpaceID]*userSpaceLoadInfo
|
||||
mostBlockStgIDs []clitypes.UserSpaceID
|
||||
}
|
||||
|
||||
type userSpaceLoadInfo struct {
|
||||
UserSpace *clitypes.UserSpaceDetail
|
||||
AccessAmount float64
|
||||
}
|
||||
|
||||
func (j *ChangeRedundancy) changeOne(ctx *changeRedundancyContext, pkg clitypes.PackageDetail) error {
|
||||
log := logger.WithType[ChangeRedundancy]("TickTock")
|
||||
db2 := ctx.ticktock.db
|
||||
|
||||
// allUserSpaces是复用的，所以需要先清空
|
||||
for _, space := range ctx.allUserSpaces {
|
||||
space.AccessAmount = 0
|
||||
}
|
||||
|
||||
pkgAccessStats, err := db2.PackageAccessStat().GetByPackageID(db2.DefCtx(), pkg.Package.PackageID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("get package access stats: %w", err)
|
||||
}
|
||||
|
||||
for _, stat := range pkgAccessStats {
|
||||
info, ok := ctx.allUserSpaces[stat.UserSpaceID]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
info.AccessAmount = stat.Amount
|
||||
}
|
||||
|
||||
lastObjID := clitypes.ObjectID(0)
|
||||
for {
|
||||
objs, err := db.DoTx31(db2, db2.Object().BatchGetDetailsPaged, pkg.Package.PackageID, lastObjID, BatchGetObjectDetailCount)
|
||||
if err != nil {
|
||||
return fmt.Errorf("get object details: %w", err)
|
||||
}
|
||||
if len(objs) == 0 {
|
||||
break
|
||||
}
|
||||
lastObjID = objs[len(objs)-1].Object.ObjectID
|
||||
|
||||
var allUpdatings []db.UpdatingObjectRedundancy
|
||||
var allSysEvts []datamap.SysEventBody
|
||||
|
||||
ctx.mostBlockStgIDs = j.summaryRepObjectBlockUserSpaces(ctx, objs, 2)
|
||||
|
||||
// // TODO 加锁
|
||||
// builder := reqbuilder.NewBuilder()
|
||||
// for _, storage := range newRepStgs {
|
||||
// builder.Shard().Buzy(storage.Storage.Storage.StorageID)
|
||||
// }
|
||||
// for _, storage := range newECStgs {
|
||||
// builder.Shard().Buzy(storage.Storage.Storage.StorageID)
|
||||
// }
|
||||
// mutex, err := builder.MutexLock(execCtx.Args.DistLock)
|
||||
// if err != nil {
|
||||
// log.Warnf("acquiring dist lock: %s", err.Error())
|
||||
// return
|
||||
// }
|
||||
// defer mutex.Unlock()
|
||||
|
||||
var willShrinks []clitypes.ObjectDetail
|
||||
|
||||
for _, obj := range objs {
|
||||
newRed, selectedStorages := j.chooseRedundancy(ctx, obj)
|
||||
// 冗余策略不需要调整，就检查是否需要收缩
|
||||
if newRed == nil {
|
||||
willShrinks = append(willShrinks, obj)
|
||||
continue
|
||||
}
|
||||
|
||||
updating, evt, err := j.doChangeRedundancy(ctx, obj, newRed, selectedStorages)
|
||||
if updating != nil {
|
||||
allUpdatings = append(allUpdatings, *updating)
|
||||
}
|
||||
if evt != nil {
|
||||
allSysEvts = append(allSysEvts, evt)
|
||||
}
|
||||
if err != nil {
|
||||
log.WithField("ObjectID", obj.Object.ObjectID).Warnf("%s, its redundancy wont be changed", err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
udpatings, sysEvts, err := j.doRedundancyShrink(ctx, pkg, willShrinks)
|
||||
if err != nil {
|
||||
log.Warnf("redundancy shrink: %s", err.Error())
|
||||
return err
|
||||
}
|
||||
allUpdatings = append(allUpdatings, udpatings...)
|
||||
allSysEvts = append(allSysEvts, sysEvts...)
|
||||
|
||||
if len(allUpdatings) > 0 {
|
||||
err := db.DoTx10(db2, db2.Object().BatchUpdateRedundancy, allUpdatings)
|
||||
if err != nil {
|
||||
log.Warnf("update object redundancy: %s", err.Error())
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
for _, e := range allSysEvts {
|
||||
ctx.ticktock.evtPub.Publish(e)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -0,0 +1,5 @@
|
|||
package ticktock
|
||||
|
||||
type Config struct {
|
||||
ECFileSizeThreshold int64 `json:"ecFileSizeThreshold"`
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -0,0 +1,959 @@
|
|||
package ticktock
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math"
|
||||
"math/rand"
|
||||
"sync"
|
||||
|
||||
"github.com/samber/lo"
|
||||
"gitlink.org.cn/cloudream/common/pkgs/bitmap"
|
||||
"gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec"
|
||||
"gitlink.org.cn/cloudream/common/pkgs/logger"
|
||||
"gitlink.org.cn/cloudream/common/utils/lo2"
|
||||
"gitlink.org.cn/cloudream/common/utils/math2"
|
||||
"gitlink.org.cn/cloudream/common/utils/sort2"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/db"
|
||||
clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/common/consts"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/common/models/datamap"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/ops2"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/parser"
|
||||
)
|
||||
|
||||
func (t *ChangeRedundancy) doRedundancyShrink(execCtx *changeRedundancyContext, pkg clitypes.PackageDetail, objs []clitypes.ObjectDetail) ([]db.UpdatingObjectRedundancy, []datamap.SysEventBody, error) {
|
||||
log := logger.WithType[ChangeRedundancy]("TickTock")
|
||||
|
||||
var readerStgIDs []clitypes.UserSpaceID
|
||||
for _, space := range execCtx.allUserSpaces {
|
||||
// TODO 可以考虑做成配置
|
||||
if space.AccessAmount >= float64(pkg.ObjectCount/2) {
|
||||
readerStgIDs = append(readerStgIDs, space.UserSpace.UserSpace.UserSpaceID)
|
||||
}
|
||||
}
|
||||
|
||||
// 只对ec和rep对象进行处理
|
||||
var ecObjects []clitypes.ObjectDetail
|
||||
var repObjects []clitypes.ObjectDetail
|
||||
for _, obj := range objs {
|
||||
if _, ok := obj.Object.Redundancy.(*clitypes.ECRedundancy); ok {
|
||||
ecObjects = append(ecObjects, obj)
|
||||
} else if _, ok := obj.Object.Redundancy.(*clitypes.RepRedundancy); ok {
|
||||
repObjects = append(repObjects, obj)
|
||||
}
|
||||
}
|
||||
|
||||
planBld := exec.NewPlanBuilder()
|
||||
planningStgIDs := make(map[clitypes.UserSpaceID]bool)
|
||||
|
||||
var sysEvents []datamap.SysEventBody
|
||||
|
||||
// 对于rep对象，统计出所有对象块分布最多的两个节点，用这两个节点代表所有rep对象块的分布，去进行退火算法
|
||||
var repObjectsUpdating []db.UpdatingObjectRedundancy
|
||||
repMostHubIDs := t.summaryRepObjectBlockNodes(repObjects)
|
||||
solu := t.startAnnealing(execCtx, readerStgIDs, annealingObject{
|
||||
totalBlockCount: 1,
|
||||
minBlockCnt: 1,
|
||||
pinnedAt: repMostHubIDs,
|
||||
blocks: nil,
|
||||
})
|
||||
for _, obj := range repObjects {
|
||||
repObjectsUpdating = append(repObjectsUpdating, t.makePlansForRepObject(execCtx, solu, obj, planBld, planningStgIDs))
|
||||
sysEvents = append(sysEvents, t.generateSysEventForRepObject(solu, obj)...)
|
||||
}
|
||||
|
||||
// 对于ec对象，则每个对象单独进行退火算法
|
||||
var ecObjectsUpdating []db.UpdatingObjectRedundancy
|
||||
for _, obj := range ecObjects {
|
||||
ecRed := obj.Object.Redundancy.(*clitypes.ECRedundancy)
|
||||
solu := t.startAnnealing(execCtx, readerStgIDs, annealingObject{
|
||||
totalBlockCount: ecRed.N,
|
||||
minBlockCnt: ecRed.K,
|
||||
pinnedAt: obj.PinnedAt,
|
||||
blocks: obj.Blocks,
|
||||
})
|
||||
ecObjectsUpdating = append(ecObjectsUpdating, t.makePlansForECObject(execCtx, solu, obj, planBld, planningStgIDs))
|
||||
sysEvents = append(sysEvents, t.generateSysEventForECObject(solu, obj)...)
|
||||
}
|
||||
|
||||
ioSwRets, err := t.executePlans(execCtx, planBld, planningStgIDs)
|
||||
if err != nil {
|
||||
log.Warn(err.Error())
|
||||
return nil, nil, fmt.Errorf("execute plans: %w", err)
|
||||
}
|
||||
|
||||
// 根据按照方案进行调整的结果，填充更新元数据的命令
|
||||
for i := range ecObjectsUpdating {
|
||||
t.populateECObjectEntry(&ecObjectsUpdating[i], ecObjects[i], ioSwRets)
|
||||
}
|
||||
|
||||
return append(repObjectsUpdating, ecObjectsUpdating...), sysEvents, nil
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) summaryRepObjectBlockNodes(objs []clitypes.ObjectDetail) []clitypes.UserSpaceID {
|
||||
type stgBlocks struct {
|
||||
UserSpaceID clitypes.UserSpaceID
|
||||
Count int
|
||||
}
|
||||
|
||||
stgBlocksMap := make(map[clitypes.UserSpaceID]*stgBlocks)
|
||||
for _, obj := range objs {
|
||||
cacheBlockStgs := make(map[clitypes.UserSpaceID]bool)
|
||||
for _, block := range obj.Blocks {
|
||||
if _, ok := stgBlocksMap[block.UserSpaceID]; !ok {
|
||||
stgBlocksMap[block.UserSpaceID] = &stgBlocks{
|
||||
UserSpaceID: block.UserSpaceID,
|
||||
Count: 0,
|
||||
}
|
||||
}
|
||||
stgBlocksMap[block.UserSpaceID].Count++
|
||||
cacheBlockStgs[block.UserSpaceID] = true
|
||||
}
|
||||
|
||||
for _, hubID := range obj.PinnedAt {
|
||||
if cacheBlockStgs[hubID] {
|
||||
continue
|
||||
}
|
||||
|
||||
if _, ok := stgBlocksMap[hubID]; !ok {
|
||||
stgBlocksMap[hubID] = &stgBlocks{
|
||||
UserSpaceID: hubID,
|
||||
Count: 0,
|
||||
}
|
||||
}
|
||||
stgBlocksMap[hubID].Count++
|
||||
}
|
||||
}
|
||||
|
||||
stgs := lo.Values(stgBlocksMap)
|
||||
sort2.Sort(stgs, func(left *stgBlocks, right *stgBlocks) int {
|
||||
return right.Count - left.Count
|
||||
})
|
||||
|
||||
// 只选出块数超过一半的节点，但要保证至少有两个节点
|
||||
for i := 2; i < len(stgs); i++ {
|
||||
if stgs[i].Count < len(objs)/2 {
|
||||
stgs = stgs[:i]
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return lo.Map(stgs, func(item *stgBlocks, idx int) clitypes.UserSpaceID { return item.UserSpaceID })
|
||||
}
|
||||
|
||||
type annealingState struct {
|
||||
ctx *changeRedundancyContext
|
||||
readerStgIDs []clitypes.UserSpaceID // 近期可能访问此对象的节点
|
||||
stgsSortedByReader map[clitypes.UserSpaceID][]stgDist // 拥有数据的节点到每个可能访问对象的节点按距离排序
|
||||
object annealingObject // 进行退火的对象
|
||||
blockList []objectBlock // 排序后的块分布情况
|
||||
stgBlockBitmaps map[clitypes.UserSpaceID]*bitmap.Bitmap64 // 用位图的形式表示每一个节点上有哪些块
|
||||
stgCombTree combinatorialTree // 节点组合树，用于加速计算容灾度
|
||||
|
||||
maxScore float64 // 搜索过程中得到过的最大分数
|
||||
maxScoreRmBlocks []bool // 最大分数对应的删除方案
|
||||
|
||||
rmBlocks []bool // 当前删除方案
|
||||
inversedIndex int // 当前删除方案是从上一次的方案改动哪个flag而来的
|
||||
lastDisasterTolerance float64 // 上一次方案的容灾度
|
||||
lastSpaceCost float64 // 上一次方案的冗余度
|
||||
lastMinAccessCost float64 // 上一次方案的最小访问费用
|
||||
lastScore float64 // 上一次方案的分数
|
||||
}
|
||||
|
||||
type objectBlock struct {
|
||||
Index int
|
||||
UserSpaceID clitypes.UserSpaceID
|
||||
HasEntity bool // 节点拥有实际的文件数据块
|
||||
HasShadow bool // 如果节点拥有完整文件数据，那么认为这个节点拥有所有块，这些块被称为影子块
|
||||
FileHash clitypes.FileHash // 只有在拥有实际文件数据块时，这个字段才有值
|
||||
Size int64 // 块大小
|
||||
}
|
||||
|
||||
type stgDist struct {
|
||||
UserSpaceID clitypes.UserSpaceID
|
||||
Distance float64
|
||||
}
|
||||
|
||||
type combinatorialTree struct {
|
||||
nodes []combinatorialTreeNode
|
||||
blocksMaps map[int]bitmap.Bitmap64
|
||||
stgIDToLocalStgID map[clitypes.UserSpaceID]int
|
||||
localStgIDToStgID []clitypes.UserSpaceID
|
||||
}
|
||||
|
||||
type annealingObject struct {
|
||||
totalBlockCount int
|
||||
minBlockCnt int
|
||||
pinnedAt []clitypes.UserSpaceID
|
||||
blocks []clitypes.ObjectBlock
|
||||
}
|
||||
|
||||
const (
|
||||
iterActionNone = 0
|
||||
iterActionSkip = 1
|
||||
iterActionBreak = 2
|
||||
)
|
||||
|
||||
func newCombinatorialTree(stgBlocksMaps map[clitypes.UserSpaceID]*bitmap.Bitmap64) combinatorialTree {
|
||||
tree := combinatorialTree{
|
||||
blocksMaps: make(map[int]bitmap.Bitmap64),
|
||||
stgIDToLocalStgID: make(map[clitypes.UserSpaceID]int),
|
||||
}
|
||||
|
||||
tree.nodes = make([]combinatorialTreeNode, (1 << len(stgBlocksMaps)))
|
||||
for id, mp := range stgBlocksMaps {
|
||||
tree.stgIDToLocalStgID[id] = len(tree.localStgIDToStgID)
|
||||
tree.blocksMaps[len(tree.localStgIDToStgID)] = *mp
|
||||
tree.localStgIDToStgID = append(tree.localStgIDToStgID, id)
|
||||
}
|
||||
|
||||
tree.nodes[0].localHubID = -1
|
||||
index := 1
|
||||
tree.initNode(0, &tree.nodes[0], &index)
|
||||
|
||||
return tree
|
||||
}
|
||||
|
||||
func (t *combinatorialTree) initNode(minAvaiLocalHubID int, parent *combinatorialTreeNode, index *int) {
|
||||
for i := minAvaiLocalHubID; i < len(t.stgIDToLocalStgID); i++ {
|
||||
curIndex := *index
|
||||
*index++
|
||||
bitMp := t.blocksMaps[i]
|
||||
bitMp.Or(&parent.blocksBitmap)
|
||||
|
||||
t.nodes[curIndex] = combinatorialTreeNode{
|
||||
localHubID: i,
|
||||
parent: parent,
|
||||
blocksBitmap: bitMp,
|
||||
}
|
||||
t.initNode(i+1, &t.nodes[curIndex], index)
|
||||
}
|
||||
}
|
||||
|
||||
// 获得索引指定的节点所在的层
|
||||
func (t *combinatorialTree) GetDepth(index int) int {
|
||||
depth := 0
|
||||
|
||||
// 反复判断节点在哪个子树。从左到右，子树节点的数量呈现8 4 2的变化，由此可以得到每个子树的索引值的范围
|
||||
subTreeCount := 1 << len(t.stgIDToLocalStgID)
|
||||
for index > 0 {
|
||||
if index < subTreeCount {
|
||||
// 定位到一个子树后，深度+1，然后进入这个子树，使用同样的方法再进行定位。
|
||||
// 进入子树后需要将索引值-1，因为要去掉子树的根节点
|
||||
index--
|
||||
depth++
|
||||
} else {
|
||||
// 如果索引值不在这个子树范围内，则将值减去子树的节点数量，
|
||||
// 这样每一次都可以视为使用同样的逻辑对不同大小的树进行判断。
|
||||
index -= subTreeCount
|
||||
}
|
||||
subTreeCount >>= 1
|
||||
}
|
||||
|
||||
return depth
|
||||
}
|
||||
|
||||
// 更新某一个算力中心节点的块分布位图，同时更新它对应组合树节点的所有子节点。
|
||||
// 如果更新到某个节点时，已有K个块，那么就不会再更新它的子节点
|
||||
func (t *combinatorialTree) UpdateBitmap(stgID clitypes.UserSpaceID, mp bitmap.Bitmap64, k int) {
|
||||
t.blocksMaps[t.stgIDToLocalStgID[stgID]] = mp
|
||||
// 首先定义两种遍历树节点时的移动方式：
|
||||
// 1. 竖直移动（深度增加）：从一个节点移动到它最左边的子节点。每移动一步，index+1
|
||||
// 2. 水平移动：从一个节点移动到它右边的兄弟节点。每移动一步，根据它所在的深度，index+8，+4，+2
|
||||
// LocalID从0开始，将其+1后得到移动步数steps。
|
||||
// 将移动步数拆成多部分，分配到上述的两种移动方式上，并进行任意组合，且保证第一次为至少进行一次的竖直移动，移动之后的节点都会是同一个计算中心节点。
|
||||
steps := t.stgIDToLocalStgID[stgID] + 1
|
||||
for d := 1; d <= steps; d++ {
|
||||
t.iterCombBits(len(t.stgIDToLocalStgID)-1, steps-d, 0, func(i int) {
|
||||
index := d + i
|
||||
node := &t.nodes[index]
|
||||
|
||||
newMp := t.blocksMaps[node.localHubID]
|
||||
newMp.Or(&node.parent.blocksBitmap)
|
||||
node.blocksBitmap = newMp
|
||||
if newMp.Weight() >= k {
|
||||
return
|
||||
}
|
||||
|
||||
t.iterChildren(index, func(index, parentIndex, depth int) int {
|
||||
curNode := &t.nodes[index]
|
||||
parentNode := t.nodes[parentIndex]
|
||||
|
||||
newMp := t.blocksMaps[curNode.localHubID]
|
||||
newMp.Or(&parentNode.blocksBitmap)
|
||||
curNode.blocksBitmap = newMp
|
||||
if newMp.Weight() >= k {
|
||||
return iterActionSkip
|
||||
}
|
||||
|
||||
return iterActionNone
|
||||
})
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 遍历树，找到至少拥有K个块的树节点的最大深度
|
||||
func (t *combinatorialTree) FindKBlocksMaxDepth(k int) int {
|
||||
maxDepth := -1
|
||||
t.iterChildren(0, func(index, parentIndex, depth int) int {
|
||||
if t.nodes[index].blocksBitmap.Weight() >= k {
|
||||
if maxDepth < depth {
|
||||
maxDepth = depth
|
||||
}
|
||||
return iterActionSkip
|
||||
}
|
||||
// 如果到了叶子节点，还没有找到K个块，那就认为要满足K个块，至少需要再多一个节点，即深度+1。
|
||||
// 由于遍历时采用的是深度优先的算法，因此遍历到这个叶子节点时，叶子节点再加一个节点的组合已经在前面搜索过，
|
||||
// 所以用当前叶子节点深度+1来作为当前分支的结果就可以，即使当前情况下增加任意一个节点依然不够K块，
|
||||
// 可以使用同样的思路去递推到当前叶子节点增加两个块的情况。
|
||||
if t.nodes[index].localHubID == len(t.stgIDToLocalStgID)-1 {
|
||||
if maxDepth < depth+1 {
|
||||
maxDepth = depth + 1
|
||||
}
|
||||
}
|
||||
|
||||
return iterActionNone
|
||||
})
|
||||
|
||||
if maxDepth == -1 || maxDepth > len(t.stgIDToLocalStgID) {
|
||||
return len(t.stgIDToLocalStgID)
|
||||
}
|
||||
|
||||
return maxDepth
|
||||
}
|
||||
|
||||
func (t *combinatorialTree) iterCombBits(width int, count int, offset int, callback func(int)) {
|
||||
if count == 0 {
|
||||
callback(offset)
|
||||
return
|
||||
}
|
||||
|
||||
for b := width; b >= count; b-- {
|
||||
t.iterCombBits(b-1, count-1, offset+(1<<b), callback)
|
||||
}
|
||||
}
|
||||
|
||||
func (t *combinatorialTree) iterChildren(index int, do func(index int, parentIndex int, depth int) int) {
|
||||
curNode := &t.nodes[index]
|
||||
childIndex := index + 1
|
||||
curDepth := t.GetDepth(index)
|
||||
|
||||
childCounts := len(t.stgIDToLocalStgID) - 1 - curNode.localHubID
|
||||
if childCounts == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
childTreeNodeCnt := 1 << (childCounts - 1)
|
||||
for c := 0; c < childCounts; c++ {
|
||||
act := t.itering(childIndex, index, curDepth+1, do)
|
||||
if act == iterActionBreak {
|
||||
return
|
||||
}
|
||||
|
||||
childIndex += childTreeNodeCnt
|
||||
childTreeNodeCnt >>= 1
|
||||
}
|
||||
}
|
||||
|
||||
func (t *combinatorialTree) itering(index int, parentIndex int, depth int, do func(index int, parentIndex int, depth int) int) int {
|
||||
act := do(index, parentIndex, depth)
|
||||
if act == iterActionBreak {
|
||||
return act
|
||||
}
|
||||
if act == iterActionSkip {
|
||||
return iterActionNone
|
||||
}
|
||||
|
||||
curNode := &t.nodes[index]
|
||||
childIndex := index + 1
|
||||
|
||||
childCounts := len(t.stgIDToLocalStgID) - 1 - curNode.localHubID
|
||||
if childCounts == 0 {
|
||||
return iterActionNone
|
||||
}
|
||||
|
||||
childTreeNodeCnt := 1 << (childCounts - 1)
|
||||
for c := 0; c < childCounts; c++ {
|
||||
act = t.itering(childIndex, index, depth+1, do)
|
||||
if act == iterActionBreak {
|
||||
return act
|
||||
}
|
||||
|
||||
childIndex += childTreeNodeCnt
|
||||
childTreeNodeCnt >>= 1
|
||||
}
|
||||
|
||||
return iterActionNone
|
||||
}
|
||||
|
||||
type combinatorialTreeNode struct {
|
||||
localHubID int
|
||||
parent *combinatorialTreeNode
|
||||
blocksBitmap bitmap.Bitmap64 // 选择了这个中心之后，所有中心一共包含多少种块
|
||||
}
|
||||
|
||||
type annealingSolution struct {
|
||||
blockList []objectBlock // 所有节点的块分布情况
|
||||
rmBlocks []bool // 要删除哪些块
|
||||
disasterTolerance float64 // 本方案的容灾度
|
||||
spaceCost float64 // 本方案的冗余度
|
||||
minAccessCost float64 // 本方案的最小访问费用
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) startAnnealing(ctx *changeRedundancyContext, readerStgIDs []clitypes.UserSpaceID, object annealingObject) annealingSolution {
|
||||
state := &annealingState{
|
||||
ctx: ctx,
|
||||
readerStgIDs: readerStgIDs,
|
||||
stgsSortedByReader: make(map[clitypes.UserSpaceID][]stgDist),
|
||||
object: object,
|
||||
stgBlockBitmaps: make(map[clitypes.UserSpaceID]*bitmap.Bitmap64),
|
||||
}
|
||||
|
||||
t.initBlockList(state)
|
||||
if state.blockList == nil {
|
||||
return annealingSolution{}
|
||||
}
|
||||
|
||||
t.initNodeBlockBitmap(state)
|
||||
|
||||
t.sortNodeByReaderDistance(state)
|
||||
|
||||
state.rmBlocks = make([]bool, len(state.blockList))
|
||||
state.inversedIndex = -1
|
||||
state.stgCombTree = newCombinatorialTree(state.stgBlockBitmaps)
|
||||
|
||||
state.lastScore = t.calcScore(state)
|
||||
state.maxScore = state.lastScore
|
||||
state.maxScoreRmBlocks = lo2.ArrayClone(state.rmBlocks)
|
||||
|
||||
// 模拟退火算法的温度
|
||||
curTemp := state.lastScore
|
||||
// 结束温度
|
||||
finalTemp := curTemp * 0.2
|
||||
// 冷却率
|
||||
coolingRate := 0.95
|
||||
|
||||
for curTemp > finalTemp {
|
||||
state.inversedIndex = rand.Intn(len(state.rmBlocks))
|
||||
block := state.blockList[state.inversedIndex]
|
||||
state.rmBlocks[state.inversedIndex] = !state.rmBlocks[state.inversedIndex]
|
||||
state.stgBlockBitmaps[block.UserSpaceID].Set(block.Index, !state.rmBlocks[state.inversedIndex])
|
||||
state.stgCombTree.UpdateBitmap(block.UserSpaceID, *state.stgBlockBitmaps[block.UserSpaceID], state.object.minBlockCnt)
|
||||
|
||||
curScore := t.calcScore(state)
|
||||
|
||||
dScore := curScore - state.lastScore
|
||||
// 如果新方案比旧方案得分低，且没有要求强制接受新方案，那么就将变化改回去
|
||||
if curScore == 0 || (dScore < 0 && !t.alwaysAccept(curTemp, dScore, coolingRate)) {
|
||||
state.rmBlocks[state.inversedIndex] = !state.rmBlocks[state.inversedIndex]
|
||||
state.stgBlockBitmaps[block.UserSpaceID].Set(block.Index, !state.rmBlocks[state.inversedIndex])
|
||||
state.stgCombTree.UpdateBitmap(block.UserSpaceID, *state.stgBlockBitmaps[block.UserSpaceID], state.object.minBlockCnt)
|
||||
// fmt.Printf("\n")
|
||||
} else {
|
||||
// fmt.Printf(" accept!\n")
|
||||
state.lastScore = curScore
|
||||
if state.maxScore < curScore {
|
||||
state.maxScore = state.lastScore
|
||||
state.maxScoreRmBlocks = lo2.ArrayClone(state.rmBlocks)
|
||||
}
|
||||
}
|
||||
curTemp *= coolingRate
|
||||
}
|
||||
// fmt.Printf("final: %v\n", state.maxScoreRmBlocks)
|
||||
return annealingSolution{
|
||||
blockList: state.blockList,
|
||||
rmBlocks: state.maxScoreRmBlocks,
|
||||
disasterTolerance: state.lastDisasterTolerance,
|
||||
spaceCost: state.lastSpaceCost,
|
||||
minAccessCost: state.lastMinAccessCost,
|
||||
}
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) initBlockList(ctx *annealingState) {
|
||||
blocksMap := make(map[clitypes.UserSpaceID][]objectBlock)
|
||||
|
||||
// 先生成所有的影子块
|
||||
for _, pinned := range ctx.object.pinnedAt {
|
||||
blocks := make([]objectBlock, 0, ctx.object.totalBlockCount)
|
||||
for i := 0; i < ctx.object.totalBlockCount; i++ {
|
||||
blocks = append(blocks, objectBlock{
|
||||
Index: i,
|
||||
UserSpaceID: pinned,
|
||||
HasShadow: true,
|
||||
})
|
||||
}
|
||||
blocksMap[pinned] = blocks
|
||||
}
|
||||
|
||||
// 再填充实际块
|
||||
for _, b := range ctx.object.blocks {
|
||||
blocks := blocksMap[b.UserSpaceID]
|
||||
|
||||
has := false
|
||||
for i := range blocks {
|
||||
if blocks[i].Index == b.Index {
|
||||
blocks[i].HasEntity = true
|
||||
blocks[i].FileHash = b.FileHash
|
||||
has = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if has {
|
||||
continue
|
||||
}
|
||||
|
||||
blocks = append(blocks, objectBlock{
|
||||
Index: b.Index,
|
||||
UserSpaceID: b.UserSpaceID,
|
||||
HasEntity: true,
|
||||
FileHash: b.FileHash,
|
||||
Size: b.Size,
|
||||
})
|
||||
blocksMap[b.UserSpaceID] = blocks
|
||||
}
|
||||
|
||||
var sortedBlocks []objectBlock
|
||||
for _, bs := range blocksMap {
|
||||
sortedBlocks = append(sortedBlocks, bs...)
|
||||
}
|
||||
sortedBlocks = sort2.Sort(sortedBlocks, func(left objectBlock, right objectBlock) int {
|
||||
d := left.UserSpaceID - right.UserSpaceID
|
||||
if d != 0 {
|
||||
return int(d)
|
||||
}
|
||||
|
||||
return left.Index - right.Index
|
||||
})
|
||||
|
||||
ctx.blockList = sortedBlocks
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) initNodeBlockBitmap(state *annealingState) {
|
||||
for _, b := range state.blockList {
|
||||
mp, ok := state.stgBlockBitmaps[b.UserSpaceID]
|
||||
if !ok {
|
||||
nb := bitmap.Bitmap64(0)
|
||||
mp = &nb
|
||||
state.stgBlockBitmaps[b.UserSpaceID] = mp
|
||||
}
|
||||
mp.Set(b.Index, true)
|
||||
}
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) sortNodeByReaderDistance(state *annealingState) {
|
||||
for _, r := range state.readerStgIDs {
|
||||
var nodeDists []stgDist
|
||||
|
||||
for n := range state.stgBlockBitmaps {
|
||||
if r == n {
|
||||
// 同节点时距离视为0.1
|
||||
nodeDists = append(nodeDists, stgDist{
|
||||
UserSpaceID: n,
|
||||
Distance: consts.StorageDistanceSameStorage,
|
||||
})
|
||||
} else if state.ctx.allUserSpaces[r].UserSpace.MasterHub.LocationID == state.ctx.allUserSpaces[n].UserSpace.MasterHub.LocationID {
|
||||
// 同地区时距离视为1
|
||||
nodeDists = append(nodeDists, stgDist{
|
||||
UserSpaceID: n,
|
||||
Distance: consts.StorageDistanceSameLocation,
|
||||
})
|
||||
} else {
|
||||
// 不同地区时距离视为5
|
||||
nodeDists = append(nodeDists, stgDist{
|
||||
UserSpaceID: n,
|
||||
Distance: consts.StorageDistanceOther,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
state.stgsSortedByReader[r] = sort2.Sort(nodeDists, func(left, right stgDist) int { return sort2.Cmp(left.Distance, right.Distance) })
|
||||
}
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) calcScore(state *annealingState) float64 {
|
||||
dt := t.calcDisasterTolerance(state)
|
||||
ac := t.calcMinAccessCost(state)
|
||||
sc := t.calcSpaceCost(state)
|
||||
|
||||
state.lastDisasterTolerance = dt
|
||||
state.lastMinAccessCost = ac
|
||||
state.lastSpaceCost = sc
|
||||
|
||||
dtSc := 1.0
|
||||
if dt < 1 {
|
||||
dtSc = 0
|
||||
} else if dt >= 2 {
|
||||
dtSc = 1.5
|
||||
}
|
||||
|
||||
newSc := 0.0
|
||||
if dt == 0 || ac == 0 {
|
||||
newSc = 0
|
||||
} else {
|
||||
newSc = dtSc / (sc * ac)
|
||||
}
|
||||
|
||||
// fmt.Printf("solu: %v, cur: %v, dt: %v, ac: %v, sc: %v \n", state.rmBlocks, newSc, dt, ac, sc)
|
||||
return newSc
|
||||
}
|
||||
|
||||
// 计算容灾度
|
||||
func (t *ChangeRedundancy) calcDisasterTolerance(state *annealingState) float64 {
|
||||
if state.inversedIndex != -1 {
|
||||
node := state.blockList[state.inversedIndex]
|
||||
state.stgCombTree.UpdateBitmap(node.UserSpaceID, *state.stgBlockBitmaps[node.UserSpaceID], state.object.minBlockCnt)
|
||||
}
|
||||
return float64(len(state.stgBlockBitmaps) - state.stgCombTree.FindKBlocksMaxDepth(state.object.minBlockCnt))
|
||||
}
|
||||
|
||||
// 计算最小访问数据的代价
|
||||
func (t *ChangeRedundancy) calcMinAccessCost(state *annealingState) float64 {
|
||||
cost := math.MaxFloat64
|
||||
for _, reader := range state.readerStgIDs {
|
||||
tarNodes := state.stgsSortedByReader[reader]
|
||||
gotBlocks := bitmap.Bitmap64(0)
|
||||
thisCost := 0.0
|
||||
|
||||
for _, tar := range tarNodes {
|
||||
tarNodeMp := state.stgBlockBitmaps[tar.UserSpaceID]
|
||||
|
||||
// 只需要从目的节点上获得缺少的块
|
||||
curWeigth := gotBlocks.Weight()
|
||||
// 下面的if会在拿到k个块之后跳出循环，所以or多了块也没关系
|
||||
gotBlocks.Or(tarNodeMp)
|
||||
// 但是算读取块的消耗时，不能多算，最多算读了k个块的消耗
|
||||
willGetBlocks := math2.Min(gotBlocks.Weight()-curWeigth, state.object.minBlockCnt-curWeigth)
|
||||
thisCost += float64(willGetBlocks) * float64(tar.Distance)
|
||||
|
||||
if gotBlocks.Weight() >= state.object.minBlockCnt {
|
||||
break
|
||||
}
|
||||
}
|
||||
if gotBlocks.Weight() >= state.object.minBlockCnt {
|
||||
cost = math.Min(cost, thisCost)
|
||||
}
|
||||
}
|
||||
|
||||
return cost
|
||||
}
|
||||
|
||||
// 计算冗余度
|
||||
func (t *ChangeRedundancy) calcSpaceCost(ctx *annealingState) float64 {
|
||||
blockCount := 0
|
||||
for i, b := range ctx.blockList {
|
||||
if ctx.rmBlocks[i] {
|
||||
continue
|
||||
}
|
||||
|
||||
if b.HasEntity {
|
||||
blockCount++
|
||||
}
|
||||
if b.HasShadow {
|
||||
blockCount++
|
||||
}
|
||||
}
|
||||
// 所有算力中心上拥有的块的总数 / 一个对象被分成了几个块
|
||||
return float64(blockCount) / float64(ctx.object.minBlockCnt)
|
||||
}
|
||||
|
||||
// 如果新方案得分比旧方案小，那么在一定概率内也接受新方案
|
||||
func (t *ChangeRedundancy) alwaysAccept(curTemp float64, dScore float64, coolingRate float64) bool {
|
||||
v := math.Exp(dScore / curTemp / coolingRate)
|
||||
// fmt.Printf(" -- chance: %v, temp: %v", v, curTemp)
|
||||
return v > rand.Float64()
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) makePlansForRepObject(ctx *changeRedundancyContext, solu annealingSolution, obj clitypes.ObjectDetail, planBld *exec.PlanBuilder, planningHubIDs map[clitypes.UserSpaceID]bool) db.UpdatingObjectRedundancy {
|
||||
entry := db.UpdatingObjectRedundancy{
|
||||
ObjectID: obj.Object.ObjectID,
|
||||
FileHash: obj.Object.FileHash,
|
||||
Size: obj.Object.Size,
|
||||
Redundancy: obj.Object.Redundancy,
|
||||
}
|
||||
|
||||
ft := ioswitch2.NewFromTo()
|
||||
|
||||
fromStg := ctx.allUserSpaces[obj.Blocks[0].UserSpaceID].UserSpace
|
||||
ft.AddFrom(ioswitch2.NewFromShardstore(obj.Object.FileHash, *fromStg.MasterHub, *fromStg, ioswitch2.RawStream()))
|
||||
|
||||
for i, f := range solu.rmBlocks {
|
||||
hasCache := lo.ContainsBy(obj.Blocks, func(b clitypes.ObjectBlock) bool { return b.UserSpaceID == solu.blockList[i].UserSpaceID }) ||
|
||||
lo.ContainsBy(obj.PinnedAt, func(n clitypes.UserSpaceID) bool { return n == solu.blockList[i].UserSpaceID })
|
||||
willRm := f
|
||||
|
||||
if !willRm {
|
||||
// 如果对象在退火后要保留副本的节点没有副本，则需要在这个节点创建副本
|
||||
if !hasCache {
|
||||
toStg := ctx.allUserSpaces[solu.blockList[i].UserSpaceID].UserSpace
|
||||
ft.AddTo(ioswitch2.NewToShardStore(*toStg.MasterHub, *toStg, ioswitch2.RawStream(), fmt.Sprintf("%d.0", obj.Object.ObjectID)))
|
||||
|
||||
planningHubIDs[solu.blockList[i].UserSpaceID] = true
|
||||
}
|
||||
entry.Blocks = append(entry.Blocks, clitypes.ObjectBlock{
|
||||
ObjectID: obj.Object.ObjectID,
|
||||
Index: solu.blockList[i].Index,
|
||||
UserSpaceID: solu.blockList[i].UserSpaceID,
|
||||
FileHash: obj.Object.FileHash,
|
||||
Size: solu.blockList[i].Size,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
err := parser.Parse(ft, planBld)
|
||||
if err != nil {
|
||||
// TODO 错误处理
|
||||
}
|
||||
|
||||
return entry
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) generateSysEventForRepObject(solu annealingSolution, obj clitypes.ObjectDetail) []datamap.SysEventBody {
|
||||
var blockChgs []datamap.BlockChange
|
||||
|
||||
for i, f := range solu.rmBlocks {
|
||||
hasCache := lo.ContainsBy(obj.Blocks, func(b clitypes.ObjectBlock) bool { return b.UserSpaceID == solu.blockList[i].UserSpaceID }) ||
|
||||
lo.ContainsBy(obj.PinnedAt, func(n clitypes.UserSpaceID) bool { return n == solu.blockList[i].UserSpaceID })
|
||||
willRm := f
|
||||
|
||||
if !willRm {
|
||||
// 如果对象在退火后要保留副本的节点没有副本，则需要在这个节点创建副本
|
||||
if !hasCache {
|
||||
blockChgs = append(blockChgs, &datamap.BlockChangeClone{
|
||||
BlockType: datamap.BlockTypeRaw,
|
||||
SourceUserSpaceID: obj.Blocks[0].UserSpaceID,
|
||||
TargetUserSpaceID: solu.blockList[i].UserSpaceID,
|
||||
})
|
||||
}
|
||||
} else {
|
||||
blockChgs = append(blockChgs, &datamap.BlockChangeDeleted{
|
||||
Index: 0,
|
||||
UserSpaceID: solu.blockList[i].UserSpaceID,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
transEvt := &datamap.BodyBlockTransfer{
|
||||
ObjectID: obj.Object.ObjectID,
|
||||
PackageID: obj.Object.PackageID,
|
||||
BlockChanges: blockChgs,
|
||||
}
|
||||
|
||||
var blockDist []datamap.BlockDistributionObjectInfo
|
||||
for i, f := range solu.rmBlocks {
|
||||
if !f {
|
||||
blockDist = append(blockDist, datamap.BlockDistributionObjectInfo{
|
||||
BlockType: datamap.BlockTypeRaw,
|
||||
Index: 0,
|
||||
UserSpaceID: solu.blockList[i].UserSpaceID,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
distEvt := &datamap.BodyBlockDistribution{
|
||||
ObjectID: obj.Object.ObjectID,
|
||||
PackageID: obj.Object.PackageID,
|
||||
Path: obj.Object.Path,
|
||||
Size: obj.Object.Size,
|
||||
FileHash: obj.Object.FileHash,
|
||||
FaultTolerance: solu.disasterTolerance,
|
||||
Redundancy: solu.spaceCost,
|
||||
AvgAccessCost: 0, // TODO 计算平均访问代价，从日常访问数据中统计
|
||||
BlockDistribution: blockDist,
|
||||
// TODO 不好计算传输量
|
||||
}
|
||||
|
||||
return []datamap.SysEventBody{transEvt, distEvt}
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) makePlansForECObject(ctx *changeRedundancyContext, solu annealingSolution, obj clitypes.ObjectDetail, planBld *exec.PlanBuilder, planningHubIDs map[clitypes.UserSpaceID]bool) db.UpdatingObjectRedundancy {
|
||||
entry := db.UpdatingObjectRedundancy{
|
||||
ObjectID: obj.Object.ObjectID,
|
||||
FileHash: obj.Object.FileHash,
|
||||
Size: obj.Object.Size,
|
||||
Redundancy: obj.Object.Redundancy,
|
||||
}
|
||||
|
||||
reconstrct := make(map[clitypes.UserSpaceID]*[]int)
|
||||
for i, f := range solu.rmBlocks {
|
||||
block := solu.blockList[i]
|
||||
if !f {
|
||||
entry.Blocks = append(entry.Blocks, clitypes.ObjectBlock{
|
||||
ObjectID: obj.Object.ObjectID,
|
||||
Index: block.Index,
|
||||
UserSpaceID: block.UserSpaceID,
|
||||
FileHash: block.FileHash,
|
||||
Size: block.Size,
|
||||
})
|
||||
|
||||
// 如果这个块是影子块，那么就要从完整对象里重建这个块
|
||||
if !block.HasEntity {
|
||||
re, ok := reconstrct[block.UserSpaceID]
|
||||
if !ok {
|
||||
re = &[]int{}
|
||||
reconstrct[block.UserSpaceID] = re
|
||||
}
|
||||
|
||||
*re = append(*re, block.Index)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
ecRed := obj.Object.Redundancy.(*clitypes.ECRedundancy)
|
||||
|
||||
for id, idxs := range reconstrct {
|
||||
// 依次生成每个节点上的执行计划，因为如果放到一个计划里一起生成，不能保证每个节点上的块用的都是本节点上的副本
|
||||
ft := ioswitch2.NewFromTo()
|
||||
ft.ECParam = ecRed
|
||||
ft.AddFrom(ioswitch2.NewFromShardstore(obj.Object.FileHash, *ctx.allUserSpaces[id].UserSpace.MasterHub, *ctx.allUserSpaces[id].UserSpace, ioswitch2.RawStream()))
|
||||
|
||||
for _, i := range *idxs {
|
||||
ft.AddTo(ioswitch2.NewToShardStore(*ctx.allUserSpaces[id].UserSpace.MasterHub, *ctx.allUserSpaces[id].UserSpace, ioswitch2.ECStream(i), fmt.Sprintf("%d.%d", obj.Object.ObjectID, i)))
|
||||
}
|
||||
|
||||
err := parser.Parse(ft, planBld)
|
||||
if err != nil {
|
||||
// TODO 错误处理
|
||||
continue
|
||||
}
|
||||
|
||||
planningHubIDs[id] = true
|
||||
}
|
||||
return entry
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) generateSysEventForECObject(solu annealingSolution, obj clitypes.ObjectDetail) []datamap.SysEventBody {
|
||||
var blockChgs []datamap.BlockChange
|
||||
|
||||
reconstrct := make(map[clitypes.UserSpaceID]*[]int)
|
||||
for i, f := range solu.rmBlocks {
|
||||
block := solu.blockList[i]
|
||||
if !f {
|
||||
// 如果这个块是影子块，那么就要从完整对象里重建这个块
|
||||
if !block.HasEntity {
|
||||
re, ok := reconstrct[block.UserSpaceID]
|
||||
if !ok {
|
||||
re = &[]int{}
|
||||
reconstrct[block.UserSpaceID] = re
|
||||
}
|
||||
|
||||
*re = append(*re, block.Index)
|
||||
}
|
||||
} else {
|
||||
blockChgs = append(blockChgs, &datamap.BlockChangeDeleted{
|
||||
Index: block.Index,
|
||||
UserSpaceID: block.UserSpaceID,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 由于每一个需要被重建的块都是从同中心的副本里构建出来的，所以对于每一个中心都要产生一个BlockChangeEnDecode
|
||||
for id, idxs := range reconstrct {
|
||||
var tarBlocks []datamap.Block
|
||||
for _, idx := range *idxs {
|
||||
tarBlocks = append(tarBlocks, datamap.Block{
|
||||
BlockType: datamap.BlockTypeEC,
|
||||
Index: idx,
|
||||
UserSpaceID: id,
|
||||
})
|
||||
}
|
||||
blockChgs = append(blockChgs, &datamap.BlockChangeEnDecode{
|
||||
SourceBlocks: []datamap.Block{{
|
||||
BlockType: datamap.BlockTypeRaw,
|
||||
Index: 0,
|
||||
UserSpaceID: id, // 影子块的原始对象就在同一个节点上
|
||||
}},
|
||||
TargetBlocks: tarBlocks,
|
||||
// 传输量为0
|
||||
})
|
||||
}
|
||||
|
||||
transEvt := &datamap.BodyBlockTransfer{
|
||||
ObjectID: obj.Object.ObjectID,
|
||||
PackageID: obj.Object.PackageID,
|
||||
BlockChanges: blockChgs,
|
||||
}
|
||||
|
||||
var blockDist []datamap.BlockDistributionObjectInfo
|
||||
for i, f := range solu.rmBlocks {
|
||||
if !f {
|
||||
blockDist = append(blockDist, datamap.BlockDistributionObjectInfo{
|
||||
BlockType: datamap.BlockTypeEC,
|
||||
Index: solu.blockList[i].Index,
|
||||
UserSpaceID: solu.blockList[i].UserSpaceID,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
distEvt := &datamap.BodyBlockDistribution{
|
||||
ObjectID: obj.Object.ObjectID,
|
||||
PackageID: obj.Object.PackageID,
|
||||
Path: obj.Object.Path,
|
||||
Size: obj.Object.Size,
|
||||
FileHash: obj.Object.FileHash,
|
||||
FaultTolerance: solu.disasterTolerance,
|
||||
Redundancy: solu.spaceCost,
|
||||
AvgAccessCost: 0, // TODO 计算平均访问代价，从日常访问数据中统计
|
||||
BlockDistribution: blockDist,
|
||||
// TODO 不好计算传输量
|
||||
}
|
||||
|
||||
return []datamap.SysEventBody{transEvt, distEvt}
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) executePlans(ctx *changeRedundancyContext, planBld *exec.PlanBuilder, planningStgIDs map[clitypes.UserSpaceID]bool) (map[string]exec.VarValue, error) {
|
||||
// TODO 统一加锁，有重复也没关系
|
||||
// lockBld := reqbuilder.NewBuilder()
|
||||
// for id := range planningStgIDs {
|
||||
// lockBld.Shard().Buzy(id)
|
||||
// }
|
||||
// lock, err := lockBld.MutexLock(ctx.Args.DistLock)
|
||||
// if err != nil {
|
||||
// return nil, fmt.Errorf("acquiring distlock: %w", err)
|
||||
// }
|
||||
// defer lock.Unlock()
|
||||
|
||||
wg := sync.WaitGroup{}
|
||||
|
||||
// 执行IO计划
|
||||
var ioSwRets map[string]exec.VarValue
|
||||
var ioSwErr error
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
execCtx := exec.NewExecContext()
|
||||
exec.SetValueByType(execCtx, ctx.ticktock.stgPool)
|
||||
ret, err := planBld.Execute(execCtx).Wait(context.TODO())
|
||||
if err != nil {
|
||||
ioSwErr = fmt.Errorf("executing io switch plan: %w", err)
|
||||
return
|
||||
}
|
||||
ioSwRets = ret
|
||||
}()
|
||||
|
||||
wg.Wait()
|
||||
|
||||
if ioSwErr != nil {
|
||||
return nil, ioSwErr
|
||||
}
|
||||
|
||||
return ioSwRets, nil
|
||||
}
|
||||
|
||||
func (t *ChangeRedundancy) populateECObjectEntry(entry *db.UpdatingObjectRedundancy, obj clitypes.ObjectDetail, ioRets map[string]exec.VarValue) {
|
||||
for i := range entry.Blocks {
|
||||
if entry.Blocks[i].FileHash != "" {
|
||||
continue
|
||||
}
|
||||
|
||||
key := fmt.Sprintf("%d.%d", obj.Object.ObjectID, entry.Blocks[i].Index)
|
||||
// 不应该出现key不存在的情况
|
||||
r := ioRets[key].(*ops2.ShardInfoValue)
|
||||
entry.Blocks[i].FileHash = r.Hash
|
||||
entry.Blocks[i].Size = r.Size
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,83 @@
|
|||
package ticktock
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/go-co-op/gocron/v2"
|
||||
"gitlink.org.cn/cloudream/common/pkgs/logger"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/db"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/client/internal/metacache"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/pool"
|
||||
"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/sysevent"
|
||||
)
|
||||
|
||||
type Job interface {
|
||||
Name() string
|
||||
Execute(t *TickTock)
|
||||
}
|
||||
|
||||
type cronJob struct {
|
||||
cronJob gocron.Job
|
||||
job Job
|
||||
}
|
||||
|
||||
type TickTock struct {
|
||||
cfg Config
|
||||
sch gocron.Scheduler
|
||||
jobs map[string]cronJob
|
||||
db *db.DB
|
||||
spaceMeta *metacache.UserSpaceMeta
|
||||
stgPool *pool.Pool
|
||||
evtPub *sysevent.Publisher
|
||||
}
|
||||
|
||||
func New(cfg Config, db *db.DB, spaceMeta *metacache.UserSpaceMeta, stgPool *pool.Pool, evtPub *sysevent.Publisher) *TickTock {
|
||||
sch, _ := gocron.NewScheduler()
|
||||
t := &TickTock{
|
||||
cfg: cfg,
|
||||
sch: sch,
|
||||
jobs: map[string]cronJob{},
|
||||
db: db,
|
||||
spaceMeta: spaceMeta,
|
||||
stgPool: stgPool,
|
||||
evtPub: evtPub,
|
||||
}
|
||||
t.initJobs()
|
||||
return t
|
||||
}
|
||||
|
||||
func (t *TickTock) Start() {
|
||||
t.sch.Start()
|
||||
}
|
||||
|
||||
func (t *TickTock) Stop() {
|
||||
t.sch.Shutdown()
|
||||
}
|
||||
|
||||
func (t *TickTock) RunNow(jobName string) {
|
||||
j, ok := t.jobs[jobName]
|
||||
if !ok {
|
||||
logger.Warnf("job %s not found", jobName)
|
||||
return
|
||||
}
|
||||
|
||||
j.cronJob.RunNow()
|
||||
}
|
||||
|
||||
func (t *TickTock) addJob(job Job, duration gocron.JobDefinition) {
|
||||
j, err := t.sch.NewJob(duration, gocron.NewTask(job.Execute, t))
|
||||
if err != nil {
|
||||
panic(fmt.Errorf("add job %s: %w", job.Name(), err))
|
||||
}
|
||||
|
||||
t.jobs[job.Name()] = cronJob{
|
||||
cronJob: j,
|
||||
job: job,
|
||||
}
|
||||
}
|
||||
|
||||
func (t *TickTock) initJobs() {
|
||||
t.addJob(&ChangeRedundancy{}, gocron.DailyJob(1, gocron.NewAtTimes(
|
||||
gocron.NewAtTime(0, 0, 0),
|
||||
)))
|
||||
}
|
||||
|
|
@ -223,3 +223,9 @@ type UserSpaceDetail struct {
|
|||
func (d UserSpaceDetail) String() string {
|
||||
return d.UserSpace.String()
|
||||
}
|
||||
|
||||
type PackageDetail struct {
|
||||
Package Package
|
||||
ObjectCount int64
|
||||
TotalSize int64
|
||||
}
|
||||
|
|
|
|||
|
|
@ -46,6 +46,9 @@
|
|||
"downloadStrategy": {
|
||||
"highLatencyHub": 35
|
||||
},
|
||||
"tickTock": {
|
||||
"ecFileSizeThreshold": 5242880
|
||||
},
|
||||
"http": {
|
||||
"enabled": true,
|
||||
"listen": "127.0.0.1:7890",
|
||||
|
|
|
|||
4
go.mod
4
go.mod
|
|
@ -57,7 +57,10 @@ require (
|
|||
github.com/jinzhu/now v1.1.5 // indirect
|
||||
github.com/jonboulle/clockwork v0.4.0 // indirect
|
||||
github.com/magiconair/properties v1.8.7 // indirect
|
||||
github.com/mattn/go-colorable v0.1.13 // indirect
|
||||
github.com/mattn/go-tty v0.0.3 // indirect
|
||||
github.com/pelletier/go-toml/v2 v2.2.2 // indirect
|
||||
github.com/pkg/term v1.2.0-beta.2 // indirect
|
||||
github.com/robfig/cron/v3 v3.0.1 // indirect
|
||||
github.com/rogpeppe/go-internal v1.12.0 // indirect
|
||||
github.com/sagikazarmark/locafero v0.4.0 // indirect
|
||||
|
|
@ -83,6 +86,7 @@ require (
|
|||
|
||||
require (
|
||||
github.com/antonfisher/nested-logrus-formatter v1.3.1 // indirect
|
||||
github.com/c-bata/go-prompt v0.2.6
|
||||
github.com/coreos/go-semver v0.3.1 // indirect
|
||||
github.com/coreos/go-systemd/v22 v22.5.0 // indirect
|
||||
github.com/gin-contrib/sse v0.1.0 // indirect
|
||||
|
|
|
|||
24
go.sum
24
go.sum
|
|
@ -32,6 +32,8 @@ github.com/bytedance/sonic v1.11.6 h1:oUp34TzMlL+OY1OUWxHqsdkgC/Zfc85zGqw9siXjrc
|
|||
github.com/bytedance/sonic v1.11.6/go.mod h1:LysEHSvpvDySVdC2f87zGWf6CIKJcAvqab1ZaiQtds4=
|
||||
github.com/bytedance/sonic/loader v0.1.1 h1:c+e5Pt1k/cy5wMveRDyk2X4B9hF4g7an8N3zCYjJFNM=
|
||||
github.com/bytedance/sonic/loader v0.1.1/go.mod h1:ncP89zfokxS5LZrJxl5z0UJcsk4M4yY2JpfqGeCtNLU=
|
||||
github.com/c-bata/go-prompt v0.2.6 h1:POP+nrHE+DfLYx370bedwNhsqmpCUynWPxuHi0C5vZI=
|
||||
github.com/c-bata/go-prompt v0.2.6/go.mod h1:/LMAke8wD2FsNu9EXNdHxNLbd9MedkPnCdfpU9wwHfY=
|
||||
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
|
||||
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
|
||||
github.com/cloudwego/base64x v0.1.4 h1:jwCgWpFanWmN8xoIUHa2rtzmkd5J2plF/dnLS6Xd/0Y=
|
||||
|
|
@ -161,10 +163,22 @@ github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg=
|
|||
github.com/magefile/mage v1.15.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
|
||||
github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY=
|
||||
github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0=
|
||||
github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
|
||||
github.com/mattn/go-colorable v0.1.7/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
|
||||
github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
|
||||
github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
|
||||
github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
|
||||
github.com/mattn/go-isatty v0.0.10/go.mod h1:qgIWMr58cqv1PHHyhnkY9lrL7etaEgOFcMEpPG5Rm84=
|
||||
github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
|
||||
github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
|
||||
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
|
||||
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
|
||||
github.com/mattn/go-runewidth v0.0.6/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
|
||||
github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
|
||||
github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc=
|
||||
github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
|
||||
github.com/mattn/go-tty v0.0.3 h1:5OfyWorkyO7xP52Mq7tB36ajHDG5OHrmBGIS/DtakQI=
|
||||
github.com/mattn/go-tty v0.0.3/go.mod h1:ihxohKRERHTVzN+aSVRwACLCeqIoZAWpoICkkvrWyR0=
|
||||
github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
|
||||
github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
|
||||
github.com/moby/sys/mountinfo v0.6.2 h1:BzJjoreD5BMFNmD9Rus6gdd1pLuecOFPt8wC+Vygl78=
|
||||
|
|
@ -177,6 +191,8 @@ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjY
|
|||
github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
|
||||
github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
|
||||
github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
|
||||
github.com/pkg/term v1.2.0-beta.2 h1:L3y/h2jkuBVFdWiJvNfYfKmzcCnILw7mJWm2JQuMppw=
|
||||
github.com/pkg/term v1.2.0-beta.2/go.mod h1:E25nymQcrSllhX42Ok8MRm1+hyBdHY0dCeiKZ9jpNGw=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
|
||||
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
|
|
@ -311,7 +327,14 @@ golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
|
|||
golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
|
||||
golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191008105621-543471e840be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200909081042-eff7692f9009/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200918174421-af09f7315aff/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
|
|
@ -319,6 +342,7 @@ golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBc
|
|||
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
|
|
|
|||
Loading…
Reference in New Issue