committed by
GitHub
70 changed files with 1489 additions and 680 deletions
@ -0,0 +1,103 @@ |
|||||
|
package common |
||||
|
|
||||
|
import ( |
||||
|
"errors" |
||||
|
"io/ioutil" |
||||
|
"path/filepath" |
||||
|
|
||||
|
"github.com/yedf/dtm/dtmcli" |
||||
|
"github.com/yedf/dtm/dtmcli/dtmimp" |
||||
|
"gopkg.in/yaml.v2" |
||||
|
) |
||||
|
|
||||
|
const ( |
||||
|
DtmMetricsPort = 8889 |
||||
|
) |
||||
|
|
||||
|
// MicroService config type for micro service
|
||||
|
type MicroService struct { |
||||
|
Driver string `yaml:"Driver" default:"default"` |
||||
|
Target string `yaml:"Target"` |
||||
|
EndPoint string `yaml:"EndPoint"` |
||||
|
} |
||||
|
|
||||
|
type Store struct { |
||||
|
Driver string `yaml:"Driver" default:"boltdb"` |
||||
|
Host string `yaml:"Host"` |
||||
|
Port int64 `yaml:"Port"` |
||||
|
User string `yaml:"User"` |
||||
|
Password string `yaml:"Password"` |
||||
|
MaxOpenConns int64 `yaml:"MaxOpenConns" default:"500"` |
||||
|
MaxIdleConns int64 `yaml:"MaxIdleConns" default:"500"` |
||||
|
ConnMaxLifeTime int64 `yaml:"ConnMaxLifeTime" default:"5"` |
||||
|
DataExpire int64 `yaml:"DataExpire" default:"604800"` // Trans data will expire in 7 days. only for redis/boltdb.
|
||||
|
RedisPrefix string `yaml:"RedisPrefix" default:"{}"` // Redis storage prefix. store data to only one slot in cluster
|
||||
|
} |
||||
|
|
||||
|
func (s *Store) IsDB() bool { |
||||
|
return s.Driver == dtmcli.DBTypeMysql || s.Driver == dtmcli.DBTypePostgres |
||||
|
} |
||||
|
|
||||
|
func (s *Store) GetDBConf() dtmcli.DBConf { |
||||
|
return dtmcli.DBConf{ |
||||
|
Driver: s.Driver, |
||||
|
Host: s.Host, |
||||
|
Port: s.Port, |
||||
|
User: s.User, |
||||
|
Passwrod: s.Password, |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
type configType struct { |
||||
|
Store Store `yaml:"Store"` |
||||
|
TransCronInterval int64 `yaml:"TransCronInterval" default:"3"` |
||||
|
TimeoutToFail int64 `yaml:"TimeoutToFail" default:"35"` |
||||
|
RetryInterval int64 `yaml:"RetryInterval" default:"10"` |
||||
|
HttpPort int64 `yaml:"HttpPort" default:"36789"` |
||||
|
GrpcPort int64 `yaml:"GrpcPort" default:"36790"` |
||||
|
MicroService MicroService `yaml:"MicroService"` |
||||
|
UpdateBranchSync int64 `yaml:"UpdateBranchSync"` |
||||
|
ExamplesDB dtmcli.DBConf `yaml:"ExamplesDB"` |
||||
|
} |
||||
|
|
||||
|
// Config 配置
|
||||
|
var Config = configType{} |
||||
|
|
||||
|
func MustLoadConfig() { |
||||
|
loadFromEnv("", &Config) |
||||
|
cont := []byte{} |
||||
|
for d := MustGetwd(); d != "" && d != "/"; d = filepath.Dir(d) { |
||||
|
cont1, err := ioutil.ReadFile(d + "/conf.yml") |
||||
|
if err != nil { |
||||
|
cont1, err = ioutil.ReadFile(d + "/conf.sample.yml") |
||||
|
} |
||||
|
if cont1 != nil { |
||||
|
cont = cont1 |
||||
|
break |
||||
|
} |
||||
|
} |
||||
|
if len(cont) != 0 { |
||||
|
dtmimp.Logf("config is: \n%s", string(cont)) |
||||
|
err := yaml.UnmarshalStrict(cont, &Config) |
||||
|
dtmimp.FatalIfError(err) |
||||
|
} |
||||
|
err := checkConfig() |
||||
|
dtmimp.LogIfFatalf(err != nil, `config error: '%v'. |
||||
|
check you env, and conf.yml/conf.sample.yml in current and parent path: %s. |
||||
|
please visit http://d.dtm.pub to see the config document.
|
||||
|
loaded config is: |
||||
|
%v`, err, MustGetwd(), Config) |
||||
|
} |
||||
|
|
||||
|
func checkConfig() error { |
||||
|
if Config.RetryInterval < 10 { |
||||
|
return errors.New("RetryInterval should not be less than 10") |
||||
|
} else if Config.TimeoutToFail < Config.RetryInterval { |
||||
|
return errors.New("TimeoutToFail should not be less than RetryInterval") |
||||
|
} else if Config.Store.Driver == "boltdb" { |
||||
|
return nil |
||||
|
} else if Config.Store.Driver != "redis" && (Config.Store.User == "" || Config.Store.Host == "" || Config.Store.Port == 0) { |
||||
|
return errors.New("db config not valid") |
||||
|
} |
||||
|
return nil |
||||
|
} |
||||
@ -0,0 +1,17 @@ |
|||||
|
package common |
||||
|
|
||||
|
import ( |
||||
|
"os" |
||||
|
"testing" |
||||
|
|
||||
|
"github.com/go-playground/assert/v2" |
||||
|
) |
||||
|
|
||||
|
func TestLoadFromEnv(t *testing.T) { |
||||
|
assert.Equal(t, "MICRO_SERVICE_DRIVER", toUnderscoreUpper("MicroService_Driver")) |
||||
|
|
||||
|
ms := MicroService{} |
||||
|
os.Setenv("T_DRIVER", "d1") |
||||
|
loadFromEnv("T", &ms) |
||||
|
assert.Equal(t, "d1", ms.Driver) |
||||
|
} |
||||
@ -0,0 +1,55 @@ |
|||||
|
package common |
||||
|
|
||||
|
import ( |
||||
|
"fmt" |
||||
|
"os" |
||||
|
"reflect" |
||||
|
"regexp" |
||||
|
"strings" |
||||
|
|
||||
|
"github.com/yedf/dtm/dtmcli/dtmimp" |
||||
|
) |
||||
|
|
||||
|
func loadFromEnv(prefix string, conf interface{}) { |
||||
|
rv := reflect.ValueOf(conf) |
||||
|
dtmimp.PanicIf(rv.Kind() != reflect.Ptr || rv.IsNil(), |
||||
|
fmt.Errorf("should be a valid pointer, but %s found", reflect.TypeOf(conf).Name())) |
||||
|
loadFromEnvInner(prefix, rv.Elem(), "") |
||||
|
} |
||||
|
|
||||
|
func loadFromEnvInner(prefix string, conf reflect.Value, defaultValue string) { |
||||
|
kind := conf.Kind() |
||||
|
switch kind { |
||||
|
case reflect.Struct: |
||||
|
t := conf.Type() |
||||
|
for i := 0; i < t.NumField(); i++ { |
||||
|
tag := t.Field(i).Tag |
||||
|
loadFromEnvInner(prefix+"_"+tag.Get("yaml"), conf.Field(i), tag.Get("default")) |
||||
|
} |
||||
|
case reflect.String: |
||||
|
str := os.Getenv(toUnderscoreUpper(prefix)) |
||||
|
if str == "" { |
||||
|
str = defaultValue |
||||
|
} |
||||
|
conf.Set(reflect.ValueOf(str)) |
||||
|
case reflect.Int64: |
||||
|
str := os.Getenv(toUnderscoreUpper(prefix)) |
||||
|
if str == "" { |
||||
|
str = defaultValue |
||||
|
} |
||||
|
if str == "" { |
||||
|
str = "0" |
||||
|
} |
||||
|
conf.Set(reflect.ValueOf(int64(dtmimp.MustAtoi(str)))) |
||||
|
default: |
||||
|
panic(fmt.Errorf("unsupported type: %s", conf.Type().Name())) |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
func toUnderscoreUpper(key string) string { |
||||
|
key = strings.Trim(key, "_") |
||||
|
matchFirstCap := regexp.MustCompile("([a-z])([A-Z]+)") |
||||
|
s2 := matchFirstCap.ReplaceAllString(key, "${1}_${2}") |
||||
|
// dtmimp.Logf("loading from env: %s", strings.ToUpper(s2))
|
||||
|
return strings.ToUpper(s2) |
||||
|
} |
||||
@ -1,29 +1,48 @@ |
|||||
DB: |
Store: # specify which engine to store trans status |
||||
driver: 'mysql' |
# Driver: 'boltdb' # default store engine |
||||
host: 'localhost' |
|
||||
user: 'root' |
# Driver: 'redis' |
||||
password: '' |
# Host: 'localhost' |
||||
port: '3306' |
# User: '' |
||||
|
# Password: '' |
||||
# driver: 'postgres' |
# Port: 6379 |
||||
# host: 'localhost' |
|
||||
# user: 'postgres' |
Driver: 'mysql' |
||||
# password: 'mysecretpassword' |
Host: 'localhost' |
||||
# port: '5432' |
User: 'root' |
||||
|
Password: '' |
||||
# max_open_conns: 'dbmaxopenconns' |
Port: 3306 |
||||
# max_idle_conns: 'dbmaxidleconns' |
|
||||
# conn_max_life_time: 'dbconnmaxlifetime' |
# Driver: 'postgres' |
||||
|
# Host: 'localhost' |
||||
|
# User: 'postgres' |
||||
|
# Password: 'mysecretpassword' |
||||
|
# Port: '5432' |
||||
|
|
||||
|
### following connection config is for only Driver postgres/mysql |
||||
|
# MaxOpenConns: 500 |
||||
|
# MaxIdleConns: 500 |
||||
|
# ConnMaxLifeTime 5 # default value is 5 (minutes) |
||||
|
|
||||
|
### flollowing config is only for some Driver |
||||
|
# DataExpire: 604800 # Trans data will expire in 7 days. only for redis/boltdb. |
||||
|
# RedisPrefix: '{}' # default value is '{}'. Redis storage prefix. store data to only one slot in cluster |
||||
|
|
||||
# MicroService: |
# MicroService: |
||||
# Driver: 'dtm-driver-gozero' # name of the driver to handle register/discover |
# Driver: 'dtm-driver-gozero' # name of the driver to handle register/discover |
||||
# Target: 'etcd://localhost:2379/dtmservice' # register dtm server to this url |
# Target: 'etcd://localhost:2379/dtmservice' # register dtm server to this url |
||||
# EndPoint: 'localhost:36790' |
# EndPoint: 'localhost:36790' |
||||
|
|
||||
# MicroService: |
|
||||
# Driver: 'dtm-driver-protocol1' |
|
||||
|
|
||||
# the unit of following configurations is second |
# the unit of following configurations is second |
||||
|
|
||||
# TransCronInterval: 3 # the interval to poll unfinished global transaction for every dtm process |
# TransCronInterval: 3 # the interval to poll unfinished global transaction for every dtm process |
||||
# TimeoutToFail: 35 # timeout for XA, TCC to fail. saga's timeout default to infinite, which can be overwritten in saga options |
# TimeoutToFail: 35 # timeout for XA, TCC to fail. saga's timeout default to infinite, which can be overwritten in saga options |
||||
# RetryInterval: 10 # the subtrans branch will be retried after this interval |
# RetryInterval: 10 # the subtrans branch will be retried after this interval |
||||
|
|
||||
|
### dtm can run examples, and examples will use following config to connect db |
||||
|
ExamplesDB: |
||||
|
Driver: 'mysql' |
||||
|
Host: 'localhost' |
||||
|
User: 'root' |
||||
|
Password: '' |
||||
|
Port: 3306 |
||||
|
|||||
@ -0,0 +1,244 @@ |
|||||
|
package storage |
||||
|
|
||||
|
import ( |
||||
|
"fmt" |
||||
|
"sync" |
||||
|
"time" |
||||
|
|
||||
|
"github.com/yedf/dtm/common" |
||||
|
"github.com/yedf/dtm/dtmcli/dtmimp" |
||||
|
bolt "go.etcd.io/bbolt" |
||||
|
"gorm.io/gorm" |
||||
|
) |
||||
|
|
||||
|
type BoltdbStore struct { |
||||
|
} |
||||
|
|
||||
|
var boltDb *bolt.DB = nil |
||||
|
var boltOnce sync.Once |
||||
|
|
||||
|
func boltGet() *bolt.DB { |
||||
|
boltOnce.Do(func() { |
||||
|
db, err := bolt.Open("./dtm.bolt", 0666, &bolt.Options{Timeout: 1 * time.Second}) |
||||
|
dtmimp.E2P(err) |
||||
|
boltDb = db |
||||
|
}) |
||||
|
return boltDb |
||||
|
} |
||||
|
|
||||
|
var bucketGlobal = []byte("global") |
||||
|
var bucketBranches = []byte("branches") |
||||
|
var bucketIndex = []byte("index") |
||||
|
|
||||
|
func tGetGlobal(t *bolt.Tx, gid string) *TransGlobalStore { |
||||
|
trans := TransGlobalStore{} |
||||
|
bs := t.Bucket(bucketGlobal).Get([]byte(gid)) |
||||
|
if bs == nil { |
||||
|
return nil |
||||
|
} |
||||
|
dtmimp.MustUnmarshal(bs, &trans) |
||||
|
return &trans |
||||
|
} |
||||
|
|
||||
|
func tGetBranches(t *bolt.Tx, gid string) []TransBranchStore { |
||||
|
branches := []TransBranchStore{} |
||||
|
cursor := t.Bucket(bucketBranches).Cursor() |
||||
|
for k, v := cursor.Seek([]byte(gid)); k != nil; k, v = cursor.Next() { |
||||
|
b := TransBranchStore{} |
||||
|
dtmimp.MustUnmarshal(v, &b) |
||||
|
if b.Gid != gid { |
||||
|
break |
||||
|
} |
||||
|
branches = append(branches, b) |
||||
|
} |
||||
|
return branches |
||||
|
} |
||||
|
func tPutGlobal(t *bolt.Tx, global *TransGlobalStore) { |
||||
|
bs := dtmimp.MustMarshal(global) |
||||
|
err := t.Bucket(bucketGlobal).Put([]byte(global.Gid), bs) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
|
||||
|
func tPutBranches(t *bolt.Tx, branches []TransBranchStore, start int64) { |
||||
|
if start == -1 { |
||||
|
bs := tGetBranches(t, branches[0].Gid) |
||||
|
start = int64(len(bs)) |
||||
|
} |
||||
|
for i, b := range branches { |
||||
|
k := b.Gid + fmt.Sprintf("%03d", i+int(start)) |
||||
|
v := dtmimp.MustMarshalString(b) |
||||
|
err := t.Bucket(bucketBranches).Put([]byte(k), []byte(v)) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
func tDelIndex(t *bolt.Tx, unix int64, gid string) { |
||||
|
k := fmt.Sprintf("%d-%s", unix, gid) |
||||
|
err := t.Bucket(bucketIndex).Delete([]byte(k)) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
|
||||
|
func tPutIndex(t *bolt.Tx, unix int64, gid string) { |
||||
|
k := fmt.Sprintf("%d-%s", unix, gid) |
||||
|
err := t.Bucket(bucketIndex).Put([]byte(k), []byte(gid)) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) Ping() error { |
||||
|
return nil |
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) PopulateData(skipDrop bool) { |
||||
|
if !skipDrop { |
||||
|
err := boltGet().Update(func(t *bolt.Tx) error { |
||||
|
t.DeleteBucket(bucketIndex) |
||||
|
t.DeleteBucket(bucketBranches) |
||||
|
t.DeleteBucket(bucketGlobal) |
||||
|
t.CreateBucket(bucketIndex) |
||||
|
t.CreateBucket(bucketBranches) |
||||
|
t.CreateBucket(bucketGlobal) |
||||
|
return nil |
||||
|
}) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) FindTransGlobalStore(gid string) (trans *TransGlobalStore) { |
||||
|
err := boltGet().View(func(t *bolt.Tx) error { |
||||
|
trans = tGetGlobal(t, gid) |
||||
|
return nil |
||||
|
}) |
||||
|
dtmimp.E2P(err) |
||||
|
return |
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) ScanTransGlobalStores(position *string, limit int64) []TransGlobalStore { |
||||
|
globals := []TransGlobalStore{} |
||||
|
err := boltGet().View(func(t *bolt.Tx) error { |
||||
|
cursor := t.Bucket(bucketGlobal).Cursor() |
||||
|
for k, v := cursor.First(); k != nil; k, v = cursor.Next() { |
||||
|
if string(k) == *position { |
||||
|
continue |
||||
|
} |
||||
|
g := TransGlobalStore{} |
||||
|
dtmimp.MustUnmarshal(v, &g) |
||||
|
globals = append(globals, g) |
||||
|
if len(globals) == int(limit) { |
||||
|
break |
||||
|
} |
||||
|
} |
||||
|
return nil |
||||
|
}) |
||||
|
dtmimp.E2P(err) |
||||
|
if len(globals) < int(limit) { |
||||
|
*position = "" |
||||
|
} else { |
||||
|
*position = globals[len(globals)-1].Gid |
||||
|
} |
||||
|
return globals |
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) FindBranches(gid string) []TransBranchStore { |
||||
|
var branches []TransBranchStore = nil |
||||
|
err := boltGet().View(func(t *bolt.Tx) error { |
||||
|
branches = tGetBranches(t, gid) |
||||
|
return nil |
||||
|
}) |
||||
|
dtmimp.E2P(err) |
||||
|
return branches |
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) UpdateBranchesSql(branches []TransBranchStore, updates []string) *gorm.DB { |
||||
|
return nil // not implemented
|
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) LockGlobalSaveBranches(gid string, status string, branches []TransBranchStore, branchStart int) { |
||||
|
err := boltGet().Update(func(t *bolt.Tx) error { |
||||
|
g := tGetGlobal(t, gid) |
||||
|
if g == nil { |
||||
|
return ErrNotFound |
||||
|
} |
||||
|
if g.Status != status { |
||||
|
return ErrNotFound |
||||
|
} |
||||
|
tPutBranches(t, branches, int64(branchStart)) |
||||
|
return nil |
||||
|
}) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) MaySaveNewTrans(global *TransGlobalStore, branches []TransBranchStore) error { |
||||
|
return boltGet().Update(func(t *bolt.Tx) error { |
||||
|
g := tGetGlobal(t, global.Gid) |
||||
|
if g != nil { |
||||
|
return ErrUniqueConflict |
||||
|
} |
||||
|
tPutGlobal(t, global) |
||||
|
tPutIndex(t, global.NextCronTime.Unix(), global.Gid) |
||||
|
tPutBranches(t, branches, 0) |
||||
|
return nil |
||||
|
}) |
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) ChangeGlobalStatus(global *TransGlobalStore, newStatus string, updates []string, finished bool) { |
||||
|
old := global.Status |
||||
|
global.Status = newStatus |
||||
|
err := boltGet().Update(func(t *bolt.Tx) error { |
||||
|
g := tGetGlobal(t, global.Gid) |
||||
|
if g == nil || g.Status != old { |
||||
|
return ErrNotFound |
||||
|
} |
||||
|
if finished { |
||||
|
tDelIndex(t, g.NextCronTime.Unix(), g.Gid) |
||||
|
} |
||||
|
tPutGlobal(t, global) |
||||
|
return nil |
||||
|
}) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) TouchCronTime(global *TransGlobalStore, nextCronInterval int64) { |
||||
|
oldUnix := global.NextCronTime.Unix() |
||||
|
global.NextCronTime = common.GetNextTime(nextCronInterval) |
||||
|
global.UpdateTime = common.GetNextTime(0) |
||||
|
global.NextCronInterval = nextCronInterval |
||||
|
err := boltGet().Update(func(t *bolt.Tx) error { |
||||
|
g := tGetGlobal(t, global.Gid) |
||||
|
if g == nil || g.Gid != global.Gid { |
||||
|
return ErrNotFound |
||||
|
} |
||||
|
tDelIndex(t, oldUnix, global.Gid) |
||||
|
tPutGlobal(t, global) |
||||
|
tPutIndex(t, global.NextCronTime.Unix(), global.Gid) |
||||
|
return nil |
||||
|
}) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
|
||||
|
func (s *BoltdbStore) LockOneGlobalTrans(expireIn time.Duration) *TransGlobalStore { |
||||
|
var trans *TransGlobalStore = nil |
||||
|
min := fmt.Sprintf("%d", time.Now().Add(expireIn).Unix()) |
||||
|
next := time.Now().Add(time.Duration(config.RetryInterval) * time.Second) |
||||
|
err := boltGet().Update(func(t *bolt.Tx) error { |
||||
|
cursor := t.Bucket(bucketIndex).Cursor() |
||||
|
k, v := cursor.First() |
||||
|
if k == nil || string(k) > min { |
||||
|
return ErrNotFound |
||||
|
} |
||||
|
trans = tGetGlobal(t, string(v)) |
||||
|
err := t.Bucket(bucketIndex).Delete(k) |
||||
|
dtmimp.E2P(err) |
||||
|
if trans == nil { // index exists, but global trans not exists, so retry to get next
|
||||
|
return ErrShouldRetry |
||||
|
} |
||||
|
trans.NextCronTime = &next |
||||
|
tPutGlobal(t, trans) |
||||
|
tPutIndex(t, next.Unix(), trans.Gid) |
||||
|
return nil |
||||
|
}) |
||||
|
if err == ErrNotFound { |
||||
|
return nil |
||||
|
} |
||||
|
dtmimp.E2P(err) |
||||
|
return trans |
||||
|
} |
||||
@ -0,0 +1,259 @@ |
|||||
|
package storage |
||||
|
|
||||
|
import ( |
||||
|
"context" |
||||
|
"fmt" |
||||
|
"time" |
||||
|
|
||||
|
"github.com/go-redis/redis/v8" |
||||
|
"github.com/yedf/dtm/common" |
||||
|
"github.com/yedf/dtm/dtmcli/dtmimp" |
||||
|
"gorm.io/gorm" |
||||
|
) |
||||
|
|
||||
|
var ctx context.Context = context.Background() |
||||
|
|
||||
|
type RedisStore struct { |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) Ping() error { |
||||
|
_, err := redisGet().Ping(ctx).Result() |
||||
|
return err |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) PopulateData(skipDrop bool) { |
||||
|
_, err := redisGet().FlushAll(ctx).Result() |
||||
|
dtmimp.PanicIf(err != nil, err) |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) FindTransGlobalStore(gid string) *TransGlobalStore { |
||||
|
r, err := redisGet().Get(ctx, config.Store.RedisPrefix+"_g_"+gid).Result() |
||||
|
if err == redis.Nil { |
||||
|
return nil |
||||
|
} |
||||
|
dtmimp.E2P(err) |
||||
|
trans := &TransGlobalStore{} |
||||
|
dtmimp.MustUnmarshalString(r, trans) |
||||
|
return trans |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) ScanTransGlobalStores(position *string, limit int64) []TransGlobalStore { |
||||
|
lid := uint64(0) |
||||
|
if *position != "" { |
||||
|
lid = uint64(dtmimp.MustAtoi(*position)) |
||||
|
} |
||||
|
keys, cursor, err := redisGet().Scan(ctx, lid, config.Store.RedisPrefix+"_g_*", limit).Result() |
||||
|
dtmimp.E2P(err) |
||||
|
globals := []TransGlobalStore{} |
||||
|
if len(keys) > 0 { |
||||
|
values, err := redisGet().MGet(ctx, keys...).Result() |
||||
|
dtmimp.E2P(err) |
||||
|
for _, v := range values { |
||||
|
global := TransGlobalStore{} |
||||
|
dtmimp.MustUnmarshalString(v.(string), &global) |
||||
|
globals = append(globals, global) |
||||
|
} |
||||
|
} |
||||
|
if cursor > 0 { |
||||
|
*position = fmt.Sprintf("%d", cursor) |
||||
|
} else { |
||||
|
*position = "" |
||||
|
} |
||||
|
return globals |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) FindBranches(gid string) []TransBranchStore { |
||||
|
sa, err := redisGet().LRange(ctx, config.Store.RedisPrefix+"_b_"+gid, 0, -1).Result() |
||||
|
dtmimp.E2P(err) |
||||
|
branches := make([]TransBranchStore, len(sa)) |
||||
|
for k, v := range sa { |
||||
|
dtmimp.MustUnmarshalString(v, &branches[k]) |
||||
|
} |
||||
|
return branches |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) UpdateBranchesSql(branches []TransBranchStore, updates []string) *gorm.DB { |
||||
|
return nil // not implemented
|
||||
|
} |
||||
|
|
||||
|
type argList struct { |
||||
|
List []interface{} |
||||
|
} |
||||
|
|
||||
|
func newArgList() *argList { |
||||
|
a := &argList{} |
||||
|
return a.AppendRaw(config.Store.RedisPrefix).AppendObject(config.Store.DataExpire) |
||||
|
} |
||||
|
|
||||
|
func (a *argList) AppendRaw(v interface{}) *argList { |
||||
|
a.List = append(a.List, v) |
||||
|
return a |
||||
|
} |
||||
|
|
||||
|
func (a *argList) AppendObject(v interface{}) *argList { |
||||
|
return a.AppendRaw(dtmimp.MustMarshalString(v)) |
||||
|
} |
||||
|
|
||||
|
func (a *argList) AppendBranches(branches []TransBranchStore) *argList { |
||||
|
for _, b := range branches { |
||||
|
a.AppendRaw(dtmimp.MustMarshalString(b)) |
||||
|
} |
||||
|
return a |
||||
|
} |
||||
|
|
||||
|
func handleRedisResult(ret interface{}, err error) (string, error) { |
||||
|
dtmimp.Logf("result is: '%v', err: '%v'", ret, err) |
||||
|
if err != nil && err != redis.Nil { |
||||
|
return "", err |
||||
|
} |
||||
|
s, _ := ret.(string) |
||||
|
err = map[string]error{ |
||||
|
"NOT_FOUND": ErrNotFound, |
||||
|
"UNIQUE_CONFLICT": ErrUniqueConflict, |
||||
|
}[s] |
||||
|
return s, err |
||||
|
} |
||||
|
|
||||
|
func callLua(args []interface{}, lua string) (string, error) { |
||||
|
dtmimp.Logf("calling lua. args: %v\nlua:%s", args, lua) |
||||
|
ret, err := redisGet().Eval(ctx, lua, []string{config.Store.RedisPrefix}, args...).Result() |
||||
|
return handleRedisResult(ret, err) |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) MaySaveNewTrans(global *TransGlobalStore, branches []TransBranchStore) error { |
||||
|
args := newArgList(). |
||||
|
AppendObject(global). |
||||
|
AppendRaw(global.NextCronTime.Unix()). |
||||
|
AppendBranches(branches). |
||||
|
List |
||||
|
global.Steps = nil |
||||
|
global.Payloads = nil |
||||
|
_, err := callLua(args, `-- MaySaveNewTrans |
||||
|
local gs = cjson.decode(ARGV[3]) |
||||
|
local g = redis.call('GET', ARGV[1] .. '_g_' .. gs.gid) |
||||
|
if g ~= false then |
||||
|
return 'UNIQUE_CONFLICT' |
||||
|
end |
||||
|
|
||||
|
redis.call('SET', ARGV[1] .. '_g_' .. gs.gid, ARGV[3], 'EX', ARGV[2]) |
||||
|
redis.call('ZADD', ARGV[1] .. '_u', ARGV[4], gs.gid) |
||||
|
for k = 5, table.getn(ARGV) do |
||||
|
redis.call('RPUSH', ARGV[1] .. '_b_' .. gs.gid, ARGV[k]) |
||||
|
end |
||||
|
redis.call('EXPIRE', ARGV[1] .. '_b_' .. gs.gid, ARGV[2]) |
||||
|
`) |
||||
|
return err |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) LockGlobalSaveBranches(gid string, status string, branches []TransBranchStore, branchStart int) { |
||||
|
args := newArgList(). |
||||
|
AppendObject(&TransGlobalStore{Gid: gid, Status: status}). |
||||
|
AppendRaw(branchStart). |
||||
|
AppendBranches(branches). |
||||
|
List |
||||
|
_, err := callLua(args, ` |
||||
|
local pre = ARGV[1] |
||||
|
local gs = cjson.decode(ARGV[3]) |
||||
|
local g = redis.call('GET', pre .. '_g_' .. gs.gid) |
||||
|
if (g == false) then |
||||
|
return 'NOT_FOUND' |
||||
|
end |
||||
|
local js = cjson.decode(g) |
||||
|
if js.status ~= gs.status then |
||||
|
return 'NOT_FOUND' |
||||
|
end |
||||
|
local start = ARGV[4] |
||||
|
for k = 5, table.getn(ARGV) do |
||||
|
if start == "-1" then |
||||
|
redis.call('RPUSH', pre .. '_b_' .. gs.gid, ARGV[k]) |
||||
|
else |
||||
|
redis.call('LSET', pre .. '_b_' .. gs.gid, start+k-5, ARGV[k]) |
||||
|
end |
||||
|
end |
||||
|
redis.call('EXPIRE', pre .. '_b_' .. gs.gid, ARGV[2]) |
||||
|
`) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) ChangeGlobalStatus(global *TransGlobalStore, newStatus string, updates []string, finished bool) { |
||||
|
old := global.Status |
||||
|
global.Status = newStatus |
||||
|
args := newArgList().AppendObject(global).AppendRaw(old).AppendRaw(finished).List |
||||
|
_, err := callLua(args, `-- ChangeGlobalStatus |
||||
|
local p = ARGV[1] |
||||
|
local gs = cjson.decode(ARGV[3]) |
||||
|
local old = redis.call('GET', p .. '_g_' .. gs.gid) |
||||
|
if old == false then |
||||
|
return 'NOT_FOUND' |
||||
|
end |
||||
|
local os = cjson.decode(old) |
||||
|
if os.status ~= ARGV[4] then |
||||
|
return 'NOT_FOUND' |
||||
|
end |
||||
|
redis.call('SET', p .. '_g_' .. gs.gid, ARGV[3], 'EX', ARGV[2]) |
||||
|
redis.log(redis.LOG_WARNING, 'finished: ', ARGV[5]) |
||||
|
if ARGV[5] == '1' then |
||||
|
redis.call('ZREM', p .. '_u', gs.gid) |
||||
|
end |
||||
|
`) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) LockOneGlobalTrans(expireIn time.Duration) *TransGlobalStore { |
||||
|
expired := time.Now().Add(expireIn).Unix() |
||||
|
next := time.Now().Add(time.Duration(config.RetryInterval) * time.Second).Unix() |
||||
|
args := newArgList().AppendRaw(expired).AppendRaw(next).List |
||||
|
lua := `-- LocakOneGlobalTrans |
||||
|
local k = ARGV[1] .. '_u' |
||||
|
local r = redis.call('ZRANGE', k, 0, 0, 'WITHSCORES') |
||||
|
local gid = r[1] |
||||
|
if gid == nil then |
||||
|
return 'NOT_FOUND' |
||||
|
end |
||||
|
local g = redis.call('GET', ARGV[1] .. '_g_' .. gid) |
||||
|
redis.log(redis.LOG_WARNING, 'g is: ', g, 'gid is: ', gid) |
||||
|
if g == false then |
||||
|
redis.call('ZREM', k, gid) |
||||
|
return 'NOT_FOUND' |
||||
|
end |
||||
|
|
||||
|
if tonumber(r[2]) > tonumber(ARGV[3]) then |
||||
|
return 'NOT_FOUND' |
||||
|
end |
||||
|
redis.call('ZADD', k, ARGV[4], gid) |
||||
|
return g |
||||
|
` |
||||
|
r, err := callLua(args, lua) |
||||
|
for err == ErrShouldRetry { |
||||
|
r, err = callLua(args, lua) |
||||
|
} |
||||
|
if err == ErrNotFound { |
||||
|
return nil |
||||
|
} |
||||
|
dtmimp.E2P(err) |
||||
|
global := &TransGlobalStore{} |
||||
|
dtmimp.MustUnmarshalString(r, global) |
||||
|
return global |
||||
|
} |
||||
|
|
||||
|
func (s *RedisStore) TouchCronTime(global *TransGlobalStore, nextCronInterval int64) { |
||||
|
global.NextCronTime = common.GetNextTime(nextCronInterval) |
||||
|
global.UpdateTime = common.GetNextTime(0) |
||||
|
global.NextCronInterval = nextCronInterval |
||||
|
args := newArgList().AppendObject(global).AppendRaw(global.NextCronTime.Unix()).List |
||||
|
_, err := callLua(args, `-- TouchCronTime |
||||
|
local p = ARGV[1] |
||||
|
local g = cjson.decode(ARGV[3]) |
||||
|
local old = redis.call('GET', p .. '_g_' .. g.gid) |
||||
|
if old == false then |
||||
|
return 'NOT_FOUND' |
||||
|
end |
||||
|
local os = cjson.decode(old) |
||||
|
if os.status ~= g.status then |
||||
|
return 'NOT_FOUND' |
||||
|
end |
||||
|
redis.call('ZADD', p .. '_u', ARGV[4], g.gid) |
||||
|
redis.call('SET', p .. '_g_' .. g.gid, ARGV[3], 'EX', ARGV[2]) |
||||
|
`) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
@ -0,0 +1,138 @@ |
|||||
|
package storage |
||||
|
|
||||
|
import ( |
||||
|
"fmt" |
||||
|
"math" |
||||
|
"time" |
||||
|
|
||||
|
"github.com/google/uuid" |
||||
|
"github.com/yedf/dtm/common" |
||||
|
"github.com/yedf/dtm/dtmcli/dtmimp" |
||||
|
"gorm.io/gorm" |
||||
|
"gorm.io/gorm/clause" |
||||
|
) |
||||
|
|
||||
|
type SqlStore struct { |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) Ping() error { |
||||
|
dbr := dbGet().Exec("select 1") |
||||
|
return dbr.Error |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) PopulateData(skipDrop bool) { |
||||
|
file := fmt.Sprintf("%s/storage.%s.sql", common.GetCallerCodeDir(), config.Store.Driver) |
||||
|
common.RunSQLScript(config.Store.GetDBConf(), file, skipDrop) |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) FindTransGlobalStore(gid string) *TransGlobalStore { |
||||
|
trans := &TransGlobalStore{} |
||||
|
dbr := dbGet().Model(trans).Where("gid=?", gid).First(trans) |
||||
|
if dbr.Error == gorm.ErrRecordNotFound { |
||||
|
return nil |
||||
|
} |
||||
|
dtmimp.E2P(dbr.Error) |
||||
|
return trans |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) ScanTransGlobalStores(position *string, limit int64) []TransGlobalStore { |
||||
|
globals := []TransGlobalStore{} |
||||
|
lid := math.MaxInt64 |
||||
|
if *position != "" { |
||||
|
lid = dtmimp.MustAtoi(*position) |
||||
|
} |
||||
|
dbr := dbGet().Must().Where("id < ?", lid).Order("id desc").Limit(int(limit)).Find(&globals) |
||||
|
if dbr.RowsAffected < limit { |
||||
|
*position = "" |
||||
|
} else { |
||||
|
*position = fmt.Sprintf("%d", globals[len(globals)-1].ID) |
||||
|
} |
||||
|
return globals |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) FindBranches(gid string) []TransBranchStore { |
||||
|
branches := []TransBranchStore{} |
||||
|
dbGet().Must().Where("gid=?", gid).Order("id asc").Find(&branches) |
||||
|
return branches |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) UpdateBranchesSql(branches []TransBranchStore, updates []string) *gorm.DB { |
||||
|
return dbGet().Clauses(clause.OnConflict{ |
||||
|
OnConstraint: "trans_branch_op_pkey", |
||||
|
DoUpdates: clause.AssignmentColumns(updates), |
||||
|
}).Create(branches) |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) LockGlobalSaveBranches(gid string, status string, branches []TransBranchStore, branchStart int) { |
||||
|
err := dbGet().Transaction(func(tx *gorm.DB) error { |
||||
|
g := &TransGlobalStore{} |
||||
|
dbr := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Model(g).Where("gid=? and status=?", gid, status).First(g) |
||||
|
if dbr.Error == nil { |
||||
|
dbr = tx.Save(branches) |
||||
|
} |
||||
|
return wrapError(dbr.Error) |
||||
|
}) |
||||
|
dtmimp.E2P(err) |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) MaySaveNewTrans(global *TransGlobalStore, branches []TransBranchStore) error { |
||||
|
return dbGet().Transaction(func(db1 *gorm.DB) error { |
||||
|
db := &common.DB{DB: db1} |
||||
|
dbr := db.Must().Clauses(clause.OnConflict{ |
||||
|
DoNothing: true, |
||||
|
}).Create(global) |
||||
|
if dbr.RowsAffected <= 0 { // 如果这个不是新事务,返回错误
|
||||
|
return ErrUniqueConflict |
||||
|
} |
||||
|
if len(branches) > 0 { |
||||
|
db.Must().Clauses(clause.OnConflict{ |
||||
|
DoNothing: true, |
||||
|
}).Create(&branches) |
||||
|
} |
||||
|
return nil |
||||
|
}) |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) ChangeGlobalStatus(global *TransGlobalStore, newStatus string, updates []string, finished bool) { |
||||
|
old := global.Status |
||||
|
global.Status = newStatus |
||||
|
dbr := dbGet().Must().Model(global).Where("status=? and gid=?", old, global.Gid).Select(updates).Updates(global) |
||||
|
if dbr.RowsAffected == 0 { |
||||
|
dtmimp.E2P(ErrNotFound) |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) TouchCronTime(global *TransGlobalStore, nextCronInterval int64) { |
||||
|
global.NextCronTime = common.GetNextTime(nextCronInterval) |
||||
|
global.UpdateTime = common.GetNextTime(0) |
||||
|
global.NextCronInterval = nextCronInterval |
||||
|
dbGet().Must().Model(global).Where("status=? and gid=?", global.Status, global.Gid). |
||||
|
Select([]string{"next_cron_time", "update_time", "next_cron_interval"}).Updates(global) |
||||
|
} |
||||
|
|
||||
|
func (s *SqlStore) LockOneGlobalTrans(expireIn time.Duration) *TransGlobalStore { |
||||
|
db := dbGet() |
||||
|
getTime := func(second int) string { |
||||
|
return map[string]string{ |
||||
|
"mysql": fmt.Sprintf("date_add(now(), interval %d second)", second), |
||||
|
"postgres": fmt.Sprintf("current_timestamp + interval '%d second'", second), |
||||
|
}[config.Store.Driver] |
||||
|
} |
||||
|
expire := int(expireIn / time.Second) |
||||
|
whereTime := fmt.Sprintf("next_cron_time < %s", getTime(expire)) |
||||
|
owner := uuid.NewString() |
||||
|
global := &TransGlobalStore{} |
||||
|
dbr := db.Must().Model(global). |
||||
|
Where(whereTime + "and status in ('prepared', 'aborting', 'submitted')"). |
||||
|
Limit(1). |
||||
|
Select([]string{"owner", "next_cron_time"}). |
||||
|
Updates(&TransGlobalStore{ |
||||
|
Owner: owner, |
||||
|
NextCronTime: common.GetNextTime(common.Config.RetryInterval), |
||||
|
}) |
||||
|
if dbr.RowsAffected == 0 { |
||||
|
return nil |
||||
|
} |
||||
|
dbr = db.Must().Where("owner=?", owner).First(global) |
||||
|
return global |
||||
|
} |
||||
@ -0,0 +1,54 @@ |
|||||
|
package storage |
||||
|
|
||||
|
import ( |
||||
|
"errors" |
||||
|
"time" |
||||
|
|
||||
|
"github.com/go-redis/redis/v8" |
||||
|
"github.com/yedf/dtm/dtmcli/dtmimp" |
||||
|
"gorm.io/gorm" |
||||
|
) |
||||
|
|
||||
|
var ErrNotFound = errors.New("storage: NotFound") |
||||
|
var ErrShouldRetry = errors.New("storage: ShoudRetry") |
||||
|
var ErrUniqueConflict = errors.New("storage: UniqueKeyConflict") |
||||
|
|
||||
|
type Store interface { |
||||
|
Ping() error |
||||
|
PopulateData(skipDrop bool) |
||||
|
FindTransGlobalStore(gid string) *TransGlobalStore |
||||
|
ScanTransGlobalStores(position *string, limit int64) []TransGlobalStore |
||||
|
FindBranches(gid string) []TransBranchStore |
||||
|
UpdateBranchesSql(branches []TransBranchStore, updates []string) *gorm.DB |
||||
|
LockGlobalSaveBranches(gid string, status string, branches []TransBranchStore, branchStart int) |
||||
|
MaySaveNewTrans(global *TransGlobalStore, branches []TransBranchStore) error |
||||
|
ChangeGlobalStatus(global *TransGlobalStore, newStatus string, updates []string, finished bool) |
||||
|
TouchCronTime(global *TransGlobalStore, nextCronInterval int64) |
||||
|
LockOneGlobalTrans(expireIn time.Duration) *TransGlobalStore |
||||
|
} |
||||
|
|
||||
|
var stores map[string]Store = map[string]Store{ |
||||
|
"redis": &RedisStore{}, |
||||
|
"mysql": &SqlStore{}, |
||||
|
"postgres": &SqlStore{}, |
||||
|
"boltdb": &BoltdbStore{}, |
||||
|
} |
||||
|
|
||||
|
func GetStore() Store { |
||||
|
return stores[config.Store.Driver] |
||||
|
} |
||||
|
|
||||
|
// WaitStoreUp wait for db to go up
|
||||
|
func WaitStoreUp() { |
||||
|
for err := GetStore().Ping(); err != nil; err = GetStore().Ping() { |
||||
|
time.Sleep(3 * time.Second) |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
func wrapError(err error) error { |
||||
|
if err == gorm.ErrRecordNotFound || err == redis.Nil { |
||||
|
return ErrNotFound |
||||
|
} |
||||
|
dtmimp.E2P(err) |
||||
|
return err |
||||
|
} |
||||
@ -0,0 +1,52 @@ |
|||||
|
package storage |
||||
|
|
||||
|
import ( |
||||
|
"time" |
||||
|
|
||||
|
"github.com/yedf/dtm/common" |
||||
|
"github.com/yedf/dtm/dtmcli" |
||||
|
) |
||||
|
|
||||
|
type TransGlobalStore struct { |
||||
|
common.ModelBase |
||||
|
Gid string `json:"gid,omitempty"` |
||||
|
TransType string `json:"trans_type,omitempty"` |
||||
|
Steps []map[string]string `json:"steps,omitempty" gorm:"-"` |
||||
|
Payloads []string `json:"payloads,omitempty" gorm:"-"` |
||||
|
BinPayloads [][]byte `json:"-" gorm:"-"` |
||||
|
Status string `json:"status,omitempty"` |
||||
|
QueryPrepared string `json:"query_prepared,omitempty"` |
||||
|
Protocol string `json:"protocol,omitempty"` |
||||
|
CommitTime *time.Time `json:"commit_time,omitempty"` |
||||
|
FinishTime *time.Time `json:"finish_time,omitempty"` |
||||
|
RollbackTime *time.Time `json:"rollback_time,omitempty"` |
||||
|
Options string `json:"options,omitempty"` |
||||
|
CustomData string `json:"custom_data,omitempty"` |
||||
|
NextCronInterval int64 `json:"next_cron_interval,omitempty"` |
||||
|
NextCronTime *time.Time `json:"next_cron_time,omitempty"` |
||||
|
Owner string `json:"owner,omitempty"` |
||||
|
dtmcli.TransOptions |
||||
|
} |
||||
|
|
||||
|
// TableName TableName
|
||||
|
func (*TransGlobalStore) TableName() string { |
||||
|
return "dtm.trans_global" |
||||
|
} |
||||
|
|
||||
|
// TransBranchStore branch transaction
|
||||
|
type TransBranchStore struct { |
||||
|
common.ModelBase |
||||
|
Gid string `json:"gid,omitempty"` |
||||
|
URL string `json:"url,omitempty"` |
||||
|
BinData []byte |
||||
|
BranchID string `json:"branch_id,omitempty"` |
||||
|
Op string `json:"op,omitempty"` |
||||
|
Status string `json:"status,omitempty"` |
||||
|
FinishTime *time.Time `json:"finish_time,omitempty"` |
||||
|
RollbackTime *time.Time `json:"rollback_time,omitempty"` |
||||
|
} |
||||
|
|
||||
|
// TableName TableName
|
||||
|
func (*TransBranchStore) TableName() string { |
||||
|
return "dtm.trans_branch_op" |
||||
|
} |
||||
@ -0,0 +1,16 @@ |
|||||
|
package storage |
||||
|
|
||||
|
import ( |
||||
|
"github.com/go-redis/redis/v8" |
||||
|
"github.com/yedf/dtm/common" |
||||
|
) |
||||
|
|
||||
|
var config = &common.Config |
||||
|
|
||||
|
func dbGet() *common.DB { |
||||
|
return common.DbGet(config.Store.GetDBConf()) |
||||
|
} |
||||
|
|
||||
|
func redisGet() *redis.Client { |
||||
|
return common.RedisGet() |
||||
|
} |
||||
@ -0,0 +1,7 @@ |
|||||
|
package dtmsvr |
||||
|
|
||||
|
import ( |
||||
|
_ "github.com/ychensha/dtmdriver-polaris" |
||||
|
_ "github.com/yedf/dtmdriver-gozero" |
||||
|
_ "github.com/yedf/dtmdriver-protocol1" |
||||
|
) |
||||
@ -0,0 +1,14 @@ |
|||||
|
package examples |
||||
|
|
||||
|
import "fmt" |
||||
|
|
||||
|
func Startup() { |
||||
|
InitConfig() |
||||
|
GrpcStartup() |
||||
|
BaseAppStartup() |
||||
|
} |
||||
|
|
||||
|
func InitConfig() { |
||||
|
DtmHttpServer = fmt.Sprintf("http://localhost:%d/api/dtmsvr", config.HttpPort) |
||||
|
DtmGrpcServer = fmt.Sprintf("localhost:%d", config.GrpcPort) |
||||
|
} |
||||
@ -1,34 +0,0 @@ |
|||||
version: '3.3' |
|
||||
services: |
|
||||
api: |
|
||||
image: golang:1.16.6-alpine3.14 |
|
||||
extra_hosts: |
|
||||
- 'host.docker.internal:host-gateway' |
|
||||
volumes: |
|
||||
- /etc/localtime:/etc/localtime:ro |
|
||||
- /etc/timezone:/etc/timezone:ro |
|
||||
environment: |
|
||||
IS_DOCKER: '1' |
|
||||
GOPROXY: 'https://mirrors.aliyun.com/goproxy/,direct' |
|
||||
ports: |
|
||||
- '8080:8080' |
|
||||
- '8082:8082' |
|
||||
- '58080:58080' |
|
||||
volumes: |
|
||||
- ..:/app/work |
|
||||
command: ['go', 'run', '/app/work/app/main.go', 'dev'] |
|
||||
working_dir: /app/work |
|
||||
mysql: |
|
||||
image: 'mysql:5.7' |
|
||||
volumes: |
|
||||
- /etc/localtime:/etc/localtime:ro |
|
||||
- /etc/timezone:/etc/timezone:ro |
|
||||
environment: |
|
||||
MYSQL_ALLOW_EMPTY_PASSWORD: 1 |
|
||||
command: |
|
||||
[ |
|
||||
'--character-set-server=utf8mb4', |
|
||||
'--collation-server=utf8mb4_unicode_ci', |
|
||||
] |
|
||||
ports: |
|
||||
- '3306:3306' |
|
||||
@ -0,0 +1,98 @@ |
|||||
|
package test |
||||
|
|
||||
|
import ( |
||||
|
"testing" |
||||
|
"time" |
||||
|
|
||||
|
"github.com/stretchr/testify/assert" |
||||
|
"github.com/yedf/dtm/dtmcli/dtmimp" |
||||
|
"github.com/yedf/dtm/dtmsvr/storage" |
||||
|
) |
||||
|
|
||||
|
func initTransGlobal(gid string) (*storage.TransGlobalStore, storage.Store) { |
||||
|
next := time.Now().Add(10 * time.Second) |
||||
|
g := &storage.TransGlobalStore{Gid: gid, Status: "prepared", NextCronTime: &next} |
||||
|
bs := []storage.TransBranchStore{ |
||||
|
{Gid: gid, BranchID: "01"}, |
||||
|
} |
||||
|
s := storage.GetStore() |
||||
|
err := s.MaySaveNewTrans(g, bs) |
||||
|
dtmimp.E2P(err) |
||||
|
return g, s |
||||
|
} |
||||
|
|
||||
|
func TestStoreSave(t *testing.T) { |
||||
|
gid := dtmimp.GetFuncName() |
||||
|
bs := []storage.TransBranchStore{ |
||||
|
{Gid: gid, BranchID: "01"}, |
||||
|
{Gid: gid, BranchID: "02"}, |
||||
|
} |
||||
|
g, s := initTransGlobal(gid) |
||||
|
g2 := s.FindTransGlobalStore(gid) |
||||
|
assert.NotNil(t, g2) |
||||
|
assert.Equal(t, gid, g2.Gid) |
||||
|
|
||||
|
bs2 := s.FindBranches(gid) |
||||
|
assert.Equal(t, len(bs2), int(1)) |
||||
|
assert.Equal(t, "01", bs2[0].BranchID) |
||||
|
|
||||
|
s.LockGlobalSaveBranches(gid, g.Status, []storage.TransBranchStore{bs[1]}, -1) |
||||
|
bs3 := s.FindBranches(gid) |
||||
|
assert.Equal(t, 2, len(bs3)) |
||||
|
assert.Equal(t, "02", bs3[1].BranchID) |
||||
|
assert.Equal(t, "01", bs3[0].BranchID) |
||||
|
|
||||
|
err := dtmimp.CatchP(func() { |
||||
|
s.LockGlobalSaveBranches(g.Gid, "submitted", []storage.TransBranchStore{bs[1]}, 1) |
||||
|
}) |
||||
|
assert.Equal(t, storage.ErrNotFound, err) |
||||
|
|
||||
|
s.ChangeGlobalStatus(g, "succeed", []string{}, true) |
||||
|
} |
||||
|
|
||||
|
func TestStoreChangeStatus(t *testing.T) { |
||||
|
gid := dtmimp.GetFuncName() |
||||
|
g, s := initTransGlobal(gid) |
||||
|
g.Status = "no" |
||||
|
err := dtmimp.CatchP(func() { |
||||
|
s.ChangeGlobalStatus(g, "submitted", []string{}, false) |
||||
|
}) |
||||
|
assert.Equal(t, storage.ErrNotFound, err) |
||||
|
g.Status = "prepared" |
||||
|
s.ChangeGlobalStatus(g, "submitted", []string{}, false) |
||||
|
s.ChangeGlobalStatus(g, "succeed", []string{}, true) |
||||
|
} |
||||
|
|
||||
|
func TestStoreLockTrans(t *testing.T) { |
||||
|
// lock trans will only lock unfinished trans. ensure all other trans are finished
|
||||
|
gid := dtmimp.GetFuncName() |
||||
|
g, s := initTransGlobal(gid) |
||||
|
|
||||
|
g2 := s.LockOneGlobalTrans(2 * time.Duration(config.RetryInterval) * time.Second) |
||||
|
assert.NotNil(t, g2) |
||||
|
assert.Equal(t, gid, g2.Gid) |
||||
|
|
||||
|
s.TouchCronTime(g, 3*config.RetryInterval) |
||||
|
g2 = s.LockOneGlobalTrans(2 * time.Duration(config.RetryInterval) * time.Second) |
||||
|
assert.Nil(t, g2) |
||||
|
|
||||
|
s.TouchCronTime(g, 1*config.RetryInterval) |
||||
|
g2 = s.LockOneGlobalTrans(2 * time.Duration(config.RetryInterval) * time.Second) |
||||
|
assert.NotNil(t, g2) |
||||
|
assert.Equal(t, gid, g2.Gid) |
||||
|
|
||||
|
s.ChangeGlobalStatus(g, "succeed", []string{}, true) |
||||
|
g2 = s.LockOneGlobalTrans(2 * time.Duration(config.RetryInterval) * time.Second) |
||||
|
assert.Nil(t, g2) |
||||
|
} |
||||
|
|
||||
|
func TestStoreWait(t *testing.T) { |
||||
|
storage.WaitStoreUp() |
||||
|
} |
||||
|
|
||||
|
func TestUpdateBranchSql(t *testing.T) { |
||||
|
if !config.Store.IsDB() { |
||||
|
r := storage.GetStore().UpdateBranchesSql(nil, nil) |
||||
|
assert.Nil(t, r) |
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue