Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

*: lease forwarding should resue transport

Browse Source
This commit is contained in:
Xiang Li 2016-01-25 06:56:07 +08:00
parent 72ffa74476
commit 1aa312fcce
3 changed files with 9 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
etcdserver/server.go
View File

@ -179,8 +179,9 @@ type EtcdServer struct {
// consistent index used to hold the offset of current executing entry // consistent index used to hold the offset of current executing entry
// It is initialized to 0 before executing any entry. // It is initialized to 0 before executing any entry.
consistIndex consistentIndex consistIndex consistentIndex
// versionRt used to send requests for peer version
versionRt http.RoundTripper // peerRt used to send requests (version, lease) to peers.
peerRt http.RoundTripper
reqIDGen *idutil.Generator reqIDGen *idutil.Generator
// forceVersionC is used to force the version monitor loop // forceVersionC is used to force the version monitor loop
@ -353,7 +354,7 @@ func NewServer(cfg *ServerConfig) (*EtcdServer, error) {
stats: sstats, stats: sstats,
lstats: lstats, lstats: lstats,
SyncTicker: time.Tick(500 * time.Millisecond), SyncTicker: time.Tick(500 * time.Millisecond),
versionRt: prt, peerRt: prt,
reqIDGen: idutil.NewGenerator(uint8(id), time.Now()), reqIDGen: idutil.NewGenerator(uint8(id), time.Now()),
forceVersionC: make(chan struct{}), forceVersionC: make(chan struct{}),
msgSnapC: make(chan raftpb.Message, maxInFlightMsgSnap), msgSnapC: make(chan raftpb.Message, maxInFlightMsgSnap),
@ -1223,7 +1224,7 @@ func (s *EtcdServer) monitorVersions() {
continue continue
} }
v := decideClusterVersion(getVersions(s.cluster, s.id, s.versionRt)) v := decideClusterVersion(getVersions(s.cluster, s.id, s.peerRt))
if v != nil { if v != nil {
// only keep major.minor version for comparison // only keep major.minor version for comparison
v = &semver.Version{ v = &semver.Version{

2
etcdserver/v3demo_server.go
View File

@ -140,7 +140,7 @@ func (s *EtcdServer) LeaseRenew(id lease.LeaseID) (int64, error) {
for _, url := range leader.PeerURLs { for _, url := range leader.PeerURLs {
lurl := url + "/leases" lurl := url + "/leases"
ttl, err = leasehttp.RenewHTTP(id, lurl, s.cfg.PeerTLSInfo, s.cfg.peerDialTimeout()) ttl, err = leasehttp.RenewHTTP(id, lurl, s.peerRt, s.cfg.peerDialTimeout())
if err == nil { if err == nil {
break break
} }

11
lease/leasehttp/http.go
View File

@ -23,7 +23,6 @@ import (
pb "github.com/coreos/etcd/etcdserver/etcdserverpb" pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
"github.com/coreos/etcd/lease" "github.com/coreos/etcd/lease"
"github.com/coreos/etcd/pkg/transport"
) )
// NewHandler returns an http Handler for lease renewals // NewHandler returns an http Handler for lease renewals
@ -70,20 +69,14 @@ func (h *leaseHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
} }
// RenewHTTP renews a lease at a given primary server. // RenewHTTP renews a lease at a given primary server.
func RenewHTTP(id lease.LeaseID, url string, tlsInfo transport.TLSInfo, timeout time.Duration) (int64, error) { // TODO: Batch request in future?
func RenewHTTP(id lease.LeaseID, url string, rt http.RoundTripper, timeout time.Duration) (int64, error) {
// will post lreq protobuf to leader // will post lreq protobuf to leader
lreq, err := (&pb.LeaseKeepAliveRequest{ID: int64(id)}).Marshal() lreq, err := (&pb.LeaseKeepAliveRequest{ID: int64(id)}).Marshal()
if err != nil { if err != nil {
return -1, err return -1, err
} }
// TODO creating a new transporter for each forward request
// can be expensive; in the future reuse transports and batch requests
rt, err := transport.NewTimeoutTransport(tlsInfo, timeout, 0, 0)
if err != nil {
return -1, err
}
cc := &http.Client{Transport: rt, Timeout: timeout} cc := &http.Client{Transport: rt, Timeout: timeout}
resp, err := cc.Post(url, "application/protobuf", bytes.NewReader(lreq)) resp, err := cc.Post(url, "application/protobuf", bytes.NewReader(lreq))
if err != nil { if err != nil {