mirror of
https://github.com/minio/minio.git
synced 2025-05-23 02:21:51 -04:00
fix: proxies set keep-alive timeouts to be system dependent (#10199)
Split the DialContext's one for internode and another for all other external communications especially proxy forwarders, gateway transport etc.
This commit is contained in:
parent
019fe69a57
commit
0b8255529a
@ -1631,9 +1631,6 @@ func fetchLoggerInfo(cfg config.Config) ([]madmin.Logger, []madmin.Audit) {
|
|||||||
|
|
||||||
// checkConnection - ping an endpoint , return err in case of no connection
|
// checkConnection - ping an endpoint , return err in case of no connection
|
||||||
func checkConnection(endpointStr string, timeout time.Duration) error {
|
func checkConnection(endpointStr string, timeout time.Duration) error {
|
||||||
tr := newCustomHTTPTransport(&tls.Config{RootCAs: globalRootCAs}, timeout)()
|
|
||||||
defer tr.CloseIdleConnections()
|
|
||||||
|
|
||||||
ctx, cancel := context.WithTimeout(GlobalContext, timeout)
|
ctx, cancel := context.WithTimeout(GlobalContext, timeout)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
@ -1642,7 +1639,20 @@ func checkConnection(endpointStr string, timeout time.Duration) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
client := &http.Client{Transport: tr}
|
client := &http.Client{Transport: &http.Transport{
|
||||||
|
Proxy: http.ProxyFromEnvironment,
|
||||||
|
DialContext: xhttp.NewCustomDialContext(timeout),
|
||||||
|
ResponseHeaderTimeout: 5 * time.Second,
|
||||||
|
TLSHandshakeTimeout: 5 * time.Second,
|
||||||
|
ExpectContinueTimeout: 5 * time.Second,
|
||||||
|
TLSClientConfig: &tls.Config{RootCAs: globalRootCAs},
|
||||||
|
// Go net/http automatically unzip if content-type is
|
||||||
|
// gzip disable this feature, as we are always interested
|
||||||
|
// in raw stream.
|
||||||
|
DisableCompression: true,
|
||||||
|
}}
|
||||||
|
defer client.CloseIdleConnections()
|
||||||
|
|
||||||
resp, err := client.Do(req.WithContext(ctx))
|
resp, err := client.Do(req.WithContext(ctx))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
|
@ -228,7 +228,7 @@ func newBootstrapRESTClient(endpoint Endpoint) *bootstrapRESTClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
trFn := newCustomHTTPTransport(tlsConfig, rest.DefaultRESTTimeout)
|
trFn := newInternodeHTTPTransport(tlsConfig, rest.DefaultRESTTimeout)
|
||||||
restClient := rest.NewClient(serverURL, trFn, newAuthToken)
|
restClient := rest.NewClient(serverURL, trFn, newAuthToken)
|
||||||
restClient.HealthCheckFn = func() bool {
|
restClient.HealthCheckFn = func() bool {
|
||||||
ctx, cancel := context.WithTimeout(GlobalContext, restClient.HealthCheckTimeout)
|
ctx, cancel := context.WithTimeout(GlobalContext, restClient.HealthCheckTimeout)
|
||||||
|
@ -49,20 +49,17 @@ func setInternalTCPParameters(c syscall.RawConn) error {
|
|||||||
_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPCNT, 5)
|
_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPCNT, 5)
|
||||||
|
|
||||||
// Wait time after successful probe in seconds.
|
// Wait time after successful probe in seconds.
|
||||||
// ~ cat /proc/sys/net/ipv4/tcp_keepalive_intvl (defaults to 75 secs, we reduce it to 2 secs)
|
// ~ cat /proc/sys/net/ipv4/tcp_keepalive_intvl (defaults to 75 secs, we reduce it to 3 secs)
|
||||||
// 75
|
// 75
|
||||||
_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPINTVL, 2)
|
_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, syscall.TCP_KEEPINTVL, 3)
|
||||||
|
|
||||||
// Set TCP_USER_TIMEOUT to TCP_KEEPIDLE + TCP_KEEPINTVL * TCP_KEEPCNT.
|
|
||||||
_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, unix.TCP_USER_TIMEOUT, 15)
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// DialContext is a function to make custom Dial for internode communications
|
// DialContext is a function to make custom Dial for internode communications
|
||||||
type DialContext func(ctx context.Context, network, address string) (net.Conn, error)
|
type DialContext func(ctx context.Context, network, address string) (net.Conn, error)
|
||||||
|
|
||||||
// NewCustomDialContext setups a custom dialer for internode communications
|
// NewInternodeDialContext setups a custom dialer for internode communication
|
||||||
func NewCustomDialContext(dialTimeout time.Duration) DialContext {
|
func NewInternodeDialContext(dialTimeout time.Duration) DialContext {
|
||||||
return func(ctx context.Context, network, addr string) (net.Conn, error) {
|
return func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||||
dialer := &net.Dialer{
|
dialer := &net.Dialer{
|
||||||
Timeout: dialTimeout,
|
Timeout: dialTimeout,
|
||||||
@ -73,3 +70,25 @@ func NewCustomDialContext(dialTimeout time.Duration) DialContext {
|
|||||||
return dialer.DialContext(ctx, network, addr)
|
return dialer.DialContext(ctx, network, addr)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// NewCustomDialContext setups a custom dialer for any external communication and proxies.
|
||||||
|
func NewCustomDialContext(dialTimeout time.Duration) DialContext {
|
||||||
|
return func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||||
|
dialer := &net.Dialer{
|
||||||
|
Timeout: dialTimeout,
|
||||||
|
Control: func(network, address string, c syscall.RawConn) error {
|
||||||
|
return c.Control(func(fdPtr uintptr) {
|
||||||
|
// got socket file descriptor to set parameters.
|
||||||
|
fd := int(fdPtr)
|
||||||
|
|
||||||
|
// Enable TCP fast connect
|
||||||
|
// TCPFastOpenConnect sets the underlying socket to use
|
||||||
|
// the TCP fast open connect. This feature is supported
|
||||||
|
// since Linux 4.11.
|
||||||
|
_ = syscall.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, unix.TCP_FASTOPEN_CONNECT, 1)
|
||||||
|
})
|
||||||
|
},
|
||||||
|
}
|
||||||
|
return dialer.DialContext(ctx, network, addr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
@ -33,6 +33,9 @@ func setInternalTCPParameters(c syscall.RawConn) error {
|
|||||||
// DialContext is a function to make custom Dial for internode communications
|
// DialContext is a function to make custom Dial for internode communications
|
||||||
type DialContext func(ctx context.Context, network, address string) (net.Conn, error)
|
type DialContext func(ctx context.Context, network, address string) (net.Conn, error)
|
||||||
|
|
||||||
|
// NewInternodeDialContext setups a custom dialer for internode communication
|
||||||
|
var NewInternodeDialContext = NewCustomDialContext
|
||||||
|
|
||||||
// NewCustomDialContext configures a custom dialer for internode communications
|
// NewCustomDialContext configures a custom dialer for internode communications
|
||||||
func NewCustomDialContext(dialTimeout time.Duration) DialContext {
|
func NewCustomDialContext(dialTimeout time.Duration) DialContext {
|
||||||
return func(ctx context.Context, network, addr string) (net.Conn, error) {
|
return func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||||
|
@ -153,7 +153,7 @@ func newlockRESTClient(endpoint Endpoint) *lockRESTClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
trFn := newCustomHTTPTransport(tlsConfig, rest.DefaultRESTTimeout)
|
trFn := newInternodeHTTPTransport(tlsConfig, rest.DefaultRESTTimeout)
|
||||||
restClient := rest.NewClient(serverURL, trFn, newAuthToken)
|
restClient := rest.NewClient(serverURL, trFn, newAuthToken)
|
||||||
restClient.HealthCheckFn = func() bool {
|
restClient.HealthCheckFn = func() bool {
|
||||||
ctx, cancel := context.WithTimeout(GlobalContext, restClient.HealthCheckTimeout)
|
ctx, cancel := context.WithTimeout(GlobalContext, restClient.HealthCheckTimeout)
|
||||||
|
@ -874,7 +874,7 @@ func newPeerRESTClient(peer *xnet.Host) *peerRESTClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
trFn := newCustomHTTPTransport(tlsConfig, rest.DefaultRESTTimeout)
|
trFn := newInternodeHTTPTransport(tlsConfig, rest.DefaultRESTTimeout)
|
||||||
restClient := rest.NewClient(serverURL, trFn, newAuthToken)
|
restClient := rest.NewClient(serverURL, trFn, newAuthToken)
|
||||||
|
|
||||||
// Construct a new health function.
|
// Construct a new health function.
|
||||||
|
@ -17,6 +17,7 @@
|
|||||||
package cmd
|
package cmd
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"crypto/tls"
|
"crypto/tls"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
@ -212,7 +213,10 @@ func IsServerResolvable(endpoint Endpoint) error {
|
|||||||
}
|
}
|
||||||
defer httpClient.CloseIdleConnections()
|
defer httpClient.CloseIdleConnections()
|
||||||
|
|
||||||
resp, err := httpClient.Do(req)
|
ctx, cancel := context.WithTimeout(GlobalContext, 5*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
resp, err := httpClient.Do(req.WithContext(ctx))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
@ -667,7 +667,7 @@ func newStorageRESTClient(endpoint Endpoint) *storageRESTClient {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
trFn := newCustomHTTPTransport(tlsConfig, rest.DefaultRESTTimeout)
|
trFn := newInternodeHTTPTransport(tlsConfig, rest.DefaultRESTTimeout)
|
||||||
restClient := rest.NewClient(serverURL, trFn, newAuthToken)
|
restClient := rest.NewClient(serverURL, trFn, newAuthToken)
|
||||||
restClient.HealthCheckInterval = 500 * time.Millisecond
|
restClient.HealthCheckInterval = 500 * time.Millisecond
|
||||||
restClient.HealthCheckFn = func() bool {
|
restClient.HealthCheckFn = func() bool {
|
||||||
|
27
cmd/utils.go
27
cmd/utils.go
@ -1,5 +1,5 @@
|
|||||||
/*
|
/*
|
||||||
* MinIO Cloud Storage, (C) 2015, 2016, 2017 MinIO, Inc.
|
* MinIO Cloud Storage, (C) 2015-2020 MinIO, Inc.
|
||||||
*
|
*
|
||||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
* you may not use this file except in compliance with the License.
|
* you may not use this file except in compliance with the License.
|
||||||
@ -170,7 +170,7 @@ const (
|
|||||||
// (Acceptable values range from 1 to 10000 inclusive)
|
// (Acceptable values range from 1 to 10000 inclusive)
|
||||||
globalMaxPartID = 10000
|
globalMaxPartID = 10000
|
||||||
|
|
||||||
// Default values used while communicating for internode communication.
|
// Default values used while communicating for gateway communication
|
||||||
defaultDialTimeout = 5 * time.Second
|
defaultDialTimeout = 5 * time.Second
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -449,6 +449,29 @@ func ToS3ETag(etag string) string {
|
|||||||
return etag
|
return etag
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func newInternodeHTTPTransport(tlsConfig *tls.Config, dialTimeout time.Duration) func() *http.Transport {
|
||||||
|
// For more details about various values used here refer
|
||||||
|
// https://golang.org/pkg/net/http/#Transport documentation
|
||||||
|
tr := &http.Transport{
|
||||||
|
Proxy: http.ProxyFromEnvironment,
|
||||||
|
DialContext: xhttp.NewInternodeDialContext(dialTimeout),
|
||||||
|
MaxIdleConnsPerHost: 16,
|
||||||
|
MaxIdleConns: 16,
|
||||||
|
IdleConnTimeout: 1 * time.Minute,
|
||||||
|
ResponseHeaderTimeout: 3 * time.Minute, // Set conservative timeouts for MinIO internode.
|
||||||
|
TLSHandshakeTimeout: 10 * time.Second,
|
||||||
|
ExpectContinueTimeout: 10 * time.Second,
|
||||||
|
TLSClientConfig: tlsConfig,
|
||||||
|
// Go net/http automatically unzip if content-type is
|
||||||
|
// gzip disable this feature, as we are always interested
|
||||||
|
// in raw stream.
|
||||||
|
DisableCompression: true,
|
||||||
|
}
|
||||||
|
return func() *http.Transport {
|
||||||
|
return tr
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func newCustomHTTPTransport(tlsConfig *tls.Config, dialTimeout time.Duration) func() *http.Transport {
|
func newCustomHTTPTransport(tlsConfig *tls.Config, dialTimeout time.Duration) func() *http.Transport {
|
||||||
// For more details about various values used here refer
|
// For more details about various values used here refer
|
||||||
// https://golang.org/pkg/net/http/#Transport documentation
|
// https://golang.org/pkg/net/http/#Transport documentation
|
||||||
|
@ -40,8 +40,10 @@ import (
|
|||||||
humanize "github.com/dustin/go-humanize"
|
humanize "github.com/dustin/go-humanize"
|
||||||
jsoniter "github.com/json-iterator/go"
|
jsoniter "github.com/json-iterator/go"
|
||||||
"github.com/klauspost/readahead"
|
"github.com/klauspost/readahead"
|
||||||
|
"github.com/minio/minio/cmd/config"
|
||||||
"github.com/minio/minio/cmd/logger"
|
"github.com/minio/minio/cmd/logger"
|
||||||
"github.com/minio/minio/pkg/disk"
|
"github.com/minio/minio/pkg/disk"
|
||||||
|
"github.com/minio/minio/pkg/env"
|
||||||
xioutil "github.com/minio/minio/pkg/ioutil"
|
xioutil "github.com/minio/minio/pkg/ioutil"
|
||||||
"github.com/minio/minio/pkg/mountinfo"
|
"github.com/minio/minio/pkg/mountinfo"
|
||||||
)
|
)
|
||||||
@ -93,6 +95,8 @@ type xlStorage struct {
|
|||||||
|
|
||||||
pool sync.Pool
|
pool sync.Pool
|
||||||
|
|
||||||
|
globalSync bool
|
||||||
|
|
||||||
diskMount bool // indicates if the path is an actual mount.
|
diskMount bool // indicates if the path is an actual mount.
|
||||||
|
|
||||||
diskID string
|
diskID string
|
||||||
@ -245,6 +249,7 @@ func newXLStorage(path string, hostname string) (*xlStorage, error) {
|
|||||||
return &b
|
return &b
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
globalSync: env.Get(config.EnvFSOSync, config.EnableOn) == config.EnableOn,
|
||||||
diskMount: mountinfo.IsLikelyMountPoint(path),
|
diskMount: mountinfo.IsLikelyMountPoint(path),
|
||||||
// Allow disk usage crawler to run with up to 2 concurrent
|
// Allow disk usage crawler to run with up to 2 concurrent
|
||||||
// I/O ops, if and when activeIOCount reaches this
|
// I/O ops, if and when activeIOCount reaches this
|
||||||
@ -1216,9 +1221,11 @@ func (s *xlStorage) renameLegacyMetadata(volume, path string) error {
|
|||||||
// Renaming xl.json to xl.meta should be fully synced to disk.
|
// Renaming xl.json to xl.meta should be fully synced to disk.
|
||||||
defer func() {
|
defer func() {
|
||||||
if err == nil {
|
if err == nil {
|
||||||
|
if s.globalSync {
|
||||||
// Sync to disk only upon success.
|
// Sync to disk only upon success.
|
||||||
globalSync()
|
globalSync()
|
||||||
}
|
}
|
||||||
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
if err = os.Rename(srcFilePath, dstFilePath); err != nil {
|
if err = os.Rename(srcFilePath, dstFilePath); err != nil {
|
||||||
@ -2104,8 +2111,10 @@ func (s *xlStorage) RenameData(srcVolume, srcPath, dataDir, dstVolume, dstPath s
|
|||||||
return osErrToFileErr(err)
|
return osErrToFileErr(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if s.globalSync {
|
||||||
// Sync all the previous directory operations.
|
// Sync all the previous directory operations.
|
||||||
globalSync()
|
globalSync()
|
||||||
|
}
|
||||||
|
|
||||||
for _, entry := range entries {
|
for _, entry := range entries {
|
||||||
if entry == xlStorageFormatFile {
|
if entry == xlStorageFormatFile {
|
||||||
@ -2121,8 +2130,10 @@ func (s *xlStorage) RenameData(srcVolume, srcPath, dataDir, dstVolume, dstPath s
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Sync all the metadata operations once renames are done.
|
// Sync all the metadata operations once renames are done.
|
||||||
|
if s.globalSync {
|
||||||
globalSync()
|
globalSync()
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
var oldDstDataPath string
|
var oldDstDataPath string
|
||||||
if fi.VersionID == "" {
|
if fi.VersionID == "" {
|
||||||
|
Loading…
x
Reference in New Issue
Block a user