190 lines
4.4 KiB
Go
190 lines
4.4 KiB
Go
/*
|
|
Copyright 2014 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.
|
|
*/
|
|
|
|
package etcdserver
|
|
|
|
import (
|
|
"bytes"
|
|
"fmt"
|
|
"log"
|
|
"net/http"
|
|
"time"
|
|
|
|
"github.com/coreos/etcd/etcdserver/stats"
|
|
"github.com/coreos/etcd/pkg/types"
|
|
"github.com/coreos/etcd/raft/raftpb"
|
|
)
|
|
|
|
const (
|
|
raftPrefix = "/raft"
|
|
connPerSender = 4
|
|
)
|
|
|
|
type sendHub struct {
|
|
tr *http.Transport
|
|
cl ClusterInfo
|
|
ss *stats.ServerStats
|
|
ls *stats.LeaderStats
|
|
senders map[types.ID]*sender
|
|
}
|
|
|
|
// newSendHub creates the default send hub used to transport raft messages
|
|
// to other members. The returned sendHub will update the given ServerStats and
|
|
// LeaderStats appropriately.
|
|
func newSendHub(t *http.Transport, cl ClusterInfo, ss *stats.ServerStats, ls *stats.LeaderStats) *sendHub {
|
|
h := &sendHub{
|
|
tr: t,
|
|
cl: cl,
|
|
ss: ss,
|
|
ls: ls,
|
|
senders: make(map[types.ID]*sender),
|
|
}
|
|
for _, m := range cl.Members() {
|
|
h.Add(m)
|
|
}
|
|
return h
|
|
}
|
|
|
|
func (h *sendHub) Send(msgs []raftpb.Message) {
|
|
for _, m := range msgs {
|
|
to := types.ID(m.To)
|
|
s, ok := h.senders[to]
|
|
if !ok {
|
|
if !h.cl.IsIDRemoved(to) {
|
|
log.Printf("etcdserver: send message to unknown receiver %s", to)
|
|
}
|
|
continue
|
|
}
|
|
|
|
// TODO: don't block. we should be able to have 1000s
|
|
// of messages out at a time.
|
|
data, err := m.Marshal()
|
|
if err != nil {
|
|
log.Println("sender: dropping message:", err)
|
|
return // drop bad message
|
|
}
|
|
if m.Type == raftpb.MsgApp {
|
|
h.ss.SendAppendReq(len(data))
|
|
}
|
|
|
|
// TODO (xiangli): reasonable retry logic
|
|
s.send(data)
|
|
}
|
|
}
|
|
|
|
func (h *sendHub) Stop() {
|
|
for _, s := range h.senders {
|
|
s.stop()
|
|
}
|
|
}
|
|
|
|
func (h *sendHub) Add(m *Member) {
|
|
if _, ok := h.senders[m.ID]; ok {
|
|
return
|
|
}
|
|
// TODO: considering how to switch between all available peer urls
|
|
u := fmt.Sprintf("%s%s", m.PickPeerURL(), raftPrefix)
|
|
c := &http.Client{Transport: h.tr}
|
|
fs := h.ls.Follower(m.ID.String())
|
|
s := newSender(u, h.cl.ID(), c, fs)
|
|
h.senders[m.ID] = s
|
|
}
|
|
|
|
func (h *sendHub) Remove(id types.ID) {
|
|
h.senders[id].stop()
|
|
delete(h.senders, id)
|
|
}
|
|
|
|
type sender struct {
|
|
u string
|
|
cid types.ID
|
|
c *http.Client
|
|
fs *stats.FollowerStats
|
|
q chan []byte
|
|
}
|
|
|
|
func newSender(u string, cid types.ID, c *http.Client, fs *stats.FollowerStats) *sender {
|
|
s := &sender{
|
|
u: u,
|
|
cid: cid,
|
|
c: c,
|
|
fs: fs,
|
|
q: make(chan []byte),
|
|
}
|
|
for i := 0; i < connPerSender; i++ {
|
|
go s.handle()
|
|
}
|
|
return s
|
|
}
|
|
|
|
func (s *sender) send(data []byte) {
|
|
select {
|
|
case s.q <- data:
|
|
default:
|
|
log.Printf("sender: reach the maximal serving to %s", s.u)
|
|
}
|
|
}
|
|
|
|
func (s *sender) stop() {
|
|
close(s.q)
|
|
}
|
|
|
|
func (s *sender) handle() {
|
|
for d := range s.q {
|
|
start := time.Now()
|
|
err := s.post(d)
|
|
end := time.Now()
|
|
if err != nil {
|
|
s.fs.Fail()
|
|
log.Printf("sender: %v", err)
|
|
continue
|
|
}
|
|
s.fs.Succ(end.Sub(start))
|
|
}
|
|
}
|
|
|
|
// post POSTs a data payload to a url. Returns nil if the POST succeeds,
|
|
// error on any failure.
|
|
func (s *sender) post(data []byte) error {
|
|
req, err := http.NewRequest("POST", s.u, bytes.NewBuffer(data))
|
|
if err != nil {
|
|
return fmt.Errorf("new request to %s error: %v", s.u, err)
|
|
}
|
|
req.Header.Set("Content-Type", "application/protobuf")
|
|
req.Header.Set("X-Etcd-Cluster-ID", s.cid.String())
|
|
resp, err := s.c.Do(req)
|
|
if err != nil {
|
|
return fmt.Errorf("error posting to %q: %v", req.URL.String(), err)
|
|
}
|
|
resp.Body.Close()
|
|
|
|
switch resp.StatusCode {
|
|
case http.StatusPreconditionFailed:
|
|
// TODO: shutdown the etcdserver gracefully?
|
|
log.Fatalf("etcd: conflicting cluster ID with the target cluster (%s != %s)", resp.Header.Get("X-Etcd-Cluster-ID"), s.cid)
|
|
return nil
|
|
case http.StatusForbidden:
|
|
// TODO: stop the server
|
|
log.Println("etcd: this member has been permanently removed from the cluster")
|
|
log.Fatalln("etcd: the data-dir used by this member must be removed so that this host can be re-added with a new member ID")
|
|
return nil
|
|
case http.StatusNoContent:
|
|
return nil
|
|
default:
|
|
return fmt.Errorf("unhandled status %s", http.StatusText(resp.StatusCode))
|
|
}
|
|
}
|