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
15package rafthttp
16
17import (
18	"bytes"
19	"context"
20	"io"
21	"io/ioutil"
22	"net/http"
23	"time"
24
25	"go.etcd.io/etcd/etcdserver/api/snap"
26	"go.etcd.io/etcd/pkg/httputil"
27	pioutil "go.etcd.io/etcd/pkg/ioutil"
28	"go.etcd.io/etcd/pkg/types"
29	"go.etcd.io/etcd/raft"
30
31	"github.com/dustin/go-humanize"
32	"go.uber.org/zap"
33)
34
35var (
36	// timeout for reading snapshot response body
37	snapResponseReadTimeout = 5 * time.Second
38)
39
40type snapshotSender struct {
41	from, to types.ID
42	cid      types.ID
43
44	tr     *Transport
45	picker *urlPicker
46	status *peerStatus
47	r      Raft
48	errorc chan error
49
50	stopc chan struct{}
51}
52
53func newSnapshotSender(tr *Transport, picker *urlPicker, to types.ID, status *peerStatus) *snapshotSender {
54	return &snapshotSender{
55		from:   tr.ID,
56		to:     to,
57		cid:    tr.ClusterID,
58		tr:     tr,
59		picker: picker,
60		status: status,
61		r:      tr.Raft,
62		errorc: tr.ErrorC,
63		stopc:  make(chan struct{}),
64	}
65}
66
67func (s *snapshotSender) stop() { close(s.stopc) }
68
69func (s *snapshotSender) send(merged snap.Message) {
70	start := time.Now()
71
72	m := merged.Message
73	to := types.ID(m.To).String()
74
75	body := createSnapBody(s.tr.Logger, merged)
76	defer body.Close()
77
78	u := s.picker.pick()
79	req := createPostRequest(u, RaftSnapshotPrefix, body, "application/octet-stream", s.tr.URLs, s.from, s.cid)
80
81	snapshotTotalSizeVal := uint64(merged.TotalSize)
82	snapshotTotalSize := humanize.Bytes(snapshotTotalSizeVal)
83	if s.tr.Logger != nil {
84		s.tr.Logger.Info(
85			"sending database snapshot",
86			zap.Uint64("snapshot-index", m.Snapshot.Metadata.Index),
87			zap.String("remote-peer-id", to),
88			zap.Int64("bytes", merged.TotalSize),
89			zap.String("size", snapshotTotalSize),
90		)
91	} else {
92		plog.Infof("start to send database snapshot [index: %d, to %s, size %s]...", m.Snapshot.Metadata.Index, types.ID(m.To), snapshotTotalSize)
93	}
94
95	snapshotSendInflights.WithLabelValues(to).Inc()
96	defer func() {
97		snapshotSendInflights.WithLabelValues(to).Dec()
98	}()
99
100	err := s.post(req)
101	defer merged.CloseWithError(err)
102	if err != nil {
103		if s.tr.Logger != nil {
104			s.tr.Logger.Warn(
105				"failed to send database snapshot",
106				zap.Uint64("snapshot-index", m.Snapshot.Metadata.Index),
107				zap.String("remote-peer-id", to),
108				zap.Int64("bytes", merged.TotalSize),
109				zap.String("size", snapshotTotalSize),
110				zap.Error(err),
111			)
112		} else {
113			plog.Warningf("database snapshot [index: %d, to: %s] failed to be sent out (%v)", m.Snapshot.Metadata.Index, types.ID(m.To), err)
114		}
115
116		// errMemberRemoved is a critical error since a removed member should
117		// always be stopped. So we use reportCriticalError to report it to errorc.
118		if err == errMemberRemoved {
119			reportCriticalError(err, s.errorc)
120		}
121
122		s.picker.unreachable(u)
123		s.status.deactivate(failureType{source: sendSnap, action: "post"}, err.Error())
124		s.r.ReportUnreachable(m.To)
125		// report SnapshotFailure to raft state machine. After raft state
126		// machine knows about it, it would pause a while and retry sending
127		// new snapshot message.
128		s.r.ReportSnapshot(m.To, raft.SnapshotFailure)
129		sentFailures.WithLabelValues(to).Inc()
130		snapshotSendFailures.WithLabelValues(to).Inc()
131		return
132	}
133	s.status.activate()
134	s.r.ReportSnapshot(m.To, raft.SnapshotFinish)
135
136	if s.tr.Logger != nil {
137		s.tr.Logger.Info(
138			"sent database snapshot",
139			zap.Uint64("snapshot-index", m.Snapshot.Metadata.Index),
140			zap.String("remote-peer-id", to),
141			zap.Int64("bytes", merged.TotalSize),
142			zap.String("size", snapshotTotalSize),
143		)
144	} else {
145		plog.Infof("database snapshot [index: %d, to: %s] sent out successfully", m.Snapshot.Metadata.Index, types.ID(m.To))
146	}
147
148	sentBytes.WithLabelValues(to).Add(float64(merged.TotalSize))
149	snapshotSend.WithLabelValues(to).Inc()
150	snapshotSendSeconds.WithLabelValues(to).Observe(time.Since(start).Seconds())
151}
152
153// post posts the given request.
154// It returns nil when request is sent out and processed successfully.
155func (s *snapshotSender) post(req *http.Request) (err error) {
156	ctx, cancel := context.WithCancel(context.Background())
157	req = req.WithContext(ctx)
158	defer cancel()
159
160	type responseAndError struct {
161		resp *http.Response
162		body []byte
163		err  error
164	}
165	result := make(chan responseAndError, 1)
166
167	go func() {
168		resp, err := s.tr.pipelineRt.RoundTrip(req)
169		if err != nil {
170			result <- responseAndError{resp, nil, err}
171			return
172		}
173
174		// close the response body when timeouts.
175		// prevents from reading the body forever when the other side dies right after
176		// successfully receives the request body.
177		time.AfterFunc(snapResponseReadTimeout, func() { httputil.GracefulClose(resp) })
178		body, err := ioutil.ReadAll(resp.Body)
179		result <- responseAndError{resp, body, err}
180	}()
181
182	select {
183	case <-s.stopc:
184		return errStopped
185	case r := <-result:
186		if r.err != nil {
187			return r.err
188		}
189		return checkPostResponse(r.resp, r.body, req, s.to)
190	}
191}
192
193func createSnapBody(lg *zap.Logger, merged snap.Message) io.ReadCloser {
194	buf := new(bytes.Buffer)
195	enc := &messageEncoder{w: buf}
196	// encode raft message
197	if err := enc.encode(&merged.Message); err != nil {
198		if lg != nil {
199			lg.Panic("failed to encode message", zap.Error(err))
200		} else {
201			plog.Panicf("encode message error (%v)", err)
202		}
203	}
204
205	return &pioutil.ReaderAndCloser{
206		Reader: io.MultiReader(buf, merged.ReadCloser),
207		Closer: merged.ReadCloser,
208	}
209}
210