2015-01-25 06:19:16 +03:00
|
|
|
// Copyright 2015 CoreOS, Inc.
|
|
|
|
//
|
|
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
// you may not use this file except in compliance with the License.
|
|
|
|
// You may obtain a copy of the License at
|
|
|
|
//
|
|
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
//
|
|
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
// See the License for the specific language governing permissions and
|
|
|
|
// limitations under the License.
|
2014-11-17 03:23:47 +03:00
|
|
|
|
|
|
|
package rafthttp
|
|
|
|
|
|
|
|
import (
|
2015-06-02 21:46:36 +03:00
|
|
|
"errors"
|
2015-10-11 21:49:52 +03:00
|
|
|
"fmt"
|
2014-11-17 03:23:47 +03:00
|
|
|
"io/ioutil"
|
|
|
|
"net/http"
|
2014-11-19 00:53:14 +03:00
|
|
|
"path"
|
2014-11-17 03:23:47 +03:00
|
|
|
|
2015-01-30 02:59:58 +03:00
|
|
|
"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
|
2015-01-23 04:14:01 +03:00
|
|
|
pioutil "github.com/coreos/etcd/pkg/ioutil"
|
2014-11-17 03:23:47 +03:00
|
|
|
"github.com/coreos/etcd/pkg/types"
|
|
|
|
"github.com/coreos/etcd/raft/raftpb"
|
2015-12-08 18:52:54 +03:00
|
|
|
"github.com/coreos/etcd/snap"
|
2015-05-07 22:40:22 +03:00
|
|
|
"github.com/coreos/etcd/version"
|
2014-11-17 03:23:47 +03:00
|
|
|
)
|
|
|
|
|
2014-11-25 00:32:09 +03:00
|
|
|
const (
|
2015-10-18 08:20:36 +03:00
|
|
|
// connReadLimitByte limits the number of bytes
|
|
|
|
// a single read can read out.
|
2015-10-18 09:30:30 +03:00
|
|
|
//
|
2015-10-18 08:20:36 +03:00
|
|
|
// 64KB should be large enough for not causing
|
|
|
|
// throughput bottleneck as well as small enough
|
|
|
|
// for not causing a read timeout.
|
|
|
|
connReadLimitByte = 64 * 1024
|
2014-11-25 00:32:09 +03:00
|
|
|
)
|
|
|
|
|
2014-11-19 00:53:14 +03:00
|
|
|
var (
|
2015-10-11 21:49:52 +03:00
|
|
|
RaftPrefix = "/raft"
|
|
|
|
ProbingPrefix = path.Join(RaftPrefix, "probing")
|
|
|
|
RaftStreamPrefix = path.Join(RaftPrefix, "stream")
|
|
|
|
RaftSnapshotPrefix = path.Join(RaftPrefix, "snapshot")
|
2015-06-02 21:46:36 +03:00
|
|
|
|
|
|
|
errIncompatibleVersion = errors.New("incompatible version")
|
|
|
|
errClusterIDMismatch = errors.New("cluster ID mismatch")
|
2014-11-19 00:53:14 +03:00
|
|
|
)
|
|
|
|
|
2015-02-23 11:15:45 +03:00
|
|
|
type peerGetter interface {
|
|
|
|
Get(id types.ID) Peer
|
|
|
|
}
|
|
|
|
|
2015-02-15 19:44:34 +03:00
|
|
|
type writerToResponse interface {
|
|
|
|
WriteTo(w http.ResponseWriter)
|
|
|
|
}
|
|
|
|
|
2015-10-18 08:28:49 +03:00
|
|
|
type pipelineHandler struct {
|
2014-12-29 04:44:26 +03:00
|
|
|
r Raft
|
2014-11-17 03:23:47 +03:00
|
|
|
cid types.ID
|
|
|
|
}
|
|
|
|
|
2015-10-18 09:30:30 +03:00
|
|
|
// newPipelineHandler returns a handler for handling raft messages
|
2015-10-18 08:28:49 +03:00
|
|
|
// from pipeline for RaftPrefix.
|
|
|
|
//
|
|
|
|
// The handler reads out the raft message from request body,
|
|
|
|
// and forwards it to the given raft state machine for processing.
|
|
|
|
func newPipelineHandler(r Raft, cid types.ID) http.Handler {
|
|
|
|
return &pipelineHandler{
|
|
|
|
r: r,
|
|
|
|
cid: cid,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (h *pipelineHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
2014-11-17 03:23:47 +03:00
|
|
|
if r.Method != "POST" {
|
|
|
|
w.Header().Set("Allow", "POST")
|
|
|
|
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2015-10-11 21:49:52 +03:00
|
|
|
w.Header().Set("X-Etcd-Cluster-ID", h.cid.String())
|
2014-11-17 03:23:47 +03:00
|
|
|
|
2015-10-11 21:49:52 +03:00
|
|
|
if err := checkClusterCompatibilityFromHeader(r.Header, h.cid); err != nil {
|
|
|
|
http.Error(w, err.Error(), http.StatusPreconditionFailed)
|
2014-11-17 03:23:47 +03:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2014-11-25 00:32:09 +03:00
|
|
|
// Limit the data size that could be read from the request body, which ensures that read from
|
2015-10-18 08:20:36 +03:00
|
|
|
// connection will not time out accidentally due to possible blocking in underlying implementation.
|
|
|
|
limitedr := pioutil.NewLimitedBufferReader(r.Body, connReadLimitByte)
|
2014-11-25 00:32:09 +03:00
|
|
|
b, err := ioutil.ReadAll(limitedr)
|
2014-11-17 03:23:47 +03:00
|
|
|
if err != nil {
|
2015-06-03 01:33:58 +03:00
|
|
|
plog.Errorf("failed to read raft message (%v)", err)
|
2014-11-17 03:23:47 +03:00
|
|
|
http.Error(w, "error reading raft message", http.StatusBadRequest)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
var m raftpb.Message
|
|
|
|
if err := m.Unmarshal(b); err != nil {
|
2015-06-03 01:33:58 +03:00
|
|
|
plog.Errorf("failed to unmarshal raft message (%v)", err)
|
2014-11-17 03:23:47 +03:00
|
|
|
http.Error(w, "error unmarshaling raft message", http.StatusBadRequest)
|
|
|
|
return
|
|
|
|
}
|
2014-12-29 04:44:26 +03:00
|
|
|
if err := h.r.Process(context.TODO(), m); err != nil {
|
2014-11-17 03:23:47 +03:00
|
|
|
switch v := err.(type) {
|
|
|
|
case writerToResponse:
|
|
|
|
v.WriteTo(w)
|
|
|
|
default:
|
2015-06-03 01:33:58 +03:00
|
|
|
plog.Warningf("failed to process raft message (%v)", err)
|
2014-11-17 03:23:47 +03:00
|
|
|
http.Error(w, "error processing raft message", http.StatusInternalServerError)
|
|
|
|
}
|
|
|
|
return
|
|
|
|
}
|
2015-03-01 19:17:14 +03:00
|
|
|
// Write StatusNoContet header after the message has been processed by
|
2015-09-11 01:06:56 +03:00
|
|
|
// raft, which facilitates the client to report MsgSnap status.
|
2014-11-17 03:23:47 +03:00
|
|
|
w.WriteHeader(http.StatusNoContent)
|
|
|
|
}
|
|
|
|
|
2015-10-11 21:49:52 +03:00
|
|
|
type snapshotHandler struct {
|
2015-12-08 18:52:54 +03:00
|
|
|
r Raft
|
|
|
|
snapshotter *snap.Snapshotter
|
|
|
|
cid types.ID
|
2015-10-11 21:49:52 +03:00
|
|
|
}
|
|
|
|
|
2015-12-08 18:52:54 +03:00
|
|
|
func newSnapshotHandler(r Raft, snapshotter *snap.Snapshotter, cid types.ID) http.Handler {
|
2015-10-18 08:34:48 +03:00
|
|
|
return &snapshotHandler{
|
2015-12-08 18:52:54 +03:00
|
|
|
r: r,
|
|
|
|
snapshotter: snapshotter,
|
|
|
|
cid: cid,
|
2015-10-18 08:34:48 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-10-11 21:49:52 +03:00
|
|
|
// ServeHTTP serves HTTP request to receive and process snapshot message.
|
|
|
|
//
|
|
|
|
// If request sender dies without closing underlying TCP connection,
|
|
|
|
// the handler will keep waiting for the request body until TCP keepalive
|
|
|
|
// finds out that the connection is broken after several minutes.
|
|
|
|
// This is acceptable because
|
|
|
|
// 1. snapshot messages sent through other TCP connections could still be
|
|
|
|
// received and processed.
|
|
|
|
// 2. this case should happen rarely, so no further optimization is done.
|
|
|
|
func (h *snapshotHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
|
|
if r.Method != "POST" {
|
|
|
|
w.Header().Set("Allow", "POST")
|
|
|
|
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
w.Header().Set("X-Etcd-Cluster-ID", h.cid.String())
|
|
|
|
|
|
|
|
if err := checkClusterCompatibilityFromHeader(r.Header, h.cid); err != nil {
|
|
|
|
http.Error(w, err.Error(), http.StatusPreconditionFailed)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
dec := &messageDecoder{r: r.Body}
|
|
|
|
m, err := dec.decode()
|
|
|
|
if err != nil {
|
|
|
|
msg := fmt.Sprintf("failed to decode raft message (%v)", err)
|
|
|
|
plog.Errorf(msg)
|
|
|
|
http.Error(w, msg, http.StatusBadRequest)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
if m.Type != raftpb.MsgSnap {
|
|
|
|
plog.Errorf("unexpected raft message type %s on snapshot path", m.Type)
|
|
|
|
http.Error(w, "wrong raft message type", http.StatusBadRequest)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2015-12-23 09:54:36 +03:00
|
|
|
plog.Infof("receiving database snapshot [index:%d, from %s] ...", m.Snapshot.Metadata.Index, types.ID(m.From))
|
2015-12-08 18:52:54 +03:00
|
|
|
// save incoming database snapshot.
|
|
|
|
if err := h.snapshotter.SaveDBFrom(r.Body, m.Snapshot.Metadata.Index); err != nil {
|
2015-10-11 21:49:52 +03:00
|
|
|
msg := fmt.Sprintf("failed to save KV snapshot (%v)", err)
|
|
|
|
plog.Error(msg)
|
|
|
|
http.Error(w, msg, http.StatusInternalServerError)
|
|
|
|
return
|
|
|
|
}
|
2015-12-08 18:52:54 +03:00
|
|
|
plog.Infof("received and saved database snapshot [index: %d, from: %s] successfully", m.Snapshot.Metadata.Index, types.ID(m.From))
|
2015-10-11 21:49:52 +03:00
|
|
|
|
|
|
|
if err := h.r.Process(context.TODO(), m); err != nil {
|
|
|
|
switch v := err.(type) {
|
|
|
|
// Process may return writerToResponse error when doing some
|
|
|
|
// additional checks before calling raft.Node.Step.
|
|
|
|
case writerToResponse:
|
|
|
|
v.WriteTo(w)
|
|
|
|
default:
|
|
|
|
msg := fmt.Sprintf("failed to process raft message (%v)", err)
|
|
|
|
plog.Warningf(msg)
|
|
|
|
http.Error(w, msg, http.StatusInternalServerError)
|
|
|
|
}
|
|
|
|
return
|
|
|
|
}
|
|
|
|
// Write StatusNoContet header after the message has been processed by
|
|
|
|
// raft, which facilitates the client to report MsgSnap status.
|
|
|
|
w.WriteHeader(http.StatusNoContent)
|
|
|
|
}
|
|
|
|
|
2014-11-19 00:53:14 +03:00
|
|
|
type streamHandler struct {
|
2015-02-23 11:15:45 +03:00
|
|
|
peerGetter peerGetter
|
2015-04-26 09:24:59 +03:00
|
|
|
r Raft
|
2015-02-23 11:15:45 +03:00
|
|
|
id types.ID
|
|
|
|
cid types.ID
|
2014-11-19 00:53:14 +03:00
|
|
|
}
|
|
|
|
|
2015-10-18 08:34:48 +03:00
|
|
|
func newStreamHandler(peerGetter peerGetter, r Raft, id, cid types.ID) http.Handler {
|
|
|
|
return &streamHandler{
|
|
|
|
peerGetter: peerGetter,
|
|
|
|
r: r,
|
|
|
|
id: id,
|
|
|
|
cid: cid,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-11-19 00:53:14 +03:00
|
|
|
func (h *streamHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
|
|
if r.Method != "GET" {
|
|
|
|
w.Header().Set("Allow", "GET")
|
|
|
|
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2015-06-02 21:46:36 +03:00
|
|
|
w.Header().Set("X-Server-Version", version.Version)
|
2015-10-11 21:49:52 +03:00
|
|
|
w.Header().Set("X-Etcd-Cluster-ID", h.cid.String())
|
2015-06-02 21:46:36 +03:00
|
|
|
|
2015-10-11 21:49:52 +03:00
|
|
|
if err := checkClusterCompatibilityFromHeader(r.Header, h.cid); err != nil {
|
|
|
|
http.Error(w, err.Error(), http.StatusPreconditionFailed)
|
2015-05-27 02:00:11 +03:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2015-02-15 19:44:34 +03:00
|
|
|
var t streamType
|
|
|
|
switch path.Dir(r.URL.Path) {
|
2015-10-20 09:41:01 +03:00
|
|
|
case streamTypeMsgAppV2.endpoint():
|
2015-04-03 07:55:13 +03:00
|
|
|
t = streamTypeMsgAppV2
|
2015-10-20 09:41:01 +03:00
|
|
|
case streamTypeMessage.endpoint():
|
2015-04-03 07:55:13 +03:00
|
|
|
t = streamTypeMessage
|
2015-02-15 19:44:34 +03:00
|
|
|
default:
|
2015-06-03 01:33:58 +03:00
|
|
|
plog.Debugf("ignored unexpected streaming request path %s", r.URL.Path)
|
2015-02-15 19:44:34 +03:00
|
|
|
http.Error(w, "invalid path", http.StatusNotFound)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
fromStr := path.Base(r.URL.Path)
|
2014-11-19 00:53:14 +03:00
|
|
|
from, err := types.IDFromString(fromStr)
|
|
|
|
if err != nil {
|
2015-06-03 01:33:58 +03:00
|
|
|
plog.Errorf("failed to parse from %s into ID (%v)", fromStr, err)
|
2015-02-15 19:44:34 +03:00
|
|
|
http.Error(w, "invalid from", http.StatusNotFound)
|
2014-11-19 00:53:14 +03:00
|
|
|
return
|
|
|
|
}
|
2015-04-26 09:24:59 +03:00
|
|
|
if h.r.IsIDRemoved(uint64(from)) {
|
2015-06-03 01:33:58 +03:00
|
|
|
plog.Warningf("rejected the stream from peer %s since it was removed", from)
|
2015-04-26 09:24:59 +03:00
|
|
|
http.Error(w, "removed member", http.StatusGone)
|
|
|
|
return
|
|
|
|
}
|
2015-02-23 11:15:45 +03:00
|
|
|
p := h.peerGetter.Get(from)
|
2014-12-29 04:44:26 +03:00
|
|
|
if p == nil {
|
2015-06-03 01:33:58 +03:00
|
|
|
// This may happen in following cases:
|
|
|
|
// 1. user starts a remote peer that belongs to a different cluster
|
|
|
|
// with the same cluster ID.
|
|
|
|
// 2. local etcd falls behind of the cluster, and cannot recognize
|
|
|
|
// the members that joined after its current progress.
|
2015-10-11 21:49:52 +03:00
|
|
|
plog.Errorf("failed to find member %s in cluster %s", from, h.cid)
|
2014-11-19 00:53:14 +03:00
|
|
|
http.Error(w, "error sender not found", http.StatusNotFound)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
wto := h.id.String()
|
|
|
|
if gto := r.Header.Get("X-Raft-To"); gto != wto {
|
2015-06-03 01:33:58 +03:00
|
|
|
plog.Errorf("streaming request ignored (ID mismatch got %s want %s)", gto, wto)
|
2014-11-19 00:53:14 +03:00
|
|
|
http.Error(w, "to field mismatch", http.StatusPreconditionFailed)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2015-02-15 19:44:34 +03:00
|
|
|
w.WriteHeader(http.StatusOK)
|
|
|
|
w.(http.Flusher).Flush()
|
2014-11-19 00:53:14 +03:00
|
|
|
|
2015-02-15 19:44:34 +03:00
|
|
|
c := newCloseNotifier()
|
|
|
|
conn := &outgoingConn{
|
|
|
|
t: t,
|
|
|
|
Writer: w,
|
|
|
|
Flusher: w.(http.Flusher),
|
|
|
|
Closer: c,
|
2014-11-19 00:53:14 +03:00
|
|
|
}
|
2015-02-15 19:44:34 +03:00
|
|
|
p.attachOutgoingConn(conn)
|
|
|
|
<-c.closeNotify()
|
|
|
|
}
|
2015-01-16 01:44:24 +03:00
|
|
|
|
2015-10-11 21:49:52 +03:00
|
|
|
// checkClusterCompatibilityFromHeader checks the cluster compatibility of
|
|
|
|
// the local member from the given header.
|
|
|
|
// It checks whether the version of local member is compatible with
|
|
|
|
// the versions in the header, and whether the cluster ID of local member
|
|
|
|
// matches the one in the header.
|
|
|
|
func checkClusterCompatibilityFromHeader(header http.Header, cid types.ID) error {
|
|
|
|
if err := checkVersionCompability(header.Get("X-Server-From"), serverVersion(header), minClusterVersion(header)); err != nil {
|
|
|
|
plog.Errorf("request version incompatibility (%v)", err)
|
|
|
|
return errIncompatibleVersion
|
|
|
|
}
|
|
|
|
if gcid := header.Get("X-Etcd-Cluster-ID"); gcid != cid.String() {
|
|
|
|
plog.Errorf("request cluster ID mismatch (got %s want %s)", gcid, cid)
|
|
|
|
return errClusterIDMismatch
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2015-02-15 19:44:34 +03:00
|
|
|
type closeNotifier struct {
|
|
|
|
done chan struct{}
|
2014-11-19 00:53:14 +03:00
|
|
|
}
|
|
|
|
|
2015-02-15 19:44:34 +03:00
|
|
|
func newCloseNotifier() *closeNotifier {
|
|
|
|
return &closeNotifier{
|
|
|
|
done: make(chan struct{}),
|
|
|
|
}
|
2014-11-17 03:23:47 +03:00
|
|
|
}
|
2015-02-15 19:44:34 +03:00
|
|
|
|
|
|
|
func (n *closeNotifier) Close() error {
|
|
|
|
close(n.done)
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (n *closeNotifier) closeNotify() <-chan struct{} { return n.done }
|