Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

etcdserver: Refactor standalone boostrap

Browse Source
This commit is contained in:
Marek Siarkowicz 2021-07-07 23:14:09 +02:00
parent 244e5c2cce
commit e1fa356fac
2 changed files with 44 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
server/etcdserver/raft.go
View File

@ -474,37 +474,10 @@ func boostrapRaftFromWalStandalone(cfg config.ServerConfig, snapshot *raftpb.Sna
wal := boostrapWALFromSnapshot(cfg.Logger, cfg.WALDir(), snapshot, cfg.UnsafeNoFsync) wal := boostrapWALFromSnapshot(cfg.Logger, cfg.WALDir(), snapshot, cfg.UnsafeNoFsync)
// discard the previously uncommitted entries // discard the previously uncommitted entries
for i, ent := range wal.ents { wal.ents = wal.CommitedEntries()
if ent.Index > wal.st.Commit { entries := wal.ConfigChangeEntries()
cfg.Logger.Info( // force commit config change entries
"discarding uncommitted WAL entries", wal.AppendAndCommitEntries(entries)
zap.Uint64("entry-index", ent.Index),
zap.Uint64("commit-index-from-wal", wal.st.Commit),
zap.Int("number-of-discarded-entries", len(wal.ents)-i),
)
wal.ents = wal.ents[:i]
break
}
}
// force append the configuration change entries
toAppEnts := createConfigChangeEnts(
cfg.Logger,
getIDs(cfg.Logger, snapshot, wal.ents),
uint64(wal.id),
wal.st.Term,
wal.st.Commit,
)
wal.ents = append(wal.ents, toAppEnts...)
// force commit newly appended entries
err := wal.w.Save(raftpb.HardState{}, toAppEnts)
if err != nil {
cfg.Logger.Fatal("failed to save hard state and entries", zap.Error(err))
}
if len(wal.ents) != 0 {
wal.st.Commit = wal.ents[len(wal.ents)-1].Index
}
cfg.Logger.Info( cfg.Logger.Info(
"forcing restart member", "forcing restart member",

40
server/etcdserver/storage.go
View File

@ -119,6 +119,7 @@ func boostrapWALFromSnapshot(lg *zap.Logger, waldir string, snapshot *raftpb.Sna
id := types.ID(metadata.NodeID) id := types.ID(metadata.NodeID)
cid := types.ID(metadata.ClusterID) cid := types.ID(metadata.ClusterID)
return &boostrappedWAL{ return &boostrappedWAL{
lg: lg,
w: w, w: w,
id: id, id: id,
cid: cid, cid: cid,
@ -144,6 +145,7 @@ func boostrapNewWal(cfg config.ServerConfig, nodeID, clusterID types.ID) *boostr
w.SetUnsafeNoFsync() w.SetUnsafeNoFsync()
} }
return &boostrappedWAL{ return &boostrappedWAL{
lg: cfg.Logger,
w: w, w: w,
id: nodeID, id: nodeID,
cid: clusterID, cid: clusterID,
@ -151,6 +153,8 @@ func boostrapNewWal(cfg config.ServerConfig, nodeID, clusterID types.ID) *boostr
} }
type boostrappedWAL struct { type boostrappedWAL struct {
lg *zap.Logger
w *wal.WAL w *wal.WAL
id, cid types.ID id, cid types.ID
st *raftpb.HardState st *raftpb.HardState
@ -171,3 +175,39 @@ func (wal *boostrappedWAL) MemoryStorage() *raft.MemoryStorage {
} }
return s return s
} }
func (wal *boostrappedWAL) CommitedEntries() []raftpb.Entry {
for i, ent := range wal.ents {
if ent.Index > wal.st.Commit {
wal.lg.Info(
"discarding uncommitted WAL entries",
zap.Uint64("entry-index", ent.Index),
zap.Uint64("commit-index-from-wal", wal.st.Commit),
zap.Int("number-of-discarded-entries", len(wal.ents)-i),
)
return wal.ents[:i]
}
}
return wal.ents
}
func (wal *boostrappedWAL) ConfigChangeEntries() []raftpb.Entry {
return createConfigChangeEnts(
wal.lg,
getIDs(wal.lg, wal.snapshot, wal.ents),
uint64(wal.id),
wal.st.Term,
wal.st.Commit,
)
}
func (wal *boostrappedWAL) AppendAndCommitEntries(ents []raftpb.Entry) {
wal.ents = append(wal.ents, ents...)
err := wal.w.Save(raftpb.HardState{}, ents)
if err != nil {
wal.lg.Fatal("failed to save hard state and entries", zap.Error(err))
}
if len(wal.ents) != 0 {
wal.st.Commit = wal.ents[len(wal.ents)-1].Index
}
}