chore: Update Raft migrate tool to handle new proposal format

dgraph/cmd/debug/run.go

@@ -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)
@@ -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)
 			}
 		}

dgraph/cmd/debug/wal.go

@@ -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())
@@ -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 {
@@ -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()
@@ -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)
 		}
@@ -286,7 +292,7 @@ func handleWal(db *badger.DB) error {
 				return err
 
 			default:
-				printRaft(db, store)
+				printRaft(db, store, gid)
 			}
 			store.Closer.SignalAndWait()
 		}

dgraph/cmd/raft-migrate/run.go

@@ -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 (
@@ -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")
@@ -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)
@@ -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)
 	}

dgraph/cmd/zero/raft.go

@@ -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()