server.go 21 KB

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