Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

meta/kv: support new backup #5365

Merged
merged 9 commits into from
Dec 13, 2024
Merged

meta/kv: support new backup #5365

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
9c79f07
meta/tkv: support new backup
jiefenghuang Dec 11, 2024
63c3325
fix test
jiefenghuang Dec 11, 2024
7d400c0
fix counter key
jiefenghuang Dec 12, 2024
2396eff
review: sort keys
jiefenghuang Dec 12, 2024
4146212
fix: max acl id
jiefenghuang Dec 12, 2024
43f6c1e
review
jiefenghuang Dec 12, 2024
76a4daf
Merge branch 'main' into backup-kv
davies Dec 13, 2024
2a698bb
fix conflict
jiefenghuang Dec 13, 2024
a2515ca
fix test
jiefenghuang Dec 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 25 additions & 28 deletions pkg/meta/backup.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,14 +57,17 @@ const (

var errBakEOF = fmt.Errorf("reach backup EOF")

func getMessageNameFromType(typ int) protoreflect.FullName {
func getMessageFromType(typ int) (proto.Message, error) {
var name protoreflect.FullName
if typ == segTypeFormat {
return proto.MessageName(&pb.Format{})
name = proto.MessageName(&pb.Format{})
} else if typ < segTypeMax {
return proto.MessageName(&pb.Batch{})
} else {
return ""
name = proto.MessageName(&pb.Batch{})
}
if name == "" {
return nil, fmt.Errorf("unknown message type %d", typ)
}
return createMessageByName(name)
}

func createMessageByName(name protoreflect.FullName) (proto.Message, error) {
Expand Down Expand Up @@ -128,16 +131,7 @@ func (f *bakFormat) writeFooter(w io.Writer) error {
if err := f.writeEOS(w); err != nil {
return err
}

data, err := f.footer.Marshal()
if err != nil {
return err
}
n, err := w.Write(data)
if err != nil && n != len(data) {
return fmt.Errorf("failed to write footer: err %v, write len %d, expect len %d", err, n, len(data))
}
return nil
return f.footer.Marshal(w)
}

func (f *bakFormat) writeEOS(w io.Writer) error {
Expand All @@ -164,15 +158,21 @@ type bakFooter struct {
len uint64
}

func (h *bakFooter) Marshal() ([]byte, error) {
func (h *bakFooter) Marshal(w io.Writer) error {
data, err := proto.Marshal(h.msg)
if err != nil {
return nil, fmt.Errorf("failed to marshal footer: %w", err)
return fmt.Errorf("failed to marshal footer: %w", err)
}

if n, err := w.Write(data); err != nil && n != len(data) {
return fmt.Errorf("failed to write footer data: err %w, write len %d, expect len %d", err, n, len(data))
}

h.len = uint64(len(data))
data = binary.BigEndian.AppendUint64(data, h.len)
return data, nil
if n, err := w.Write(binary.BigEndian.AppendUint64(nil, h.len)); err != nil && n != 8 {
return fmt.Errorf("failed to write footer length: err %w, write len %d, expect len 8", err, n)
}
return nil
}

func (h *bakFooter) Unmarshal(r io.ReadSeeker) error {
Expand Down Expand Up @@ -205,7 +205,7 @@ type bakSegment struct {
}

func (s *bakSegment) String() string {
return string(proto.MessageName(s.val).Name())
return fmt.Sprintf("type-%d", s.typ)
}

func (s *bakSegment) num() uint64 {
Expand Down Expand Up @@ -310,13 +310,9 @@ func (s *bakSegment) Unmarshal(r io.Reader) error {
return fmt.Errorf("failed to read segment type: %v", err)
}

if s.typ == BakMagic {
if s.typ == BakEOS {
return errBakEOF
}
name := getMessageNameFromType(int(s.typ))
if name == "" {
return fmt.Errorf("segment type %d is unknown", s.typ)
}

if err := binary.Read(r, binary.BigEndian, &s.len); err != nil {
return fmt.Errorf("failed to read segment %s length: %v", s, err)
Expand All @@ -326,12 +322,13 @@ func (s *bakSegment) Unmarshal(r io.Reader) error {
if err != nil && n != int(s.len) {
return fmt.Errorf("failed to read segment value: err %v, read len %d, expect len %d", err, n, s.len)
}
msg, err := createMessageByName(name)

msg, err := getMessageFromType(int(s.typ))
if err != nil {
return fmt.Errorf("failed to create message %s: %v", name, err)
return fmt.Errorf("failed to create message by type %d: %w", s.typ, err)
}
if err = proto.Unmarshal(data, msg); err != nil {
return fmt.Errorf("failed to unmarshal segment msg %s: %v", name, err)
return fmt.Errorf("failed to unmarshal segment msg %d: %w", s.typ, err)
}
s.val = msg
return nil
Expand Down
4 changes: 2 additions & 2 deletions pkg/meta/base_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2367,7 +2367,7 @@ func setAttr(t *testing.T, m Meta, inode Ino, attr *Attr) {
return err
})
case *kvMeta:
err = m.txn(func(tx *kvTxn) error {
err = m.txn(Background(), func(tx *kvTxn) error {
tx.set(m.inodeKey(inode), m.marshal(attr))
return nil
})
Expand Down Expand Up @@ -2834,7 +2834,7 @@ func testClone(t *testing.T, m Meta) {
t.Fatalf("remove tree error rootInode: %v", cloneDstIno)
}
removedItem = append(removedItem, m.detachedKey(cloneDstIno))
m.txn(func(tx *kvTxn) error {
m.txn(Background(), func(tx *kvTxn) error {
for _, key := range removedItem {
if buf := tx.get(key.([]byte)); buf != nil {
t.Fatalf("has keys not removed: %v", removedItem)
Expand Down
28 changes: 24 additions & 4 deletions pkg/meta/load_dump_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -402,7 +402,8 @@ func TestLoadDumpV2(t *testing.T) {
engines := map[string][]string{
"sqlite3": {"sqlite3://dev.db", "sqlite3://dev2.db"},
// "mysql": {"mysql://root:@/dev", "mysql://root:@/dev2"},
"redis": {"redis://127.0.0.1:6379/2", "redis://127.0.0.1:6379/3"},
"redis": {"redis://127.0.0.1:6379/2", "redis://127.0.0.1:6379/3"},
"badger": {"badger://" + path.Join(t.TempDir(), "jfs-load-duimp-testdb-bk1"), "badger://" + path.Join(t.TempDir(), "jfs-load-duimp-testdb-bk2")},
// "tikv": {"tikv://127.0.0.1:2379/jfs-load-dump-1", "tikv://127.0.0.1:2379/jfs-load-dump-2"},
}

Expand Down Expand Up @@ -460,9 +461,9 @@ func testSecretAndTrash(t *testing.T, addr, addr2 string) {
if m2.GetFormat().EncryptKey != m.GetFormat().EncryptKey {
t.Fatalf("encrypt key not valid: %s", m2.GetFormat().EncryptKey)
}

testDumpV2(t, m, "sqlite-non-secret.dump", &DumpOption{Threads: 10, KeepSecret: false})
m2.Reset()
m2.Shutdown()

m2 = testLoadV2(t, addr2, "sqlite-non-secret.dump")
if m2.GetFormat().EncryptKey != "removed" {
t.Fatalf("encrypt key not valid: %s", m2.GetFormat().EncryptKey)
Expand All @@ -484,6 +485,9 @@ func testSecretAndTrash(t *testing.T, addr, addr2 string) {
if cnt != len(trashs) {
t.Fatalf("trash count: %d != %d", cnt, len(trashs))
}

m.Shutdown()
m2.Shutdown()
}

/*
Expand All @@ -493,7 +497,7 @@ func BenchmarkLoadDumpV2(b *testing.B) {
engines := map[string]string{
"mysql": "mysql://root:@/dev",
"redis": "redis://127.0.0.1:6379/2",
"tikv": "tikv://127.0.0.1:2379/jfs-load-dump-1",
"tikv": "tikv://127.0.0.1:2379/jfs-load-dump-1",
}

sample := "../../1M_files_in_one_dir.dump"
Expand Down Expand Up @@ -550,6 +554,22 @@ func BenchmarkLoadDumpV2(b *testing.B) {
b.Fatalf("dump meta: %s", err)
}
fp.Sync()

b.StopTimer()
bak := &bakFormat{}
fp2, err := os.Open(path)
if err != nil {
b.Fatalf("open file: %s", path)
}
defer fp2.Close()
footer, err := bak.readFooter(fp2)
if err != nil {
b.Fatalf("read footer: %s", err)
}
for name, info := range footer.msg.Infos {
b.Logf("segment: %s, num: %d", name, info.Num)
}
b.StartTimer()
})

b.Run("LoadV2 "+name, func(b *testing.B) {
Expand Down
Loading
Loading