diff --git a/weed/server/raft_hashicorp.go b/weed/server/raft_hashicorp.go index 575478aec..df557c64c 100644 --- a/weed/server/raft_hashicorp.go +++ b/weed/server/raft_hashicorp.go @@ -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) diff --git a/weed/server/raft_migration_test.go b/weed/server/raft_migration_test.go new file mode 100644 index 000000000..778570d01 --- /dev/null +++ b/weed/server/raft_migration_test.go @@ -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") + } +} diff --git a/weed/server/raft_server.go b/weed/server/raft_server.go index c98d43250..84cf3c4bc 100644 --- a/weed/server/raft_server.go +++ b/weed/server/raft_server.go @@ -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,