fix: httpclient timeout value cleanup, use notimeout client for oss (#4527)

* fix: httpclient timeout value cleanup, use notimeout client for oss

* Adaptive timeout client

* more test cases

* remove resolve timeout test case

* use adptive timeout only if timeout = 0

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)
+}

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() {

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("<html>ABCDEFD</html>\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("<html>ABCDEFD</html>\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)
+}

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 {