master: carry cluster identity from legacy raft into hashicorp raft

On a master's first start on hashicorp raft, import TopologyId and MaxVolumeId from an existing seaweedfs/raft (legacy) snapshot so the cluster keeps its identity instead of generating a fresh one. A marker in the hashicorp stable store makes the import one-time; later restarts skip it. With no legacy state the marker is still set, so a fresh install never re-scans.
2026-09-20 13:30:46 +02:00 · 2026-06-05 14:45:33 -07:00
parent ab7be7867d
commit c5121f3dba
3 changed files with 201 additions and 10 deletions
@@ -24,6 +24,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/topology"
+	"github.com/seaweedfs/seaweedfs/weed/util/version"
 	"google.golang.org/grpc"
 )

@@ -31,8 +32,57 @@ const (
 	ldbFile            = "logs.dat"
 	sdbFile            = "stable.dat"
 	updatePeersTimeout = 15 * time.Minute
+
+	// legacyMigrationKey marks, in the hashicorp stable store, that this
+	// master's hashicorp raft state has already absorbed any pre-existing
+	// legacy seaweedfs/raft state. Its presence makes the import one-time so
+	// later restarts never re-read stale legacy state.
+	legacyMigrationKey = "seaweedfs.legacy.raft.migrated"
 )

+// migrateLegacyRaftStateIfNeeded performs the one-time legacy -> hashicorp raft
+// migration. It first checks the migration marker in the hashicorp stable
+// store; if already migrated it does nothing. Otherwise, when a legacy
+// seaweedfs/raft state is present, it imports the cluster identity
+// (TopologyId) and MaxVolumeId from it, then marks the migration done. With no
+// legacy state it just records the marker. Either way the master ends up
+// marked migrated, so the import runs at most once per cluster.
+func migrateLegacyRaftStateIfNeeded(sdb *boltdb.BoltStore, dataDir string, topo *topology.Topology) {
+	if _, err := sdb.Get([]byte(legacyMigrationKey)); err == nil {
+		return // marker present: already migrated
+	}
+	if legacyRaftStateExists(dataDir) {
+		glog.V(0).Infof("first hashicorp-raft start with legacy raft state in %s; migrating", dataDir)
+		importLegacyRaftState(dataDir, topo)
+	}
+	if err := sdb.Set([]byte(legacyMigrationKey), []byte(version.Version())); err != nil {
+		glog.Warningf("failed to record legacy raft migration marker: %v", err)
+	}
+}
+
+// importLegacyRaftState seeds TopologyId and MaxVolumeId from the latest legacy
+// snapshot. TopologyId is the cluster identity and must survive the engine
+// switch; MaxVolumeId is otherwise rebuilt from volume heartbeats, but carrying
+// it avoids reusing an id whose volume was deleted before the migration.
+func importLegacyRaftState(dataDir string, topo *topology.Topology) {
+	state, ok := readLegacyRaftSnapshotState(dataDir)
+	if !ok {
+		glog.V(0).Infof("legacy raft state present but no readable snapshot; " +
+			"TopologyId/MaxVolumeId will be rebuilt from volume heartbeats")
+		return
+	}
+	var cmd topology.MaxVolumeIdCommand
+	if err := json.Unmarshal(state, &cmd); err != nil {
+		glog.Warningf("failed to parse legacy raft snapshot state: %v", err)
+		return
+	}
+	topo.UpAdjustMaxVolumeId(cmd.MaxVolumeId)
+	if cmd.TopologyId != "" {
+		topo.SetTopologyId(cmd.TopologyId)
+	}
+	glog.V(0).Infof("migrated legacy raft state: MaxVolumeId=%d TopologyId=%s", cmd.MaxVolumeId, cmd.TopologyId)
+}
+
 func getPeerIdx(self pb.ServerAddress, mapPeers map[string]pb.ServerAddress) int {
 	peerIDs := make([]string, 0, len(mapPeers))
 	seen := make(map[string]struct{}, len(mapPeers))
@@ -212,6 +262,11 @@ func NewHashicorpRaftServer(option *RaftServerOption) (*RaftServer, error) {
 		return nil, fmt.Errorf("boltdb.NewBoltStore(%q): %v", filepath.Join(baseDir, "stable.dat"), err)
 	}

+	// Carry cluster identity forward when upgrading from legacy seaweedfs/raft.
+	// Runs once, before raft starts, so the seeded TopologyId is in place
+	// before ensureTopologyId would otherwise generate a fresh one.
+	migrateLegacyRaftStateIfNeeded(sdb, baseDir, option.Topo)
+
 	fss, err := raft.NewFileSnapshotStore(baseDir, 3, os.Stderr)
 	if err != nil {
 		return nil, fmt.Errorf("raft.NewFileSnapshotStore(%q, ...): %v", baseDir, err)
@@ -0,0 +1,116 @@
+package weed_server
+
+import (
+	"encoding/json"
+	"fmt"
+	"hash/crc32"
+	"os"
+	"path"
+	"path/filepath"
+	"testing"
+
+	boltdb "github.com/hashicorp/raft-boltdb/v2"
+	"github.com/seaweedfs/seaweedfs/weed/sequence"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/topology"
+)
+
+// writeLegacySnapshot writes a seaweedfs/raft (legacy) snapshot carrying the
+// given FSM state, in the on-disk format readLegacyRaftSnapshotState expects.
+func writeLegacySnapshot(t *testing.T, dataDir string, maxVolumeId needle.VolumeId, topologyId string) {
+	t.Helper()
+	snapshotDir := path.Join(dataDir, "snapshot")
+	if err := os.MkdirAll(snapshotDir, 0755); err != nil {
+		t.Fatal(err)
+	}
+	state, err := json.Marshal(topology.MaxVolumeIdCommand{MaxVolumeId: maxVolumeId, TopologyId: topologyId})
+	if err != nil {
+		t.Fatal(err)
+	}
+	body, err := json.Marshal(struct {
+		State json.RawMessage `json:"state"`
+	}{State: state})
+	if err != nil {
+		t.Fatal(err)
+	}
+	f, err := os.Create(path.Join(snapshotDir, "0_1.ss"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer f.Close()
+	if _, err := fmt.Fprintf(f, "%08x\n", crc32.ChecksumIEEE(body)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := f.Write(body); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func openStableStore(t *testing.T, dataDir string) *boltdb.BoltStore {
+	t.Helper()
+	sdb, err := boltdb.NewBoltStore(filepath.Join(dataDir, sdbFile))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { sdb.Close() })
+	return sdb
+}
+
+func newTestTopology() *topology.Topology {
+	return topology.NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+}
+
+func TestMigrateLegacyRaftState_ImportsIdentityOnce(t *testing.T) {
+	dir := t.TempDir()
+	writeLegacySnapshot(t, dir, 4242, "cluster-abc")
+
+	sdb := openStableStore(t, dir)
+	topo := newTestTopology()
+
+	migrateLegacyRaftStateIfNeeded(sdb, dir, topo)
+
+	if got := topo.GetTopologyId(); got != "cluster-abc" {
+		t.Fatalf("TopologyId = %q, want cluster-abc", got)
+	}
+	if got := topo.GetMaxVolumeId(); got != needle.VolumeId(4242) {
+		t.Fatalf("MaxVolumeId = %d, want 4242", got)
+	}
+	if v, err := sdb.Get([]byte(legacyMigrationKey)); err != nil || len(v) == 0 {
+		t.Fatalf("migration marker not set: v=%q err=%v", v, err)
+	}
+
+	// Second run on a fresh topology must NOT re-import: the marker gates it.
+	topo2 := newTestTopology()
+	migrateLegacyRaftStateIfNeeded(sdb, dir, topo2)
+	if got := topo2.GetTopologyId(); got != "" {
+		t.Fatalf("re-import after marker set: TopologyId = %q, want empty", got)
+	}
+}
+
+func TestMigrateLegacyRaftState_NoLegacyJustMarks(t *testing.T) {
+	dir := t.TempDir()
+	sdb := openStableStore(t, dir)
+	topo := newTestTopology()
+
+	migrateLegacyRaftStateIfNeeded(sdb, dir, topo)
+
+	if got := topo.GetTopologyId(); got != "" {
+		t.Fatalf("TopologyId = %q, want empty (nothing to import)", got)
+	}
+	if v, err := sdb.Get([]byte(legacyMigrationKey)); err != nil || len(v) == 0 {
+		t.Fatalf("migration marker not set on fresh install: v=%q err=%v", v, err)
+	}
+}
+
+func TestLegacyRaftStateExists(t *testing.T) {
+	dir := t.TempDir()
+	if legacyRaftStateExists(dir) {
+		t.Fatal("empty dir reported as having legacy state")
+	}
+	if err := os.WriteFile(path.Join(dir, "log"), []byte("x"), 0644); err != nil {
+		t.Fatal(err)
+	}
+	if !legacyRaftStateExists(dir) {
+		t.Fatal("legacy log file not detected")
+	}
+}
@@ -249,35 +249,47 @@ func (s *RaftServer) Peers() (members []string) {
 	return
 }

-// recoverTopologyIdFromSnapshot reads the TopologyId from the latest
-// seaweedfs/raft snapshot before state cleanup.
-func recoverTopologyIdFromSnapshot(dataDir string, topo *topology.Topology) {
+// legacyRaftStateExists reports whether a seaweedfs/raft (legacy) state is
+// present in dataDir. The legacy engine writes these names; hashicorp raft
+// uses logs.dat/stable.dat/snapshots, so there is no overlap.
+func legacyRaftStateExists(dataDir string) bool {
+	for _, name := range []string{"snapshot", "log", "conf", "state"} {
+		if _, err := os.Stat(path.Join(dataDir, name)); err == nil {
+			return true
+		}
+	}
+	return false
+}
+
+// readLegacyRaftSnapshotState returns the raw FSM state JSON from the latest
+// seaweedfs/raft (legacy) snapshot, if a valid one exists.
+func readLegacyRaftSnapshotState(dataDir string) (json.RawMessage, bool) {
 	snapshotDir := path.Join(dataDir, "snapshot")
 	dir, err := os.Open(snapshotDir)
 	if err != nil {
-		return
+		return nil, false
 	}
 	defer dir.Close()
 	filenames, err := dir.Readdirnames(-1)
 	if err != nil || len(filenames) == 0 {
-		return
+		return nil, false
 	}

 	sort.Strings(filenames)
 	file, err := os.Open(path.Join(snapshotDir, filenames[len(filenames)-1]))
 	if err != nil {
-		return
+		return nil, false
 	}
 	defer file.Close()

 	// Snapshot format: 8-hex-digit CRC32 checksum, newline, JSON body.
 	var checksum uint32
 	if _, err := fmt.Fscanf(file, "%08x\n", &checksum); err != nil {
-		return
+		return nil, false
 	}
 	b, err := io.ReadAll(file)
 	if err != nil || crc32.ChecksumIEEE(b) != checksum {
-		return
+		return nil, false
 	}

 	// The snapshot JSON wraps the FSM state in a "state" field.
@@ -285,9 +297,17 @@ func recoverTopologyIdFromSnapshot(dataDir string, topo *topology.Topology) {
 		State json.RawMessage `json:"state"`
 	}
 	if err := json.Unmarshal(b, &snap); err != nil || len(snap.State) == 0 {
-		return
+		return nil, false
+	}
+	return snap.State, true
+}
+
+// recoverTopologyIdFromSnapshot reads the TopologyId from the latest
+// seaweedfs/raft snapshot before state cleanup.
+func recoverTopologyIdFromSnapshot(dataDir string, topo *topology.Topology) {
+	if state, ok := readLegacyRaftSnapshotState(dataDir); ok {
+		recoverTopologyIdFromState(state, topo)
 	}
-	recoverTopologyIdFromState(snap.State, topo)
 }

 // HasExistingState returns true when the raft log already contains entries,