Skip to content

Commit

Permalink
chore: Update Raft migrate tool to handle new proposal format (#7123)
Browse files Browse the repository at this point in the history 
This PR updates the raftmigrate tool. We recently updated raft proposal format for alpha and zero, see #7059 .
The tool now can migrate raftwal from old proposal format to new proposal format wherein we are encoding the proposal.Key within the entry.Data.
This change will help to migrate raftwal from [RC-2,RC-3] to RC-4 onwards.
  • Loading branch information
@rahulgurnani
rahulgurnani committed Dec 16, 2020
1 parent 2e2228a commit ab0f3ea
Show file tree
Hide file tree
Showing 4 changed files with 84 additions and 54 deletions.
3 changes: 2 additions & 1 deletion dgraph/cmd/debug/run.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -907,6 +907,7 @@ func run() {
store, err := raftwal.InitEncrypted(dir, opt.key)
x.Check(err)
fmt.Printf("RaftID: %+v\n", store.Uint(raftwal.RaftId))
isZero := store.Uint(raftwal.GroupId) == 0

// TODO: Fix the pending logic.
pending := make(map[uint64]bool)
Expand All @@ -916,7 +917,7 @@ func run() {
entries, err := store.Entries(start, last+1, 64<<20)
x.Check(err)
for _, e := range entries {
printEntry(e, pending)
printEntry(e, pending, isZero)
start = x.Max(start, e.Index)
}
}
Expand Down
28 changes: 17 additions & 11 deletions dgraph/cmd/debug/wal.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ import (
"go.etcd.io/etcd/raft/raftpb"
)

func printEntry(es raftpb.Entry, pending map[uint64]bool) {
func printEntry(es raftpb.Entry, pending map[uint64]bool, isZero bool) {
var buf bytes.Buffer
defer func() {
fmt.Printf("%s\n", buf.Bytes())
Expand All @@ -46,15 +46,21 @@ func printEntry(es raftpb.Entry, pending map[uint64]bool) {
if len(es.Data) == 0 {
return
}
var pr pb.Proposal
var zpr pb.ZeroProposal
if err := pr.Unmarshal(es.Data[8:]); err == nil {
printAlphaProposal(&buf, &pr, pending)
} else if err := zpr.Unmarshal(es.Data[8:]); err == nil {
printZeroProposal(&buf, &zpr)
var err error
if isZero {
var zpr pb.ZeroProposal
if err = zpr.Unmarshal(es.Data[8:]); err == nil {
printZeroProposal(&buf, &zpr)
return
}
} else {
fmt.Fprintf(&buf, " Unable to parse Proposal: %v", err)
var pr pb.Proposal
if err = pr.Unmarshal(es.Data[8:]); err == nil {
printAlphaProposal(&buf, &pr, pending)
return
}
}
fmt.Fprintf(&buf, " Unable to parse Proposal: %v", err)
}

type RaftStore interface {
Expand Down Expand Up @@ -116,7 +122,7 @@ func printBasic(store RaftStore) (uint64, uint64) {
return startIdx, lastIdx
}

func printRaft(db *badger.DB, store *raftmigrate.OldDiskStorage) {
func printRaft(db *badger.DB, store *raftmigrate.OldDiskStorage, groupId uint32) {
startIdx, lastIdx := printBasic(store)

commitTs := db.MaxVersion()
Expand Down Expand Up @@ -149,7 +155,7 @@ func printRaft(db *badger.DB, store *raftmigrate.OldDiskStorage) {
log.Fatalf("Unable to set data: %+v", err)
}
default:
printEntry(ent, pending)
printEntry(ent, pending, groupId == 0)
}
startIdx = x.Max(startIdx, ent.Index)
}
Expand Down Expand Up @@ -286,7 +292,7 @@ func handleWal(db *badger.DB) error {
return err

default:
printRaft(db, store)
printRaft(db, store, gid)
}
store.Closer.SignalAndWait()
}
Expand Down
104 changes: 65 additions & 39 deletions dgraph/cmd/raft-migrate/run.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,12 +21,13 @@ import (
"math"
"os"

"github.com/dgraph-io/badger/v2"
"github.com/dgraph-io/dgraph/ee/enc"
"github.com/dgraph-io/dgraph/protos/pb"
"github.com/dgraph-io/dgraph/raftwal"
"github.com/dgraph-io/dgraph/x"
"github.com/spf13/cobra"
"github.com/spf13/viper"
"go.etcd.io/etcd/raft/raftpb"
)

var (
Expand All @@ -50,13 +51,37 @@ func init() {

flag := RaftMigrate.Cmd.Flags()
flag.StringP("old-dir", "", "", "Path to the old (z)w directory.")
flag.IntP("old-node-id", "", 1,
"Node ID of the old node. This will be the node ID of the new node.")
flag.IntP("old-group-id", "", 0, "Group ID of the old node. This is used to open the old wal.")
flag.StringP("new-dir", "", "", "Path to the new (z)w directory.")
enc.RegisterFlags(flag)
}

func updateEntry(entry raftpb.Entry) raftpb.Entry {
// Raft commits an empty entry on becoming leader.
if entry.Type == raftpb.EntryConfChange || len(entry.Data) == 0 {
return entry
}

data := make([]byte, 8+len(entry.Data))
// First 8 bytes are used to store key in current raftwal format. We don't need the key in
// migrate tool. In zero, we don't use the key and in alpha the key used for not applying the
// same proposal twice.
copy(data[8:], entry.Data)
entry.Data = data
return entry
}

func parseAndConvertSnapshot(snap *raftpb.Snapshot) {
var ms pb.MembershipState
var zs pb.ZeroSnapshot
var err error
x.Check(ms.Unmarshal(snap.Data))
zs.State = &ms
zs.Index = snap.Metadata.Index
// It is okay to not set zs.CheckpointTs as it is used for purgeBelow.
snap.Data, err = zs.Marshal()
x.Check(err)
}

func run(conf *viper.Viper) error {
oldDir := conf.GetString("old-dir")
newDir := conf.GetString("new-dir")
Expand All @@ -68,49 +93,26 @@ func run(conf *viper.Viper) error {
log.Fatal("--new-dir not specified.")
}

nodeId := conf.GetInt("old-node-id")
groupId := conf.GetInt("old-group-id")

var err error
if encKey, err = enc.ReadKey(conf); err != nil {
log.Fatalf("Failed to read encryption file: %s", err)
}

// Copied over from zero/run.go
kvOpt := badger.LSMOnlyOptions(oldDir).
WithSyncWrites(false).
WithValueLogFileSize(64 << 20).
WithIndexCacheSize(100 << 20).
WithEncryptionKey(encKey)

kv, err := badger.OpenManaged(kvOpt)
x.Checkf(err, "Error while opening WAL store")
defer kv.Close()

raftID, err := RaftId(kv)
x.Check(err)
oldWal := Init(kv, uint64(nodeId), uint32(groupId))
oldWal, err := raftwal.InitEncrypted(oldDir, encKey)
x.Checkf(err, "failed to initialize old wal: %s", err)
defer oldWal.Close()

firstIndex, err := oldWal.FirstIndex()
x.Checkf(err, "failed to read FirstIndex from old wal: %s", err)

lastIndex, err := oldWal.LastIndex()
x.Checkf(err, "failed to read LastIndex from the old wal: %s", err)

// TODO(ibrahim): Do we need this??
// The new directory has one less entry if we don't do this.
lastIndex++

fmt.Printf("Fetching entries from low: %d to high: %d\n", firstIndex, lastIndex)
// Should we batch this up?
oldEntries, err := oldWal.Entries(firstIndex, lastIndex, math.MaxUint64)
x.Checkf(err, "failed to read entries from low:%d high:%d err:%s", firstIndex, lastIndex, err)
oldEntries, err := oldWal.Entries(firstIndex, lastIndex+1, math.MaxUint64)

snapshot, err := oldWal.Snapshot()
x.Checkf(err, "failed to read snaphot %s", err)
newEntries := make([]raftpb.Entry, len(oldEntries))
for i, entry := range oldEntries {
newEntries[i] = updateEntry(entry)
}

hs, err := oldWal.HardState()
x.Checkf(err, "failed to read hardstate %s", err)
x.Checkf(err, "failed to read entries from low:%d high:%d err:%s", firstIndex, lastIndex, err)

if _, err := os.Stat(newDir); os.IsNotExist(err) {
os.Mkdir(newDir, 0777)
Expand All @@ -119,16 +121,40 @@ func run(conf *viper.Viper) error {
newWal, err := raftwal.InitEncrypted(newDir, encKey)
x.Check(err)

fmt.Printf("Setting raftID to: %+v\n", raftID)
// Set the raft ID
raftID := oldWal.Uint(raftwal.RaftId)
fmt.Printf("Setting raftID to: %+v\n", raftID)
newWal.SetUint(raftwal.RaftId, raftID)

// Set the Group ID
groupID := oldWal.Uint(raftwal.GroupId)
fmt.Printf("Setting GroupID to: %+v\n", groupID)
newWal.SetUint(raftwal.GroupId, groupID)

// Set the checkpoint index
checkPoint, err := oldWal.Checkpoint()
x.Checkf(err, "failed to read checkpoint %s", err)
newWal.SetUint(raftwal.CheckpointIndex, checkPoint)

snapshot, err := oldWal.Snapshot()
x.Checkf(err, "failed to read snaphot %s", err)
if groupID == 0 {
// We earlier used to store MembershipState in raftpb.Snapshot. Now we store ZeroSnapshot in
// case of zero.
fmt.Println("Parsing and converting zero-snapshot")
parseAndConvertSnapshot(&snapshot)
}

hs, err := oldWal.HardState()
x.Checkf(err, "failed to read hardstate %s", err)

fmt.Printf("Saving num of oldEntries:%+v\nsnapshot %+v\nhardstate = %+v\n",
len(oldEntries), snapshot, hs)
if err := newWal.Save(&hs, oldEntries, &snapshot); err != nil {
len(newEntries), snapshot, hs)
if err := newWal.Save(&hs, newEntries, &snapshot); err != nil {
log.Fatalf("failed to save new state. hs: %+v, snapshot: %+v, oldEntries: %+v, err: %s",
hs, oldEntries, snapshot, err)
}

if err := newWal.Close(); err != nil {
log.Fatalf("Failed to close new wal: %s", err)
}
Expand Down
3 changes: 0 additions & 3 deletions dgraph/cmd/zero/raft.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -333,9 +333,6 @@ func (n *node) applyProposal(e raftpb.Entry) (uint64, error) {
if err := p.Unmarshal(e.Data[8:]); err != nil {
return key, err
}
if key == 0 {
return key, errInvalidProposal
}
span := otrace.FromContext(n.Proposals.Ctx(key))

n.server.Lock()
Expand Down

0 comments on commit ab0f3ea

Please sign in to comment.