JCS-pub/client/internal/ticktock/change_redundancy.go

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"
	"gitlink.org.cn/cloudream/jcs-pub/client/internal/publock"
	jcstypes "gitlink.org.cn/cloudream/jcs-pub/common/types"
	"gitlink.org.cn/cloudream/jcs-pub/common/types/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.Infof("job start")
	defer func() {
		log.Infof("job end, time: %v", time.Since(startTime))
	}()

	if !t.cluster.IsMaster() {
		log.Infof("not master, skip")
		return
	}

	ctx := &changeRedundancyContext{
		ticktock:      t,
		allUserSpaces: make(map[jcstypes.UserSpaceID]*userSpaceUsageInfo),
	}

	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] = &userSpaceUsageInfo{
			UserSpace: space,
		}
	}
	if len(ctx.allUserSpaces) == 0 {
		log.Warnf("no user space found")
		return
	}

	lastPkgID := jcstypes.PackageID(0)

loop:
	for {
		pkgIDs, err := db.DoTx21(t.db, t.db.Package().BatchGetIDPaged, lastPkgID, BatchGetPackageDetailCount)
		if err != nil {
			log.Warnf("get package ids: %v", err)
			return
		}
		if len(pkgIDs) == 0 {
			break
		}
		lastPkgID = pkgIDs[len(pkgIDs)-1]

		for _, id := range pkgIDs {
			// 如果执行超过两个小时，则停止
			if time.Since(startTime) > time.Hour*2 {
				break loop
			}

			lock, err := t.pubLock.BeginMutex().Package().Buzy(id).End().Lock(publock.WithTimeout(time.Second * 10))
			if err != nil {
				log.Warnf("lock package: %v", err)
				continue
			}

			detail, err := db.DoTx11(t.db, t.db.Package().GetDetail, id)
			if err != nil {
				log.Warnf("get package detail: %v", err)
				lock.Unlock()
				continue
			}
			if detail.Package.Pinned {
				lock.Unlock()
				continue
			}

			err = j.changeOne(ctx, detail)

			lock.Unlock()
			if err != nil {
				log.Warnf("change redundancy: %v", err)
				return
			}
		}

	}
}

type changeRedundancyContext struct {
	ticktock        *TickTock
	allUserSpaces   map[jcstypes.UserSpaceID]*userSpaceUsageInfo
	mostBlockStgIDs []jcstypes.UserSpaceID
}

type userSpaceUsageInfo struct {
	UserSpace    *jcstypes.UserSpaceDetail
	AccessAmount float64
}

func (j *ChangeRedundancy) changeOne(ctx *changeRedundancyContext, pkg jcstypes.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 := jcstypes.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

		reen := ctx.ticktock.pubLock.BeginReentrant()

		var allUpdatings []db.UpdatingObjectRedundancy
		var allSysEvts []datamap.SysEventBody

		ctx.mostBlockStgIDs = j.summaryRepObjectBlockUserSpaces(ctx, objs, 2)

		var willShrinks []jcstypes.ObjectDetail

		for _, obj := range objs {
			newRed, selectedSpaces := j.chooseRedundancy(ctx, obj)
			// 冗余策略不需要调整，就检查是否需要收缩
			if newRed == nil {
				willShrinks = append(willShrinks, obj)
				continue
			}

			for _, space := range selectedSpaces {
				reen.UserSpace().Buzy(space.UserSpace.UserSpace.UserSpaceID)
			}
			err := reen.Lock()
			if err != nil {
				log.WithField("ObjectID", obj.Object.ObjectID).Warnf("acquire lock: %s", err.Error())
				continue
			}

			updating, evt, err := j.doChangeRedundancy(ctx, obj, newRed, selectedSpaces)
			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())
				continue
			}
		}

		udpatings, sysEvts, err := j.doRedundancyShrink(ctx, pkg, willShrinks, reen)
		if err != nil {
			log.Warnf("redundancy shrink: %s", err.Error())
		} else {
			allUpdatings = append(allUpdatings, udpatings...)
			allSysEvts = append(allSysEvts, sysEvts...)
		}

		if len(allUpdatings) > 0 {
			err := db.DoTx10(db2, db2.Object().BatchUpdateRedundancy, allUpdatings)
			if err != nil {
				reen.Unlock()
				log.Warnf("update object redundancy: %s", err.Error())
				return err
			}
		}
		reen.Unlock()

		for _, e := range allSysEvts {
			ctx.ticktock.evtPub.Publish(e)
		}
	}

	return nil
}