server.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770
  1. /*
  2. Copyright 2014 CoreOS, Inc.
  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. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. package etcdserver
  14. import (
  15. "encoding/json"
  16. "errors"
  17. "io/ioutil"
  18. "log"
  19. "math/rand"
  20. "net/http"
  21. "os"
  22. "path"
  23. "regexp"
  24. "strconv"
  25. "sync/atomic"
  26. "time"
  27. "github.com/coreos/etcd/Godeps/_workspace/src/code.google.com/p/go.net/context"
  28. "github.com/coreos/etcd/discovery"
  29. pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
  30. "github.com/coreos/etcd/etcdserver/stats"
  31. "github.com/coreos/etcd/pkg/pbutil"
  32. "github.com/coreos/etcd/raft"
  33. "github.com/coreos/etcd/raft/raftpb"
  34. "github.com/coreos/etcd/snap"
  35. "github.com/coreos/etcd/store"
  36. "github.com/coreos/etcd/wait"
  37. "github.com/coreos/etcd/wal"
  38. )
  39. const (
  40. // owner can make/remove files inside the directory
  41. privateDirMode = 0700
  42. defaultSyncTimeout = time.Second
  43. DefaultSnapCount = 10000
  44. // TODO: calculate based on heartbeat interval
  45. defaultPublishRetryInterval = 5 * time.Second
  46. StoreAdminPrefix = "/0"
  47. StoreKeysPrefix = "/1"
  48. )
  49. var (
  50. ErrUnknownMethod = errors.New("etcdserver: unknown method")
  51. ErrStopped = errors.New("etcdserver: server stopped")
  52. ErrRemoved = errors.New("etcdserver: server removed")
  53. ErrIDRemoved = errors.New("etcdserver: ID removed")
  54. ErrIDExists = errors.New("etcdserver: ID exists")
  55. ErrIDNotFound = errors.New("etcdserver: ID not found")
  56. storeMembersPrefix = path.Join(StoreAdminPrefix, "members")
  57. storeRemovedMembersPrefix = path.Join(StoreAdminPrefix, "removed_members")
  58. storeMemberAttributeRegexp = regexp.MustCompile(path.Join(storeMembersPrefix, "[[:xdigit:]]{1,16}", attributesSuffix))
  59. )
  60. func init() {
  61. rand.Seed(time.Now().UnixNano())
  62. }
  63. type sendFunc func(m []raftpb.Message)
  64. type Response struct {
  65. Event *store.Event
  66. Watcher store.Watcher
  67. err error
  68. }
  69. type Storage interface {
  70. // Save function saves ents and state to the underlying stable storage.
  71. // Save MUST block until st and ents are on stable storage.
  72. Save(st raftpb.HardState, ents []raftpb.Entry)
  73. // SaveSnap function saves snapshot to the underlying stable storage.
  74. SaveSnap(snap raftpb.Snapshot)
  75. // TODO: WAL should be able to control cut itself. After implement self-controled cut,
  76. // remove it in this interface.
  77. // Cut cuts out a new wal file for saving new state and entries.
  78. Cut() error
  79. }
  80. type Server interface {
  81. // Start performs any initialization of the Server necessary for it to
  82. // begin serving requests. It must be called before Do or Process.
  83. // Start must be non-blocking; any long-running server functionality
  84. // should be implemented in goroutines.
  85. Start()
  86. // Stop terminates the Server and performs any necessary finalization.
  87. // Do and Process cannot be called after Stop has been invoked.
  88. Stop()
  89. // Do takes a request and attempts to fulfil it, returning a Response.
  90. Do(ctx context.Context, r pb.Request) (Response, error)
  91. // Process takes a raft message and applies it to the server's raft state
  92. // machine, respecting any timeout of the given context.
  93. Process(ctx context.Context, m raftpb.Message) error
  94. // AddMember attempts to add a member into the cluster. It will return
  95. // ErrIDRemoved if member ID is removed from the cluster, or return
  96. // ErrIDExists if member ID exists in the cluster.
  97. AddMember(ctx context.Context, memb Member) error
  98. // RemoveMember attempts to remove a member from the cluster. It will
  99. // return ErrIDRemoved if member ID is removed from the cluster, or return
  100. // ErrIDNotFound if member ID is not in the cluster.
  101. RemoveMember(ctx context.Context, id uint64) error
  102. }
  103. type Stats interface {
  104. // SelfStats returns the struct representing statistics of this server
  105. SelfStats() []byte
  106. // LeaderStats returns the statistics of all followers in the cluster
  107. // if this server is leader. Otherwise, nil is returned.
  108. LeaderStats() []byte
  109. // StoreStats returns statistics of the store backing this EtcdServer
  110. StoreStats() []byte
  111. // UpdateRecvApp updates the underlying statistics in response to a receiving an Append request
  112. UpdateRecvApp(from uint64, length int64)
  113. }
  114. type RaftTimer interface {
  115. Index() uint64
  116. Term() uint64
  117. }
  118. // EtcdServer is the production implementation of the Server interface
  119. type EtcdServer struct {
  120. w wait.Wait
  121. done chan struct{}
  122. stopped chan struct{}
  123. id uint64
  124. attributes Attributes
  125. Cluster *Cluster
  126. node raft.Node
  127. store store.Store
  128. stats *stats.ServerStats
  129. lstats *stats.LeaderStats
  130. // send specifies the send function for sending msgs to members. send
  131. // MUST NOT block. It is okay to drop messages, since clients should
  132. // timeout and reissue their messages. If send is nil, server will
  133. // panic.
  134. send sendFunc
  135. storage Storage
  136. Ticker <-chan time.Time
  137. SyncTicker <-chan time.Time
  138. snapCount uint64 // number of entries to trigger a snapshot
  139. // Cache of the latest raft index and raft term the server has seen
  140. raftIndex uint64
  141. raftTerm uint64
  142. }
  143. // NewServer creates a new EtcdServer from the supplied configuration. The
  144. // configuration is considered static for the lifetime of the EtcdServer.
  145. func NewServer(cfg *ServerConfig) *EtcdServer {
  146. if err := os.MkdirAll(cfg.SnapDir(), privateDirMode); err != nil {
  147. log.Fatalf("etcdserver: cannot create snapshot directory: %v", err)
  148. }
  149. ss := snap.New(cfg.SnapDir())
  150. st := store.New()
  151. var w *wal.WAL
  152. var n raft.Node
  153. var id uint64
  154. haveWAL := wal.Exist(cfg.WALDir())
  155. switch {
  156. case !haveWAL && cfg.ClusterState == ClusterStateValueExisting:
  157. cl := getClusterFromPeers(cfg.Cluster.PeerURLs())
  158. if err := cfg.Cluster.ValidateAndAssignIDs(cl.Members()); err != nil {
  159. log.Fatalf("etcdserver: %v", err)
  160. }
  161. cfg.Cluster.SetID(cl.id)
  162. cfg.Cluster.SetStore(st)
  163. id, n, w = startNode(cfg, nil)
  164. case !haveWAL && cfg.ClusterState == ClusterStateValueNew:
  165. if err := cfg.VerifyBootstrapConfig(); err != nil {
  166. log.Fatalf("etcdserver: %v", err)
  167. }
  168. m := cfg.Cluster.MemberByName(cfg.Name)
  169. if cfg.ShouldDiscover() {
  170. d, err := discovery.New(cfg.DiscoveryURL, m.ID, cfg.Cluster.String())
  171. if err != nil {
  172. log.Fatalf("etcdserver: cannot init discovery %v", err)
  173. }
  174. s, err := d.Discover()
  175. if err != nil {
  176. log.Fatalf("etcdserver: %v", err)
  177. }
  178. if cfg.Cluster, err = NewClusterFromString(cfg.Cluster.name, s); err != nil {
  179. log.Fatalf("etcdserver: %v", err)
  180. }
  181. }
  182. cfg.Cluster.SetStore(st)
  183. id, n, w = startNode(cfg, cfg.Cluster.MemberIDs())
  184. case haveWAL:
  185. if cfg.ShouldDiscover() {
  186. log.Printf("etcdserver: warn: ignoring discovery: etcd has already been initialized and has a valid log in %q", cfg.WALDir())
  187. }
  188. var index uint64
  189. snapshot, err := ss.Load()
  190. if err != nil && err != snap.ErrNoSnapshot {
  191. log.Fatal(err)
  192. }
  193. if snapshot != nil {
  194. log.Printf("etcdserver: recovering from snapshot at index %d", snapshot.Index)
  195. st.Recovery(snapshot.Data)
  196. index = snapshot.Index
  197. }
  198. cfg.Cluster = NewClusterFromStore(cfg.Cluster.name, st)
  199. id, n, w = restartNode(cfg, index, snapshot)
  200. default:
  201. log.Fatalf("etcdserver: unsupported bootstrap config")
  202. }
  203. sstats := &stats.ServerStats{
  204. Name: cfg.Name,
  205. ID: idAsHex(id),
  206. }
  207. lstats := stats.NewLeaderStats(idAsHex(id))
  208. s := &EtcdServer{
  209. store: st,
  210. node: n,
  211. id: id,
  212. attributes: Attributes{Name: cfg.Name, ClientURLs: cfg.ClientURLs.StringSlice()},
  213. Cluster: cfg.Cluster,
  214. storage: struct {
  215. *wal.WAL
  216. *snap.Snapshotter
  217. }{w, ss},
  218. stats: sstats,
  219. lstats: lstats,
  220. send: Sender(cfg.Transport, cfg.Cluster, sstats, lstats),
  221. Ticker: time.Tick(100 * time.Millisecond),
  222. SyncTicker: time.Tick(500 * time.Millisecond),
  223. snapCount: cfg.SnapCount,
  224. }
  225. return s
  226. }
  227. // Start prepares and starts server in a new goroutine. It is no longer safe to
  228. // modify a server's fields after it has been sent to Start.
  229. // It also starts a goroutine to publish its server information.
  230. func (s *EtcdServer) Start() {
  231. s.start()
  232. go s.publish(defaultPublishRetryInterval)
  233. }
  234. // start prepares and starts server in a new goroutine. It is no longer safe to
  235. // modify a server's fields after it has been sent to Start.
  236. // This function is just used for testing.
  237. func (s *EtcdServer) start() {
  238. if s.snapCount == 0 {
  239. log.Printf("etcdserver: set snapshot count to default %d", DefaultSnapCount)
  240. s.snapCount = DefaultSnapCount
  241. }
  242. s.w = wait.New()
  243. s.done = make(chan struct{})
  244. s.stopped = make(chan struct{})
  245. s.stats.Initialize()
  246. // TODO: if this is an empty log, writes all peer infos
  247. // into the first entry
  248. go s.run()
  249. }
  250. func (s *EtcdServer) Process(ctx context.Context, m raftpb.Message) error {
  251. if s.Cluster.IsIDRemoved(m.From) {
  252. return ErrRemoved
  253. }
  254. return s.node.Step(ctx, m)
  255. }
  256. func (s *EtcdServer) run() {
  257. var syncC <-chan time.Time
  258. // snapi indicates the index of the last submitted snapshot request
  259. var snapi, appliedi uint64
  260. var nodes []uint64
  261. for {
  262. select {
  263. case <-s.Ticker:
  264. s.node.Tick()
  265. case rd := <-s.node.Ready():
  266. if rd.SoftState != nil {
  267. nodes = rd.SoftState.Nodes
  268. if rd.RaftState == raft.StateLeader {
  269. syncC = s.SyncTicker
  270. } else {
  271. syncC = nil
  272. }
  273. }
  274. s.storage.Save(rd.HardState, rd.Entries)
  275. s.storage.SaveSnap(rd.Snapshot)
  276. s.send(rd.Messages)
  277. // TODO(bmizerany): do this in the background, but take
  278. // care to apply entries in a single goroutine, and not
  279. // race them.
  280. // TODO: apply configuration change into ClusterStore.
  281. if len(rd.CommittedEntries) != 0 {
  282. appliedi = s.apply(rd.CommittedEntries, nodes)
  283. }
  284. if rd.Snapshot.Index > snapi {
  285. snapi = rd.Snapshot.Index
  286. }
  287. // recover from snapshot if it is more updated than current applied
  288. if rd.Snapshot.Index > appliedi {
  289. if err := s.store.Recovery(rd.Snapshot.Data); err != nil {
  290. panic("TODO: this is bad, what do we do about it?")
  291. }
  292. appliedi = rd.Snapshot.Index
  293. }
  294. if appliedi-snapi > s.snapCount {
  295. s.snapshot(appliedi, nodes)
  296. snapi = appliedi
  297. }
  298. case <-syncC:
  299. s.sync(defaultSyncTimeout)
  300. case <-s.done:
  301. close(s.stopped)
  302. return
  303. }
  304. }
  305. }
  306. // Stop stops the server gracefully, and shuts down the running goroutine.
  307. // Stop should be called after a Start(s), otherwise it will block forever.
  308. func (s *EtcdServer) Stop() {
  309. s.node.Stop()
  310. close(s.done)
  311. <-s.stopped
  312. }
  313. // Do interprets r and performs an operation on s.store according to r.Method
  314. // and other fields. If r.Method is "POST", "PUT", "DELETE", or a "GET" with
  315. // Quorum == true, r will be sent through consensus before performing its
  316. // respective operation. Do will block until an action is performed or there is
  317. // an error.
  318. func (s *EtcdServer) Do(ctx context.Context, r pb.Request) (Response, error) {
  319. if r.ID == 0 {
  320. panic("r.ID cannot be 0")
  321. }
  322. if r.Method == "GET" && r.Quorum {
  323. r.Method = "QGET"
  324. }
  325. switch r.Method {
  326. case "POST", "PUT", "DELETE", "QGET":
  327. data, err := r.Marshal()
  328. if err != nil {
  329. return Response{}, err
  330. }
  331. ch := s.w.Register(r.ID)
  332. s.node.Propose(ctx, data)
  333. select {
  334. case x := <-ch:
  335. resp := x.(Response)
  336. return resp, resp.err
  337. case <-ctx.Done():
  338. s.w.Trigger(r.ID, nil) // GC wait
  339. return Response{}, ctx.Err()
  340. case <-s.done:
  341. return Response{}, ErrStopped
  342. }
  343. case "GET":
  344. switch {
  345. case r.Wait:
  346. wc, err := s.store.Watch(r.Path, r.Recursive, r.Stream, r.Since)
  347. if err != nil {
  348. return Response{}, err
  349. }
  350. return Response{Watcher: wc}, nil
  351. default:
  352. ev, err := s.store.Get(r.Path, r.Recursive, r.Sorted)
  353. if err != nil {
  354. return Response{}, err
  355. }
  356. return Response{Event: ev}, nil
  357. }
  358. default:
  359. return Response{}, ErrUnknownMethod
  360. }
  361. }
  362. func (s *EtcdServer) SelfStats() []byte {
  363. return s.stats.JSON()
  364. }
  365. func (s *EtcdServer) LeaderStats() []byte {
  366. // TODO(jonboulle): need to lock access to lstats, set it to nil when not leader, ...
  367. return s.lstats.JSON()
  368. }
  369. func (s *EtcdServer) StoreStats() []byte {
  370. return s.store.JsonStats()
  371. }
  372. func (s *EtcdServer) UpdateRecvApp(from uint64, length int64) {
  373. s.stats.RecvAppendReq(idAsHex(from), int(length))
  374. }
  375. func (s *EtcdServer) AddMember(ctx context.Context, memb Member) error {
  376. // TODO: move Member to protobuf type
  377. b, err := json.Marshal(memb)
  378. if err != nil {
  379. return err
  380. }
  381. cc := raftpb.ConfChange{
  382. ID: GenID(),
  383. Type: raftpb.ConfChangeAddNode,
  384. NodeID: memb.ID,
  385. Context: b,
  386. }
  387. return s.configure(ctx, cc)
  388. }
  389. func (s *EtcdServer) RemoveMember(ctx context.Context, id uint64) error {
  390. cc := raftpb.ConfChange{
  391. ID: GenID(),
  392. Type: raftpb.ConfChangeRemoveNode,
  393. NodeID: id,
  394. }
  395. return s.configure(ctx, cc)
  396. }
  397. // Implement the RaftTimer interface
  398. func (s *EtcdServer) Index() uint64 {
  399. return atomic.LoadUint64(&s.raftIndex)
  400. }
  401. func (s *EtcdServer) Term() uint64 {
  402. return atomic.LoadUint64(&s.raftTerm)
  403. }
  404. // configure sends configuration change through consensus then performs it.
  405. // It will block until the change is performed or there is an error.
  406. func (s *EtcdServer) configure(ctx context.Context, cc raftpb.ConfChange) error {
  407. ch := s.w.Register(cc.ID)
  408. if err := s.node.ProposeConfChange(ctx, cc); err != nil {
  409. log.Printf("configure error: %v", err)
  410. s.w.Trigger(cc.ID, nil)
  411. return err
  412. }
  413. select {
  414. case x := <-ch:
  415. if err, ok := x.(error); ok {
  416. return err
  417. }
  418. if x != nil {
  419. log.Panicf("unexpected return type")
  420. }
  421. return nil
  422. case <-ctx.Done():
  423. s.w.Trigger(cc.ID, nil) // GC wait
  424. return ctx.Err()
  425. case <-s.done:
  426. return ErrStopped
  427. }
  428. }
  429. // sync proposes a SYNC request and is non-blocking.
  430. // This makes no guarantee that the request will be proposed or performed.
  431. // The request will be cancelled after the given timeout.
  432. func (s *EtcdServer) sync(timeout time.Duration) {
  433. ctx, cancel := context.WithTimeout(context.Background(), timeout)
  434. req := pb.Request{
  435. Method: "SYNC",
  436. ID: GenID(),
  437. Time: time.Now().UnixNano(),
  438. }
  439. data := pbutil.MustMarshal(&req)
  440. // There is no promise that node has leader when do SYNC request,
  441. // so it uses goroutine to propose.
  442. go func() {
  443. s.node.Propose(ctx, data)
  444. cancel()
  445. }()
  446. }
  447. // publish registers server information into the cluster. The information
  448. // is the JSON representation of this server's member struct, updated with the
  449. // static clientURLs of the server.
  450. // The function keeps attempting to register until it succeeds,
  451. // or its server is stopped.
  452. func (s *EtcdServer) publish(retryInterval time.Duration) {
  453. b, err := json.Marshal(s.attributes)
  454. if err != nil {
  455. log.Printf("etcdserver: json marshal error: %v", err)
  456. return
  457. }
  458. req := pb.Request{
  459. ID: GenID(),
  460. Method: "PUT",
  461. Path: path.Join(memberStoreKey(s.id), attributesSuffix),
  462. Val: string(b),
  463. }
  464. for {
  465. ctx, cancel := context.WithTimeout(context.Background(), retryInterval)
  466. _, err := s.Do(ctx, req)
  467. cancel()
  468. switch err {
  469. case nil:
  470. log.Printf("etcdserver: published %+v to the cluster", s.attributes)
  471. return
  472. case ErrStopped:
  473. log.Printf("etcdserver: aborting publish because server is stopped")
  474. return
  475. default:
  476. log.Printf("etcdserver: publish error: %v", err)
  477. }
  478. }
  479. }
  480. func getExpirationTime(r *pb.Request) time.Time {
  481. var t time.Time
  482. if r.Expiration != 0 {
  483. t = time.Unix(0, r.Expiration)
  484. }
  485. return t
  486. }
  487. func (s *EtcdServer) apply(es []raftpb.Entry, nodes []uint64) uint64 {
  488. var applied uint64
  489. for i := range es {
  490. e := es[i]
  491. switch e.Type {
  492. case raftpb.EntryNormal:
  493. var r pb.Request
  494. pbutil.MustUnmarshal(&r, e.Data)
  495. s.w.Trigger(r.ID, s.applyRequest(r))
  496. case raftpb.EntryConfChange:
  497. var cc raftpb.ConfChange
  498. pbutil.MustUnmarshal(&cc, e.Data)
  499. s.w.Trigger(cc.ID, s.applyConfChange(cc, nodes))
  500. default:
  501. panic("unexpected entry type")
  502. }
  503. atomic.StoreUint64(&s.raftIndex, e.Index)
  504. atomic.StoreUint64(&s.raftTerm, e.Term)
  505. applied = e.Index
  506. }
  507. return applied
  508. }
  509. // applyRequest interprets r as a call to store.X and returns a Response interpreted
  510. // from store.Event
  511. func (s *EtcdServer) applyRequest(r pb.Request) Response {
  512. f := func(ev *store.Event, err error) Response {
  513. return Response{Event: ev, err: err}
  514. }
  515. expr := getExpirationTime(&r)
  516. switch r.Method {
  517. case "POST":
  518. return f(s.store.Create(r.Path, r.Dir, r.Val, true, expr))
  519. case "PUT":
  520. exists, existsSet := getBool(r.PrevExist)
  521. switch {
  522. case existsSet:
  523. if exists {
  524. return f(s.store.Update(r.Path, r.Val, expr))
  525. }
  526. return f(s.store.Create(r.Path, r.Dir, r.Val, false, expr))
  527. case r.PrevIndex > 0 || r.PrevValue != "":
  528. return f(s.store.CompareAndSwap(r.Path, r.PrevValue, r.PrevIndex, r.Val, expr))
  529. default:
  530. if storeMemberAttributeRegexp.MatchString(r.Path) {
  531. id := parseMemberID(path.Dir(r.Path))
  532. m := s.Cluster.Member(id)
  533. if m == nil {
  534. log.Fatalf("fetch member %x should never fail", id)
  535. }
  536. if err := json.Unmarshal([]byte(r.Val), &m.Attributes); err != nil {
  537. log.Fatalf("unmarshal %s should never fail: %v", r.Val, err)
  538. }
  539. }
  540. return f(s.store.Set(r.Path, r.Dir, r.Val, expr))
  541. }
  542. case "DELETE":
  543. switch {
  544. case r.PrevIndex > 0 || r.PrevValue != "":
  545. return f(s.store.CompareAndDelete(r.Path, r.PrevValue, r.PrevIndex))
  546. default:
  547. return f(s.store.Delete(r.Path, r.Dir, r.Recursive))
  548. }
  549. case "QGET":
  550. return f(s.store.Get(r.Path, r.Recursive, r.Sorted))
  551. case "SYNC":
  552. s.store.DeleteExpiredKeys(time.Unix(0, r.Time))
  553. return Response{}
  554. default:
  555. // This should never be reached, but just in case:
  556. return Response{err: ErrUnknownMethod}
  557. }
  558. }
  559. func (s *EtcdServer) applyConfChange(cc raftpb.ConfChange, nodes []uint64) error {
  560. if err := s.checkConfChange(cc, nodes); err != nil {
  561. cc.NodeID = raft.None
  562. s.node.ApplyConfChange(cc)
  563. return err
  564. }
  565. s.node.ApplyConfChange(cc)
  566. switch cc.Type {
  567. case raftpb.ConfChangeAddNode:
  568. m := new(Member)
  569. if err := json.Unmarshal(cc.Context, m); err != nil {
  570. panic("unexpected unmarshal error")
  571. }
  572. if cc.NodeID != m.ID {
  573. panic("unexpected nodeID mismatch")
  574. }
  575. s.Cluster.AddMember(m)
  576. case raftpb.ConfChangeRemoveNode:
  577. s.Cluster.RemoveMember(cc.NodeID)
  578. }
  579. return nil
  580. }
  581. func (s *EtcdServer) checkConfChange(cc raftpb.ConfChange, nodes []uint64) error {
  582. if s.Cluster.IsIDRemoved(cc.NodeID) {
  583. return ErrIDRemoved
  584. }
  585. switch cc.Type {
  586. case raftpb.ConfChangeAddNode:
  587. if containsUint64(nodes, cc.NodeID) {
  588. return ErrIDExists
  589. }
  590. case raftpb.ConfChangeRemoveNode:
  591. if !containsUint64(nodes, cc.NodeID) {
  592. return ErrIDNotFound
  593. }
  594. default:
  595. panic("unexpected ConfChange type")
  596. }
  597. return nil
  598. }
  599. // TODO: non-blocking snapshot
  600. func (s *EtcdServer) snapshot(snapi uint64, snapnodes []uint64) {
  601. d, err := s.store.Save()
  602. // TODO: current store will never fail to do a snapshot
  603. // what should we do if the store might fail?
  604. if err != nil {
  605. panic("TODO: this is bad, what do we do about it?")
  606. }
  607. s.node.Compact(snapi, snapnodes, d)
  608. s.storage.Cut()
  609. }
  610. func getClusterFromPeers(urls []string) *Cluster {
  611. for _, u := range urls {
  612. resp, err := http.Get(u + "/members")
  613. if err != nil {
  614. log.Printf("etcdserver: get /members on %s: %v", u, err)
  615. continue
  616. }
  617. b, err := ioutil.ReadAll(resp.Body)
  618. if err != nil {
  619. log.Printf("etcdserver: read body error: %v", err)
  620. continue
  621. }
  622. var membs []*Member
  623. if err := json.Unmarshal(b, &membs); err != nil {
  624. log.Printf("etcdserver: unmarshal body error: %v", err)
  625. continue
  626. }
  627. id, err := strconv.ParseUint(resp.Header.Get("X-Etcd-Cluster-ID"), 16, 64)
  628. if err != nil {
  629. log.Printf("etcdserver: parse uint error: %v", err)
  630. continue
  631. }
  632. return NewClusterFromMembers("", id, membs)
  633. }
  634. log.Fatalf("etcdserver: could not retrieve cluster information from the given urls")
  635. return nil
  636. }
  637. func startNode(cfg *ServerConfig, ids []uint64) (id uint64, n raft.Node, w *wal.WAL) {
  638. var err error
  639. // TODO: remove the discoveryURL when it becomes part of the source for
  640. // generating nodeID.
  641. member := cfg.Cluster.MemberByName(cfg.Name)
  642. metadata := pbutil.MustMarshal(&pb.Metadata{NodeID: member.ID, ClusterID: cfg.Cluster.ID()})
  643. if w, err = wal.Create(cfg.WALDir(), metadata); err != nil {
  644. log.Fatal(err)
  645. }
  646. peers := make([]raft.Peer, len(ids))
  647. for i, id := range ids {
  648. ctx, err := json.Marshal((*cfg.Cluster).Member(id))
  649. if err != nil {
  650. log.Fatal(err)
  651. }
  652. peers[i] = raft.Peer{ID: id, Context: ctx}
  653. }
  654. id = member.ID
  655. log.Printf("etcdserver: start node %x in cluster %x", id, cfg.Cluster.ID())
  656. n = raft.StartNode(id, peers, 10, 1)
  657. return
  658. }
  659. func restartNode(cfg *ServerConfig, index uint64, snapshot *raftpb.Snapshot) (id uint64, n raft.Node, w *wal.WAL) {
  660. var err error
  661. // restart a node from previous wal
  662. if w, err = wal.OpenAtIndex(cfg.WALDir(), index); err != nil {
  663. log.Fatal(err)
  664. }
  665. wmetadata, st, ents, err := w.ReadAll()
  666. if err != nil {
  667. log.Fatal(err)
  668. }
  669. var metadata pb.Metadata
  670. pbutil.MustUnmarshal(&metadata, wmetadata)
  671. id = metadata.NodeID
  672. cfg.Cluster.SetID(metadata.ClusterID)
  673. log.Printf("etcdserver: restart member %x in cluster %x at commit index %d", id, cfg.Cluster.ID(), st.Commit)
  674. n = raft.RestartNode(id, 10, 1, snapshot, st, ents)
  675. return
  676. }
  677. // TODO: move the function to /id pkg maybe?
  678. // GenID generates a random id that is not equal to 0.
  679. func GenID() (n uint64) {
  680. for n == 0 {
  681. n = uint64(rand.Int63())
  682. }
  683. return
  684. }
  685. func getBool(v *bool) (vv bool, set bool) {
  686. if v == nil {
  687. return false, false
  688. }
  689. return *v, true
  690. }
  691. func containsUint64(a []uint64, x uint64) bool {
  692. for _, v := range a {
  693. if v == x {
  694. return true
  695. }
  696. }
  697. return false
  698. }
  699. func idAsHex(id uint64) string {
  700. return strconv.FormatUint(id, 16)
  701. }