datasync

.gitignore

@@ -0,0 +1,40 @@
+# Compiled Object files, Static and Dynamic libs (Shared Objects)
+*.o
+*.a
+*.so
+
+# Folders
+_obj
+_test
+.idea
+.vscode
+.github
+data
+
+# Architecture specific extensions/prefixes
+*.[568vq]
+[568vq].out
+
+*.cgo1.go
+*.cgo2.c
+_cgo_defun.c
+_cgo_gotypes.go
+_cgo_export.*
+
+_testmain.go
+
+test/*
+!test/*.*
+!test/*/
+
+test/*/*
+!test/*/*.*
+!test/*/*/
+
+*.exe
+*.test
+*.prof
+*.iml
+*.class
+*.tmp
+*.zip

datasync/build.sh

@@ -0,0 +1,9 @@
+
+
+installpath=`go env GOPATH | awk -F ':' '{print $1}'`/bin
+
+GOOS=linux go install
+
+cp datasync.conf ${installpath}/linux_amd64
+cp datasync.sh ${installpath}/linux_amd64
+

datasync/datasync.conf

@@ -0,0 +1,68 @@
+
+[datasync]
+from.odbserver=127.0.0.1:11001
+from.keyspace=matrix
+
+to.odbserver=127.0.0.1:11001
+to.keyspace=oktest 
+
+# 可以配置多个 类名 或 查询语句，查询结果中必须包含 class id 字段
+# 查询语句中可以指定 where 条件，不支持聚合、排序、limit 等子句
+# 不配置将读取所有类数据
+from.data=/test/
+# from.data=/test/bucketpromdb
+from.data=select * from /test/alert_status where vtime>'2022-03-10 16:53:43'
+
+# 不同步的类
+deny.class=/matrix/jobs/
+deny.class=/matrix/ldap
+deny.class=/matrix/group
+deny.class=/matrix/perms/
+deny.class=/matrix/filesystem
+deny.class=/matrix/portal/
+deny.class=/matrix/
+deny.class=/m3event/
+deny.class=/m3entity/
+deny.class=/testdata1
+deny.class=/testclasscache
+deny.class=/test_odbinsert
+deny.class=/testtagdir
+deny.class=/testdc1/
+deny.class=/testdc2/
+deny.class=/testruledata
+deny.class=/aywl/
+
+
+# 只同步指定的最近一段时间的对象数据，默认 365d 
+data.time.since=3000d
+# 只同步指定的最近一段时间的Bucket数据，默认 30d 
+bucket.time.since=30d
+
+# 指定建类语句等初始化mql语句，可以是mql文件或mql语句，默认根据原始元数据信息自动建类
+#to.init.mql= 
+
+# 类映射
+# mapping.class.fromclass=toclass
+
+# overwrite / ignore / error
+# exists=overwrite
+
+# 每秒写入数据量，同步上限
+# limit.rate=50
+
+# 对象数据同步并发数，默认cpu数
+concur.threads=2
+# Bucket数据同步并发数，默认cpu数
+bucket.threads=2
+# 每次读取数据的最大数量，默认50，立即生效
+pagesize=10
+# 每次读取数据的时间分段，默认一天，立即生效
+pagetime=1d
+# 每次读取bucket数据的时间分段，默认1h，立即生效
+bucket.pagetime=1d
+# 用于存放同步进度记录
+data.dir=data
+# 持续运行间隔，数据同步到当前时间后，间隔指定时间轮询执行
+# 默认 0 运行一次，数据同步到当前时间后退出
+# 立即生效
+run.interval=5s

datasync/datasync.sh

@@ -0,0 +1,56 @@
+
+export LANG=zh_CN.utf8
+export LC_ALL=zh_CN.utf8
+
+#改变工作目录到当前脚本所在路径
+if [[ "$0" =~ / ]]; then cd "${0%/*}"; fi
+
+export CWD=`pwd`
+
+if [[ "$MATRIXROOT" == "" ]]; then
+    export MATRIXROOT="/opt/matrix"
+fi
+
+export STDERROR=$MATRIXROOT/var/logs/datasync/out.log
+
+printusage(){
+    echo "type '${0} stop' to stop it"
+    echo "  or '${0} restart' to restart it"
+    echo "  or '${0} status' to check status"
+    echo "cat $STDERROR for detail information"
+}
+
+#检查正在运行的实例
+pid=`ps -ef | grep "$CWD/datasync" | grep -v grep | awk '{print $2}'`
+if [[ "$1" == "stop" || "$1" == "restart" ]]; then
+    if [[ "${pid}" != "" ]]; then
+        echo "stop last running datasync, pid=${pid}"
+        kill -9 ${pid}
+        while [[ "${pid}" != "" ]]; do
+            sleep 1
+            pid=`ps -ef | grep "$CWD/datasync" | grep -v grep | awk '{print $2}'`
+        done
+    fi
+    if [[ "$1" == "stop" ]]; then
+        exit 0
+    fi
+elif [[ "$1" == "status" ]]; then
+    if [[ "${pid}" == "" ]]; then
+        echo "datasync is shutdown"
+    else
+        echo "datasync is running, pid=${pid}"
+    fi
+    exit 0
+elif [[ "$1" == "help" ]]; then
+    printusage
+    exit 0
+elif [[ "${pid}" != "" ]]; then
+    echo "datasync is running, pid=${pid}"
+    printusage
+    exit 0
+fi
+
+#开始启动
+nohup $CWD/datasync "$@" 1>/dev/null 2>${STDERROR} & 
+echo "datasync is started"
+printusage

datasync/datasync/datasync.go

