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-26 08:38:57 +03:00
|
|
|
package rafthttp
|
|
|
|
|
|
|
|
import (
|
2014-12-29 04:44:26 +03:00
|
|
|
"log"
|
2014-11-26 08:38:57 +03:00
|
|
|
"net/http"
|
2014-12-29 04:44:26 +03:00
|
|
|
"sync"
|
2014-11-26 08:38:57 +03:00
|
|
|
|
2015-01-30 02:59:58 +03:00
|
|
|
"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
|
2014-11-26 08:38:57 +03:00
|
|
|
"github.com/coreos/etcd/etcdserver/stats"
|
|
|
|
"github.com/coreos/etcd/pkg/types"
|
2015-03-01 19:17:14 +03:00
|
|
|
"github.com/coreos/etcd/raft"
|
2014-11-26 08:38:57 +03:00
|
|
|
"github.com/coreos/etcd/raft/raftpb"
|
|
|
|
)
|
|
|
|
|
2014-12-29 04:44:26 +03:00
|
|
|
type Raft interface {
|
2014-11-26 08:38:57 +03:00
|
|
|
Process(ctx context.Context, m raftpb.Message) error
|
2015-03-01 18:02:24 +03:00
|
|
|
ReportUnreachable(id uint64)
|
2015-03-01 19:17:14 +03:00
|
|
|
ReportSnapshot(id uint64, status raft.SnapshotStatus)
|
2014-11-26 08:38:57 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
type Transporter interface {
|
2015-03-03 21:22:41 +03:00
|
|
|
// Handler returns the HTTP handler of the transporter.
|
|
|
|
// A transporter HTTP handler handles the HTTP requests
|
|
|
|
// from remote peers.
|
|
|
|
// The handler MUST be used to handle RaftPrefix(/raft)
|
|
|
|
// endpoint.
|
2014-11-26 08:38:57 +03:00
|
|
|
Handler() http.Handler
|
2015-03-03 21:22:41 +03:00
|
|
|
// Send sends out the given messages to the remote peers.
|
|
|
|
// Each message has a To field, which is an id that maps
|
|
|
|
// to an existing peer in the transport.
|
|
|
|
// If the id cannot be found in the transport, the message
|
|
|
|
// will be ignored.
|
2014-11-26 08:38:57 +03:00
|
|
|
Send(m []raftpb.Message)
|
2015-03-03 21:22:41 +03:00
|
|
|
// AddPeer adds a peer with given peer urls into the transport.
|
|
|
|
// It is the caller's responsibility to ensure the urls are all vaild,
|
|
|
|
// or it panics.
|
|
|
|
// Peer urls are used to connect to the remote peer.
|
2014-11-26 08:38:57 +03:00
|
|
|
AddPeer(id types.ID, urls []string)
|
2015-03-03 21:22:41 +03:00
|
|
|
// RemovePeer removes the peer with given id.
|
2014-11-26 08:38:57 +03:00
|
|
|
RemovePeer(id types.ID)
|
2015-03-03 21:22:41 +03:00
|
|
|
// RemoveAllPeers removes all the existing peers in the transport.
|
2015-02-12 21:48:06 +03:00
|
|
|
RemoveAllPeers()
|
2015-03-03 21:22:41 +03:00
|
|
|
// UpdatePeer updates the peer urls of the peer with the given id.
|
|
|
|
// It is the caller's responsibility to ensure the urls are all vaild,
|
|
|
|
// or it panics.
|
2014-11-26 08:38:57 +03:00
|
|
|
UpdatePeer(id types.ID, urls []string)
|
2015-03-03 21:22:41 +03:00
|
|
|
// Stop closes the connections and stops the transporter.
|
2014-11-26 08:38:57 +03:00
|
|
|
Stop()
|
|
|
|
}
|
|
|
|
|
2014-12-29 23:20:52 +03:00
|
|
|
type transport struct {
|
2014-12-29 23:08:13 +03:00
|
|
|
roundTripper http.RoundTripper
|
|
|
|
id types.ID
|
|
|
|
clusterID types.ID
|
|
|
|
raft Raft
|
|
|
|
serverStats *stats.ServerStats
|
|
|
|
leaderStats *stats.LeaderStats
|
2014-11-26 08:38:57 +03:00
|
|
|
|
2015-02-26 11:13:35 +03:00
|
|
|
mu sync.RWMutex // protect the peer map
|
|
|
|
peers map[types.ID]Peer // remote peers
|
2015-01-03 07:00:29 +03:00
|
|
|
errorc chan error
|
2014-11-26 08:38:57 +03:00
|
|
|
}
|
|
|
|
|
2015-01-03 07:00:29 +03:00
|
|
|
func NewTransporter(rt http.RoundTripper, id, cid types.ID, r Raft, errorc chan error, ss *stats.ServerStats, ls *stats.LeaderStats) Transporter {
|
2014-12-29 23:20:52 +03:00
|
|
|
return &transport{
|
2014-12-29 23:08:13 +03:00
|
|
|
roundTripper: rt,
|
|
|
|
id: id,
|
|
|
|
clusterID: cid,
|
|
|
|
raft: r,
|
|
|
|
serverStats: ss,
|
|
|
|
leaderStats: ls,
|
2015-02-26 11:13:35 +03:00
|
|
|
peers: make(map[types.ID]Peer),
|
2015-01-03 07:00:29 +03:00
|
|
|
errorc: errorc,
|
2014-12-29 23:08:13 +03:00
|
|
|
}
|
2014-12-29 04:44:26 +03:00
|
|
|
}
|
|
|
|
|
2014-12-29 23:20:52 +03:00
|
|
|
func (t *transport) Handler() http.Handler {
|
2015-02-15 19:44:34 +03:00
|
|
|
pipelineHandler := NewHandler(t.raft, t.clusterID)
|
|
|
|
streamHandler := newStreamHandler(t, t.id, t.clusterID)
|
2014-11-26 08:38:57 +03:00
|
|
|
mux := http.NewServeMux()
|
2015-02-15 19:44:34 +03:00
|
|
|
mux.Handle(RaftPrefix, pipelineHandler)
|
|
|
|
mux.Handle(RaftStreamPrefix+"/", streamHandler)
|
2014-12-29 04:44:26 +03:00
|
|
|
return mux
|
|
|
|
}
|
|
|
|
|
2015-02-23 11:15:45 +03:00
|
|
|
func (t *transport) Get(id types.ID) Peer {
|
2014-12-29 04:44:26 +03:00
|
|
|
t.mu.RLock()
|
|
|
|
defer t.mu.RUnlock()
|
|
|
|
return t.peers[id]
|
|
|
|
}
|
|
|
|
|
2014-12-29 23:20:52 +03:00
|
|
|
func (t *transport) Send(msgs []raftpb.Message) {
|
2014-12-29 04:44:26 +03:00
|
|
|
for _, m := range msgs {
|
|
|
|
// intentionally dropped message
|
|
|
|
if m.To == 0 {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
to := types.ID(m.To)
|
|
|
|
p, ok := t.peers[to]
|
|
|
|
if !ok {
|
|
|
|
log.Printf("etcdserver: send message to unknown receiver %s", to)
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
if m.Type == raftpb.MsgApp {
|
2014-12-29 23:08:13 +03:00
|
|
|
t.serverStats.SendAppendReq(m.Size())
|
2014-12-29 04:44:26 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
p.Send(m)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-12-29 23:20:52 +03:00
|
|
|
func (t *transport) Stop() {
|
2014-12-29 04:44:26 +03:00
|
|
|
for _, p := range t.peers {
|
|
|
|
p.Stop()
|
|
|
|
}
|
2014-12-29 23:08:13 +03:00
|
|
|
if tr, ok := t.roundTripper.(*http.Transport); ok {
|
2014-12-29 04:44:26 +03:00
|
|
|
tr.CloseIdleConnections()
|
|
|
|
}
|
2014-11-26 08:38:57 +03:00
|
|
|
}
|
|
|
|
|
2015-02-27 18:54:06 +03:00
|
|
|
func (t *transport) AddPeer(id types.ID, us []string) {
|
2014-12-29 04:44:26 +03:00
|
|
|
t.mu.Lock()
|
|
|
|
defer t.mu.Unlock()
|
2015-03-24 09:19:38 +03:00
|
|
|
// There is no need to build connection to itself because local message
|
|
|
|
// is not sent through transport.
|
|
|
|
if id == t.id {
|
|
|
|
return
|
|
|
|
}
|
2014-12-29 04:44:26 +03:00
|
|
|
if _, ok := t.peers[id]; ok {
|
|
|
|
return
|
|
|
|
}
|
2015-02-27 18:54:06 +03:00
|
|
|
urls, err := types.NewURLs(us)
|
2014-12-29 04:44:26 +03:00
|
|
|
if err != nil {
|
2015-02-27 18:54:06 +03:00
|
|
|
log.Panicf("newURLs %+v should never fail: %+v", us, err)
|
2014-12-29 04:44:26 +03:00
|
|
|
}
|
2014-12-29 23:08:13 +03:00
|
|
|
fs := t.leaderStats.Follower(id.String())
|
2015-02-27 18:54:06 +03:00
|
|
|
t.peers[id] = startPeer(t.roundTripper, urls, t.id, id, t.clusterID, t.raft, fs, t.errorc)
|
2014-12-29 04:44:26 +03:00
|
|
|
}
|
|
|
|
|
2014-12-29 23:20:52 +03:00
|
|
|
func (t *transport) RemovePeer(id types.ID) {
|
2014-12-29 04:44:26 +03:00
|
|
|
t.mu.Lock()
|
|
|
|
defer t.mu.Unlock()
|
2015-03-24 09:19:38 +03:00
|
|
|
if id == t.id {
|
|
|
|
return
|
|
|
|
}
|
2015-02-12 21:48:06 +03:00
|
|
|
t.removePeer(id)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (t *transport) RemoveAllPeers() {
|
|
|
|
t.mu.Lock()
|
|
|
|
defer t.mu.Unlock()
|
|
|
|
for id, _ := range t.peers {
|
|
|
|
t.removePeer(id)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// the caller of this function must have the peers mutex.
|
|
|
|
func (t *transport) removePeer(id types.ID) {
|
2015-02-12 05:37:02 +03:00
|
|
|
if peer, ok := t.peers[id]; ok {
|
|
|
|
peer.Stop()
|
|
|
|
} else {
|
|
|
|
log.Panicf("rafthttp: unexpected removal of unknown peer '%d'", id)
|
|
|
|
}
|
2014-12-29 04:44:26 +03:00
|
|
|
delete(t.peers, id)
|
2015-02-10 22:20:12 +03:00
|
|
|
delete(t.leaderStats.Followers, id.String())
|
2014-12-29 04:44:26 +03:00
|
|
|
}
|
|
|
|
|
2015-02-27 18:54:06 +03:00
|
|
|
func (t *transport) UpdatePeer(id types.ID, us []string) {
|
2014-12-29 04:44:26 +03:00
|
|
|
t.mu.Lock()
|
|
|
|
defer t.mu.Unlock()
|
2015-03-24 09:19:38 +03:00
|
|
|
if id == t.id {
|
|
|
|
return
|
|
|
|
}
|
2014-12-29 04:44:26 +03:00
|
|
|
// TODO: return error or just panic?
|
|
|
|
if _, ok := t.peers[id]; !ok {
|
|
|
|
return
|
|
|
|
}
|
2015-02-27 18:54:06 +03:00
|
|
|
urls, err := types.NewURLs(us)
|
2014-12-29 04:44:26 +03:00
|
|
|
if err != nil {
|
2015-02-27 18:54:06 +03:00
|
|
|
log.Panicf("newURLs %+v should never fail: %+v", us, err)
|
2014-12-29 04:44:26 +03:00
|
|
|
}
|
2015-02-27 18:54:06 +03:00
|
|
|
t.peers[id].Update(urls)
|
2014-12-29 04:44:26 +03:00
|
|
|
}
|
|
|
|
|
2014-12-29 23:20:52 +03:00
|
|
|
type Pausable interface {
|
|
|
|
Pause()
|
|
|
|
Resume()
|
|
|
|
}
|
|
|
|
|
2014-12-29 04:44:26 +03:00
|
|
|
// for testing
|
2014-12-29 23:20:52 +03:00
|
|
|
func (t *transport) Pause() {
|
2014-12-29 04:44:26 +03:00
|
|
|
for _, p := range t.peers {
|
2015-02-26 11:13:35 +03:00
|
|
|
p.(Pausable).Pause()
|
2014-12-29 04:44:26 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-12-29 23:20:52 +03:00
|
|
|
func (t *transport) Resume() {
|
2014-12-29 04:44:26 +03:00
|
|
|
for _, p := range t.peers {
|
2015-02-26 11:13:35 +03:00
|
|
|
p.(Pausable).Resume()
|
2014-12-29 04:44:26 +03:00
|
|
|
}
|
|
|
|
}
|