From 400e573013a5422f901581c51c4634c5f924c328 Mon Sep 17 00:00:00 2001 From: Yicheng Qin Date: Sat, 22 Nov 2014 22:24:10 -0800 Subject: [PATCH 1/4] rafthttp: log start and stop of streaming --- rafthttp/sender.go | 1 - rafthttp/streamer.go | 12 ++++++++++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/rafthttp/sender.go b/rafthttp/sender.go index 6ed909338..e6ae56b9b 100644 --- a/rafthttp/sender.go +++ b/rafthttp/sender.go @@ -176,7 +176,6 @@ func (s *sender) initStream(from, to types.ID, term uint64) { return } s.strmCln = strmCln - log.Printf("rafthttp: start stream client with %s in term %d", to, term) } func (s *sender) tryStream(m raftpb.Message) bool { diff --git a/rafthttp/streamer.go b/rafthttp/streamer.go index ee5489234..f7fe08b5b 100644 --- a/rafthttp/streamer.go +++ b/rafthttp/streamer.go @@ -59,6 +59,7 @@ func startStreamServer(w WriteFlusher, to types.ID, term uint64, fs *stats.Follo done: make(chan struct{}), } go s.handle(w) + log.Printf("rafthttp: stream server to %s at term %d starts", to, term) return s } @@ -85,7 +86,10 @@ func (s *streamServer) stop() { func (s *streamServer) stopNotify() <-chan struct{} { return s.done } func (s *streamServer) handle(w WriteFlusher) { - defer close(s.done) + defer func() { + close(s.done) + log.Printf("rafthttp: stream server to %s at term %d is closed", s.to, s.term) + }() ew := &entryWriter{w: w} for ents := range s.q { @@ -145,6 +149,7 @@ func (s *streamClient) start(tr http.RoundTripper, u string, cid types.ID) error } s.closer = resp.Body go s.handle(resp.Body) + log.Printf("rafthttp: stream client to %s at term %d starts", s.to, s.term) return nil } @@ -163,7 +168,10 @@ func (s *streamClient) isStopped() bool { } func (s *streamClient) handle(r io.Reader) { - defer close(s.done) + defer func() { + close(s.done) + log.Printf("rafthttp: stream client to %s at term %d is closed", s.to, s.term) + }() er := &entryReader{r: r} for { From ad58122e3cb81adf2ebcb84fb33fed2af16a6b9a Mon Sep 17 00:00:00 2001 From: Yicheng Qin Date: Sun, 23 Nov 2014 23:21:33 -0800 Subject: [PATCH 2/4] pkg/transport: fix dialer typo --- pkg/transport/{timeout_dailer.go => timeout_dialer.go} | 0 pkg/transport/{timeout_dailer_test.go => timeout_dialer_test.go} | 0 2 files changed, 0 insertions(+), 0 deletions(-) rename pkg/transport/{timeout_dailer.go => timeout_dialer.go} (100%) rename pkg/transport/{timeout_dailer_test.go => timeout_dialer_test.go} (100%) diff --git a/pkg/transport/timeout_dailer.go b/pkg/transport/timeout_dialer.go similarity index 100% rename from pkg/transport/timeout_dailer.go rename to pkg/transport/timeout_dialer.go diff --git a/pkg/transport/timeout_dailer_test.go b/pkg/transport/timeout_dialer_test.go similarity index 100% rename from pkg/transport/timeout_dailer_test.go rename to pkg/transport/timeout_dialer_test.go From 3e55834c3835bbb6c67139ca215b7cb44fb71c24 Mon Sep 17 00:00:00 2001 From: Yicheng Qin Date: Sun, 23 Nov 2014 23:37:54 -0800 Subject: [PATCH 3/4] *: set read/write timeout for raft transport and listener --- etcdmain/etcd.go | 5 ++-- pkg/transport/timeout_listener.go | 15 +++++++++++ pkg/transport/timeout_transport.go | 42 ++++++++++++++++++++++++++++++ rafthttp/sender.go | 3 +++ 4 files changed, 63 insertions(+), 2 deletions(-) create mode 100644 pkg/transport/timeout_transport.go diff --git a/etcdmain/etcd.go b/etcdmain/etcd.go index da26390b8..ca9a73c7a 100644 --- a/etcdmain/etcd.go +++ b/etcdmain/etcd.go @@ -35,6 +35,7 @@ import ( "github.com/coreos/etcd/pkg/transport" "github.com/coreos/etcd/pkg/types" "github.com/coreos/etcd/proxy" + "github.com/coreos/etcd/rafthttp" "github.com/coreos/etcd/version" ) @@ -209,7 +210,7 @@ func startEtcd() (<-chan struct{}, error) { return nil, fmt.Errorf("cannot write to data directory: %v", err) } - pt, err := transport.NewTransport(peerTLSInfo) + pt, err := transport.NewTimeoutTransport(peerTLSInfo, rafthttp.ConnReadTimeout, rafthttp.ConnWriteTimeout) if err != nil { return nil, err } @@ -230,7 +231,7 @@ func startEtcd() (<-chan struct{}, error) { plns := make([]net.Listener, 0) for _, u := range lpurls { var l net.Listener - l, err = transport.NewListener(u.Host, u.Scheme, peerTLSInfo) + l, err = transport.NewTimeoutListener(u.Host, u.Scheme, peerTLSInfo, rafthttp.ConnReadTimeout, rafthttp.ConnWriteTimeout) if err != nil { return nil, err } diff --git a/pkg/transport/timeout_listener.go b/pkg/transport/timeout_listener.go index 0c4917531..2aca95b80 100644 --- a/pkg/transport/timeout_listener.go +++ b/pkg/transport/timeout_listener.go @@ -21,6 +21,21 @@ import ( "time" ) +// NewTimeoutListener returns a listener that listens on the given address. +// If read/write on the accepted connection blocks longer than its time limit, +// it will return timeout error. +func NewTimeoutListener(addr string, scheme string, info TLSInfo, rdtimeoutd, wtimeoutd time.Duration) (net.Listener, error) { + ln, err := NewListener(addr, scheme, info) + if err != nil { + return nil, err + } + return &rwTimeoutListener{ + Listener: ln, + rdtimeoutd: rdtimeoutd, + wtimeoutd: wtimeoutd, + }, nil +} + type rwTimeoutListener struct { net.Listener wtimeoutd time.Duration diff --git a/pkg/transport/timeout_transport.go b/pkg/transport/timeout_transport.go new file mode 100644 index 000000000..2151ad48d --- /dev/null +++ b/pkg/transport/timeout_transport.go @@ -0,0 +1,42 @@ +/* + 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 transport + +import ( + "net" + "net/http" + "time" +) + +// NewTimeoutTransport returns a transport created using the given TLS info. +// If read/write on the created connection blocks longer than its time limit, +// it will return timeout error. +func NewTimeoutTransport(info TLSInfo, rdtimeoutd, wtimeoutd time.Duration) (*http.Transport, error) { + tr, err := NewTransport(info) + if err != nil { + return nil, err + } + tr.Dial = (&rwTimeoutDialer{ + Dialer: net.Dialer{ + Timeout: 30 * time.Second, + KeepAlive: 30 * time.Second, + }, + rdtimeoutd: rdtimeoutd, + wtimeoutd: wtimeoutd, + }).Dial + return tr, nil +} diff --git a/rafthttp/sender.go b/rafthttp/sender.go index e6ae56b9b..008e70673 100644 --- a/rafthttp/sender.go +++ b/rafthttp/sender.go @@ -35,6 +35,9 @@ const ( senderBufSize = connPerSender * 4 appRespBatchMs = 50 + + ConnReadTimeout = 5 * time.Second + ConnWriteTimeout = 5 * time.Second ) type Sender interface { From 1e797c1e389f3645b4bbeb956c28c6fdf848bbf9 Mon Sep 17 00:00:00 2001 From: Yicheng Qin Date: Mon, 24 Nov 2014 13:32:09 -0800 Subject: [PATCH 4/4] rafthttp: limit the data size read from connection each time --- rafthttp/http.go | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/rafthttp/http.go b/rafthttp/http.go index 052b0a6ce..ba8dc5e02 100644 --- a/rafthttp/http.go +++ b/rafthttp/http.go @@ -17,6 +17,7 @@ package rafthttp import ( + "io" "io/ioutil" "log" "net/http" @@ -30,6 +31,10 @@ import ( "github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context" ) +const ( + ConnReadLimitByte = 64 * 1024 +) + var ( RaftPrefix = "/raft" RaftStreamPrefix = path.Join(RaftPrefix, "stream") @@ -83,7 +88,10 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - b, err := ioutil.ReadAll(r.Body) + // Limit the data size that could be read from the request body, which ensures that read from + // connection will not time out accidentally due to possible block in underlying implementation. + limitedr := io.LimitReader(r.Body, ConnReadLimitByte) + b, err := ioutil.ReadAll(limitedr) if err != nil { log.Println("rafthttp: error reading raft message:", err) http.Error(w, "error reading raft message", http.StatusBadRequest)