diff --git a/pkg/mcclient/mcclient.go b/pkg/mcclient/mcclient.go index 259177da5e..9966f6c49b 100644 --- a/pkg/mcclient/mcclient.go +++ b/pkg/mcclient/mcclient.go @@ -57,7 +57,7 @@ func NewClient(authUrl string, timeout int, debug bool, insecure bool, certFile, } tlsConf.InsecureSkipVerify = insecure - tr := httputils.GetTransport(insecure, 5*time.Second) + tr := httputils.GetTransport(insecure) tr.TLSClientConfig = tlsConf tr.IdleConnTimeout = 5 * time.Second tr.TLSHandshakeTimeout = 10 * time.Second diff --git a/pkg/multicloud/aliyun/aliyun.go b/pkg/multicloud/aliyun/aliyun.go index a26263e45b..d68762221c 100644 --- a/pkg/multicloud/aliyun/aliyun.go +++ b/pkg/multicloud/aliyun/aliyun.go @@ -236,8 +236,9 @@ func (client *SAliyunClient) getOssClient(regionId string) (*oss.Client, error) // The ClientOption Proxy, AuthProxy lacks the feature NO_PROXY has // which can be used to whitelist ips, domains from http_proxy, // https_proxy setting + // oss use no timeout client so as to send/download large files cliOpts := []oss.ClientOption{ - oss.HTTPClient(httputils.GetDefaultClient()), + oss.HTTPClient(httputils.GetAdaptiveTimeoutClient()), } ep := getOSSExternalDomain(regionId) cli, err := oss.New(ep, client.accessKey, client.secret, cliOpts...) diff --git a/pkg/multicloud/objectstore/ceph/adminapi.go b/pkg/multicloud/objectstore/ceph/adminapi.go index 01e255e54b..f32cd786eb 100644 --- a/pkg/multicloud/objectstore/ceph/adminapi.go +++ b/pkg/multicloud/objectstore/ceph/adminapi.go @@ -49,8 +49,9 @@ func newCephAdminApi(ak, sk, ep string, debug bool, adminPath string) *SCephAdmi accessKey: ak, secret: sk, endpoint: ep, - client: httputils.GetDefaultClient(), - debug: debug, + // ceph use no timeout client so as to download/upload large files + client: httputils.GetAdaptiveTimeoutClient(), + debug: debug, } } diff --git a/pkg/multicloud/objectstore/objectstore.go b/pkg/multicloud/objectstore/objectstore.go index 8c7c306a6b..e9327ce89f 100644 --- a/pkg/multicloud/objectstore/objectstore.go +++ b/pkg/multicloud/objectstore/objectstore.go @@ -17,7 +17,6 @@ package objectstore import ( "net/url" "os" - "time" "yunion.io/x/jsonutils" "yunion.io/x/log" @@ -80,7 +79,7 @@ func NewObjectStoreClientAndFetch(providerId string, providerName string, endpoi return nil, errors.Wrap(err, "minio.New") } - tr := httputils.GetTransport(true, time.Second*5) + tr := httputils.GetTransport(true) cli.SetCustomTransport(tr) client.client = cli diff --git a/pkg/multicloud/objectstore/xsky/adminapi.go b/pkg/multicloud/objectstore/xsky/adminapi.go index eb8000b442..02d46404aa 100644 --- a/pkg/multicloud/objectstore/xsky/adminapi.go +++ b/pkg/multicloud/objectstore/xsky/adminapi.go @@ -46,8 +46,9 @@ func newXskyAdminApi(user, passwd, ep string, debug bool) *SXskyAdminApi { endpoint: ep, username: user, password: passwd, - client: httputils.GetDefaultClient(), - debug: debug, + // xsky use notimeout client so as to download/upload large files + client: httputils.GetAdaptiveTimeoutClient(), + debug: debug, } } diff --git a/pkg/multicloud/ucloud/ufile.go b/pkg/multicloud/ucloud/ufile.go index f2793d1521..6fc0cd264c 100644 --- a/pkg/multicloud/ucloud/ufile.go +++ b/pkg/multicloud/ucloud/ufile.go @@ -191,7 +191,8 @@ func (self *SFile) SetMeta(ctx context.Context, meta http.Header) error { } func doRequest(req *http.Request) (jsonutils.JSONObject, error) { - res, err := httputils.GetDefaultClient().Do(req) + // ufile request use no timeout client so as to download/upload large files + res, err := httputils.GetAdaptiveTimeoutClient().Do(req) if err != nil { return nil, errors.Wrap(err, "httpclient Do") } diff --git a/pkg/util/httputils/delegate.go b/pkg/util/httputils/delegate.go new file mode 100644 index 0000000000..1edeb99207 --- /dev/null +++ b/pkg/util/httputils/delegate.go @@ -0,0 +1,104 @@ +// Copyright 2019 Yunion +// +// 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 httputils + +import ( + "net" + "time" +) + +type connDelegate struct { + conn net.Conn + socketTimeout time.Duration + finalTimeout time.Duration +} + +func getConnDelegate(conn net.Conn, socketTimeout, finalTimeout time.Duration) *connDelegate { + return &connDelegate{ + conn: conn, + socketTimeout: socketTimeout, + finalTimeout: finalTimeout, + } +} + +func (delegate *connDelegate) Read(b []byte) (n int, err error) { + delegate.SetReadDeadline(time.Now().Add(delegate.socketTimeout)) + n, err = delegate.conn.Read(b) + delegate.SetReadDeadline(time.Now().Add(delegate.finalTimeout)) + return n, err +} + +func (delegate *connDelegate) Write(b []byte) (n int, err error) { + delegate.SetWriteDeadline(time.Now().Add(delegate.socketTimeout)) + n, err = delegate.conn.Write(b) + finalTimeout := time.Now().Add(delegate.finalTimeout) + delegate.SetWriteDeadline(finalTimeout) + delegate.SetReadDeadline(finalTimeout) + return n, err +} + +func (delegate *connDelegate) Close() error { + return delegate.conn.Close() +} + +func (delegate *connDelegate) LocalAddr() net.Addr { + return delegate.conn.LocalAddr() +} + +func (delegate *connDelegate) RemoteAddr() net.Addr { + return delegate.conn.RemoteAddr() +} + +// SetDeadline sets the read and write deadlines associated +// with the connection. It is equivalent to calling both +// SetReadDeadline and SetWriteDeadline. +// +// A deadline is an absolute time after which I/O operations +// fail with a timeout (see type Error) instead of +// blocking. The deadline applies to all future and pending +// I/O, not just the immediately following call to Read or +// Write. After a deadline has been exceeded, the connection +// can be refreshed by setting a deadline in the future. +// +// An idle timeout can be implemented by repeatedly extending +// the deadline after successful Read or Write calls. +// +// A zero value for t means I/O operations will not time out. +// +// Note that if a TCP connection has keep-alive turned on, +// which is the default unless overridden by Dialer.KeepAlive +// or ListenConfig.KeepAlive, then a keep-alive failure may +// also return a timeout error. On Unix systems a keep-alive +// failure on I/O can be detected using +// errors.Is(err, syscall.ETIMEDOUT). +func (delegate *connDelegate) SetDeadline(t time.Time) error { + return delegate.conn.SetDeadline(t) +} + +// SetReadDeadline sets the deadline for future Read calls +// and any currently-blocked Read call. +// A zero value for t means Read will not time out. +func (delegate *connDelegate) SetReadDeadline(t time.Time) error { + return delegate.conn.SetReadDeadline(t) +} + +// SetWriteDeadline sets the deadline for future Write calls +// and any currently-blocked Write call. +// Even if write times out, it may return n > 0, indicating that +// some of the data was successfully written. +// A zero value for t means Write will not time out. +func (delegate *connDelegate) SetWriteDeadline(t time.Time) error { + return delegate.conn.SetWriteDeadline(t) +} diff --git a/pkg/util/httputils/httputils.go b/pkg/util/httputils/httputils.go index 4a3704a45e..f7b8777b20 100644 --- a/pkg/util/httputils/httputils.go +++ b/pkg/util/httputils/httputils.go @@ -142,25 +142,105 @@ func GetAddrPort(urlStr string) (string, int, error) { } } -func GetTransport(insecure bool, timeout time.Duration) *http.Transport { - return &http.Transport{ +func GetTransport(insecure bool) *http.Transport { + return getTransport(insecure, false) +} + +func adptiveDial(network, addr string) (net.Conn, error) { + conn, err := net.DialTimeout(network, addr, 10*time.Second) + if err != nil { + return nil, err + } + return getConnDelegate(conn, 10*time.Second, 20*time.Second), nil +} + +func getTransport(insecure bool, adaptive bool) *http.Transport { + tr := &http.Transport{ Proxy: http.ProxyFromEnvironment, - DialContext: (&net.Dialer{ - Timeout: timeout, - KeepAlive: timeout, - }).DialContext, - IdleConnTimeout: 90 * time.Second, - TLSHandshakeTimeout: 10 * time.Second, - ExpectContinueTimeout: 1 * time.Second, + // 一个空闲连接保持连接的时间 + // IdleConnTimeout is the maximum amount of time an idle + // (keep-alive) connection will remain idle before closing + // itself. + // Zero means no limit. + IdleConnTimeout: 60 * time.Second, + // 建立TCP连接后，等待TLS握手的超时时间 + // TLSHandshakeTimeout specifies the maximum amount of time waiting to + // wait for a TLS handshake. Zero means no timeout. + TLSHandshakeTimeout: 10 * time.Second, + // 发送请求后，等待服务端http响应的超时时间 + // ResponseHeaderTimeout, if non-zero, specifies the amount of + // time to wait for a server's response headers after fully + // writing the request (including its body, if any). This + // time does not include the time to read the response body. + ResponseHeaderTimeout: 10 * time.Second, + // 当请求携带Expect: 100-continue时，等待服务端100响应的超时时间 + // ExpectContinueTimeout, if non-zero, specifies the amount of + // time to wait for a server's first response headers after fully + // writing the request headers if the request has an + // "Expect: 100-continue" header. Zero means no timeout and + // causes the body to be sent immediately, without + // waiting for the server to approve. + // This time does not include the time to send the request header. + ExpectContinueTimeout: 5 * time.Second, TLSClientConfig: &tls.Config{InsecureSkipVerify: insecure}, } + if adaptive { + tr.Dial = adptiveDial + } else { + tr.DialContext = (&net.Dialer{ + // 建立TCP连接超时时间 + // Timeout is the maximum amount of time a dial will wait for + // a connect to complete. If Deadline is also set, it may fail + // earlier. + // + // The default is no timeout. + // + // When using TCP and dialing a host name with multiple IP + // addresses, the timeout may be divided between them. + // + // With or without a timeout, the operating system may impose + // its own earlier timeout. For instance, TCP timeouts are + // often around 3 minutes. + Timeout: 10 * time.Second, + // + // KeepAlive specifies the interval between keep-alive + // probes for an active network connection. + // If zero, keep-alive probes are sent with a default value + // (currently 15 seconds), if supported by the protocol and operating + // system. Network protocols or operating systems that do + // not support keep-alives ignore this field. + // If negative, keep-alive probes are disabled. + KeepAlive: 5 * time.Second, // send keep-alive probe every 5 seconds + }).DialContext + } + return tr } func GetClient(insecure bool, timeout time.Duration) *http.Client { - tr := GetTransport(insecure, timeout) + adaptive := false + if timeout == 0 { + adaptive = true + } + tr := getTransport(insecure, adaptive) return &http.Client{ Transport: tr, - Timeout: timeout, + // 一个完整http request的超时时间 + // Timeout specifies a time limit for requests made by this + // Client. The timeout includes connection time, any + // redirects, and reading the response body. The timer remains + // running after Get, Head, Post, or Do return and will + // interrupt reading of the Response.Body. + // + // A Timeout of zero means no timeout. + // + // The Client cancels requests to the underlying Transport + // as if the Request's Context ended. + // + // For compatibility, the Client will also use the deprecated + // CancelRequest method on Transport if found. New + // RoundTripper implementations should use the Request's Context + // for cancellation instead of implementing CancelRequest. + Timeout: timeout, } } @@ -168,6 +248,10 @@ func GetTimeoutClient(timeout time.Duration) *http.Client { return GetClient(true, timeout) } +func GetAdaptiveTimeoutClient() *http.Client { + return GetClient(true, 0) +} + var defaultHttpClient *http.Client func init() { diff --git a/pkg/util/httputils/httputils_test.go b/pkg/util/httputils/httputils_test.go index dfdf0527bd..ed98838b0c 100644 --- a/pkg/util/httputils/httputils_test.go +++ b/pkg/util/httputils/httputils_test.go @@ -17,12 +17,19 @@ package httputils import ( "context" "encoding/json" + "fmt" + "net" "net/http" "net/http/httptest" + "net/url" + "reflect" "testing" + "time" "yunion.io/x/jsonutils" "yunion.io/x/pkg/errors" + + "yunion.io/x/onecloud/pkg/util/netutils2" ) type SErrorMsg struct { @@ -141,3 +148,162 @@ func TestErrorCause(t *testing.T) { t.Errorf("wrapErro.Cause should be err: %#v != %#v", errors.Cause(wrapError), err) } } + +type ResponseHeaderTimeoutHandler struct{} + +func (h *ResponseHeaderTimeoutHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + time.Sleep(12 * time.Second) + w.Write([]byte("hello")) +} + +func TestResponseHeaderTimeout(t *testing.T) { + s := getTestServer(t, &ResponseHeaderTimeoutHandler{}) + go func() { + s.ListenAndServe() + }() + + clientWait(t) + cli := GetAdaptiveTimeoutClient() + resp, err := cli.Get(fmt.Sprintf("http://%s", s.Addr)) + if err == nil { + t.Errorf("Read shoud error") + } else if !err.(*url.Error).Timeout() { + t.Errorf("Read error %s %s, should be url.Error.Timeout", err, reflect.TypeOf(err)) + } else { + t.Logf("Read error %s %s", err, reflect.TypeOf(err)) + } + CloseResponse(resp) + s.Close() +} + +type clientTimeoutHandler struct{} + +func (h *clientTimeoutHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(200) + for i := 0; i < 4096; i++ { + for j := 0; j < 4096; j++ { + w.Write([]byte("ABCDEFD\n")) + } + time.Sleep(1 * time.Second) + } +} + +func clientWait(t *testing.T) { + t.Logf("wait 2 seconds for server ...") + time.Sleep(2 * time.Second) + t.Logf("Start client ...") +} + +func getTestServer(t *testing.T, h http.Handler) *http.Server { + port, err := netutils2.GetFreePort() + if err != nil { + t.Fatalf("fail to found free port %s", err) + } + t.Logf("Start serve at %d", port) + return &http.Server{ + Addr: fmt.Sprintf("127.0.0.1:%d", port), + Handler: h, + ReadTimeout: 300 * time.Second, + WriteTimeout: 300 * time.Second, + MaxHeaderBytes: 1 << 20, + } +} + +func TestClientTimeout(t *testing.T) { + s := getTestServer(t, &clientTimeoutHandler{}) + go func() { + s.ListenAndServe() + }() + + clientWait(t) + cli := GetTimeoutClient(15 * time.Second) + resp, err := cli.Get(fmt.Sprintf("http://%s", s.Addr)) + if err != nil { + t.Errorf("Read error %s", err) + } else { + buf := make([]byte, 4096) + offset := 0 + for { + var n int + n, err = resp.Body.Read(buf) + if n > 0 { + offset += n + } + if err != nil { + t.Logf("read error %s %s", err, reflect.TypeOf(err)) + break + } + } + t.Logf("read offset %d", offset) + CloseResponse(resp) + } + s.Close() + if err == nil { + t.Errorf("read should error") + } +} + +type idleTimeoutHandler struct{} + +func (h *idleTimeoutHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(200) + for i := 0; i < 4096; i++ { + for j := 0; j < 4096; j++ { + w.Write([]byte("ABCDEFD\n")) + } + // sleep 15 seconds, make read timeout + time.Sleep(15 * time.Second) + } +} + +func TestIdleTimeout(t *testing.T) { + s := getTestServer(t, &idleTimeoutHandler{}) + go func() { + s.ListenAndServe() + }() + + t.Logf("%s", s.Addr) + clientWait(t) + cli := GetAdaptiveTimeoutClient() + resp, err := cli.Get(fmt.Sprintf("http://%s", s.Addr)) + if err != nil { + t.Errorf("Read error %s", err) + } else { + buf := make([]byte, 4096) + offset := 0 + for { + var n int + n, err = resp.Body.Read(buf) + if n > 0 { + offset += n + } + if err != nil { + t.Logf("read error %s %s", err, reflect.TypeOf(err)) + break + } + } + t.Logf("read offset %d", offset) + CloseResponse(resp) + } + s.Close() + if err == nil { + t.Errorf("read shoud error") + } else if !err.(*net.OpError).Timeout() { + t.Errorf("error shoud be net.OpsError.Timeout, not %s", err) + } else { + t.Logf("error found %s", err) + } +} + +func TestDialTimeout(t *testing.T) { + cli := GetAdaptiveTimeoutClient() + resp, err := cli.Get(fmt.Sprintf("http://192.0.0.1:48481")) + if err == nil { + t.Errorf("Read shoud error") + } else if !err.(*url.Error).Timeout() { + t.Errorf("Read error %s %s, should be url.Error.Timeout", err, reflect.TypeOf(err)) + } else { + t.Logf("Read error %s %s", err, reflect.TypeOf(err)) + } + CloseResponse(resp) +} diff --git a/pkg/util/netutils2/netutils.go b/pkg/util/netutils2/netutils.go index 2f7bfe13ce..23f8a656b3 100644 --- a/pkg/util/netutils2/netutils.go +++ b/pkg/util/netutils2/netutils.go @@ -31,7 +31,6 @@ import ( "yunion.io/x/pkg/util/regutils" "yunion.io/x/onecloud/pkg/cloudcommon/types" - "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/util/procutils" "yunion.io/x/onecloud/pkg/util/regutils2" ) @@ -130,7 +129,7 @@ func GetMainNicFromDeployApi(nics []*types.SServerNic) (*types.SServerNic, error if mainNic != nil { return mainNic, nil } - return nil, errors.Wrap(httperrors.ErrInvalidStatus, "no valid nic") + return nil, errors.Wrap(errors.ErrInvalidStatus, "no valid nic") } func GetMainNic(nics []jsonutils.JSONObject) (jsonutils.JSONObject, error) { @@ -172,7 +171,7 @@ func GetMainNic(nics []jsonutils.JSONObject) (jsonutils.JSONObject, error) { if mainNic != nil { return mainNic, nil } - return nil, errors.Wrap(httperrors.ErrInvalidStatus, "no valid nic") + return nil, errors.Wrap(errors.ErrInvalidStatus, "no valid nic") } func Netlen2Mask(netmasklen int) string {