snapshot_merge.go 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172
  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. package etcdserver
  15. import (
  16. "io"
  17. "log"
  18. "github.com/coreos/etcd/mvcc/backend"
  19. "github.com/coreos/etcd/raft/raftpb"
  20. "github.com/coreos/etcd/snap"
  21. )
  22. // createMergedSnapshotMessage creates a snapshot message that contains: raft status (term, conf),
  23. // a snapshot of v2 store inside raft.Snapshot as []byte, a snapshot of v3 KV in the top level message
  24. // as ReadCloser.
  25. func (s *EtcdServer) createMergedSnapshotMessage(m raftpb.Message, snapi uint64, confState raftpb.ConfState) snap.Message {
  26. snapt, err := s.r.raftStorage.Term(snapi)
  27. if err != nil {
  28. log.Panicf("get term should never fail: %v", err)
  29. }
  30. // get a snapshot of v2 store as []byte
  31. clone := s.store.Clone()
  32. d, err := clone.SaveNoCopy()
  33. if err != nil {
  34. plog.Panicf("store save should never fail: %v", err)
  35. }
  36. dbsnap := s.be.Snapshot()
  37. // get a snapshot of v3 KV as readCloser
  38. rc := newSnapshotReaderCloser(dbsnap)
  39. // put the []byte snapshot of store into raft snapshot and return the merged snapshot with
  40. // KV readCloser snapshot.
  41. snapshot := raftpb.Snapshot{
  42. Metadata: raftpb.SnapshotMetadata{
  43. Index: snapi,
  44. Term: snapt,
  45. ConfState: confState,
  46. },
  47. Data: d,
  48. }
  49. m.Snapshot = snapshot
  50. return *snap.NewMessage(m, rc, dbsnap.Size())
  51. }
  52. func newSnapshotReaderCloser(snapshot backend.Snapshot) io.ReadCloser {
  53. pr, pw := io.Pipe()
  54. go func() {
  55. n, err := snapshot.WriteTo(pw)
  56. if err == nil {
  57. plog.Infof("wrote database snapshot out [total bytes: %d]", n)
  58. }
  59. pw.CloseWithError(err)
  60. snapshot.Close()
  61. }()
  62. return pr
  63. }