1// Copyright 2015 The etcd Authors 2// 3// Licensed under the Apache License, Version 2.0 (the "License"); 4// you may not use this file except in compliance with the License. 5// You may obtain a copy of the License at 6// 7// http://www.apache.org/licenses/LICENSE-2.0 8// 9// Unless required by applicable law or agreed to in writing, software 10// distributed under the License is distributed on an "AS IS" BASIS, 11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12// See the License for the specific language governing permissions and 13// limitations under the License. 14 15package raft 16 17import ( 18 "context" 19 "errors" 20 21 pb "go.etcd.io/etcd/raft/v3/raftpb" 22) 23 24type SnapshotStatus int 25 26const ( 27 SnapshotFinish SnapshotStatus = 1 28 SnapshotFailure SnapshotStatus = 2 29) 30 31var ( 32 emptyState = pb.HardState{} 33 34 // ErrStopped is returned by methods on Nodes that have been stopped. 35 ErrStopped = errors.New("raft: stopped") 36) 37 38// SoftState provides state that is useful for logging and debugging. 39// The state is volatile and does not need to be persisted to the WAL. 40type SoftState struct { 41 Lead uint64 // must use atomic operations to access; keep 64-bit aligned. 42 RaftState StateType 43} 44 45func (a *SoftState) equal(b *SoftState) bool { 46 return a.Lead == b.Lead && a.RaftState == b.RaftState 47} 48 49// Ready encapsulates the entries and messages that are ready to read, 50// be saved to stable storage, committed or sent to other peers. 51// All fields in Ready are read-only. 52type Ready struct { 53 // The current volatile state of a Node. 54 // SoftState will be nil if there is no update. 55 // It is not required to consume or store SoftState. 56 *SoftState 57 58 // The current state of a Node to be saved to stable storage BEFORE 59 // Messages are sent. 60 // HardState will be equal to empty state if there is no update. 61 pb.HardState 62 63 // ReadStates can be used for node to serve linearizable read requests locally 64 // when its applied index is greater than the index in ReadState. 65 // Note that the readState will be returned when raft receives msgReadIndex. 66 // The returned is only valid for the request that requested to read. 67 ReadStates []ReadState 68 69 // Entries specifies entries to be saved to stable storage BEFORE 70 // Messages are sent. 71 Entries []pb.Entry 72 73 // Snapshot specifies the snapshot to be saved to stable storage. 74 Snapshot pb.Snapshot 75 76 // CommittedEntries specifies entries to be committed to a 77 // store/state-machine. These have previously been committed to stable 78 // store. 79 CommittedEntries []pb.Entry 80 81 // Messages specifies outbound messages to be sent AFTER Entries are 82 // committed to stable storage. 83 // If it contains a MsgSnap message, the application MUST report back to raft 84 // when the snapshot has been received or has failed by calling ReportSnapshot. 85 Messages []pb.Message 86 87 // MustSync indicates whether the HardState and Entries must be synchronously 88 // written to disk or if an asynchronous write is permissible. 89 MustSync bool 90} 91 92func isHardStateEqual(a, b pb.HardState) bool { 93 return a.Term == b.Term && a.Vote == b.Vote && a.Commit == b.Commit 94} 95 96// IsEmptyHardState returns true if the given HardState is empty. 97func IsEmptyHardState(st pb.HardState) bool { 98 return isHardStateEqual(st, emptyState) 99} 100 101// IsEmptySnap returns true if the given Snapshot is empty. 102func IsEmptySnap(sp pb.Snapshot) bool { 103 return sp.Metadata.Index == 0 104} 105 106func (rd Ready) containsUpdates() bool { 107 return rd.SoftState != nil || !IsEmptyHardState(rd.HardState) || 108 !IsEmptySnap(rd.Snapshot) || len(rd.Entries) > 0 || 109 len(rd.CommittedEntries) > 0 || len(rd.Messages) > 0 || len(rd.ReadStates) != 0 110} 111 112// appliedCursor extracts from the Ready the highest index the client has 113// applied (once the Ready is confirmed via Advance). If no information is 114// contained in the Ready, returns zero. 115func (rd Ready) appliedCursor() uint64 { 116 if n := len(rd.CommittedEntries); n > 0 { 117 return rd.CommittedEntries[n-1].Index 118 } 119 if index := rd.Snapshot.Metadata.Index; index > 0 { 120 return index 121 } 122 return 0 123} 124 125// Node represents a node in a raft cluster. 126type Node interface { 127 // Tick increments the internal logical clock for the Node by a single tick. Election 128 // timeouts and heartbeat timeouts are in units of ticks. 129 Tick() 130 // Campaign causes the Node to transition to candidate state and start campaigning to become leader. 131 Campaign(ctx context.Context) error 132 // Propose proposes that data be appended to the log. Note that proposals can be lost without 133 // notice, therefore it is user's job to ensure proposal retries. 134 Propose(ctx context.Context, data []byte) error 135 // ProposeConfChange proposes a configuration change. Like any proposal, the 136 // configuration change may be dropped with or without an error being 137 // returned. In particular, configuration changes are dropped unless the 138 // leader has certainty that there is no prior unapplied configuration 139 // change in its log. 140 // 141 // The method accepts either a pb.ConfChange (deprecated) or pb.ConfChangeV2 142 // message. The latter allows arbitrary configuration changes via joint 143 // consensus, notably including replacing a voter. Passing a ConfChangeV2 144 // message is only allowed if all Nodes participating in the cluster run a 145 // version of this library aware of the V2 API. See pb.ConfChangeV2 for 146 // usage details and semantics. 147 ProposeConfChange(ctx context.Context, cc pb.ConfChangeI) error 148 149 // Step advances the state machine using the given message. ctx.Err() will be returned, if any. 150 Step(ctx context.Context, msg pb.Message) error 151 152 // Ready returns a channel that returns the current point-in-time state. 153 // Users of the Node must call Advance after retrieving the state returned by Ready. 154 // 155 // NOTE: No committed entries from the next Ready may be applied until all committed entries 156 // and snapshots from the previous one have finished. 157 Ready() <-chan Ready 158 159 // Advance notifies the Node that the application has saved progress up to the last Ready. 160 // It prepares the node to return the next available Ready. 161 // 162 // The application should generally call Advance after it applies the entries in last Ready. 163 // 164 // However, as an optimization, the application may call Advance while it is applying the 165 // commands. For example. when the last Ready contains a snapshot, the application might take 166 // a long time to apply the snapshot data. To continue receiving Ready without blocking raft 167 // progress, it can call Advance before finishing applying the last ready. 168 Advance() 169 // ApplyConfChange applies a config change (previously passed to 170 // ProposeConfChange) to the node. This must be called whenever a config 171 // change is observed in Ready.CommittedEntries, except when the app decides 172 // to reject the configuration change (i.e. treats it as a noop instead), in 173 // which case it must not be called. 174 // 175 // Returns an opaque non-nil ConfState protobuf which must be recorded in 176 // snapshots. 177 ApplyConfChange(cc pb.ConfChangeI) *pb.ConfState 178 179 // TransferLeadership attempts to transfer leadership to the given transferee. 180 TransferLeadership(ctx context.Context, lead, transferee uint64) 181 182 // ReadIndex request a read state. The read state will be set in the ready. 183 // Read state has a read index. Once the application advances further than the read 184 // index, any linearizable read requests issued before the read request can be 185 // processed safely. The read state will have the same rctx attached. 186 // Note that request can be lost without notice, therefore it is user's job 187 // to ensure read index retries. 188 ReadIndex(ctx context.Context, rctx []byte) error 189 190 // Status returns the current status of the raft state machine. 191 Status() Status 192 // ReportUnreachable reports the given node is not reachable for the last send. 193 ReportUnreachable(id uint64) 194 // ReportSnapshot reports the status of the sent snapshot. The id is the raft ID of the follower 195 // who is meant to receive the snapshot, and the status is SnapshotFinish or SnapshotFailure. 196 // Calling ReportSnapshot with SnapshotFinish is a no-op. But, any failure in applying a 197 // snapshot (for e.g., while streaming it from leader to follower), should be reported to the 198 // leader with SnapshotFailure. When leader sends a snapshot to a follower, it pauses any raft 199 // log probes until the follower can apply the snapshot and advance its state. If the follower 200 // can't do that, for e.g., due to a crash, it could end up in a limbo, never getting any 201 // updates from the leader. Therefore, it is crucial that the application ensures that any 202 // failure in snapshot sending is caught and reported back to the leader; so it can resume raft 203 // log probing in the follower. 204 ReportSnapshot(id uint64, status SnapshotStatus) 205 // Stop performs any necessary termination of the Node. 206 Stop() 207} 208 209type Peer struct { 210 ID uint64 211 Context []byte 212} 213 214// StartNode returns a new Node given configuration and a list of raft peers. 215// It appends a ConfChangeAddNode entry for each given peer to the initial log. 216// 217// Peers must not be zero length; call RestartNode in that case. 218func StartNode(c *Config, peers []Peer) Node { 219 if len(peers) == 0 { 220 panic("no peers given; use RestartNode instead") 221 } 222 rn, err := NewRawNode(c) 223 if err != nil { 224 panic(err) 225 } 226 rn.Bootstrap(peers) 227 228 n := newNode(rn) 229 230 go n.run() 231 return &n 232} 233 234// RestartNode is similar to StartNode but does not take a list of peers. 235// The current membership of the cluster will be restored from the Storage. 236// If the caller has an existing state machine, pass in the last log index that 237// has been applied to it; otherwise use zero. 238func RestartNode(c *Config) Node { 239 rn, err := NewRawNode(c) 240 if err != nil { 241 panic(err) 242 } 243 n := newNode(rn) 244 go n.run() 245 return &n 246} 247 248type msgWithResult struct { 249 m pb.Message 250 result chan error 251} 252 253// node is the canonical implementation of the Node interface 254type node struct { 255 propc chan msgWithResult 256 recvc chan pb.Message 257 confc chan pb.ConfChangeV2 258 confstatec chan pb.ConfState 259 readyc chan Ready 260 advancec chan struct{} 261 tickc chan struct{} 262 done chan struct{} 263 stop chan struct{} 264 status chan chan Status 265 266 rn *RawNode 267} 268 269func newNode(rn *RawNode) node { 270 return node{ 271 propc: make(chan msgWithResult), 272 recvc: make(chan pb.Message), 273 confc: make(chan pb.ConfChangeV2), 274 confstatec: make(chan pb.ConfState), 275 readyc: make(chan Ready), 276 advancec: make(chan struct{}), 277 // make tickc a buffered chan, so raft node can buffer some ticks when the node 278 // is busy processing raft messages. Raft node will resume process buffered 279 // ticks when it becomes idle. 280 tickc: make(chan struct{}, 128), 281 done: make(chan struct{}), 282 stop: make(chan struct{}), 283 status: make(chan chan Status), 284 rn: rn, 285 } 286} 287 288func (n *node) Stop() { 289 select { 290 case n.stop <- struct{}{}: 291 // Not already stopped, so trigger it 292 case <-n.done: 293 // Node has already been stopped - no need to do anything 294 return 295 } 296 // Block until the stop has been acknowledged by run() 297 <-n.done 298} 299 300func (n *node) run() { 301 var propc chan msgWithResult 302 var readyc chan Ready 303 var advancec chan struct{} 304 var rd Ready 305 306 r := n.rn.raft 307 308 lead := None 309 310 for { 311 if advancec != nil { 312 readyc = nil 313 } else if n.rn.HasReady() { 314 // Populate a Ready. Note that this Ready is not guaranteed to 315 // actually be handled. We will arm readyc, but there's no guarantee 316 // that we will actually send on it. It's possible that we will 317 // service another channel instead, loop around, and then populate 318 // the Ready again. We could instead force the previous Ready to be 319 // handled first, but it's generally good to emit larger Readys plus 320 // it simplifies testing (by emitting less frequently and more 321 // predictably). 322 rd = n.rn.readyWithoutAccept() 323 readyc = n.readyc 324 } 325 326 if lead != r.lead { 327 if r.hasLeader() { 328 if lead == None { 329 r.logger.Infof("raft.node: %x elected leader %x at term %d", r.id, r.lead, r.Term) 330 } else { 331 r.logger.Infof("raft.node: %x changed leader from %x to %x at term %d", r.id, lead, r.lead, r.Term) 332 } 333 propc = n.propc 334 } else { 335 r.logger.Infof("raft.node: %x lost leader %x at term %d", r.id, lead, r.Term) 336 propc = nil 337 } 338 lead = r.lead 339 } 340 341 select { 342 // TODO: maybe buffer the config propose if there exists one (the way 343 // described in raft dissertation) 344 // Currently it is dropped in Step silently. 345 case pm := <-propc: 346 m := pm.m 347 m.From = r.id 348 err := r.Step(m) 349 if pm.result != nil { 350 pm.result <- err 351 close(pm.result) 352 } 353 case m := <-n.recvc: 354 // filter out response message from unknown From. 355 if pr := r.prs.Progress[m.From]; pr != nil || !IsResponseMsg(m.Type) { 356 r.Step(m) 357 } 358 case cc := <-n.confc: 359 _, okBefore := r.prs.Progress[r.id] 360 cs := r.applyConfChange(cc) 361 // If the node was removed, block incoming proposals. Note that we 362 // only do this if the node was in the config before. Nodes may be 363 // a member of the group without knowing this (when they're catching 364 // up on the log and don't have the latest config) and we don't want 365 // to block the proposal channel in that case. 366 // 367 // NB: propc is reset when the leader changes, which, if we learn 368 // about it, sort of implies that we got readded, maybe? This isn't 369 // very sound and likely has bugs. 370 if _, okAfter := r.prs.Progress[r.id]; okBefore && !okAfter { 371 var found bool 372 outer: 373 for _, sl := range [][]uint64{cs.Voters, cs.VotersOutgoing} { 374 for _, id := range sl { 375 if id == r.id { 376 found = true 377 break outer 378 } 379 } 380 } 381 if !found { 382 propc = nil 383 } 384 } 385 select { 386 case n.confstatec <- cs: 387 case <-n.done: 388 } 389 case <-n.tickc: 390 n.rn.Tick() 391 case readyc <- rd: 392 n.rn.acceptReady(rd) 393 advancec = n.advancec 394 case <-advancec: 395 n.rn.Advance(rd) 396 rd = Ready{} 397 advancec = nil 398 case c := <-n.status: 399 c <- getStatus(r) 400 case <-n.stop: 401 close(n.done) 402 return 403 } 404 } 405} 406 407// Tick increments the internal logical clock for this Node. Election timeouts 408// and heartbeat timeouts are in units of ticks. 409func (n *node) Tick() { 410 select { 411 case n.tickc <- struct{}{}: 412 case <-n.done: 413 default: 414 n.rn.raft.logger.Warningf("%x A tick missed to fire. Node blocks too long!", n.rn.raft.id) 415 } 416} 417 418func (n *node) Campaign(ctx context.Context) error { return n.step(ctx, pb.Message{Type: pb.MsgHup}) } 419 420func (n *node) Propose(ctx context.Context, data []byte) error { 421 return n.stepWait(ctx, pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}}) 422} 423 424func (n *node) Step(ctx context.Context, m pb.Message) error { 425 // ignore unexpected local messages receiving over network 426 if IsLocalMsg(m.Type) { 427 // TODO: return an error? 428 return nil 429 } 430 return n.step(ctx, m) 431} 432 433func confChangeToMsg(c pb.ConfChangeI) (pb.Message, error) { 434 typ, data, err := pb.MarshalConfChange(c) 435 if err != nil { 436 return pb.Message{}, err 437 } 438 return pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Type: typ, Data: data}}}, nil 439} 440 441func (n *node) ProposeConfChange(ctx context.Context, cc pb.ConfChangeI) error { 442 msg, err := confChangeToMsg(cc) 443 if err != nil { 444 return err 445 } 446 return n.Step(ctx, msg) 447} 448 449func (n *node) step(ctx context.Context, m pb.Message) error { 450 return n.stepWithWaitOption(ctx, m, false) 451} 452 453func (n *node) stepWait(ctx context.Context, m pb.Message) error { 454 return n.stepWithWaitOption(ctx, m, true) 455} 456 457// Step advances the state machine using msgs. The ctx.Err() will be returned, 458// if any. 459func (n *node) stepWithWaitOption(ctx context.Context, m pb.Message, wait bool) error { 460 if m.Type != pb.MsgProp { 461 select { 462 case n.recvc <- m: 463 return nil 464 case <-ctx.Done(): 465 return ctx.Err() 466 case <-n.done: 467 return ErrStopped 468 } 469 } 470 ch := n.propc 471 pm := msgWithResult{m: m} 472 if wait { 473 pm.result = make(chan error, 1) 474 } 475 select { 476 case ch <- pm: 477 if !wait { 478 return nil 479 } 480 case <-ctx.Done(): 481 return ctx.Err() 482 case <-n.done: 483 return ErrStopped 484 } 485 select { 486 case err := <-pm.result: 487 if err != nil { 488 return err 489 } 490 case <-ctx.Done(): 491 return ctx.Err() 492 case <-n.done: 493 return ErrStopped 494 } 495 return nil 496} 497 498func (n *node) Ready() <-chan Ready { return n.readyc } 499 500func (n *node) Advance() { 501 select { 502 case n.advancec <- struct{}{}: 503 case <-n.done: 504 } 505} 506 507func (n *node) ApplyConfChange(cc pb.ConfChangeI) *pb.ConfState { 508 var cs pb.ConfState 509 select { 510 case n.confc <- cc.AsV2(): 511 case <-n.done: 512 } 513 select { 514 case cs = <-n.confstatec: 515 case <-n.done: 516 } 517 return &cs 518} 519 520func (n *node) Status() Status { 521 c := make(chan Status) 522 select { 523 case n.status <- c: 524 return <-c 525 case <-n.done: 526 return Status{} 527 } 528} 529 530func (n *node) ReportUnreachable(id uint64) { 531 select { 532 case n.recvc <- pb.Message{Type: pb.MsgUnreachable, From: id}: 533 case <-n.done: 534 } 535} 536 537func (n *node) ReportSnapshot(id uint64, status SnapshotStatus) { 538 rej := status == SnapshotFailure 539 540 select { 541 case n.recvc <- pb.Message{Type: pb.MsgSnapStatus, From: id, Reject: rej}: 542 case <-n.done: 543 } 544} 545 546func (n *node) TransferLeadership(ctx context.Context, lead, transferee uint64) { 547 select { 548 // manually set 'from' and 'to', so that leader can voluntarily transfers its leadership 549 case n.recvc <- pb.Message{Type: pb.MsgTransferLeader, From: transferee, To: lead}: 550 case <-n.done: 551 case <-ctx.Done(): 552 } 553} 554 555func (n *node) ReadIndex(ctx context.Context, rctx []byte) error { 556 return n.step(ctx, pb.Message{Type: pb.MsgReadIndex, Entries: []pb.Entry{{Data: rctx}}}) 557} 558 559func newReady(r *raft, prevSoftSt *SoftState, prevHardSt pb.HardState) Ready { 560 rd := Ready{ 561 Entries: r.raftLog.unstableEntries(), 562 CommittedEntries: r.raftLog.nextEnts(), 563 Messages: r.msgs, 564 } 565 if softSt := r.softState(); !softSt.equal(prevSoftSt) { 566 rd.SoftState = softSt 567 } 568 if hardSt := r.hardState(); !isHardStateEqual(hardSt, prevHardSt) { 569 rd.HardState = hardSt 570 } 571 if r.raftLog.unstable.snapshot != nil { 572 rd.Snapshot = *r.raftLog.unstable.snapshot 573 } 574 if len(r.readStates) != 0 { 575 rd.ReadStates = r.readStates 576 } 577 rd.MustSync = MustSync(r.hardState(), prevHardSt, len(rd.Entries)) 578 return rd 579} 580 581// MustSync returns true if the hard state and count of Raft entries indicate 582// that a synchronous write to persistent storage is required. 583func MustSync(st, prevst pb.HardState, entsnum int) bool { 584 // Persistent state on all servers: 585 // (Updated on stable storage before responding to RPCs) 586 // currentTerm 587 // votedFor 588 // log entries[] 589 return entsnum != 0 || st.Vote != prevst.Vote || st.Term != prevst.Term 590} 591