diff --git a/.gitignore b/.gitignore index def000a1..d864178d 100644 --- a/.gitignore +++ b/.gitignore @@ -9,5 +9,3 @@ dist/ tmp/ data/ *.log -*.rdb -*.aof diff --git a/cmd/redis-shake/main.go b/cmd/redis-shake/main.go index 106d044a..ed5700c5 100644 --- a/cmd/redis-shake/main.go +++ b/cmd/redis-shake/main.go @@ -60,6 +60,15 @@ func main() { } theReader = reader.NewRDBReader(opts) log.Infof("create RdbReader: %v", opts.Filepath) + } else if v.IsSet("aof_reader") { + opts := new(reader.AOFReaderOptions) + defaults.SetDefaults(opts) + err := v.UnmarshalKey("aof_reader", opts) + if err != nil { + log.Panicf("failed to read the AOFReader config entry. err: %v", err) + } + theReader = reader.NewAOFReader(opts) + log.Infof("create AOFReader: %v", opts.Filepath) } else { log.Panicf("no reader config entry found") } diff --git a/docs/src/en/reader/aof_reader.md b/docs/src/en/reader/aof_reader.md new file mode 100644 index 00000000..9779c3c8 --- /dev/null +++ b/docs/src/en/reader/aof_reader.md @@ -0,0 +1,19 @@ +# aof_reader + +## Introduction + +Can use ` aof_ Reader 'to read data from the AOF file and then write it to the target end. +It is commonly used to recover data from backup files and also supports data flash back. + +## configuration + +```toml +[aof_reader] +aoffilepath="/tmp/appendonly.aof.manifest" #or single-aof: /tmp/appendonly.aof" +aoftimestamp="0" +``` + +*An absolute path should be passed in. + +##The main process is as follows: +![aof_reader.jpg](/public/aof_reader.jpg) \ No newline at end of file diff --git a/docs/src/public/aof_reader.jpg b/docs/src/public/aof_reader.jpg new file mode 100644 index 00000000..1a3db65e Binary files /dev/null and b/docs/src/public/aof_reader.jpg differ diff --git a/docs/src/zh/guide/config.md b/docs/src/zh/guide/config.md index b39b41d6..206a5575 100644 --- a/docs/src/zh/guide/config.md +++ b/docs/src/zh/guide/config.md @@ -34,6 +34,7 @@ RedisShake 提供了不同的 Reader 用来对接不同的源端，配置详见 * [Sync Reader](../reader/sync_reader.md) * [Scan Reader](../reader/scan_reader.md) * [RDB Reader](../reader/rdb_reader.md) +* [AOF Reader](../reader/aof_reader.md) ## writer 配置 diff --git a/docs/src/zh/reader/aof_reader.md b/docs/src/zh/reader/aof_reader.md new file mode 100644 index 00000000..2be803c6 --- /dev/null +++ b/docs/src/zh/reader/aof_reader.md @@ -0,0 +1,18 @@ +# aof_reader + +## 介绍 + +可以使用 `aof_reader` 来从 AOF 文件中读取数据，然后写入目标端。常见于从备份文件中恢复数据，还支持数据闪回。 + +## 配置 + +```toml +[aof_reader] +aoffilepath="/tmp/appendonly.aof.manifest" #或者单aof文件 "/tmp/appendonly.aof" +aoftimestamp="0" +``` + +* 应传入绝对路径。 + +## 主要流程如下： +![aof_reader.jpg](/public/aof_reader.jpg) \ No newline at end of file diff --git a/go.mod b/go.mod index f8098742..03f9bdc3 100644 --- a/go.mod +++ b/go.mod @@ -27,7 +27,7 @@ require ( github.com/spf13/pflag v1.0.5 // indirect github.com/subosito/gotenv v1.4.2 // indirect golang.org/x/sys v0.12.0 // indirect - golang.org/x/text v0.7.0 // indirect + golang.org/x/text v0.12.0 // indirect gopkg.in/ini.v1 v1.67.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 4727f8bb..2d2336c2 100644 --- a/go.sum +++ b/go.sum @@ -336,8 +336,8 @@ golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo= -golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.12.0 h1:k+n5B8goJNdU7hSvEtMUz3d1Q6D/XW4COJSJR6fN0mc= +golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= diff --git a/internal/aof/aof.go b/internal/aof/aof.go new file mode 100644 index 00000000..78625cde --- /dev/null +++ b/internal/aof/aof.go @@ -0,0 +1,155 @@ +package aof + +import ( + "bufio" + "io" + "os" + "strconv" + "strings" + + "RedisShake/internal/entry" + "RedisShake/internal/log" +) + +const ( + AOFNotExist = 1 + AOFOpenErr = 3 + AOFOK = 0 + AOFEmpty = 2 + AOFFailed = 4 + AOFTruncated = 5 + SizeMax = 128 +) + +type Loader struct { + filPath string + ch chan *entry.Entry +} + +func NewLoader(filPath string, ch chan *entry.Entry) *Loader { + ld := new(Loader) + ld.ch = ch + ld.filPath = filPath + return ld +} + +func ReadCompleteLine(reader *bufio.Reader) ([]byte, error) { + line, isPrefix, err := reader.ReadLine() + if err != nil { + return nil, err + } + + for isPrefix { + var additional []byte + additional, isPrefix, err = reader.ReadLine() + if err != nil { + return nil, err + } + line = append(line, additional...) + } + + return line, err +} + +func (ld *Loader) LoadSingleAppendOnlyFile(AOFTimeStamp int64) int { + ret := AOFOK + AOFFilepath := ld.filPath + fp, err := os.Open(AOFFilepath) + if err != nil { + if os.IsNotExist(err) { + if _, err := os.Stat(AOFFilepath); err == nil || !os.IsNotExist(err) { + log.Infof("Fatal error: can't open the append log File %v for reading: %v", AOFFilepath, err.Error()) + return AOFOpenErr + } else { + log.Infof("The append log File %v doesn't exist: %v", AOFFilepath, err.Error()) + return AOFNotExist + } + + } + defer fp.Close() + + stat, _ := fp.Stat() + if stat.Size() == 0 { + return AOFEmpty + } + } + reader := bufio.NewReader(fp) + for { + + line, err := ReadCompleteLine(reader) + { + if err != nil { + if err == io.EOF { + break + } else { + log.Infof("Unrecoverable error reading the append only File %v: %v", AOFFilepath, err) + ret = AOFFailed + return ret + } + } else { + _, errs := fp.Seek(0, io.SeekCurrent) + if errs != nil { + log.Infof("Unrecoverable error reading the append only File %v: %v", AOFFilepath, errs) + ret = AOFFailed + return ret + } + } + + if line[0] == '#' { + if AOFTimeStamp != 0 && strings.HasPrefix(string(line), "#TS:") { + var ts int64 + ts, err = strconv.ParseInt(strings.TrimPrefix(string(line), "#TS:"), 10, 64) + if err != nil { + log.Panicf("Invalid timestamp annotation") + } + + if ts > AOFTimeStamp { + ret = AOFTruncated + log.Infof("Reached recovery timestamp: %s, subsequent data will no longer be read.", line) + return ret + } + } + continue + } + if line[0] != '*' { + log.Panicf("Bad File format reading the append only File %v:make a backup of your AOF File, then use ./redis-check-AOF --fix ", AOFFilepath) + } + argc, _ := strconv.ParseInt(string(line[1:]), 10, 64) + if argc < 1 { + log.Panicf("Bad File format reading the append only File %v:make a backup of your AOF File, then use ./redis-check-AOF --fix ", AOFFilepath) + } + if argc > int64(SizeMax) { + log.Panicf("Bad File format reading the append only File %v:make a backup of your AOF File, then use ./redis-check-AOF --fix ", AOFFilepath) + } + e := entry.NewEntry() + var argv []string + + for j := 0; j < int(argc); j++ { + line, err := ReadCompleteLine(reader) + if err != nil || line[0] != '$' { + log.Infof("Bad File format reading the append only File %v:make a backup of your AOF File, then use ./redis-check-AOF --fix ", AOFFilepath) + ret = AOFFailed + return ret + } + v64, _ := strconv.ParseInt(string(line[1:]), 10, 64) + argString := make([]byte, v64+2) + argString, err = ReadCompleteLine(reader) + if err != nil { + log.Infof("Unrecoverable error reading the append only File %v: %v", AOFFilepath, err) + ret = AOFFailed + return ret + } + argString = argString[:v64] + argv = append(argv, string(argString)) + } + + for _, value := range argv { + e.Argv = append(e.Argv, value) + } + ld.ch <- e + + } + + } + return ret +} diff --git a/internal/reader/aof_reader.go b/internal/reader/aof_reader.go new file mode 100644 index 00000000..b4631034 --- /dev/null +++ b/internal/reader/aof_reader.go @@ -0,0 +1,104 @@ +package reader + +import ( + "RedisShake/internal/aof" + "path/filepath" + + "RedisShake/internal/entry" + "RedisShake/internal/log" + "RedisShake/internal/utils" + + "github.com/dustin/go-humanize" +) + +type AOFReaderOptions struct { + Filepath string `mapstructure:"filepath" default:""` + AOFTimestamp int64 `mapstructure:"timestamp" default:"0"` +} + +type aofReader struct { + path string + ch chan *entry.Entry + + stat struct { + AOFName string `json:"aof_name"` + AOFStatus string `json:"aof_status"` + AOFFilepath string `json:"aof_file_path"` + AOFFileSizeBytes int64 `json:"aof_file_size_bytes"` + AOFFileSizeHuman string `json:"aof_file_size_human"` + AOFFileSentBytes int64 `json:"aof_file_sent_bytes"` + AOFFileSentHuman string `json:"aof_file_sent_human"` + AOFPercent string `json:"aof_percent"` + AOFTimestamp int64 `json:"aof_time_stamp"` + } +} + +func (r *aofReader) Status() interface{} { + return r.stat +} + +func (r *aofReader) StatusString() string { + return r.stat.AOFStatus +} + +func (r *aofReader) StatusConsistent() bool { + return r.stat.AOFFileSentBytes == r.stat.AOFFileSizeBytes +} + +func NewAOFReader(opts *AOFReaderOptions) Reader { + log.Infof("NewAOFReader: path=[%s]", opts.Filepath) + absolutePath, err := filepath.Abs(opts.Filepath) + if err != nil { + log.Panicf("NewAOFReader: filepath.Abs error: %s", err.Error()) + } + log.Infof("NewAOFReader: absolute path=[%s]", absolutePath) + r := &aofReader{ + path: absolutePath, + ch: make(chan *entry.Entry), + } + r.stat.AOFName = "aof_reader" + r.stat.AOFStatus = "init" + r.stat.AOFFilepath = absolutePath + r.stat.AOFFileSizeBytes = int64(utils.GetFileSize(absolutePath)) + r.stat.AOFFileSizeHuman = humanize.Bytes(uint64(r.stat.AOFFileSizeBytes)) + r.stat.AOFTimestamp = opts.AOFTimestamp + return r +} + +func (r *aofReader) StartRead() chan *entry.Entry { + //init entry + r.ch = make(chan *entry.Entry, 1024) + + // start read aof + go func() { + aofFileInfo := NewAOFFileInfo(r.path, r.ch) + // try load manifest file + aofFileInfo.AOFLoadManifestFromDisk() + manifestInfo := aofFileInfo.AOFManifest + if manifestInfo == nil { // load single aof file + log.Infof("start send single AOF path=[%s]", r.path) + aofLoader := aof.NewLoader(r.path, r.ch) + ret := aofLoader.LoadSingleAppendOnlyFile(r.stat.AOFTimestamp) + if ret == AOFOk || ret == AOFTruncated { + log.Infof("The AOF File was successfully loaded") + } else { + log.Infof("There was an error opening the AOF File.") + } + log.Infof("Send single AOF finished. path=[%s]", r.path) + close(r.ch) + } else { + aofLoader := NewAOFFileInfo(r.path, r.ch) + ret := aofLoader.LoadAppendOnlyFile(manifestInfo, r.stat.AOFTimestamp) + if ret == AOFOk || ret == AOFTruncated { + log.Infof("The AOF File was successfully loaded") + } else { + log.Infof("There was an error opening the AOF File.") + } + log.Infof("Send multi-part AOF finished. path=[%s]", r.path) + close(r.ch) + } + + }() + + return r.ch +} \ No newline at end of file diff --git a/internal/reader/parsing_aof.go b/internal/reader/parsing_aof.go new file mode 100644 index 00000000..e6ca00fd --- /dev/null +++ b/internal/reader/parsing_aof.go @@ -0,0 +1,735 @@ +package reader + +import ( + "RedisShake/internal/aof" + "RedisShake/internal/entry" + "RedisShake/internal/log" + "bufio" + "bytes" + "container/list" + "fmt" + "io" + "os" + "path" + "path/filepath" + "strconv" + "strings" + "time" + "unicode" +) + +const ( + AOFManifestFileTypeBase = "b" /* Base File */ + AOFManifestTypeHist = "h" /* History File */ + AOFManifestTypeIncr = "i" /* INCR File */ + AOFNotExist = 1 + AOFOpenErr = 3 + AOFOk = 0 + AOFEmpty = 2 + AOFFailed = 4 + AOFTruncated = 5 + AOFManifestKeyFileName = "File" + AOFManifestKeyFileSeq = "seq" + AOFManifestKeyFileType = "type" +) + +func Ustime() int64 { + tv := time.Now() + ust := int64(tv.UnixNano()) / 1000 + return ust + +} + +func MakePath(Paths string, FileName string) string { + return path.Join(Paths, FileName) +} + +func StringNeedsRepr(s string) int { + sLen := len(s) + point := 0 + for sLen > 0 { + if s[point] == '\\' || s[point] == '"' || s[point] == '\n' || s[point] == '\r' || + s[point] == '\t' || s[point] == '\a' || s[point] == '\b' || !unicode.IsPrint(rune(s[point])) || unicode.IsSpace(rune(s[point])) { + return 1 + } + sLen-- + point++ + } + + return 0 +} + +type INFO struct { + AOFDirName string + AOFUseRDBPreamble int // TODO:not support parsing rdb preamble + AOFManifest *AOFManifest + AOFFileName string + AOFCurrentSize int64 + AOFRewriteBaseSize int64 + updateLoadingFile string + ch chan *entry.Entry +} + +func (aofInfo *INFO) GetAOFDirName() string { + return aofInfo.AOFDirName +} + +func NewAOFFileInfo(aofFilePath string, ch chan *entry.Entry) *INFO { + return &INFO{ + AOFDirName: filepath.Dir(aofFilePath), + AOFUseRDBPreamble: 0, + AOFManifest: nil, + AOFFileName: filepath.Base(aofFilePath), + AOFCurrentSize: 0, + AOFRewriteBaseSize: 0, + ch: ch, + } +} + +func (a *AOFInfo) GetAOFInfoName() string { + return a.FileName +} + +func DirExists(dName string) int { + _, err := os.Stat(dName) + if err != nil { + return 0 + } + + return 1 +} + +func FileExist(FileName string) int { + _, err := os.Stat(FileName) + if err != nil { + return 0 + } + + return 1 +} + +func IsHexDigit(c byte) bool { + return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || + (c >= 'A' && c <= 'F') +} + +func HexDigitToInt(c byte) int { + switch c { + case '0': + return 0 + case '1': + return 1 + case '2': + return 2 + case '3': + return 3 + case '4': + return 4 + case '5': + return 5 + case '6': + return 6 + case '7': + return 7 + case '8': + return 8 + case '9': + return 9 + case 'a', 'A': + return 10 + case 'b', 'B': + return 11 + case 'c', 'C': + return 12 + case 'd', 'D': + return 13 + case 'e', 'E': + return 14 + case 'f', 'F': + return 15 + default: + return 0 + } +} + +func SplitArgs(line string) ([]string, int) { + var p = line + var Current string + var vector []string + argc := 0 + i := 0 + lens := len(p) + for { //SKIP BLANKS + for i < lens && unicode.IsSpace(rune(p[i])) { + i++ + } + if i < lens { + inq := false // Set to true if we are in "quotes" + insq := false // Set to true if we are in 'single quotes' + done := false + + for !done { + if inq { + + if p[i] == '\\' && (p[i+1]) == 'x' && IsHexDigit(p[i+2]) && IsHexDigit(p[i+3]) { + _, err1 := strconv.ParseInt(string(p[i+2]), 16, 64) + _, err2 := strconv.ParseInt(string(p[i+3]), 16, 64) + if err1 == nil && err2 == nil { + hexadecimal := (HexDigitToInt(p[i+2]) * 16) + HexDigitToInt(p[i+3]) + Current = Current + fmt.Sprint(hexadecimal) + i += 3 + } + + } else if p[i] == '\\' && i+1 < lens { + var c byte + i++ + switch p[i] { + case 'n': + c = '\n' + case 'r': + c = 'r' + case 'a': + c = '\a' + default: + c = p[i] + } + Current += string(c) + } else if p[i] == '"' { + if i+1 < lens && !unicode.IsSpace(rune(p[i+1])) { + return nil, 0 + } + done = true + } else if i >= lens { + return nil, 0 + } else { + Current += string(p[i]) + } + } else if insq { + if p[i] == '\\' && p[i+1] == '\'' { + i++ + Current += "'" + } else if p[i] == '\'' { + if i+1 < lens && !unicode.IsSpace(rune(p[i+1])) { + return nil, 0 + } + done = true + } else if i >= lens { + return nil, 0 + } else { + Current += string(p[i]) + } + + } else { + switch p[i] { + case ' ', '\n', '\r', '\t', '\000': + done = true + case '"': + inq = true + case '\'': + insq = true + default: + Current += string(p[i]) + } + } + if i < lens { + i++ + } + if i == lens { + done = true + } + } + + vector = append(vector, Current) + argc++ + Current = "" + + } else { + return vector, argc + } + + } +} + +func StringCatPrintf(s string, fmtStr string, args ...interface{}) string { + result := fmt.Sprintf(fmtStr, args...) + if s == "" { + return result + } else { + return s + result + } +} + +func StringCatRepr(s string, p string, length int) string { + s = s + ("\"") + for i := 0; i < length; i++ { + switch p[i] { + case '\\', '"': + s = StringCatPrintf(s, "\\%c", p[i]) + case '\n': + s = s + "\\n" + case '\r': + s = s + "\\r" + case '\t': + s = s + "\\t" + case '\a': + s = s + "\\a" + case '\b': + s = s + "\\b" + default: + if strconv.IsPrint(rune(p[i])) { + s = s + string(p[i]) + } else { + s = s + "\\x%02x" + } + } + } + return s + "\"" +} + +func (aofInfo *INFO) UpdateLoadingFileName(FileName string) { + aofInfo.updateLoadingFile = FileName +} + +// AOFInfo AOF manifest definition +type AOFInfo struct { + FileName string + FileSeq int64 + AOFFileType string +} + +func AOFInfoCreate() *AOFInfo { + return new(AOFInfo) +} + +func AOFInfoFormat(buf string, ai *AOFInfo) string { + var aofManifestTostring string + if StringNeedsRepr(ai.FileName) == 1 { + aofManifestTostring = StringCatRepr("", ai.FileName, len(ai.FileName)) + } + var ret string + if aofManifestTostring != "" { + ret = StringCatPrintf(buf, "%s %s %s %d %s %s\n", AOFManifestKeyFileName, aofManifestTostring, AOFManifestKeyFileSeq, ai.FileSeq, AOFManifestKeyFileType, ai.AOFFileType) + } else { + ret = StringCatPrintf(buf, "%s %s %s %d %s %s\n", AOFManifestKeyFileName, ai.FileName, AOFManifestKeyFileSeq, ai.FileSeq, AOFManifestKeyFileType, ai.AOFFileType) + } + return ret +} + +func PathIsBaseName(Path string) bool { + return strings.IndexByte(Path, '/') == -1 && strings.IndexByte(Path, '\\') == -1 +} + +func AOFLoadManifestFromFile(amFilepath string) *AOFManifest { + var maxSeq int64 + am := AOFManifestCreate() + fp, err := os.Open(amFilepath) + if err != nil { + log.Panicf("Fatal error:can't open the AOF manifest %v for reading: %v", amFilepath, err) + } + defer fp.Close() + var argv []string + var ai *AOFInfo + var line string + lineNum := 0 + fpReader := bufio.NewReader(fp) + for { + buf, err := fpReader.ReadString('\n') + if err != nil { + if err == io.EOF { + if lineNum == 0 { + log.Infof("Found an empty AOF manifest") + am = nil + return am + } else { + break + } + + } else { + log.Infof("Reading the manifest file, at line %d", lineNum) + log.Infof("Read AOF manifest failed") + am = nil + return am + + } + } + + lineNum++ + if buf[0] == '#' { + continue + } + if !strings.Contains(buf, "\n") { + log.Infof("Reading the manifest file, at line %d", lineNum) + log.Infof("The AOF manifest File contains too long line") + return nil + } + line = strings.Trim(buf, " \t\r\n") + if len(line) == 0 { + log.Infof("Reading the manifest file, at line %d", lineNum) + log.Infof("Invalid AOF manifest File format") + return nil + } + argc := 0 + argv, argc = SplitArgs(line) + + if argc < 6 || argc%2 != 0 { + log.Infof("Reading the manifest file, at line %d", lineNum) + log.Infof("Invalid AOF manifest File format") + am = nil + return am + } + ai = AOFInfoCreate() + for i := 0; i < argc; i += 2 { + if strings.EqualFold(argv[i], AOFManifestKeyFileName) { + ai.FileName = argv[i+1] + if !PathIsBaseName(ai.FileName) { + log.Infof("Reading the manifest file, at line %d", lineNum) + log.Panicf("File can't be a path, just a Filename") + } + } else if strings.EqualFold(argv[i], AOFManifestKeyFileSeq) { + ai.FileSeq, _ = strconv.ParseInt(argv[i+1], 10, 64) + } else if strings.EqualFold(argv[i], AOFManifestKeyFileType) { + ai.AOFFileType = string(argv[i+1][0]) + } + } + if ai.FileName == "" || ai.FileSeq == 0 || ai.AOFFileType == "" { + log.Infof("Reading the manifest file, at line %d", lineNum) + log.Panicf("Invalid AOF manifest File format") + } + if ai.AOFFileType == AOFManifestFileTypeBase { + if am.BaseAOFInfo != nil { + log.Infof("Reading the manifest file, at line %d", lineNum) + log.Panicf("Found duplicate Base File information") + } + am.BaseAOFInfo = ai + am.CurrBaseFileSeq = ai.FileSeq + } else if ai.AOFFileType == AOFManifestTypeHist { + am.HistoryList.PushBack(ai) + } else if ai.AOFFileType == AOFManifestTypeIncr { + if ai.FileSeq <= maxSeq { + log.Infof("Reading the manifest file, at line %d", lineNum) + log.Panicf("Found a non-monotonic sequence number") + } + am.incrAOFList.PushBack(ai) + am.CurrIncrFileSeq = ai.FileSeq + maxSeq = ai.FileSeq + } else { + log.Infof("Reading the manifest file, at line %d", lineNum) + log.Panicf("Unknown AOF File type") + } + ai = nil + } + return am +} + +type AOFManifest struct { + BaseAOFInfo *AOFInfo + incrAOFList *list.List + HistoryList *list.List + CurrBaseFileSeq int64 + CurrIncrFileSeq int64 + Dirty int64 +} + +func AOFManifestCreate() *AOFManifest { + am := &AOFManifest{ + incrAOFList: list.New(), + HistoryList: list.New(), + } + return am +} + +func GetAOFManifestAsString(am *AOFManifest) string { + if am == nil { + panic("am is nil") + } + var buf string + if am.BaseAOFInfo != nil { + buf = AOFInfoFormat(buf, am.BaseAOFInfo) + } + for ln := am.HistoryList.Front(); ln != nil; ln = ln.Next() { + buf = AOFInfoFormat(buf, ln.Value.(*AOFInfo)) + } + for ln := am.incrAOFList.Front(); ln != nil; ln = ln.Next() { + buf = AOFInfoFormat(buf, ln.Value.(*AOFInfo)) + } + return buf + +} + +func (aofInfo *INFO) AOFLoadManifestFromDisk() { + if DirExists(aofInfo.AOFDirName) == 0 { + log.Infof("The AOF Directory %v doesn't exist", aofInfo.AOFDirName) + return + } + aofInfo.AOFManifest = AOFManifestCreate() + amFilepath := path.Join(aofInfo.AOFDirName, aofInfo.AOFFileName) + if FileExist(amFilepath) == 0 { + log.Infof("The AOF Directory %v doesn't exist", aofInfo.AOFDirName) + return + } + + am := AOFLoadManifestFromFile(amFilepath) + aofInfo.AOFManifest = am +} + +func (aofInfo *INFO) GetAOFManifestFileName() string { + return aofInfo.AOFFileName +} + +func (aofInfo *INFO) AOFFileExist(FileName string) int { + Filepath := path.Join(aofInfo.AOFDirName, FileName) + ret := FileExist(Filepath) + return ret +} + +func (aofInfo *INFO) GetAppendOnlyFileSize(FileName string, status *int) int64 { + var size int64 + + AOFFilePath := path.Join(aofInfo.AOFDirName, FileName) + + stat, err := os.Stat(AOFFilePath) + if err != nil { + if status != nil { + if os.IsNotExist(err) { + *status = AOFNotExist + } else { + *status = AOFOpenErr + } + } + log.Panicf("Unable to obtain the AOF File %v length. stat: %v", FileName, err.Error()) + size = 0 + } else { + if status != nil { + *status = AOFOk + } + size = stat.Size() + } + return size +} + +func (aofInfo *INFO) GetBaseAndIncrAppendOnlyFilesSize(am *AOFManifest, status *int) int64 { + var size int64 + if am.BaseAOFInfo != nil { + if am.BaseAOFInfo.AOFFileType != AOFManifestFileTypeBase { + log.Panicf("File type must be Base.") + } + size += aofInfo.GetAppendOnlyFileSize(am.BaseAOFInfo.FileName, status) + if *status != AOFOk { + return 0 + } + } + + for ln := am.HistoryList.Front(); ln != nil; ln = ln.Next() { + ai := ln.Value.(*AOFInfo) + if ai.AOFFileType != AOFManifestTypeIncr { + log.Panicf("File type must be Incr") + } + size += aofInfo.GetAppendOnlyFileSize(ai.FileName, status) + if *status != AOFOk { + return 0 + } + } + return size +} + +func GetBaseAndIncrAppendOnlyFilesNum(am *AOFManifest) int { + num := 0 + if am.BaseAOFInfo != nil { + num++ + } + if am.incrAOFList != nil { + num += am.incrAOFList.Len() + } + return num +} + +func GetHistoryAndIncrAppendOnlyFilesNum(am *AOFManifest) int { + num := 0 + if am.HistoryList != nil { + num += am.HistoryList.Len() + } + if am.incrAOFList != nil { + num += am.incrAOFList.Len() + } + return num +} + +func (aofInfo *INFO) LoadAppendOnlyFile(am *AOFManifest, AOFTimeStamp int64) int { + if am == nil { + log.Panicf("AOFManifest is null") + } + status := AOFOk + ret := AOFOk + var start int64 + var totalSize int64 = 0 + var BaseSize int64 = 0 + var AOFName string + var totalNum, AOFNum int + + if am.BaseAOFInfo == nil && am.incrAOFList == nil { + return AOFNotExist + } + + totalNum = GetBaseAndIncrAppendOnlyFilesNum(am) + if totalNum <= 0 { + log.Panicf("Assertion failed: IncrAppendOnlyFilestotalNum > 0") + } + + totalSize = aofInfo.GetBaseAndIncrAppendOnlyFilesSize(am, &status) + if status != AOFOk { + if status == AOFNotExist { + status = AOFFailed + } + return status + } else if totalSize == 0 { + return AOFEmpty + } + + log.Infof("The AOF File starts loading.") + if am.BaseAOFInfo != nil { + if am.BaseAOFInfo.AOFFileType == AOFManifestFileTypeBase { + AOFName = am.BaseAOFInfo.FileName + aofInfo.UpdateLoadingFileName(AOFName) + BaseSize = aofInfo.GetAppendOnlyFileSize(AOFName, nil) + start = Ustime() + ret = aofInfo.ParsingSingleAppendOnlyFile(AOFName, 0) //Currently, RDB files cannot be restored at a point in time. + if ret == AOFOk || (ret == AOFTruncated) { + log.Infof("DB loaded from Base File %v: %.3f seconds", AOFName, float64(Ustime()-start)/1000000) + } + if ret == AOFEmpty { + ret = AOFOk + } + if ret == AOFOpenErr || ret == AOFFailed { + if ret == AOFOk || ret == AOFTruncated { + log.Infof("The AOF File was successfully loaded") + } else { + if ret == AOFOpenErr { + log.Panicf("There was an error opening the AOF File.") + } else { + log.Panicf("Failed to open AOF File.") + } + } + return ret + } + } + totalNum-- + } else { + totalNum = GetHistoryAndIncrAppendOnlyFilesNum(am) + log.Infof("The BaseAOF file does not exist. Start loading the HistoryAOF and IncrAOF files.") + if am.HistoryList.Len() > 0 { + for ln := am.HistoryList.Front(); ln != nil; ln = ln.Next() { + ai := ln.Value.(*AOFInfo) + if ai.AOFFileType != AOFManifestTypeHist { + log.Panicf("The manifestType must be Hist") + } + AOFName = ai.FileName + aofInfo.UpdateLoadingFileName(AOFName) + AOFNum++ + start = Ustime() + ret = aofInfo.ParsingSingleAppendOnlyFile(AOFName, AOFTimeStamp) + if ret == AOFOk || (ret == AOFTruncated) { + log.Infof("DB loaded from History File %v: %.3f seconds", AOFName, float64(Ustime()-start)/1000000) + return ret + } + if ret == AOFEmpty { + ret = AOFOk + } + if ret == AOFOpenErr || ret == AOFFailed { + if ret == AOFOpenErr { + log.Panicf("There was an error opening the AOF File.") + } else { + log.Infof("Failed to open AOF File.") + } + return ret + } + totalNum-- + } + } + + } + + if am.incrAOFList.Len() > 0 { + for ln := am.incrAOFList.Front(); ln != nil; ln = ln.Next() { + ai := ln.Value.(*AOFInfo) + if ai.AOFFileType != AOFManifestTypeIncr { + log.Panicf("The manifestType must be Incr") + } + AOFName = ai.FileName + aofInfo.UpdateLoadingFileName(AOFName) + AOFNum++ + start = Ustime() + ret = aofInfo.ParsingSingleAppendOnlyFile(AOFName, AOFTimeStamp) + if ret == AOFOk || (ret == AOFTruncated) { + log.Infof("DB loaded from incr File %v: %.3f seconds", AOFName, float64(Ustime()-start)/1000000) + return ret + } + if ret == AOFEmpty { + ret = AOFOk + } + if ret == AOFOpenErr || ret == AOFFailed { + if ret == AOFOpenErr { + log.Panicf("There was an error opening the AOF File.") + } else { + log.Infof("Failed to open AOF File.") + } + return ret + } + totalNum-- + } + } + if totalNum == 0 { + log.Infof("All AOF files have been sent.") + } else { + log.Panicf("There are still %d AOF files that were not successfully sent.", totalNum) + } + aofInfo.AOFCurrentSize = totalSize + aofInfo.AOFRewriteBaseSize = BaseSize + + log.Infof("The AOF File loading end.") + return ret + +} + +func (aofInfo *INFO) ParsingSingleAppendOnlyFile(FileName string, AOFTimeStamp int64) int { + ret := AOFOk + AOFFilepath := path.Join(aofInfo.AOFDirName, FileName) + println(AOFFilepath) + fp, err := os.Open(AOFFilepath) + if err != nil { + if os.IsNotExist(err) { + if _, err := os.Stat(AOFFilepath); err == nil || !os.IsNotExist(err) { + log.Infof("Fatal error: can't open the append log File %v for reading: %v", FileName, err.Error()) + return AOFOpenErr + } else { + log.Infof("The append log File %v doesn't exist: %v", FileName, err.Error()) + return AOFNotExist + } + + } + + stat, _ := fp.Stat() + if stat.Size() == 0 { + return AOFEmpty + } + } + defer fp.Close() + sig := make([]byte, 5) + if n, err := fp.Read(sig); err != nil || n != 5 || !bytes.Equal(sig, []byte("REDIS")) { + if _, err := fp.Seek(0, 0); err != nil { + log.Infof("Unrecoverable error reading the append only File %v: %v", FileName, err) + ret = AOFFailed + return ret + } + } else { //Skipped RDB checksum and has not been processed yet. + log.Infof("Reading RDB Base File on AOF loading...") + rdbOpt := RdbReaderOptions{Filepath: AOFFilepath} + ldRDB := NewRDBReader(&rdbOpt) + ldRDB.StartRead() + return AOFOk + } + // load single aof file + aofSingleReader := aof.NewLoader(MakePath(aofInfo.AOFDirName, FileName), aofInfo.ch) + ret = aofSingleReader.LoadSingleAppendOnlyFile(AOFTimeStamp) + return ret + +} diff --git a/shake.toml b/shake.toml index 39d7cbf2..6ec20198 100644 --- a/shake.toml +++ b/shake.toml @@ -22,6 +22,10 @@ sync_aof = true # set to false if you don't want to sync aof # [rdb_reader] # filepath = "/tmp/dump.rdb" +# [aof_reader] +# filepath = "/tmp/.aof" +# timestamp = 0 # subsecond + [redis_writer] cluster = false # set to true if target is a redis cluster address = "127.0.0.1:6380" # when cluster is true, set address to one of the cluster node diff --git a/tests/cases/aof.py b/tests/cases/aof.py new file mode 100644 index 00000000..c18d545c --- /dev/null +++ b/tests/cases/aof.py @@ -0,0 +1,313 @@ +import pybbt as p + +import helpers as h +import os +#format aof command +def format_command(*args): + cmd = f"*{len(args)}\r\n" + for a in args: + cmd += f"${len(a)}\r\n{a}\r\n" + return cmd + + +def append_to_file(write_file, strings): + with open(write_file, "w+") as fp: + for string in strings: + fp.write(string) + +def create_aof_dir(dir_path): + os.makedirs(dir_path, exist_ok=True) + +def get_aof_file_relative_path(): + if h.REDIS_SERVER_VERSION == 7.0: + aof_file = "/appendonlydir/appendonly.aof.manifest" + else: + aof_file = "/appendonly.aof" + return aof_file + +def test(src, dst): + cross_slots_cmd = not (src.is_cluster() or dst.is_cluster()) + inserter = h.DataInserter() + inserter.add_data(src, cross_slots_cmd=cross_slots_cmd) + inserter.add_data(src, cross_slots_cmd=cross_slots_cmd) + p.ASSERT_TRUE(src.do("save")) + + opts = h.ShakeOpts.create_aof_opts(f"{src.dir}{get_aof_file_relative_path()}", dst) + h.Shake.run_once(opts) + # check data + inserter.check_data(dst, cross_slots_cmd=cross_slots_cmd) + p.ASSERT_EQ(src.dbsize(), dst.dbsize()) + +def test_base_file(dst): + #creat manifest file + current_directory = p.get_case_context().dir + "_own" + create_aof_dir(current_directory + "/appendonlydir") + manifest_filepath = current_directory + "/appendonlydir/appendonly.aof.manifest" + commands = [] + commands += "file appendonly.aof.1.base.aof seq 1 type b\n" + append_to_file(manifest_filepath, commands) + + #creat aof file + base_file_path = current_directory + "/appendonlydir/appendonly.aof.1.base.aof" + commands = [] + commands += format_command("set", "k1", "v1") + commands += format_command("set", "k2", "v2") + append_to_file(base_file_path, commands) + + #start shake redis + opts = h.ShakeOpts.create_aof_opts(f"{current_directory}{get_aof_file_relative_path()}", dst) + p.log(f"opts: {opts}") + h.Shake.run_once(opts) + + #check data + pip = dst.pipeline() + pip.get("k1") + pip.get("k2") + ret = pip.execute() + p.ASSERT_EQ(ret, [b"v1", b"v2"]) + p.ASSERT_EQ(dst.dbsize(), 2) + + +def test_error(src, dst): + #set aof + ret = src.do("CONFIG SET", "appendonly", "yes") + p.log(f"aof_ret: {ret}") + cross_slots_cmd = not (src.is_cluster() or dst.is_cluster()) + inserter = h.DataInserter() + inserter.add_data(src, cross_slots_cmd=cross_slots_cmd) + p.ASSERT_TRUE(src.do("save")) + #destroy file + file_path = src.dir + get_aof_file_relative_path() + with open(file_path, "r+") as file: + destroy_data = "xxxxs" + file.seek(0, 0) + file.write(destroy_data) + + + opts = h.ShakeOpts.create_aof_opts(f"{src.dir}/appendonlydir/appendonly.aof.manifest", dst) + p.log(f"opts: {opts}") + h.Shake.run_once(opts) + + #cant restore + p.ASSERT_EQ( dst.dbsize(), 0) + + + +def test_rm_file(src, dst): + cross_slots_cmd = not (src.is_cluster() or dst.is_cluster()) + inserter = h.DataInserter() + inserter.add_data(src, cross_slots_cmd=cross_slots_cmd) + p.ASSERT_TRUE(src.do("save")) + #rm file + file_path = src.dir + "/appendonlydir/appendonly.aof.1.base.rdb" + os.remove(file_path) + opts = h.ShakeOpts.create_aof_opts(f"{src.dir}{get_aof_file_relative_path()}", dst) + h.Shake.run_once(opts) + #cant restore + p.ASSERT_EQ(dst.dbsize(), 0) + +def test_history_file(src, dst): + + cross_slots_cmd = not (src.is_cluster() or dst.is_cluster()) + inserter = h.DataInserter() + for i in range(1000): + inserter.add_data(src, cross_slots_cmd=cross_slots_cmd) + p.ASSERT_TRUE(src.do("BGREWRITEAOF")) + + opts = h.ShakeOpts.create_aof_opts(f"{src.dir}{get_aof_file_relative_path()}", dst) + h.Shake.run_once(opts) + # check data + inserter.check_data(dst, cross_slots_cmd=cross_slots_cmd) + p.ASSERT_EQ(src.dbsize(), dst.dbsize()) + + +def test_base_file_timestamp(dst): # base file play back all + #creat manifest file + current_directory = p.get_case_context().dir + "_own" + create_aof_dir(current_directory) + manifest_filepath = current_directory + "/appendonlydir/appendonly.aof.manifest" + commands = [] + commands += "file appendonly.aof.1.base.aof seq 1 type b\n" + append_to_file(manifest_filepath, commands) + + #creat aof file + base_file_path = current_directory + "/appendonlydir/appendonly.aof.1.base.aof" + commands = [] + commands += "#TS1233\r\n" + commands += format_command("set", "k1", "v1") + commands += "#TS1234\r\n" + commands += format_command("set", "k2", "v2") + commands += "#TS1235\r\n" + commands += format_command("set", "k3", "v3") + append_to_file(base_file_path, commands) + + #start shake redis + opts = h.ShakeOpts.create_aof_opts(f"{current_directory}{get_aof_file_relative_path()}", dst, 1234) + p.log(f"opts: {opts}") + h.Shake.run_once(opts) + + #check data + pip = dst.pipeline() + pip.get("k1") + pip.get("k2") + pip.get("k3") + ret = pip.execute() + p.ASSERT_EQ(ret, [b"v1",b"v2",b"v3",]) + p.ASSERT_EQ(dst.dbsize(), 3) + +def test_base_and_incr_timestamp(dst): + + #creat manifest file + current_directory = p.get_case_context().dir + "_own" + create_aof_dir(current_directory + "/appendonlydir") + manifest_filepath = current_directory + "/appendonlydir/appendonly.aof.manifest" + commands = [] + commands += "file appendonly.aof.1.base.aof seq 1 type b\n" + commands += "file appendonly.aof.1.incr.aof seq 1 type i\n" + commands += "file appendonly.aof.2.incr.aof seq 2 type i\n" + append_to_file(manifest_filepath, commands) + + #creat aof base file + base_file_path = current_directory + "/appendonlydir/appendonly.aof.1.base.aof" + commands = [] + commands += format_command("set", "k1", "v1") + append_to_file(base_file_path, commands) + + commands = [] + #create aof incr file + incr1_file_path = current_directory + "/appendonlydir/appendonly.aof.1.incr.aof" + commands += "#TS1233\r\n" + commands += format_command("set", "k2", "v2") + append_to_file(incr1_file_path , commands) + + commands = [] + incr2_file_path = current_directory + "/appendonlydir/appendonly.aof.2.incr.aof" + commands += "#TS1235\r\n" + commands += format_command("set", "k3", "v3") + append_to_file(incr2_file_path , commands) + #start shake redis + opts = h.ShakeOpts.create_aof_opts(f"{current_directory}{get_aof_file_relative_path()}", dst, 1234) + p.log(f"opts: {opts}") + h.Shake.run_once(opts) + + #check data + pip = dst.pipeline() + pip.get("k1") + pip.get("k2") + ret = pip.execute() + p.ASSERT_EQ(ret, [b"v1",b"v2"]) + p.ASSERT_EQ(dst.dbsize(), 2) + + + + +@p.subcase() +def aof_to_standalone(): + if h.REDIS_SERVER_VERSION < 7.0: + return + src = h.Redis() + #set aof + ret = src.do("CONFIG SET", "appendonly", "yes") + p.log(f"aof_ret: {ret}") + + ret = src.do("CONFIG SET", "aof-timestamp-enabled", "yes") + p.log(f"aof_ret: {ret}") + dst = h.Redis() + test(src, dst) +@p.subcase() +def aof_to_standalone_base_file(): + if h.REDIS_SERVER_VERSION < 7.0: + return + dst = h.Redis() + test_base_file(dst) + + +@p.subcase() +def aof_to_standalone_rm_file(): + if h.REDIS_SERVER_VERSION < 7.0: + return + src = h.Redis() + #set aof + ret = src.do("CONFIG SET", "appendonly", "yes") + dst = h.Redis() + test_rm_file(src, dst) + +@p.subcase() +def aof_to_standalone_error(): + if h.REDIS_SERVER_VERSION < 7.0: + return + src = h.Redis() + #set aof + ret = src.do("CONFIG SET", "appendonly", "yes") + dst = h.Redis() + test_error(src, dst) + +@p.subcase() +def aof_to_cluster(): + if h.REDIS_SERVER_VERSION < 7.0: + return + src = h.Redis() + #set aof + ret = src.do("CONFIG SET", "appendonly", "yes") + p.log(f"aof_ret: {ret}") + dst = h.Cluster() + test(src, dst) + +@p.subcase() +def aof_to_standalone_single(): + if h.REDIS_SERVER_VERSION >= 7.0: + return + src = h.Redis() + #set preamble no + ret = src.do("CONFIG SET", "aof-use-rdb-preamble", "no") + p.log(f"aof_ret: {ret}") + #set aof + ret = src.do("CONFIG SET", "appendonly", "yes") + p.log(f"aof_ret: {ret}") + dst = h.Redis() + test(src, dst) + +@p.subcase() +def aof_to_standalone_timestamp(): + if h.REDIS_SERVER_VERSION < 7.0: + return + dst = h.Redis() + + ret = dst.do("FLUSHALL") + test_base_file_timestamp(dst) + ret = dst.do("FLUSHALL") + test_base_and_incr_timestamp(dst) + ret = dst.do("FLUSHALL") + +def aof_to_standalone_history_file(): + if h.REDIS_SERVER_VERSION < 7.0: + return + src = h.Redis() + #set aof + #set hist + ret = src.do("CONFIG SET", "aof-disable-auto-gc", "yes") + p.log(f"aof_ret: {ret}") + + ret = src.do("CONFIG SET", "appendonly", "yes") + p.log(f"aof_ret: {ret}") + + ret = src.do("CONFIG SET", "aof-timestamp-enabled", "yes") + p.log(f"aof_ret: {ret}") + + dst = h.Redis() + test_history_file(src, dst) + +@p.case(tags=["sync"]) +def main(): + aof_to_standalone() # base + incr aof-multi + aof_to_standalone_base_file() # base file aof-multi + aof_to_standalone_single() #single aof + aof_to_standalone_error() # error aof file + aof_to_standalone_rm_file() # rm aof file + aof_to_standalone_history_file() # history + incr aof-multi + aof_to_cluster() #test cluster + aof_to_standalone_timestamp() #set timestamp aof-multi + + +if __name__ == '__main__': + main() diff --git a/tests/helpers/constant.py b/tests/helpers/constant.py index c6f62792..3d529c45 100644 --- a/tests/helpers/constant.py +++ b/tests/helpers/constant.py @@ -7,8 +7,6 @@ BASE_PATH = f"{Path(__file__).parent.parent.parent.absolute()}" # project path PATH_REDIS_SHAKE = f"{BASE_PATH}/bin/redis-shake" PATH_REDIS_SERVER = shutil.which('redis-server') - -# REDIS_SERVER_VERSION output = subprocess.check_output(f"{PATH_REDIS_SERVER} --version", shell=True) output_str = output.decode("utf-8") REDIS_SERVER_VERSION = float(output_str.split("=")[1].split(" ")[0][:3]) diff --git a/tests/helpers/redis.py b/tests/helpers/redis.py index aa83bee5..17e817b5 100644 --- a/tests/helpers/redis.py +++ b/tests/helpers/redis.py @@ -17,7 +17,6 @@ def __init__(self, args=None): self.port = get_free_port() self.dir = f"{self.case_ctx.dir}/redis_{self.port}" args.extend(["--port", str(self.port)]) - if REDIS_SERVER_MODULES_ENABLED: args.extend(["--loadmodule", "tairstring_module.so"]) args.extend(["--loadmodule", "tairhash_module.so"]) diff --git a/tests/helpers/shake.py b/tests/helpers/shake.py index 34c7c525..7664b3f2 100644 --- a/tests/helpers/shake.py +++ b/tests/helpers/shake.py @@ -50,7 +50,17 @@ def create_rdb_opts(rdb_path: str, dts: Redis) -> typing.Dict: } } return d - + + @staticmethod + def create_aof_opts(aof_path: str, dts: Redis, timestamp: int = 0) -> typing.Dict: + d = { + "aof_reader": {"filepath": aof_path, "timestamp": timestamp}, + "redis_writer": { + "cluster": dts.is_cluster(), + "address": dts.get_address() + } + } + return d class Shake: def __init__(self, opts: typing.Dict):