storage.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154
  1. // Copyright 2015 CoreOS, Inc.
  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. package etcdserver
  15. import (
  16. "io"
  17. "log"
  18. "os"
  19. "path"
  20. pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
  21. "github.com/coreos/etcd/migrate"
  22. "github.com/coreos/etcd/pkg/pbutil"
  23. "github.com/coreos/etcd/pkg/types"
  24. "github.com/coreos/etcd/raft/raftpb"
  25. "github.com/coreos/etcd/snap"
  26. "github.com/coreos/etcd/version"
  27. "github.com/coreos/etcd/wal"
  28. "github.com/coreos/etcd/wal/walpb"
  29. )
  30. type Storage interface {
  31. // Save function saves ents and state to the underlying stable storage.
  32. // Save MUST block until st and ents are on stable storage.
  33. Save(st raftpb.HardState, ents []raftpb.Entry) error
  34. // SaveSnap function saves snapshot to the underlying stable storage.
  35. SaveSnap(snap raftpb.Snapshot) error
  36. // Close closes the Storage and performs finalization.
  37. Close() error
  38. }
  39. type storage struct {
  40. *wal.WAL
  41. *snap.Snapshotter
  42. }
  43. func NewStorage(w *wal.WAL, s *snap.Snapshotter) Storage {
  44. return &storage{w, s}
  45. }
  46. // SaveSnap saves the snapshot to disk and release the locked
  47. // wal files since they will not be used.
  48. func (st *storage) SaveSnap(snap raftpb.Snapshot) error {
  49. walsnap := walpb.Snapshot{
  50. Index: snap.Metadata.Index,
  51. Term: snap.Metadata.Term,
  52. }
  53. err := st.WAL.SaveSnapshot(walsnap)
  54. if err != nil {
  55. return err
  56. }
  57. err = st.Snapshotter.SaveSnap(snap)
  58. if err != nil {
  59. return err
  60. }
  61. err = st.WAL.ReleaseLockTo(snap.Metadata.Index)
  62. if err != nil {
  63. return err
  64. }
  65. return nil
  66. }
  67. func readWAL(waldir string, snap walpb.Snapshot) (w *wal.WAL, id, cid types.ID, st raftpb.HardState, ents []raftpb.Entry) {
  68. var (
  69. err error
  70. wmetadata []byte
  71. )
  72. repaired := false
  73. for {
  74. if w, err = wal.Open(waldir, snap); err != nil {
  75. log.Fatalf("etcdserver: open wal error: %v", err)
  76. }
  77. if wmetadata, st, ents, err = w.ReadAll(); err != nil {
  78. w.Close()
  79. // we can only repair ErrUnexpectedEOF and we never repair twice.
  80. if repaired || err != io.ErrUnexpectedEOF {
  81. log.Fatalf("etcdserver: read wal error (%v) and cannot be repaired", err)
  82. }
  83. if !wal.Repair(waldir) {
  84. log.Fatalf("etcdserver: WAL error (%v) cannot be repaired", err)
  85. } else {
  86. log.Printf("etcdserver: repaired WAL error (%v)", err)
  87. repaired = true
  88. }
  89. continue
  90. }
  91. break
  92. }
  93. var metadata pb.Metadata
  94. pbutil.MustUnmarshal(&metadata, wmetadata)
  95. id = types.ID(metadata.NodeID)
  96. cid = types.ID(metadata.ClusterID)
  97. return
  98. }
  99. // upgradeWAL converts an older version of the etcdServer data to the newest version.
  100. // It must ensure that, after upgrading, the most recent version is present.
  101. func upgradeDataDir(baseDataDir string, name string, ver version.DataDirVersion) error {
  102. switch ver {
  103. case version.DataDir0_4:
  104. log.Print("etcdserver: converting v0.4 log to v2.0")
  105. err := migrate.Migrate4To2(baseDataDir, name)
  106. if err != nil {
  107. log.Fatalf("etcdserver: failed migrating data-dir: %v", err)
  108. return err
  109. }
  110. fallthrough
  111. case version.DataDir2_0:
  112. err := makeMemberDir(baseDataDir)
  113. if err != nil {
  114. return err
  115. }
  116. fallthrough
  117. case version.DataDir2_0_1:
  118. fallthrough
  119. default:
  120. log.Printf("etcdserver: datadir is valid for the 2.0.1 format")
  121. }
  122. return nil
  123. }
  124. func makeMemberDir(dir string) error {
  125. membdir := path.Join(dir, "member")
  126. _, err := os.Stat(membdir)
  127. switch {
  128. case err == nil:
  129. return nil
  130. case !os.IsNotExist(err):
  131. return err
  132. }
  133. if err := os.MkdirAll(membdir, 0700); err != nil {
  134. return err
  135. }
  136. names := []string{"snap", "wal"}
  137. for _, name := range names {
  138. if err := os.Rename(path.Join(dir, name), path.Join(membdir, name)); err != nil {
  139. return err
  140. }
  141. }
  142. return nil
  143. }