@@ -0,0 +1,618 @@
+package datasync
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"regexp"
+	"runtime"
+	"strings"
+	"sync"
+	"time"
+
+	"git.wecise.com/wecise/odb-go/dbo"
+	"git.wecise.com/wecise/odb-go/odb"
+	"git.wecise.com/wecise/odb-go/odbc"
+	"github.com/scylladb/go-set/strset"
+	"github.com/spf13/cast"
+	"github.com/wecisecode/util/merrs"
+	"github.com/wecisecode/util/mfmt"
+	"github.com/wecisecode/util/rc"
+)
+
+var mcfg = odbc.Config
+var logger = odbc.Logger
+
+type DataSync struct {
+	odbcFrom        odb.Client
+	odbcTo          odb.Client
+	schemaFrom      *dbo.Schema
+	schemaTo        *dbo.Schema
+	fromodbserver   string
+	fromkeyspace    string
+	fromdc          string
+	toodbserver     string
+	tokeyspace      string
+	todc            string
+	fromdata        []string
+	classmapping    map[string]string
+	datatimesince   time.Duration
+	buckettimesince time.Duration
+	ctx             context.Context
+	cancel          context.CancelFunc
+	wg              *sync.WaitGroup
+	mutex           sync.Mutex
+	errs            []error
+	ctrlrc          *rc.RoutinesController
+	objectrc        *rc.RoutinesController
+	bucketrc        *rc.RoutinesController
+	syncstatus      *SyncStatus
+}
+
+func NewDataSync() *DataSync {
+	return &DataSync{}
+}
+
+func (ds *DataSync) Init() (err error) {
+	ds.fromodbserver = mcfg.GetString("datasync.from.odbserver")
+	ds.fromkeyspace = mcfg.GetString("datasync.from.keyspace")
+	ds.toodbserver = mcfg.GetString("datasync.to.odbserver")
+	ds.tokeyspace = mcfg.GetString("datasync.to.keyspace")
+	if ds.fromodbserver == "" ||
+		ds.fromkeyspace == "" ||
+		ds.toodbserver == "" ||
+		ds.tokeyspace == "" {
+		return odbc.NoConfError.New("need configure settings: datasync.from.odbserver, datasync.from.keyspace, datasync.to.odbserver, datasync.to.keyspace")
+	}
+	ds.fromdc = mcfg.GetString("datasync.from.dc")
+	ds.todc = mcfg.GetString("datasync.to.dc")
+	ds.fromdata = mcfg.GetStrings("datasync.from.data")
+	ds.classmapping = mcfg.GetMapping("datasync.mapping.class")
+	ds.odbcFrom, err = odb.NewClient(&odb.Config{
+		Keyspace: ds.fromkeyspace,
+		Hosts:    strings.Split(ds.fromodbserver, ","),
+	})
+	if err != nil {
+		if strings.Contains(err.Error(), "error: EOF") {
+			println("\n!!!should add your ip to odbserver(" + ds.fromodbserver + ") whitelist!!!\n")
+			os.Exit(1)
+		}
+		return merrs.New(err)
+	}
+	ds.odbcTo, err = odb.NewClient(&odb.Config{
+		Keyspace: ds.tokeyspace,
+		Hosts:    strings.Split(ds.toodbserver, ","),
+	})
+	if err != nil {
+		if strings.Contains(err.Error(), "error: EOF") {
+			println("\n!!!should add your ip to odbserver(" + ds.toodbserver + ") whitelist!!!\n")
+			os.Exit(1)
+		}
+		return merrs.New(err)
+	}
+	ds.schemaFrom = dbo.NewSchema(ds.odbcFrom)
+	ds.schemaTo = dbo.NewSchema(ds.odbcTo)
+	ctrlthreads := mcfg.GetInt("datasync.ctrl.threads", runtime.GOMAXPROCS(0))
+	ds.ctrlrc = rc.NewRoutinesControllerLimit("", ctrlthreads, ctrlthreads*2)
+	concurthreads := mcfg.GetInt("datasync.concur.threads", runtime.GOMAXPROCS(0))
+	ds.objectrc = rc.NewRoutinesControllerLimit("", concurthreads, concurthreads*2)
+	bucketthreads := mcfg.GetInt("datasync.bucket.threads", runtime.GOMAXPROCS(0))
+	ds.bucketrc = rc.NewRoutinesControllerLimit("", bucketthreads, bucketthreads*2)
+	ds.datatimesince = mcfg.GetDuration("datasync.data.time.since", "365d")
+	ds.buckettimesince = mcfg.GetDuration("datasync.bucket.time.since", "30d")
+	return nil
+}
+
+func (ds *DataSync) Run() (done <-chan error) {
+	ret := make(chan error, 1)
+	key := regexp.MustCompile(`\W`).ReplaceAllString(strings.Split(ds.fromodbserver, ",")[0]+"_"+ds.fromkeyspace+"_"+strings.Split(ds.toodbserver, ",")[0]+"_"+ds.tokeyspace, "_")
+	ds.syncstatus = NewSyncStatus(key)
+	if !mcfg.GetBool("reload") {
+		ds.syncstatus.Load()
+	}
+	go ds.run(ret)
+	return ret
+}
+
+func (ds *DataSync) run(ret chan error) {
+	fromdatas := []string{}
+	for _, fromdata := range ds.fromdata {
+		fromdata = strings.TrimSpace(fromdata)
+		if len(fromdata) > 0 {
+			fromdatas = append(fromdatas, fromdata)
+		}
+	}
+	if len(fromdatas) == 0 {
+		cis, e := ds.odbcFrom.ClassInfo("/", true)
+		if e != nil {
+			ret <- e
+			return
+		}
+		for _, ci := range cis {
+			fromdatas = append(fromdatas, ci.Fullname)
+		}
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ds.ctx = ctx
+	ds.cancel = cancel
+	for {
+		ds.wg = &sync.WaitGroup{}
+		ds.syncstatus.Resume()
+		for _, fromdata := range fromdatas {
+			mqlfrom := fromdata
+			ds.startsyncproc(ds.wg, ds.ctrlrc, func() error {
+				return ds.syncdata(mqlfrom)
+			})
+		}
+		ds.wg.Wait()
+		ds.syncstatus.WaitSaveDone()
+		if len(ds.errs) > 0 {
+			ret <- merrs.New(ds.errs)
+			return
+		}
+		logger.Info("total sync data", ds.syncstatus.TotalChunks(), "chunks", ds.syncstatus.TotalRecords(), "records,", "use time:", mfmt.FormatDuration(ds.syncstatus.TotalUseTime()))
+		interval := mcfg.GetDuration("datasync.run.interval", 0)
+		if interval > 0 {
+			time.Sleep(interval)
+		} else {
+			break
+		}
+	}
+	ret <- nil
+}
+
+func (ds *DataSync) startsyncproc(wg *sync.WaitGroup, rc *rc.RoutinesController, proc func() error) {
+	wg.Add(1)
+	e := rc.ConcurCall(1, func() {
+		defer wg.Done()
+		if ds.ctx.Err() == nil {
+			e := proc()
+			if e != nil {
+				if !merrs.ContextError.Contains(e) {
+					ds.mutex.Lock()
+					ds.errs = append(ds.errs, e)
+					ds.mutex.Unlock()
+				}
+				ds.cancel()
+			}
+		}
+	})
+	if e != nil {
+		ds.mutex.Lock()
+		ds.errs = append(ds.errs, e)
+		ds.mutex.Unlock()
+		wg.Done()
+	}
+}
+
+// 同步一块数据
+// mqlfrom 可以是类名 或 查询语句
+func (ds *DataSync) syncdata(mqlfrom string) error {
+	// 同一格式化为查询语句
+	mqlfrom = FormatMQL(mqlfrom)
+	// 已完成同步进度
+	fromclass, fields, condition, e := ds.LastSyncProgress(mqlfrom)
+	if e != nil {
+		return e
+	}
+	cifroms, e := ds.schemaFrom.LoadClassinfos(fromclass)
+	if e != nil {
+		return e
+	}
+	for _, v := range cifroms {
+		cifrom := v
+		ds.startsyncproc(ds.wg, ds.objectrc, func() error {
+			return ds.syncclassdata(cifrom, fields, condition)
+		})
+	}
+	return nil
+}
+
+var reselect = regexp.MustCompile(`(?is)select\s.*`)
+var reselectfromclass = regexp.MustCompile(`(?is)select\s+(.*)\s+from\s+(\S+)(?:\s+(.*)\s*)?`)
+var commentexprs = regexp.MustCompile(`(?s)\/\*(?:[^\*]|\*+[^\*\/])*\*+\/`)
+var commentexprs_2 = regexp.MustCompile(`(?ms)(?:^|\n)\-\-[^\n]*(?:\n|$)`)
+var commentexprs_3 = regexp.MustCompile(`(?ms)(?:^|\n)//[^\n]*(?:\n|$)`)
+
+func FormatMQL(mql string) string {
+	mql = commentexprs.ReplaceAllString(mql, "")
+	mql = commentexprs_2.ReplaceAllString(mql, "")
+	mql = commentexprs_3.ReplaceAllString(mql, "")
+	mql = strings.TrimSpace(mql)
+	if !reselect.MatchString(mql) {
+		mql = "select * from " + mql
+	}
+	return mql
+}
+
+func (ds *DataSync) LastSyncProgress(mql string) (class string, fields, condition string, err error) {
+	selstmt := reselectfromclass.FindStringSubmatch(mql)
+	if len(selstmt) < 3 {
+		return "", "", "", merrs.New("from.data select statement error", []string{"mql", mql})
+	}
+	fields = selstmt[1]
+	class = selstmt[2]
+	condition = selstmt[3]
+	if class == "" {
+		return "", "", "", merrs.New("from.data select statement error", []string{"mql", mql})
+	}
+	return class, fields, condition, nil
+}
+
+var rewhere = regexp.MustCompile(`(?is)\swhere\s(.*)`)
+
+func mqlAddVtimeRange(mql, sbeginvtime, sendvtime string) string {
+	mqlseg := mql
+	segwhere := fmt.Sprint(" where vtime>='", sbeginvtime, "' and vtime<'", sendvtime, "'")
+	if rewhere.MatchString(mql) {
+		mqlseg = rewhere.ReplaceAllString(mqlseg, segwhere+" and ($1)")
+	} else {
+		mqlseg = mqlseg + segwhere
+	}
+	return mqlseg
+}
+
+func (ds *DataSync) syncclassdata(cifrom *dbo.ClassInfoHelper, fields, condition string) error {
+	denyclass := strset.New(mcfg.GetStrings("datasync.deny.class")...)
+	if denyclass.Has(cifrom.Fullname) || denyclass.Has(cifrom.Fullname+"/") {
+		return nil
+	}
+	checkdenyclass := cifrom
+	for checkdenyclass != nil {
+		if denyclass.Has(checkdenyclass.BaseClassFullname() + "/") {
+			return nil
+		}
+		checkdenyclass = ds.schemaFrom.GetClassInfo(checkdenyclass.BaseClassFullname())
+	}
+	// 确定目标类已创建
+	toclass := ds.classmapping[cifrom.Fullname]
+	if toclass == "" {
+		toclass = cifrom.Fullname
+	}
+	cito, e := ds.assureToClass(toclass, cifrom)
+	if e != nil {
+		return e
+	}
+	//
+	mqlfrom := "select " + fields + " from " + cifrom.Fullname
+	if condition != "" {
+		mqlfrom += " " + condition
+	}
+	dc := ds.syncstatus.DoneCount(mqlfrom)
+	isrunning := <-dc.isrunning
+	if isrunning {
+		dc.isrunning <- isrunning
+		return nil
+	}
+	dc.isrunning <- true
+	defer func() {
+		<-dc.isrunning
+		dc.isrunning <- false
+	}()
+	recordscount := dc.RecordsCount
+	sfromvtime := dc.FromVtime
+	slastdatavtime := dc.LastDataVtime
+	slastsyncvtime := dc.LastSyncVtime
+	// 分段获取数据
+	fromvtime, _ := time.Parse("2006-01-02 15:04:05", sfromvtime)
+	lastdatavtime, _ := time.Parse("2006-01-02 15:04:05.000000", slastdatavtime)
+	lastsyncvtime, _ := time.Parse("2006-01-02 15:04:05", slastsyncvtime)
+	sincevtime := time.Now().Add(-ds.datatimesince)
+	if fromvtime.Before(sincevtime) || lastsyncvtime.Before(fromvtime) || lastdatavtime.Before(fromvtime) {
+		ssincevtime := sincevtime.Format("2006-01-02 00:00:00")
+		firstdatavtime := time.Now()
+		sfirstdatavtime := firstdatavtime.Format("2006-01-02 15:04:05.000000")
+		for i := 0; ; i++ {
+			mqlseg := mqlAddVtimeRange(mqlfrom, ssincevtime, sfirstdatavtime)
+			mqlchunk := mqlseg + fmt.Sprint(" order by vtime limit 1")
+			logger.Debug("check first data vtime:", ds.odbcFrom.Config().Keyspace, mqlchunk)
+			// 读取源数据
+			r, e := ds.odbcFrom.Query(mqlchunk).WithContext(ds.ctx).Do()
+			if e != nil {
+				return e
+			}
+			if len(r.Data) == 0 {
+				// 没有更多数据
+				if i == 0 {
+					logger.Info("check first data vtime:", ds.odbcFrom.Config().Keyspace, cifrom.Fullname, "no data")
+					ds.syncstatus.RemoveDoneCount(mqlfrom)
+					return nil
+				}
+				logger.Info("check first data vtime:", ds.odbcFrom.Config().Keyspace, cifrom.Fullname, sfirstdatavtime, mqlfrom)
+				break
+			}
+			firstdata := r.Data[0]
+			firstdatavtime = firstdata["vtime"].(time.Time)
+			sfirstdatavtime = firstdatavtime.Format("2006-01-02 15:04:05.000000")
+			logger.Debug("check first data vtime:", ds.odbcFrom.Config().Keyspace, firstdata["class"], firstdata["id"], sfirstdatavtime)
+		}
+		fromvtime = firstdatavtime
+		lastdatavtime = firstdatavtime
+		lastsyncvtime = firstdatavtime
+		sfromvtime = fromvtime.Format("2006-01-02 15:04:05")
+		slastdatavtime = lastdatavtime.Format("2006-01-02 15:04:05.000000")
+		slastsyncvtime = lastsyncvtime.Format("2006-01-02 15:04:05")
+		recordscount = 0
+		// 初始化DataCount进度信息
+		dc.RecordsCount = recordscount
+		dc.FromVtime = sfromvtime
+		dc.LastDataVtime = slastdatavtime
+		dc.LastSyncVtime = slastsyncvtime
+	}
+	// 继续执行相关bucket数据同步
+	e = ds.syncbucketdatacontinue(cifrom, cito, mqlfrom)
+	if e != nil {
+		return e
+	}
+	// 继续执行对象数据同步
+	nextvtime := lastsyncvtime
+	for {
+		nextvtime = lastsyncvtime.Add(mcfg.GetDuration("datasync.pagetime", "1d"))
+		if time.Now().Before(nextvtime) {
+			break
+		}
+		snextvtime := nextvtime.Format("2006-01-02 15:04:05")
+		mqlseg := mqlAddVtimeRange(mqlfrom, slastsyncvtime, snextvtime)
+		offset := 0
+		for {
+			mqlchunk := mqlseg + fmt.Sprint(" limit ", offset, ",", mcfg.GetInt("datasync.pagesize", 50))
+			logger.Debug(mqlchunk)
+			// 读取源数据
+			r, e := ds.odbcFrom.Query(mqlchunk).WithContext(ds.ctx).Do()
+			if e != nil {
+				return e
+			}
+			if len(r.Data) == 0 {
+				// 没有更多数据
+				break
+			}
+			// 写入目标数据
+			for _, data := range r.Data {
+				e = ds.insertData(mqlfrom, cifrom, cito, data)
+				if e != nil {
+					return e
+				}
+				datavtime := data["vtime"].(time.Time)
+				if lastdatavtime.Before(datavtime) {
+					lastdatavtime = datavtime
+				}
+			}
+			offset += len(r.Data)
+		}
+		lastsyncvtime = nextvtime
+		slastsyncvtime = lastsyncvtime.Format("2006-01-02 15:04:05")
+		recordscount += int64(offset)
+		//
+		dc.RecordsCount = recordscount
+		dc.LastDataVtime = lastdatavtime.Format("2006-01-02 15:04:05.000000")
+		dc.LastSyncVtime = slastsyncvtime
+		ds.syncstatus.Save(mqlfrom, dc)
+	}
+	return nil
+}
+
+func (ds *DataSync) insertData(mqlfrom string, cifrom, cito *dbo.ClassInfoHelper, data map[string]any) error {
+	logger.Debug(data["class"], data["id"], data["vtime"])
+	vals := []any{}
+	for _, fn := range cito.Fieldslist {
+		if cito.Fieldinfos[fn].Fieldtype == "bucket" {
+			vals = append(vals, nil)
+			continue
+		}
+		v := data[fn]
+		if v == nil {
+			i := strings.Index(fn, ":")
+			if i >= 0 {
+				fn = fn[i+1:]
+			}
+			v = data[fn]
+		}
+		vals = append(vals, v)
+	}
+	_, e := ds.odbcTo.Query(cito.Insertmql, vals...).WithContext(ds.ctx).Do()
+	if e != nil {
+		return merrs.New(e, []string{"mql", cito.Insertmql, "vals", fmt.Sprint(data)})
+	}
+	e = ds.syncbucketdatanew(cifrom, cito, mqlfrom, data)
+	if e != nil {
+		return e
+	}
+	return nil
+}
+
+func (ds *DataSync) syncbucketdatanew(cifrom, cito *dbo.ClassInfoHelper, mqlfrom string, data map[string]any) error {
+	if ds.buckettimesince == 0 {
+		return nil
+	}
+	for _, bf := range cifrom.BucketFields {
+		if _, has := data[bf]; has {
+			ds.startsyncproc(ds.wg, ds.bucketrc, func() error {
+				oid := data["id"].(string)
+				buckettype := cast.ToString(cifrom.Fieldinfos[bf].Fieldoption["type"])
+				key := buckettype + ":" + cifrom.Fullname + ":" + bf + "[" + oid + "]"
+				dc := ds.syncstatus.DoneCount(key)
+				dc.BucketClass = cifrom.Fullname
+				dc.BucketField = bf
+				dc.BucketObjID = oid
+				return ds.syncbucketdata(mqlfrom, cifrom, cito, key, dc)
+			})
+		}
+	}
+	return nil
+}
+
+func (ds *DataSync) syncbucketdatacontinue(cifrom, cito *dbo.ClassInfoHelper, mqlfrom string) error {
+	if ds.buckettimesince == 0 {
+		return nil
+	}
+
+	mqlchunk := mqlfrom + fmt.Sprint(" limit 1")
+	logger.Debug("check data fields:", ds.odbcFrom.Config().Keyspace, mqlchunk)
+	// 读取源数据
+	r, e := ds.odbcFrom.Query(mqlchunk).WithContext(ds.ctx).Do()
+	if e != nil {
+		return e
+	}
+	if len(r.Data) == 0 {
+		return nil
+	}
+	data := r.Data[0]
+	for _, bf := range cifrom.BucketFields {
+		if _, has := data[bf]; has {
+			e := func() error {
+				bucketdonecount := ds.syncstatus.DoneCountCopy(func(k string, v *DoneCount) bool {
+					return v.BucketClass == cifrom.Fullname && v.BucketField == bf
+				})
+				for k, v := range bucketdonecount {
+					key := k
+					dc := v
+					ds.startsyncproc(ds.wg, ds.bucketrc, func() error {
+						return ds.syncbucketdata(mqlfrom, cifrom, cito, key, dc)
+					})
+				}
+				return nil
+			}()
+			if e != nil {
+				return e
+			}
+		}
+	}
+	return nil
+}
+
+func (ds *DataSync) syncbucketdata(mqlfrom string, cifrom, cito *dbo.ClassInfoHelper, dckey string, dc *DoneCount) error {
+	isrunning := <-dc.isrunning
+	if isrunning {
+		dc.isrunning <- isrunning
+		return nil
+	}
+	dc.isrunning <- true
+	defer func() {
+		<-dc.isrunning
+		dc.isrunning <- false
+	}()
+
+	bucketType := cast.ToString(cifrom.Fieldinfos[dc.BucketField].Fieldoption["type"])
+
+	logger.Debug("to sync", bucketType, "data", dc.BucketClass, dc.BucketField, "id:", dc.BucketObjID, mqlfrom)
+
+	recordscount := dc.RecordsCount
+	sfromvtime := dc.FromVtime
+	slastdatavtime := dc.LastDataVtime
+	slastsyncvtime := dc.LastSyncVtime
+	// 分段获取数据
+	fromvtime, _ := time.Parse("2006-01-02 15:04:05", sfromvtime)
+	lastdatavtime, _ := time.Parse("2006-01-02 15:04:05.000000", slastdatavtime)
+	lastsyncvtime, _ := time.Parse("2006-01-02 15:04:05", slastsyncvtime)
+	sincevtime := time.Now().Add(-ds.buckettimesince)
+	if fromvtime.Before(sincevtime) || lastsyncvtime.Before(fromvtime) || lastdatavtime.Before(fromvtime) {
+		fromvtime = sincevtime
+		lastdatavtime = sincevtime
+		lastsyncvtime = sincevtime
+		sfromvtime = fromvtime.Format("2006-01-02 15:04:05")
+		slastdatavtime = lastdatavtime.Format("2006-01-02 15:04:05.000000")
+		slastsyncvtime = lastsyncvtime.Format("2006-01-02 15:04:05")
+		recordscount = 0
+		// 初始化DataCount进度信息
+		dc.RecordsCount = recordscount
+		dc.FromVtime = sfromvtime
+		dc.LastDataVtime = slastdatavtime
+		dc.LastSyncVtime = slastsyncvtime
+	}
+	// 继续执行数据同步
+	count := 0
+	nextvtime := lastsyncvtime
+	for {
+		nextvtime = lastsyncvtime.Add(mcfg.GetDuration("datasync.bucket.pagetime", "1h"))
+		if time.Now().Before(nextvtime) {
+			break
+		}
+		snextvtime := nextvtime.Format("2006-01-02 15:04:05")
+		offset := 0
+		{
+			mqlchunk := "select " + dc.BucketField + ".time('" + slastsyncvtime + "','" + snextvtime + "')" + " from " + dc.BucketClass + " where id=?"
+			logger.Debug(mqlchunk, dc.BucketObjID)
+			// 读取源数据
+			r, e := ds.odbcFrom.Query(mqlchunk, dc.BucketObjID).WithContext(ds.ctx).Do()
+			if e != nil {
+				return e
+			}
+			if len(r.Data) == 0 {
+				return merrs.New("bucket host data not found id="+dc.BucketObjID, merrs.Map{"mql": mqlchunk})
+			}
+			idata := r.Data[0][dc.BucketField]
+			if idata != nil {
+				data := cast.ToSlice(idata)
+				ms := []map[string]any{}
+				for i := 0; i < len(data); i++ {
+					dat := cast.ToSlice(data[i])
+					if len(dat) >= 3 {
+						m := map[string]any{}
+						m["timestamp"] = dat[0]
+						m["name"] = dat[1]
+						m["value"] = dat[2]
+						if len(dat) >= 4 {
+							tm := cast.ToStringMap(dat[3])
+							for k, v := range tm {
+								m[k] = v
+							}
+						}
+						ms = append(ms, m)
+					}
+				}
+				// 写入目标数据
+				mqlinsert := "insert into " + dc.BucketClass + "(id," + dc.BucketField + ") values(?,?)"
+				_, e = ds.odbcTo.Query(mqlinsert, dc.BucketObjID, ms).WithContext(ds.ctx).Do()
+				if e != nil {
+					return e
+				}
+				lastdatavtime = nextvtime
+				offset += len(data)
+			}
+		}
+		lastsyncvtime = nextvtime
+		slastsyncvtime = lastsyncvtime.Format("2006-01-02 15:04:05")
+		recordscount += int64(offset)
+		count += offset
+		//
+		dc.RecordsCount = recordscount
+		dc.LastDataVtime = lastdatavtime.Format("2006-01-02 15:04:05.000000")
+		dc.LastSyncVtime = slastsyncvtime
+		ds.syncstatus.Save(dckey, dc)
+	}
+	logger.Debug("end sync", bucketType, "data", dc.BucketClass, dc.BucketField, "id:", dc.BucketObjID)
+
+	if count > 0 {
+		logger.Info("sync", bucketType, "data", dc.BucketClass, dc.BucketField, "id:", dc.BucketObjID, count, "records", "to time", dc.LastSyncVtime)
+	}
+	return nil
+}
+
+func (ds *DataSync) assureToClass(toclass string, cifrom *dbo.ClassInfoHelper) (cito *dbo.ClassInfoHelper, err error) {
+	if toclass != cifrom.Classfullname {
+		return nil, merrs.New("not support class mapping", []string{"toclass", toclass, "fromclass", cifrom.Classfullname})
+	}
+	cis, e := ds.odbcTo.ClassInfo(toclass, false)
+	if e != nil && !merrs.NotExistError.Contains(e) && !strings.Contains(e.Error(), "not exists") {
+		return nil, merrs.New(e)
+	}
+	if len(cis) == 0 {
+		_, e = ds.odbcTo.Query(cifrom.DDL).WithContext(ds.ctx).Do()
+		if e != nil {
+			return nil, merrs.New(e)
+		}
+		cis, e = ds.odbcTo.ClassInfo(toclass, false)
+		if e != nil {
+			return nil, merrs.New(e)
+		}
+		if len(cis) == 0 {
+			return nil, merrs.New("len(cis) == 0")
+		}
+	}
+	cito, e = ds.schemaTo.NewClassinfo(cis[0])
+	if e != nil {
+		return nil, merrs.New(e)
+	}
+	return cito, nil
+}

datasync/datasync/syncstatus.go

@@ -0,0 +1,192 @@
+package datasync
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/wecisecode/util/mio"
+	"github.com/wecisecode/util/rc"
+)
+
+type DoneCount struct {
+	BucketClass   string    `json:",omitempty"`
+	BucketField   string    `json:",omitempty"`
+	BucketObjID   string    `json:",omitempty"`
+	RecordsCount  int64     `json:",omitempty"`
+	FromVtime     string    `json:",omitempty"`
+	LastDataVtime string    `json:",omitempty"`
+	LastSyncVtime string    `json:",omitempty"`
+	RetryCount    int64     `json:",omitempty"`
+	lastlogtime   time.Time `json:"-"`
+	isrunning     chan bool `json:"-"`
+}
+
+type CountStatus struct {
+	TotalUseTime time.Duration         `json:",omitempty"`
+	DoneCount    map[string]*DoneCount `json:",omitempty"`
+	mutex        sync.RWMutex          `json:"-"`
+}
+
+type SyncStatus struct {
+	filepath string
+	//
+	starttime   time.Time
+	countstatus *CountStatus
+	//
+	rc           *rc.RoutinesController
+	lasterror    error
+	lastsavetime time.Time
+	waitdone     chan any
+}
+
+func NewSyncStatus(key string) *SyncStatus {
+	datadir := mcfg.GetString("datasync.data.dir", "/opt/matrix/var/datasync")
+	statusfile := mcfg.GetString("datasync.statusfile", key+".status.txt")
+	if !strings.HasPrefix(statusfile, "/") {
+		statusfile = filepath.Join(datadir, statusfile)
+	}
+	return &SyncStatus{
+		filepath: statusfile,
+		countstatus: &CountStatus{
+			DoneCount: map[string]*DoneCount{},
+		},
+		rc:       rc.NewRoutinesController("", 1),
+		waitdone: make(chan any, 1),
+	}
+}
+
+func (syncstatus *SyncStatus) Resume() {
+	syncstatus.countstatus.mutex.Lock()
+	defer syncstatus.countstatus.mutex.Unlock()
+	syncstatus.starttime = time.Now().Add(-syncstatus.countstatus.TotalUseTime)
+}
+
+func (syncstatus *SyncStatus) TotalUseTime() time.Duration {
+	syncstatus.countstatus.mutex.RLock()
+	defer syncstatus.countstatus.mutex.RUnlock()
+	return syncstatus.countstatus.TotalUseTime
+}
+
+func (syncstatus *SyncStatus) TotalChunks() int {
+	syncstatus.countstatus.mutex.RLock()
+	defer syncstatus.countstatus.mutex.RUnlock()
+	return len(syncstatus.countstatus.DoneCount)
+}
+
+func (syncstatus *SyncStatus) TotalRecords() int {
+	syncstatus.countstatus.mutex.RLock()
+	defer syncstatus.countstatus.mutex.RUnlock()
+	n := 0
+	for _, v := range syncstatus.countstatus.DoneCount {
+		n += int(v.RecordsCount)
+	}
+	return n
+}
+
+func (syncstatus *SyncStatus) DoneCountCopy(filter func(k string, v *DoneCount) bool) map[string]*DoneCount {
+	syncstatus.countstatus.mutex.RLock()
+	defer syncstatus.countstatus.mutex.RUnlock()
+	m := map[string]*DoneCount{}
+	for k, v := range syncstatus.countstatus.DoneCount {
+		if filter(k, v) {
+			m[k] = v
+		}
+	}
+	return m
+}
+
+func (syncstatus *SyncStatus) DoneCount(key string) *DoneCount {
+	syncstatus.countstatus.mutex.Lock()
+	defer syncstatus.countstatus.mutex.Unlock()
+	dc := syncstatus.countstatus.DoneCount[key]
+	if dc == nil {
+		dc = &DoneCount{}
+		syncstatus.countstatus.DoneCount[key] = dc
+	}
+	if dc.isrunning == nil {
+		dc.isrunning = make(chan bool, 1)
+		dc.isrunning <- false
+	}
+	return dc
+}
+
+func (syncstatus *SyncStatus) RemoveDoneCount(key string) {
+	syncstatus.countstatus.mutex.Lock()
+	defer syncstatus.countstatus.mutex.Unlock()
+	delete(syncstatus.countstatus.DoneCount, key)
+}
+
+func (syncstatus *SyncStatus) Load() error {
+	syncstatus.countstatus.mutex.Lock()
+	defer syncstatus.countstatus.mutex.Unlock()
+	logger.Info("load progress from", syncstatus.filepath)
+	syncstatusbs, e := mio.ReadFile(syncstatus.filepath)
+	if e != nil && !os.IsNotExist(e) {
+		return e
+	}
+	if len(syncstatusbs) > 0 {
+		e = json.Unmarshal(syncstatusbs, syncstatus.countstatus)
+		if e != nil {
+			logger.Warn(e)
+		}
+	}
+	for _, dc := range syncstatus.countstatus.DoneCount {
+		dc.isrunning = make(chan bool, 1)
+		dc.isrunning <- false
+	}
+	return nil
+}
+
+func (syncstatus *SyncStatus) WaitSaveDone() {
+	syncstatus.waitdone <- 1
+	syncstatus.rc.WaitDone()
+	<-syncstatus.waitdone
+}
+
+func (syncstatus *SyncStatus) Save(key string, dc *DoneCount) (err error) {
+	syncstatus.countstatus.mutex.Lock()
+	syncstatus.countstatus.DoneCount[key] = dc
+	syncstatus.countstatus.TotalUseTime = time.Since(syncstatus.starttime)
+	outputloginfo := false
+	if time.Since(dc.lastlogtime) > 5*time.Second {
+		dc.lastlogtime = time.Now()
+		outputloginfo = true
+	}
+	syncstatus.countstatus.mutex.Unlock()
+	if outputloginfo {
+		logger.Info("sync data:", key, "vtime:", dc.FromVtime, "~", dc.LastSyncVtime, "records:", dc.RecordsCount, "lastvtime:", dc.LastDataVtime)
+	}
+	syncstatus.rc.CallLast2Only(func() {
+		if !syncstatus.lastsavetime.Equal(time.Time{}) {
+			interval := 10 * time.Second
+			realinterval := time.Since(syncstatus.lastsavetime)
+			if realinterval < interval {
+				t := time.NewTimer(interval - realinterval)
+				select {
+				case <-t.C:
+				case v := <-syncstatus.waitdone:
+					syncstatus.waitdone <- v
+				}
+			}
+		}
+		syncstatus.countstatus.mutex.RLock()
+		syncstatusbs, e := json.MarshalIndent(syncstatus.countstatus, "", "  ")
+		syncstatus.countstatus.mutex.RUnlock()
+		if e != nil {
+			syncstatus.lasterror = e
+			return
+		}
+		e = mio.WriteFile(syncstatus.filepath, syncstatusbs, true)
+		if e != nil {
+			syncstatus.lasterror = e
+			return
+		}
+		syncstatus.lastsavetime = time.Now()
+		// fmt.Println(syncstatus.lastsavetime)
+	})
+	return syncstatus.lasterror
+}

