MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-02-26 12:59:17 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix: add deadlines for all synchronous REST callers (#19741)

Browse Source 
add deadlines that can be dynamically changed via
the drive max timeout values.

Bonus: optimize "file not found" case and hung drives/network - circuit break the check and return right
away instead of waiting.
This commit is contained in:
Harshavardhana 2024-05-15 09:52:29 -07:00 committed by GitHub
parent c05ca63158
commit d3db7d31a3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 56 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.github/workflows/mint/nginx-4-node.conf vendored
View File

@ -23,10 +23,9 @@ http {
# include /etc/nginx/conf.d/*.conf;
upstream minio {
server minio1:9000;
server minio2:9000;
server minio3:9000;
server minio4:9000;
server minio1:9000 max_fails=1 fail_timeout=10s;
server minio2:9000 max_fails=1 fail_timeout=10s;
server minio3:9000 max_fails=1 fail_timeout=10s;
}
upstream console {

16
.github/workflows/mint/nginx-8-node.conf vendored
View File

@ -23,14 +23,14 @@ http {
# include /etc/nginx/conf.d/*.conf;
upstream minio {
server minio1:9000;
server minio2:9000;
server minio3:9000;
server minio4:9000;
server minio5:9000;
server minio6:9000;
server minio7:9000;
server minio8:9000;
server minio1:9000 max_fails=1 fail_timeout=10s;
server minio2:9000 max_fails=1 fail_timeout=10s;
server minio3:9000 max_fails=1 fail_timeout=10s;
server minio4:9000 max_fails=1 fail_timeout=10s;
server minio5:9000 max_fails=1 fail_timeout=10s;
server minio6:9000 max_fails=1 fail_timeout=10s;
server minio7:9000 max_fails=1 fail_timeout=10s;
server minio8:9000 max_fails=1 fail_timeout=10s;
}
upstream console {

8
.github/workflows/mint/nginx.conf vendored
View File

@ -23,10 +23,10 @@ http {
# include /etc/nginx/conf.d/*.conf;
upstream minio {
server minio1:9000;
server minio2:9000;
server minio3:9000;
server minio4:9000;
server minio1:9000 max_fails=1 fail_timeout=10s;
server minio2:9000 max_fails=1 fail_timeout=10s;
server minio3:9000 max_fails=1 fail_timeout=10s;
server minio4:9000 max_fails=1 fail_timeout=10s;
}
upstream console {

14
cmd/erasure-object.go
View File

@ -899,6 +899,20 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s
if success {
validResp++
}
if totalResp >= minDisks && opts.FastGetObjInfo {
rw.Lock()
ok := countErrs(errs, errFileNotFound) >= minDisks || countErrs(errs, errFileVersionNotFound) >= minDisks
rw.Unlock()
if ok {
err = errFileNotFound
if opts.VersionID != "" {
err = errFileVersionNotFound
}
break
}
}
if totalResp < er.setDriveCount {
if !opts.FastGetObjInfo {
continue

21
cmd/storage-rest-client.go
View File

@ -391,6 +391,9 @@ func (client *storageRESTClient) CreateFile(ctx context.Context, origvolume, vol
}
func (client *storageRESTClient) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) error {
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
defer cancel()
_, err := storageWriteMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{
DiskID: *client.diskID.Load(),
OrigVolume: origvolume,
@ -402,6 +405,9 @@ func (client *storageRESTClient) WriteMetadata(ctx context.Context, origvolume,
}
func (client *storageRESTClient) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo, opts UpdateMetadataOpts) error {
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
defer cancel()
_, err := storageUpdateMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{
DiskID: *client.diskID.Load(),
Volume: volume,
@ -413,6 +419,9 @@ func (client *storageRESTClient) UpdateMetadata(ctx context.Context, volume, pat
}
func (client *storageRESTClient) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool, opts DeleteOptions) (err error) {
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
defer cancel()
_, err = storageDeleteVersionRPC.Call(ctx, client.gridConn, &DeleteVersionHandlerParams{
DiskID: *client.diskID.Load(),
Volume: volume,
@ -426,6 +435,9 @@ func (client *storageRESTClient) DeleteVersion(ctx context.Context, volume, path
// WriteAll - write all data to a file.
func (client *storageRESTClient) WriteAll(ctx context.Context, volume string, path string, b []byte) error {
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
defer cancel()
_, err := storageWriteAllRPC.Call(ctx, client.gridConn, &WriteAllHandlerParams{
DiskID: *client.diskID.Load(),
Volume: volume,
@ -497,6 +509,9 @@ func readMsgpReaderPoolPut(r *msgp.Reader) {
}
func (client *storageRESTClient) ReadVersion(ctx context.Context, origvolume, volume, path, versionID string, opts ReadOptions) (fi FileInfo, err error) {
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
defer cancel()
// Use websocket when not reading data.
if !opts.ReadData {
resp, err := storageReadVersionRPC.Call(ctx, client.gridConn, grid.NewMSSWith(map[string]string{
@ -537,6 +552,9 @@ func (client *storageRESTClient) ReadVersion(ctx context.Context, origvolume, vo
// ReadXL - reads all contents of xl.meta of a file.
func (client *storageRESTClient) ReadXL(ctx context.Context, volume string, path string, readData bool) (rf RawFileInfo, err error) {
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
defer cancel()
// Use websocket when not reading data.
if !readData {
resp, err := storageReadXLRPC.Call(ctx, client.gridConn, grid.NewMSSWith(map[string]string{
@ -570,6 +588,9 @@ func (client *storageRESTClient) ReadXL(ctx context.Context, volume string, path
// ReadAll - reads all contents of a file.
func (client *storageRESTClient) ReadAll(ctx context.Context, volume string, path string) ([]byte, error) {
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
defer cancel()
gridBytes, err := storageReadAllRPC.Call(ctx, client.gridConn, &ReadAllHandlerParams{
DiskID: *client.diskID.Load(),
Volume: volume,

7
internal/config/drive/drive.go
View File

@ -25,6 +25,10 @@ import (
"github.com/minio/pkg/v2/env"
)
const (
envMaxDriveTimeout = "MINIO_DRIVE_MAX_TIMEOUT"
)
// DefaultKVS - default KVS for drive
var DefaultKVS = config.KVS{
config.KV{
@ -65,8 +69,9 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
if err = config.CheckValidKeys(config.DriveSubSys, kvs, DefaultKVS); err != nil {
return cfg, err
}
// if not set. Get default value from environment
d := kvs.GetWithDefault(MaxTimeout, DefaultKVS)
d := env.Get(envMaxDriveTimeout, kvs.GetWithDefault(MaxTimeout, DefaultKVS))
if d == "" {
d = env.Get("_MINIO_DRIVE_MAX_TIMEOUT", "")
if d == "" {