blob: 41777681321fe00f3f2e6aac6b7c9d96177209ec [file] [log] [blame]
khenaidoo59ce9dd2019-11-11 13:05:32 -05001// 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 etcdserver
16
17import (
18 "io"
19
20 "go.etcd.io/etcd/etcdserver/api/snap"
21 "go.etcd.io/etcd/mvcc/backend"
22 "go.etcd.io/etcd/raft/raftpb"
23
24 humanize "github.com/dustin/go-humanize"
25 "go.uber.org/zap"
26)
27
28// createMergedSnapshotMessage creates a snapshot message that contains: raft status (term, conf),
29// a snapshot of v2 store inside raft.Snapshot as []byte, a snapshot of v3 KV in the top level message
30// as ReadCloser.
31func (s *EtcdServer) createMergedSnapshotMessage(m raftpb.Message, snapt, snapi uint64, confState raftpb.ConfState) snap.Message {
32 // get a snapshot of v2 store as []byte
33 clone := s.v2store.Clone()
34 d, err := clone.SaveNoCopy()
35 if err != nil {
36 if lg := s.getLogger(); lg != nil {
37 lg.Panic("failed to save v2 store data", zap.Error(err))
38 } else {
39 plog.Panicf("store save should never fail: %v", err)
40 }
41 }
42
43 // commit kv to write metadata(for example: consistent index).
44 s.KV().Commit()
45 dbsnap := s.be.Snapshot()
46 // get a snapshot of v3 KV as readCloser
47 rc := newSnapshotReaderCloser(s.getLogger(), dbsnap)
48
49 // put the []byte snapshot of store into raft snapshot and return the merged snapshot with
50 // KV readCloser snapshot.
51 snapshot := raftpb.Snapshot{
52 Metadata: raftpb.SnapshotMetadata{
53 Index: snapi,
54 Term: snapt,
55 ConfState: confState,
56 },
57 Data: d,
58 }
59 m.Snapshot = snapshot
60
61 return *snap.NewMessage(m, rc, dbsnap.Size())
62}
63
64func newSnapshotReaderCloser(lg *zap.Logger, snapshot backend.Snapshot) io.ReadCloser {
65 pr, pw := io.Pipe()
66 go func() {
67 n, err := snapshot.WriteTo(pw)
68 if err == nil {
69 if lg != nil {
70 lg.Info(
71 "sent database snapshot to writer",
72 zap.Int64("bytes", n),
73 zap.String("size", humanize.Bytes(uint64(n))),
74 )
75 } else {
76 plog.Infof("wrote database snapshot out [total bytes: %d]", n)
77 }
78 } else {
79 if lg != nil {
80 lg.Warn(
81 "failed to send database snapshot to writer",
82 zap.String("size", humanize.Bytes(uint64(n))),
83 zap.Error(err),
84 )
85 } else {
86 plog.Warningf("failed to write database snapshot out [written bytes: %d]: %v", n, err)
87 }
88 }
89 pw.CloseWithError(err)
90 err = snapshot.Close()
91 if err != nil {
92 if lg != nil {
93 lg.Panic("failed to close database snapshot", zap.Error(err))
94 } else {
95 plog.Panicf("failed to close database snapshot: %v", err)
96 }
97 }
98 }()
99 return pr
100}