From 51ae40f16939249eaac75201c24636d1716431af Mon Sep 17 00:00:00 2001 From: bug-superman <2545080316@qq.com> Date: Mon, 19 Jun 2023 00:49:31 +0800 Subject: [PATCH 1/6] put on aof airs --- cmd/redis-shake/main.go | 2 +- internal/aof/aof.go | 19 +++++++++++++++++++ internal/aof/aof_check.go | 1 + internal/config/config.go | 2 ++ internal/reader/aof.reader.go | 18 ++++++++++++++++++ 5 files changed, 41 insertions(+), 1 deletion(-) create mode 100644 internal/aof/aof.go create mode 100644 internal/aof/aof_check.go create mode 100644 internal/reader/aof.reader.go diff --git a/cmd/redis-shake/main.go b/cmd/redis-shake/main.go index 1f16b5a1..b21e2da4 100644 --- a/cmd/redis-shake/main.go +++ b/cmd/redis-shake/main.go @@ -82,7 +82,7 @@ func main() { var theReader reader.Reader if config.Config.Type == "sync" { theReader = reader.NewPSyncReader(source.Address, source.Username, source.Password, source.IsTLS, source.ElastiCachePSync) - } else if config.Config.Type == "restore" { + } else if config.Config.Type == "restore" { // TODO: new aof reader theReader = reader.NewRDBReader(source.RDBFilePath) } else if config.Config.Type == "scan" { theReader = reader.NewScanReader(source.Address, source.Username, source.Password, source.IsTLS) diff --git a/internal/aof/aof.go b/internal/aof/aof.go new file mode 100644 index 00000000..18b4cca1 --- /dev/null +++ b/internal/aof/aof.go @@ -0,0 +1,19 @@ +package aof + +import "github.com/alibaba/RedisShake/internal/entry" + +// TODO: 待填充 +type Loader struct { +} + +func NewLoader(filPath string, ch chan *entry.Entry) *Loader { + ld := new(Loader) + return ld +} + +func (ld *Loader) ParseRDB() int { + // 加载aof目录 + // 进行check_aof， aof + // 加载清单 + return 0 +} diff --git a/internal/aof/aof_check.go b/internal/aof/aof_check.go new file mode 100644 index 00000000..7b20395f --- /dev/null +++ b/internal/aof/aof_check.go @@ -0,0 +1 @@ +package aof diff --git a/internal/config/config.go b/internal/config/config.go index eb303abc..e4b4fbea 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -20,6 +20,7 @@ type tomlSource struct { // restore mode RDBFilePath string `toml:"rdb_file_path"` + AOFFilePath string `toml:"aof_file_path"` // add the aof path } type tomlTarget struct { @@ -74,6 +75,7 @@ func init() { Config.Source.ElastiCachePSync = "" // restore Config.Source.RDBFilePath = "" + Config.Source.AOFFilePath = "" // target Config.Target.Type = "standalone" diff --git a/internal/reader/aof.reader.go b/internal/reader/aof.reader.go new file mode 100644 index 00000000..d1951785 --- /dev/null +++ b/internal/reader/aof.reader.go @@ -0,0 +1,18 @@ +package reader + +// this file references rdb_reader.go + +import "github.com/alibaba/RedisShake/internal/entry" + +type aofReader struct { + path string + ch chan *entry.Entry +} + +func NewAOFReader(path string) Reader { + return nil +} + +func (r *aofReader) StartRead() chan *entry.Entry { + return nil +} From 389c602a20115cb9faba29b9fc5822ac65ec5c1a Mon Sep 17 00:00:00 2001 From: bug-superman <2545080316@qq.com> Date: Sat, 24 Jun 2023 22:34:48 +0800 Subject: [PATCH 2/6] add the aof mainifest --- cmd/redis-shake/main.go | 7 ++++++- internal/aof/aof.go | 39 ++++++++++++++++++++++++++++++++--- internal/aof/aof_check.go | 5 +++++ internal/reader/aof.reader.go | 9 +++++++- 4 files changed, 55 insertions(+), 5 deletions(-) diff --git a/cmd/redis-shake/main.go b/cmd/redis-shake/main.go index b21e2da4..b80dfe31 100644 --- a/cmd/redis-shake/main.go +++ b/cmd/redis-shake/main.go @@ -83,7 +83,12 @@ func main() { if config.Config.Type == "sync" { theReader = reader.NewPSyncReader(source.Address, source.Username, source.Password, source.IsTLS, source.ElastiCachePSync) } else if config.Config.Type == "restore" { // TODO: new aof reader - theReader = reader.NewRDBReader(source.RDBFilePath) + if source.RDBFilePath != "" { + theReader = reader.NewRDBReader(source.RDBFilePath) + } else { + theReader = reader.NewAOFReader(source.AOFFilePath) + } + } else if config.Config.Type == "scan" { theReader = reader.NewScanReader(source.Address, source.Username, source.Password, source.IsTLS) } else { diff --git a/internal/aof/aof.go b/internal/aof/aof.go index 18b4cca1..81f91785 100644 --- a/internal/aof/aof.go +++ b/internal/aof/aof.go @@ -1,19 +1,52 @@ package aof -import "github.com/alibaba/RedisShake/internal/entry" +import ( + "container/list" + "github.com/alibaba/RedisShake/internal/entry" +) -// TODO: 待填充 +type AofFileType string + +const ( + AofFileTypeBase AofFileType = "b" /* Base file */ + AofFileTypeHist AofFileType = "h" /* History file */ + AofFileTypeIncr AofFileType = "i" /* INCR file */ +) + +/* AOF manifest definition */ +type aofInfo struct { + fileName string + fileSeq int64 + aofFileType AofFileType +} + +type aofManifest struct { + baseAofInfo *aofInfo + incrAofList *list.List + historyList *list.List + currBaseFileSeq int64 + currIncrFIleSeq int64 + dirty int64 +} + +// TODO: 待填充完整loader 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 } +// TODO：完成checAofMain后写单测进行测试 func (ld *Loader) ParseRDB() int { // 加载aof目录 // 进行check_aof， aof - // 加载清单 + checkAofMain(ld.filPath) + // TODO：执行加载 return 0 } diff --git a/internal/aof/aof_check.go b/internal/aof/aof_check.go index 7b20395f..bf7d7b41 100644 --- a/internal/aof/aof_check.go +++ b/internal/aof/aof_check.go @@ -1 +1,6 @@ package aof + +// TODO：参考飞书 +func checkAofMain(path string) { + +} diff --git a/internal/reader/aof.reader.go b/internal/reader/aof.reader.go index d1951785..c4fedcf4 100644 --- a/internal/reader/aof.reader.go +++ b/internal/reader/aof.reader.go @@ -2,17 +2,24 @@ package reader // this file references rdb_reader.go -import "github.com/alibaba/RedisShake/internal/entry" +import ( + "github.com/alibaba/RedisShake/internal/entry" +) type aofReader struct { path string ch chan *entry.Entry } +// TODO:待完善参考rdb reader func NewAOFReader(path string) Reader { + return nil } func (r *aofReader) StartRead() chan *entry.Entry { + //调用 aof中的函数 + // aof.NewLoader() + // aof.ParseRDB() return nil } From 3a869c550a8cefa8ee52aa446ac647c68255721b Mon Sep 17 00:00:00 2001 From: bug-superman <2545080316@qq.com> Date: Sat, 24 Jun 2023 22:36:09 +0800 Subject: [PATCH 3/6] fix the file name --- internal/reader/{aof.reader.go => aof_reader.go} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename internal/reader/{aof.reader.go => aof_reader.go} (100%) diff --git a/internal/reader/aof.reader.go b/internal/reader/aof_reader.go similarity index 100% rename from internal/reader/aof.reader.go rename to internal/reader/aof_reader.go From c1f93871e51bf8f5bedd58464d1aab5a18e94bc2 Mon Sep 17 00:00:00 2001 From: bug-superman <2545080316@qq.com> Date: Sat, 24 Jun 2023 22:49:06 +0800 Subject: [PATCH 4/6] add the aof check --- cmd/redis-shake/main.go | 2 +- internal/aof/aof.go | 12 ++++++------ internal/aof/aof_check.go | 28 ++++++++++++++++++++++++++-- 3 files changed, 33 insertions(+), 9 deletions(-) diff --git a/cmd/redis-shake/main.go b/cmd/redis-shake/main.go index b80dfe31..70f5fad7 100644 --- a/cmd/redis-shake/main.go +++ b/cmd/redis-shake/main.go @@ -86,7 +86,7 @@ func main() { if source.RDBFilePath != "" { theReader = reader.NewRDBReader(source.RDBFilePath) } else { - theReader = reader.NewAOFReader(source.AOFFilePath) + theReader = reader.NewAOFReader(source.AOFFilePath) // 如果是mp-aof 用户传入 manifest文件的地址 ，其他的传递aof的地址 } } else if config.Config.Type == "scan" { diff --git a/internal/aof/aof.go b/internal/aof/aof.go index 81f91785..13f4b202 100644 --- a/internal/aof/aof.go +++ b/internal/aof/aof.go @@ -5,19 +5,19 @@ import ( "github.com/alibaba/RedisShake/internal/entry" ) -type AofFileType string +type AofManifestFileType string const ( - AofFileTypeBase AofFileType = "b" /* Base file */ - AofFileTypeHist AofFileType = "h" /* History file */ - AofFileTypeIncr AofFileType = "i" /* INCR file */ + AofManifestFileTypeBase AofManifestFileType = "b" /* Base file */ + AofManifestTypeHist AofManifestFileType = "h" /* History file */ + AofManifestTypeIncr AofManifestFileType = "i" /* INCR file */ ) /* AOF manifest definition */ type aofInfo struct { fileName string fileSeq int64 - aofFileType AofFileType + aofFileType AofManifestFileType } type aofManifest struct { @@ -46,7 +46,7 @@ func NewLoader(filPath string, ch chan *entry.Entry) *Loader { func (ld *Loader) ParseRDB() int { // 加载aof目录 // 进行check_aof， aof - checkAofMain(ld.filPath) + CheckAofMain(ld.filPath) // TODO：执行加载 return 0 } diff --git a/internal/aof/aof_check.go b/internal/aof/aof_check.go index bf7d7b41..9b9f645b 100644 --- a/internal/aof/aof_check.go +++ b/internal/aof/aof_check.go @@ -1,6 +1,30 @@ package aof -// TODO：参考飞书 -func checkAofMain(path string) { +type AofFileType string +const ( + aofResp AofFileType = "AOF_RESP" + aofRdbPreamble AofFileType = "AOF_RDB_PREAMBLE" + aofMultiPart AofFileType = "AOF_MULTI_PART" +) + +// check 里面的主函数 +func CheckAofMain(aofFilePath string) (checkResult bool, fileType AofFileType, err error) { + /* + getAofType := getInputAofFileTye(filePath) // 获取aof文件的类型 + switch getAOFType { + case AOF_MULTI_PART: + checkResult, err = checkMultiPartAof(dirpath, filepath, fix) + return checkResult, aofMultiPart, nil + case AOF_RESP: + checkResult, err := checkOldStyleAof(filepath) + return checkResult, aofResp, nil + case AOF_RDB_PREAMBLE: + checkResult, err := checkOldStyleAof(filepath) + return checkResult, aofRdbPreamble, nil + } + return result, err + */ + //TODO: mock result + return true, aofMultiPart, nil } From a405cd59a8b4d583aaf182e507a015ff41b400d4 Mon Sep 17 00:00:00 2001 From: hwy1314 <1426317404@qq.com> Date: Sat, 22 Jul 2023 14:23:07 +0800 Subject: [PATCH 5/6] 7/22 --- internal/aof/aof.go | 304 +++++++++++++++++++- internal/aof/aof_check.go | 575 +++++++++++++++++++++++++++++++++++++- 2 files changed, 867 insertions(+), 12 deletions(-) diff --git a/internal/aof/aof.go b/internal/aof/aof.go index 13f4b202..0f1eb09f 100644 --- a/internal/aof/aof.go +++ b/internal/aof/aof.go @@ -1,29 +1,283 @@ package aof import ( - "container/list" + "bytes" + "encoding/binary" + "log" + "strconv" + "unicode" + "github.com/alibaba/RedisShake/internal/entry" ) -type AofManifestFileType string - const ( - AofManifestFileTypeBase AofManifestFileType = "b" /* Base file */ - AofManifestTypeHist AofManifestFileType = "h" /* History file */ - AofManifestTypeIncr AofManifestFileType = "i" /* INCR file */ + AofManifestFileTypeBase = "b" /* Base file */ + AofManifestTypeHist = "h" /* History file */ + AofManifestTypeIncr = "i" /* INCR file */ ) /* AOF manifest definition */ type aofInfo struct { fileName string fileSeq int64 - aofFileType AofManifestFileType + aofFileType string +} + +func IntToBytes(n int) []byte { + data := int64(n) + bytebuf := bytes.NewBuffer([]byte{}) + binary.Write(bytebuf, binary.BigEndian, data) + return bytebuf.Bytes() +} + +func aofInfoCreate() *aofInfo { + return new(aofInfo) +} + +func StringNeedsRepr(s string) int { + len := len(s) + point := 0 + for len > 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 + } + len-- + point++ + } + + return 0 +} + +/*如果字符串中包含要转义的字符，则返回一 + *通过sdscatrer（），否则为零。 + * + *通常，这应该用于以某种方式帮助保护聚合字符串 + *与sdsspliargs（）兼容。因此，空间也将 + *被视为需要逃跑。 + */ +/*func stringTrim(s string,cset string)string{ + slen:=len(s) + sp:=0 + ep:=slen-1 + + for sp<=ep&&strings.ContainsRune(cset,rune(s[sp])){ + sp++ + } + + for ep>sp&&strings.ContainsRune(cset,rune(s[sp])){ + ep-- + } + trimmed:=s[sp:ep+1] + trimmedlen:=len(trimmed) + if sp<0||ep= 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': + done = true + break + case '"': + inq = true + break + case '\'': + insq = true + break + default: + current += string(p[i]) + break + } + } + if i < lens { + i++ + } + if i == lens { + done = true + } + } + + vector = append(vector, current) + argc++ + current = "" + + } else { + return vector, argc + } + + } +} + +func stringcatlen(s string, t []byte, lent int) string { + curlen := len(s) + + if curlen == 0 { + return "" + } + + buf := make([]byte, curlen+lent) + + copy(buf[:curlen], []byte(s)) + copy(buf[curlen:], t) + //buf[curlen+lent] = 0 ///0 + return string(buf) +} + +func aofInfoDup(orig *aofInfo) *aofInfo { + if orig == nil { + log.Fatal("Assertion failed: orig != nil") + } + ai := aofInfoCreate() + ai.fileName = orig.fileName + ai.fileSeq = orig.fileSeq + ai.aofFileType = orig.aofFileType + return ai +} + +type lists struct { + head, tail *listNode + len uint64 +} + +type listNode struct { + prev *listNode + next *listNode + value interface{} +} + +func listCreate() *lists { + lists := &lists{} + lists.head = nil + lists.tail = nil + lists.len = 0 + return lists } type aofManifest struct { baseAofInfo *aofInfo - incrAofList *list.List - historyList *list.List + incrAofList *lists + historyList *lists currBaseFileSeq int64 currIncrFIleSeq int64 dirty int64 @@ -35,6 +289,30 @@ type Loader struct { ch chan *entry.Entry } +func listAddNodeTail(lists *lists, value interface{}) *lists { + node := &listNode{ + value: value, + prev: nil, + next: nil, + } + listLinkNodeTail(lists, node) + return lists +} + +func listLinkNodeTail(lists *lists, node *listNode) { + if lists.len == 0 { + lists.head = node + lists.tail = node + node.prev = nil + node.next = nil + } else { + node.prev = lists.tail + node.next = nil + lists.tail.next = node + lists.tail = node + } + lists.len++ +} func NewLoader(filPath string, ch chan *entry.Entry) *Loader { ld := new(Loader) ld.ch = ch @@ -50,3 +328,11 @@ func (ld *Loader) ParseRDB() int { // TODO：执行加载 return 0 } + +func aofManifestcreate() *aofManifest { + am := &aofManifest{ + incrAofList: listCreate(), + historyList: listCreate(), + } + return am +} diff --git a/internal/aof/aof_check.go b/internal/aof/aof_check.go index 9b9f645b..2a86be5e 100644 --- a/internal/aof/aof_check.go +++ b/internal/aof/aof_check.go @@ -1,11 +1,39 @@ package aof +import ( + "bufio" + "fmt" + "io" + "log" + "math" + "os" + "path" + "strconv" + "strings" + //"time" +) + type AofFileType string +var errors [1044]byte +var line int64 = 1 +var epos int64 + const ( - aofResp AofFileType = "AOF_RESP" - aofRdbPreamble AofFileType = "AOF_RDB_PREAMBLE" - aofMultiPart AofFileType = "AOF_MULTI_PART" + aofResp AofFileType = "AOF_RESP" + aofRdbPreamble AofFileType = "AOF_RDB_PREAMBLE" + aofMultiPart AofFileType = "AOF_MULTI_PART" + rdbCheckMode = 1 + MANIFEST_MAX_LINE = 1024 + AOF_CHECK_OK = 0 + AOF_CHECK_EMPTY = 1 + AOF_CHECK_TRUNCATED = 2 + AOF_CHECK_TIMESTAMP_TRUNCATED = 3 + toTimestamp = 0 + AOF_MANIFEST_KEY_FILE_NAME = "file" + AOF_MANIFEST_KEY_FILE_SEQ = "seq" + AOF_MANIFEST_KEY_FILE_TYPE = "type" + AOF_ANNOTATION_LINE_MAX_LEN = 1024 ) // check 里面的主函数 @@ -28,3 +56,544 @@ func CheckAofMain(aofFilePath string) (checkResult bool, fileType AofFileType, e //TODO: mock result return true, aofMultiPart, nil } + +func getInputAofFileType(aofFilepath string) AofFileType { + if filelsManifest(aofFilepath) { + return "AOF_MULTI_PART" + } else if fileIsRDB(aofFilepath) { + return "AOF_RDB_PREAMBLE" + } else { + return "AOF_RESP" + } +} + +func filelsManifest(aofFilepath string) bool { + var is_manifest bool = false + fp, err := os.Open(aofFilepath) + if err != nil { + fmt.Printf("Cannot open file %s:%s\n", aofFilepath, err.Error()) + os.Exit(1) + } + sb, err := os.Stat(aofFilepath) + if err != nil { + fmt.Printf("cannot stat file: %s\n", aofFilepath) + os.Exit(1) + } + size := sb.Size() + if size == 0 { + fp.Close() + return false + } + reader := bufio.NewReader(fp) + for { + lines, err := reader.ReadString('\n') + if err != nil { + if err == io.EOF { + break + } else { + fmt.Printf("cannot read file: %s\n", aofFilepath) + os.Exit(1) + } + } + if lines[0] == '#' { + continue + } else if lines[:4] == "file" { + is_manifest = true + } + } + fp.Close() + return is_manifest +} + +func fileIsRDB(aofFilepath string) bool { + fp, err := os.Open(aofFilepath) + if err != nil { + fmt.Printf("Cannot open file %s:%s\n", aofFilepath, err.Error()) + os.Exit(1) + } + sb, err := os.Stat(aofFilepath) + if err != nil { + fmt.Printf("cannot stat file: %s\n", aofFilepath) + os.Exit(1) + } + size := sb.Size() + if size == 0 { + fp.Close() + return false + } + if size >= 8 { + sig := make([]byte, 5) + _, err := fp.Read(sig) + if err == nil && string(sig) == "REDIS" { + fp.Close() + return true + } + } + fp.Close() + return false +} + +func printAofStyle(ret int, aofFileName string, aofType string) { + switch ret { + case AOF_CHECK_OK: + fmt.Printf("%s %s is valid\n", aofType, aofFileName) + case AOF_CHECK_EMPTY: + fmt.Printf("%s %s is empty\n", aofType, aofFileName) + case AOF_CHECK_TIMESTAMP_TRUNCATED: + fmt.Printf("Successfully truncated AOF %s to timestamp %d\n", aofFileName, toTimestamp) + case AOF_CHECK_TRUNCATED: + fmt.Printf("Successfully truncated AOF %s\n", aofFileName) + } + +} + +func makePath(paths string, filename string) string { + return path.Join(paths, filename) +} + +func pathIsBaseName(path string) bool { + return strings.IndexByte(path, '/') == -1 && strings.IndexByte(path, '\\') == -1 +} + +func readArgc(fp *os.File, target *int64) int { + return readLong(fp, '*', target) +} + +func readString(fp *os.File, target *string) int { + var len int64 + *target = "" + if readLong(fp, '$', &len) == 0 { + return 0 + } + + if len < 0 || len > math.MaxInt64-2 { + fmt.Printf("Expected to read string of %d bytes, which is not in the suitable range\n", len) + return 0 + } + + // Increase length to also consume \r\n + len += 2 + data := make([]byte, len) + if readBytes(fp, &data, len) == 0 { + return 0 + } + + if consumeNewline(data[len-2:]) == 0 { + return 0 + } + + *target = string(data[:len-2]) + return 1 +} + +func readBytes(fp *os.File, target *[]byte, length int64) int { + var real int64 + epos, _ = fp.Seek(0, io.SeekCurrent) + n, err := fp.Read(*target) + real = int64(n) + if err != nil || real != length { + fmt.Printf("Expected to read %d bytes, got %d bytes\n", length, real) + return 0 + } + return 1 +} + +func consumeNewline(buf []byte) int { + if buf[0] != '\r' || buf[1] != '\n' { + fmt.Printf("Expected \\r\\n, got: %02x%02x", buf[0], buf[1]) + return 0 + } + line += 1 + return 1 +} + +func readLong(fp *os.File, prefix byte, target *int64) int { + buf := make([]byte, 128) + var err error + epos, err = fp.Seek(0, io.SeekCurrent) + if err != nil { + fmt.Printf("Failed to get current position, aborting...\n") + os.Exit(1) + } + reader := bufio.NewReader(fp) + if _, err := reader.ReadBytes('\n'); err != nil { + return 0 + } + buf, err = reader.ReadBytes('\n') + if err != nil { + fmt.Println("Failed to read line from file") + return 0 + } + if buf[0] != prefix { + fmt.Printf("Expected prefix '%c', got: '%c'\n", prefix, buf[0]) + return 0 + } + value, err := strconv.ParseInt(string(buf[1:len(buf)-1]), 10, 64) //去除了换行符*8* + if err != nil { + fmt.Println("Failed to parse long value") + return 0 + } + *target = value + line += 1 + return 1 + +} +func aofLoadManifestFromFile(am_filepath string) *aofManifest { + var maxseq int64 + am := aofManifestcreate() + fp, err := os.Open(am_filepath) + if err != nil { + log.Fatalf("Fatal error:can't open the AOF manifest %s for reading: %s", am_filepath, err) + } + var argv []string + var ai *aofInfo + var line string + linenum := 0 + reader := bufio.NewReader(fp) + for { + buf, err := reader.ReadString('\n') + if err != nil { + if err == io.EOF { + if linenum == 0 { + log.Fatalf("Found an empty AOF manifest") + } else { + break + } + + } else { + log.Fatalf("Read AOF manifest failed") + } + } + linenum++ + if buf[0] == '#' { + continue + } + if !strings.Contains(buf, "\n") { + log.Fatalf("The AOF manifest file contains too long line") + } + line = strings.Trim(buf, " \t\r\n") + if len(line) == 0 { + log.Fatalf("Invalid AOF manifest file format") + } + argc := 0 + argv, argc = splitArgs(line) + + if argc < 6 || argc%2 != 0 { + log.Fatalf("Invalid AOF manifest file format") + } + ai = aofInfoCreate() + for i := 0; i < argc; i += 2 { + if strings.EqualFold(argv[i], AOF_MANIFEST_KEY_FILE_NAME) { + ai.fileName = string(argv[i+1]) + if !pathIsBaseName(string(ai.fileName)) { + log.Fatalf("File can't be a path, just a filename") + } + } else if strings.EqualFold(argv[i], AOF_MANIFEST_KEY_FILE_SEQ) { + ai.fileSeq, _ = strconv.ParseInt(argv[i+1], 10, 64) + } else if strings.EqualFold(argv[i], AOF_MANIFEST_KEY_FILE_TYPE) { + ai.aofFileType = string(argv[i+1][0]) + } + } + if ai.fileName == "" || ai.fileSeq == 0 || ai.aofFileType == "" { + log.Fatalf("Invalid AOF manifest file format") + } + //==nil + if ai.aofFileType == AofManifestFileTypeBase { + if am.baseAofInfo != nil { + log.Fatalf("Found duplicate base file information") + } + am.baseAofInfo = ai + am.currBaseFileSeq = ai.fileSeq + } else if ai.aofFileType == AofManifestTypeHist { + am.historyList = listAddNodeTail(am.historyList, ai) + } else if ai.aofFileType == AofManifestTypeIncr { + if ai.fileSeq <= maxseq { + log.Fatalf("Found a non-monotonic sequence number") + } + am.incrAofList = listAddNodeTail(am.historyList, ai) + am.currIncrFIleSeq = ai.fileSeq + maxseq = ai.fileSeq + } else { + log.Fatalf("Unknown AOF file type") + } + line = " " + ai = nil + } + fp.Close() + return am +} + +func processRESP(fp *os.File, filename string, outMulti *int) int { + var argc int64 + var str string + + if readArgc(fp, &argc) == 0 { + return 0 + } + + for i := int64(0); i < argc; i++ { + if readString(fp, &str) == 0 { + return 0 + } + if i == 0 { + if strings.EqualFold(str, "multi") { + (*outMulti)++ + if (*outMulti) > 0 { + err := fmt.Errorf("Unexpected MULTI in AOF %s", filename) + fmt.Println(err.Error()) + return 0 + } + } else if strings.EqualFold(str, "exec") { + (*outMulti)-- + if (*outMulti) != 0 { + err := fmt.Errorf("Unexpected EXEC in AOF %s", filename) + fmt.Println(err.Error()) + return 0 + } + } + } + } + + return 1 +} + +// 截断可能有问题 +func processAnnotations(fp *os.File, filename string, lastFile bool) int { + + epos, err := fp.Seek(0, io.SeekCurrent) + if err != nil { + fmt.Printf("Failed to get current position, aborting...\n") + os.Exit(1) + } + reader := bufio.NewReader(fp) + buf, _, err := reader.ReadLine() + if err != nil { + fmt.Printf("Failed to read annotations from AOF %s, aborting...\n", filename) + os.Exit(1) + } + + if toTimestamp != 0 && strings.HasPrefix(string(buf), "#TS:") { + var ts int64 + ts, err = strconv.ParseInt(strings.TrimPrefix(string(buf), "#TS:"), 10, 64) + if err != nil { + fmt.Println("Invalid timestamp annotation") + os.Exit(1) + } + if ts <= toTimestamp { + return 1 + } + if epos == 0 { + fmt.Printf("AOF %s has nothing before timestamp %ld, aborting...\n", filename, toTimestamp) + os.Exit(1) + } + if lastFile == false { + fmt.Printf("Failed to truncate AOF %s to timestamp %ld to offset %ld because it is not the last file.\n", filename, toTimestamp, epos) + fmt.Println("If you insist, please delete all files after this file according to the manifest file and delete the corresponding records in manifest file manually. Then re-run redis-check-aof.") + os.Exit(1) + } + // Truncate remaining AOF if exceeding 'toTimestamp' + if err := fp.Truncate(epos); err != nil { + fmt.Printf("Failed to truncate AOF %s to timestamp %ld\n", filename, toTimestamp) + os.Exit(1) + } else { + return 0 + } + } + return 1 +} + +/* +func checkRdbMain(argc int, argv *[]string, fp *os.File) int { + + if argc != 2 && fp == nil { + fmt.Fprintf(os.Stderr, "Usage: %s \n", argv[0]) + os.Exit(1) + } else if argv[1] == "-v" || argv[1] == "--version" { + version := checkRdbVersion() + fmt.Printf("redis-check-rdb %s\n", version) + os.Exit(0) + } + + tv := time.Now() + sec := int64(tv.Unix()) + usec := int64(tv.Nanosecond()) / 1000 + pid := int64(os.Getpid()) + seed := ((sec * 1000000) + usec) ^ pid + init_genrand64(seed) + rdbCheckMode = 1 + rdbCheckInfo("Checking RDB file %s", args[1]) + rdbCheckSetupSignals() + retval := redis_Check_RDB(args[1], fp) + if retval == 0 { + rdbCheckInfo("\\o/ RDB looks OK! \\o/") + rdbShowGenericInfo() + } + if fp != nil { + if retval == 0 { + return 0 + } else { + return 1 + } + } + os.Exit(retval) + } +*/ +func checkMultipartAof(dirpath string, manifestFilepath string, fix int) { + totalNum := 0 + aofNum := 0 + var ret int + am := aofLoadManifestFromFile(manifestFilepath) + if am.baseAofInfo != nil { + totalNum++ + } + if am.incrAofList != nil { + totalNum += int(am.incrAofList.len) + } + if am.baseAofInfo != nil { + aofFilename := am.baseAofInfo.fileName + aofFilepath := makePath(dirpath, aofFilename) + lastFile := (aofNum + 1) == totalNum + aofPreable := fileIsRDB(aofFilepath) + if aofPreable { + fmt.Printf("Start to check BASE AOF (RDB format).\n") + } else { + fmt.Printf("Start to check BASE AOF (AOF format).\n") + } + ret = checkSingleAof(aofFilename, aofFilepath, lastFile, fix, aofPreable) + printAofStyle(ret, aofFilename, "BASE AOF") + + } + if am.incrAofList.len != 0 { + fmt.Printf("start to check INCR INCR files.") + var ln *listNode + ln = am.incrAofList.head + for ln != nil { + ai := ln.value.(*aofInfo) + aofFilename := ai.fileName + aofFilepath := makePath(dirpath, aofFilename) + lastFile := (aofNum + 1) == totalNum + ret = checkSingleAof(aofFilename, aofFilepath, lastFile, fix, false) + printAofStyle(ret, aofFilename, "INCR AOF") + //stringfree(aofFilepath) + ln = ln.next + } + } + + //aofManifestFree(am) + fmt.Println("All AOF files and manifest are vaild") +} + +func checkOldStyleAof(aofFilepath string, fix int, preamble bool) { + fmt.Printf("Start checking Old-Style AOF\n") + var ret = checkSingleAof(aofFilepath, aofFilepath, true, fix, preamble) + printAofStyle(ret, aofFilepath, "AOF") + +} + +func checkSingleAof(aofFilename, aofFilepath string, lastFile bool, fix int, preamble bool) int { + var pos, diff int64 + multi := 0 + buf := make([]byte, 2) + + fp, err := os.OpenFile(aofFilepath, os.O_RDWR, 0666) + if err != nil { + log.Fatalf("Cannot open file %s:%s,aborting...\n", aofFilepath, err) + } + sb, err := fp.Stat() + if err != nil { + log.Fatalf("Cannot stat file: %s,aborting...\n", aofFilename) + } + size := sb.Size() + if size == 0 { + return AOF_CHECK_EMPTY + } + if preamble != false { + argv := []string{aofFilepath} + err := checkRdbMain(2, argv, fp) + if err == -1 { + log.Fatal("RDB preamble of AOF file is not sane, aborting.") + } else { + fmt.Println("RDB preamble is OK, proceeding with AOF tail...") + } + } + for { + if multi != 0 { + var err error + pos, err = fp.Seek(0, io.SeekCurrent) + if err != nil { + log.Fatalf(("Failed to seek in AOF %s: %s"), aofFilename, err) + } + } + if _, err := fp.Read(buf); err != nil { + if err == io.EOF { + break + } + log.Fatalf("Failed to read from AOF %s, aborting...\n", aofFilename) + } + if _, err := fp.Seek(-1, io.SeekCurrent); err != nil { + log.Fatalf("Failed to fseek in AOF %s: %s", aofFilename, err) + } + + switch buf[0] { + case '#': + if processAnnotations(fp, aofFilepath, lastFile) == 0 { + fp.Close() + return AOF_CHECK_TIMESTAMP_TRUNCATED + } + case '*': + if processRESP(fp, aofFilepath, &multi) == 0 { + break + } + default: + fmt.Printf("AOF %s format error\n", aofFilename) + break + } + } + if _, err := fp.Stat(); err == nil && multi == 1 && len(errors) == 0 { + if _, err := fp.Seek(0, io.SeekEnd); err != nil { + if _, err := fp.Seek(0, io.SeekCurrent); err == io.EOF { + fmt.Println("Reached EOF before reading EXEC for MULTI") + } + } + } + + if len(errors) > 0 { + fmt.Println(errors) + } + + diff = size - pos + if diff == 0 && toTimestamp == 1 { + fmt.Printf("Truncate nothing in AOF %s to timestamp %d\n", aofFilename, toTimestamp) + return AOF_CHECK_OK + } + fmt.Printf("AOF analyzed: filename=%s, size=%d, ok_up_to=%d, ok_up_to_line=%d, diff=%d\n", aofFilename, size, pos, line, diff) + if diff > 0 { + if fix == 1 { + if lastFile == false { + fmt.Printf("Failed to truncate AOF %s because it is not the last file\n", aofFilename) + os.Exit(1) + } + + fmt.Printf("this will shrink the AOF %s from %d bytes,with %d bytes,to %d bytes\n", aofFilename, size, diff, pos) + fmt.Print("Continue? [y/N]: ") + reader := bufio.NewReader(os.Stdin) + input, err := reader.ReadString('\n') + if err != nil || strings.ToLower(string(input[0])) != "y" { + fmt.Println("Aborting...") + os.Exit(1) + } + + if err := fp.Truncate(pos); err != nil { + fmt.Printf("Failed to truncate AOF %s\n", aofFilename) + os.Exit(1) + } else { + return AOF_CHECK_TRUNCATED + } + } else { + fmt.Printf("AOF %s is not valid.Use the --fix potion to try fixing it.\n", aofFilename) + os.Exit(1) + } + } + fp.Close() + return AOF_CHECK_OK +} From e091adc608491504163c6e753dd7eb782ff48ac6 Mon Sep 17 00:00:00 2001 From: hwy1314 <1426317404@qq.com> Date: Tue, 25 Jul 2023 10:01:20 +0800 Subject: [PATCH 6/6] 7/25 aof.go --- internal/aof/aof.go | 207 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 206 insertions(+), 1 deletion(-) diff --git a/internal/aof/aof.go b/internal/aof/aof.go index 0f1eb09f..bfa205b7 100644 --- a/internal/aof/aof.go +++ b/internal/aof/aof.go @@ -3,17 +3,21 @@ package aof import ( "bytes" "encoding/binary" + "fmt" "log" "strconv" "unicode" "github.com/alibaba/RedisShake/internal/entry" + "golang.org/x/tools/cmd/getgo/server" ) const ( AofManifestFileTypeBase = "b" /* Base file */ AofManifestTypeHist = "h" /* History file */ AofManifestTypeIncr = "i" /* INCR file */ + RDB_FORMAT_SUFFIX = ".rdb" + AOF_FORMAT_SUFFIX = ".aof" ) /* AOF manifest definition */ @@ -34,7 +38,13 @@ func aofInfoCreate() *aofInfo { return new(aofInfo) } -func StringNeedsRepr(s string) int { +var Aof_Info aofInfo = *aofInfoCreate() + +func (a *aofInfo) get_AofInfo_Name() string { + return a.fileName +} + +func stringNeedsRepr(s string) int { len := len(s) point := 0 for len > 0 { @@ -255,6 +265,11 @@ func aofInfoDup(orig *aofInfo) *aofInfo { return ai } +type listIter struct { + next *listNode + direction int +} + type lists struct { head, tail *listNode len uint64 @@ -273,6 +288,23 @@ func listCreate() *lists { lists.len = 0 return lists } +func listNext(iter *listIter) *listNode { + current := iter.next + + if current != nil { + if iter.direction == 0 { + iter.next = current.next + } else { + iter.next = current.prev + } + } + return current +} + +func (list *lists) ListsRewind(li *listIter) { + li.next = list.head + li.direction = 0 +} type aofManifest struct { baseAofInfo *aofInfo @@ -329,6 +361,52 @@ func (ld *Loader) ParseRDB() int { return 0 } +func stringcatprintf(s string, fmtStr string, args ...interface{}) string { + result := fmt.Sprintf(fmtStr, args...) + return s + result +} + +func stringcatrepr(s string, p string, length int) string { + s = s + string("\"") + 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 aofInfoFormat(buf string, ai *aofInfo) string { + var filenameRepr string + if stringNeedsRepr(ai.fileName) == 1 { + filenameRepr = stringcatrepr("", ai.fileName, len(ai.fileName)) + } + var ret string + if filenameRepr != "" { + ret = stringcatprintf(buf, "%s %s %s %d %s %c\n", AOF_MANIFEST_KEY_FILE_NAME, filenameRepr, AOF_MANIFEST_KEY_FILE_SEQ, ai.fileSeq, AOF_MANIFEST_KEY_FILE_TYPE, ai.aofFileType) + } else { + ret = stringcatprintf(buf, "%s %s %s %d %s %c\n", AOF_MANIFEST_KEY_FILE_NAME, ai.fileName, AOF_MANIFEST_KEY_FILE_SEQ, ai.fileSeq, AOF_MANIFEST_KEY_FILE_TYPE, ai.aofFileType) + } + return ret +} + func aofManifestcreate() *aofManifest { am := &aofManifest{ incrAofList: listCreate(), @@ -336,3 +414,130 @@ func aofManifestcreate() *aofManifest { } return am } + +func listDup(orig *lists) *lists { + var copy *lists + var iter listIter + var node *listNode + copy = listCreate() + if copy == nil { + return nil + } + copy.ListsRewind(&iter) + node = listNext(&iter) + var value interface{} + for node != nil { + value = node.value + } + + if listAddNodeTail(copy, value) == nil { + return nil + } + return copy +} + +func AOFManifestDup(orig *aofManifest) *aofManifest { + if orig == nil { + panic("orig is nil") + } + + am := &aofManifest{ + currBaseFileSeq: orig.currBaseFileSeq, + currIncrFIleSeq: orig.currIncrFIleSeq, + dirty: orig.dirty, + } + + if orig.baseAofInfo != nil { + am.baseAofInfo = aofInfoDup(orig.baseAofInfo) + } + + am.incrAofList = listDup(orig.incrAofList) + am.historyList = listDup(orig.historyList) + + if am.incrAofList == nil || am.historyList == nil { + panic("IncrAOFlist or HistoryAOFlist is nil") + } + return am +} + +func getAofManifestAsString(am *aofManifest) string { + if am == nil { + panic("am is nil") + } + var buf string + var ln *listNode + var li listIter + + if am.baseAofInfo != nil { + buf = aofInfoFormat(buf, am.baseAofInfo) + } + am.historyList.ListsRewind(&li) + ln = listNext(&li) + for ln != nil { + ai, ok := ln.value.(*aofInfo) + if ok { + buf = aofInfoFormat(buf, ai) + } + ln = listNext(&li) + } + + am.incrAofList.ListsRewind(&li) + ln = listNext(&li) + for ln != nil { + ai, ok := ln.value.(*aofInfo) + if ok { + buf = aofInfoFormat(buf, ai) + } + ln = listNext(&li) + } + + return buf + +} + +func getNewBaseFileNameAndMarkPreAsHistory(am *aofManifest) string { + if am == nil { + log.Fatal("aofManifest is nil") + } + if am.baseAofInfo != nil { + if am.baseAofInfo.aofFileType != AofManifestFileTypeBase { + log.Fatal("base_aof_info has invalid file_type") + } + am.baseAofInfo.aofFileType = AofManifestTypeHist + } + var formatSuffix string + if server.aofUseRdbPreamble { + formatSuffix = RDB_FORMAT_SUFFIX + } else { + formatSuffix = AOF_FORMAT_SUFFIX + } + ai := aofInfoCreate() + ai.fileName = fmt.Sprintf("%s.%d%s%d", Aof_Info.get_AofInfo_Name(), am.currBaseFileSeq+1, BASE_FILE_SUFFIX, formatSuffix) + ai.fileSeq = am.currBaseFileSeq + 1 + ai.aofFileType = AofManifestFileTypeBase + am.baseAofInfo = ai + am.dirty = 1 + return am.baseAofInfo.fileName +} + +// server 未处理 +func aofLoadManifestFromDisk() { + aof_manifest := aofManifestcreate() + if !difExists(server.aof_dirname) { + fmt.Printf("The AOF directory %s doesn't exist\n", server.AofDirname) + return + } + + am_name := getAofManifestFileName() + am_filepath := makePath(server.aof_dirname, am_name) + if !fileExist(am_filepath) { + fmt.Printf("The AOF directory %s doesn't exist\n", server.AofDirname) + return + } + + am := aofLoadManifestFromFile(am_filepath) + if am != nil { + aofManifestFreeAndUpdate(am) + } + +}