datasync/main.go

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"os"
+	"time"
+
+	"git.wecise.com/wecise/odb-go/odbc"
+	"git.wecise.com/wecise/odbtools/datasync/datasync"
+)
+
+var mcfg, logger = odbc.SetDefaultAppName("datasync")
+
+func main() {
+	println("cat", logger.FileOutPath(), "for detail information")
+	odbtools := datasync.NewDataSync()
+	e := odbtools.Init()
+	if e != nil {
+		logger.Error(e)
+		os.Exit(1)
+		return
+	}
+	done := odbtools.Run()
+	if done == nil {
+		os.Exit(0)
+		return
+	}
+	e = <-done
+	if e != nil {
+		logger.Error(e)
+		os.Exit(1)
+		return
+	}
+	time.Sleep(500 * time.Millisecond)
+}

go.mod

@@ -0,0 +1,49 @@
+module git.wecise.com/wecise/odbtools
+
+go 1.22.0
+
+require (
+	git.wecise.com/wecise/odb-go v0.0.0-20250328085256-ba8b2fb06fe1
+	github.com/scylladb/go-set v1.0.3-0.20200225121959-cc7b2070d91e
+	github.com/spf13/cast v1.7.1
+	github.com/wecisecode/util v0.3.2
+)
+
+require (
+	github.com/bluele/gcache v0.0.2 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/coreos/go-semver v0.3.0 // indirect
+	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
+	github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/fatih/color v1.18.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
+	github.com/gomodule/redigo v1.8.5 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/redis/go-redis/v9 v9.0.5 // indirect
+	github.com/spacemonkeygo/errors v0.0.0-20201030155909-2f5f890dbc62 // indirect
+	github.com/tidwall/uhatools v0.5.1 // indirect
+	github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
+	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
+	go.etcd.io/etcd/api/v3 v3.5.18 // indirect
+	go.etcd.io/etcd/client/pkg/v3 v3.5.18 // indirect
+	go.etcd.io/etcd/client/v3 v3.5.18 // indirect
+	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/multierr v1.6.0 // indirect
+	go.uber.org/zap v1.17.0 // indirect
+	golang.org/x/net v0.35.0 // indirect
+	golang.org/x/sys v0.30.0 // indirect
+	golang.org/x/text v0.22.0 // indirect
+	google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect
+	google.golang.org/grpc v1.59.0 // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
+	gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
+	gopkg.in/ini.v1 v1.67.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

go.sum

@@ -0,0 +1,149 @@
+git.wecise.com/wecise/odb-go v0.0.0-20250328085256-ba8b2fb06fe1 h1:HiUKPb6BWR4EqAx5LZGOHiWdUzQfUj8tJvey+wpr0BA=
+git.wecise.com/wecise/odb-go v0.0.0-20250328085256-ba8b2fb06fe1/go.mod h1:7DJYChz1HA8QZ2lslxCa0507/G47FdzmjTqpdHnKxSo=
+github.com/bluele/gcache v0.0.2 h1:WcbfdXICg7G/DGBh1PFfcirkWOQV+v077yF1pSy3DGw=
+github.com/bluele/gcache v0.0.2/go.mod h1:m15KV+ECjptwSPxKhOhQoAFQVtUFjTVkc3H8o0t/fp0=
+github.com/bsm/ginkgo/v2 v2.7.0 h1:ItPMPH90RbmZJt5GtkcNvIRuGEdwlBItdNVoyzaNQao=
+github.com/bsm/ginkgo/v2 v2.7.0/go.mod h1:AiKlXPm7ItEHNc/2+OkrNG4E0ITzojb9/xWzvQ9XZ9w=
+github.com/bsm/gomega v1.26.0 h1:LhQm+AFcgV2M0WyKroMASzAzCAJVpAxQXv4SaI9a69Y=
+github.com/bsm/gomega v1.26.0/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM=
+github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
+github.com/coreos/go-systemd/v22 v22.3.2 h1:D9/bQk5vlXQFZ6Kwuu6zaiXJ9oTPe68++AzAJc1DzSI=
+github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da h1:aIftn67I1fkbMa512G+w+Pxci9hJPB8oMnkcP3iZF38=
+github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/fatih/color v1.18.0 h1:S8gINlzdQ840/4pfAwic/ZE0djQEH3wM94VfqLTZcOM=
+github.com/fatih/color v1.18.0/go.mod h1:4FelSpRwEGDpQ12mAdzqdOukCy4u8WUtOY6lkT/6HfU=
+github.com/fatih/set v0.2.1 h1:nn2CaJyknWE/6txyUDGwysr3G5QC6xWB/PtVjPBbeaA=
+github.com/fatih/set v0.2.1/go.mod h1:+RKtMCH+favT2+3YecHGxcc0b4KyVWA1QWWJUs4E0CI=
+github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
+github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
+github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
+github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/gomodule/redigo v1.8.2/go.mod h1:P9dn9mFrCBvWhGE1wpxx6fgq7BAeLBk+UUUzlpkBYO0=
+github.com/gomodule/redigo v1.8.5 h1:nRAxCa+SVsyjSBrtZmG/cqb6VbTmuRzpg/PoTFlpumc=
+github.com/gomodule/redigo v1.8.5/go.mod h1:P9dn9mFrCBvWhGE1wpxx6fgq7BAeLBk+UUUzlpkBYO0=
+github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
+github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
+github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+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.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/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/redis/go-redis/v9 v9.0.5 h1:CuQcn5HIEeK7BgElubPP8CGtE0KakrnbBSTLjathl5o=
+github.com/redis/go-redis/v9 v9.0.5/go.mod h1:WqMKv5vnQbRuZstUwxQI195wHy+t4PuXDOjzMvcuQHk=
+github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
+github.com/scylladb/go-set v1.0.3-0.20200225121959-cc7b2070d91e h1:7q6NSFZDeGfvvtIRwBrU/aegEYJYmvev0cHAwo17zZQ=
+github.com/scylladb/go-set v1.0.3-0.20200225121959-cc7b2070d91e/go.mod h1:DkpGd78rljTxKAnTDPFqXSGxvETQnJyuSOQwsHycqfs=
+github.com/spacemonkeygo/errors v0.0.0-20201030155909-2f5f890dbc62 h1:X5+jSi+pL+sc2/Sp9mtrmRqDDnKkm9YVy1ik/jJDRD0=
+github.com/spacemonkeygo/errors v0.0.0-20201030155909-2f5f890dbc62/go.mod h1:7NL9UAYQnRM5iKHUCld3tf02fKb5Dft+41+VckASUy0=
+github.com/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y=
+github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/tidwall/uhatools v0.5.1 h1:zMm7nDCZ4dbF+GQIonXpm6ST1V2JHZB4WgtqR84CFl4=
+github.com/tidwall/uhatools v0.5.1/go.mod h1:A0rmLPzOam2WhWA02UFtC3WRWcQJygCH/TwXR11Jd3w=
+github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
+github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
+github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
+github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
+github.com/wecisecode/util v0.3.2 h1:u1VFgXfmutw/5YZyCAz6+ddtsQ5Ojcdpl3Mt7vYrsnc=
+github.com/wecisecode/util v0.3.2/go.mod h1:fASrRbyMPjssR1owWRMZIBCUSRYo5YkcxOk+kG59byY=
+github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+go.etcd.io/etcd/api/v3 v3.5.18 h1:Q4oDAKnmwqTo5lafvB+afbgCDF7E35E4EYV2g+FNGhs=
+go.etcd.io/etcd/api/v3 v3.5.18/go.mod h1:uY03Ob2H50077J7Qq0DeehjM/A9S8PhVfbQ1mSaMopU=
+go.etcd.io/etcd/client/pkg/v3 v3.5.18 h1:mZPOYw4h8rTk7TeJ5+3udUkfVGBqc+GCjOJYd68QgNM=
+go.etcd.io/etcd/client/pkg/v3 v3.5.18/go.mod h1:BxVf2o5wXG9ZJV+/Cu7QNUiJYk4A29sAhoI5tIRsCu4=
+go.etcd.io/etcd/client/v3 v3.5.18 h1:nvvYmNHGumkDjZhTHgVU36A9pykGa2K4lAJ0yY7hcXA=
+go.etcd.io/etcd/client/v3 v3.5.18/go.mod h1:kmemwOsPU9broExyhYsBxX4spCTDX3yLgPMWtpBXG6E=
+go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
+go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
+go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
+go.uber.org/zap v1.17.0 h1:MTjgFu6ZLKvY6Pvaqk97GlxNBuMpV4Hy/3P6tRGlI2U=
+go.uber.org/zap v1.17.0/go.mod h1:MXVU+bhUf/A7Xi2HNOnopQOrmycQ5Ih87HtOu4q5SSo=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.35.0 h1:T5GQRQb2y08kTAByq9L4/bz8cipCdA8FbRTXewonqY8=
+golang.org/x/net v0.35.0/go.mod h1:EglIi67kWsHKlRzzVMUD93VMSWGFOMSZgxFjparz1Qk=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/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-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
+golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d h1:VBu5YqKPv6XiJ199exd8Br+Aetz+o08F+PLMnwJQHAY=
+google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d/go.mod h1:yZTlhN0tQnXo3h00fuXNCxJdLdIdnVFVBaRJ5LWBbw4=
+google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d h1:DoPTO70H+bcDXcd39vOqb2viZxgqeBeSGtZ55yZU4/Q=
+google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d/go.mod h1:KjSP20unUpOx5kyQUFa7k4OJg0qeJ7DEZflGDu2p6Bk=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d h1:uvYuEyMHKNt+lT4K3bN6fGswmK8qSvcreM3BwjDh+y4=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d/go.mod h1:+Bk1OCOj40wS2hwAMA+aCW9ypzm63QTBBHp6lQ3p+9M=
+google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk=
+google.golang.org/grpc v1.59.0/go.mod h1:aUPDwccQo6OTjy7Hct4AfBPD1GptF4fyUjIkQ9YtF98=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA=
+gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=