pogpp/frostfs-s3-gw
1
0
Fork 0
forked from TrueCloudLab/frostfs-s3-gw
Code Pull requests Activity

[#103] Return 504 http code on timeout errors

Browse source 
Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>
This commit is contained in:
Denis Kirillov 2023-06-13 17:47:31 +03:00
parent 8fcaf76f41
commit 462589fc0c
9 changed files with 128 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
CHANGELOG.md
View file

@ -43,6 +43,7 @@ This document outlines major changes between releases.
- Support new system attributes (#64)
- Changed values for `frostfs_s3_gw_state_health` metric (#91)
- Support multiple tree service endpoints (#74)
- Timeout errors has code 504 now (#103)
### Removed
- Drop `tree.service` param (now endpoints from `peers` section are used) (#133)

7
api/errors/errors.go
View file

@ -175,6 +175,7 @@ const (
// Add new extended error codes here.
ErrInvalidObjectName
ErrOperationTimedOut
ErrGatewayTimeout
ErrOperationMaxedOut
ErrInvalidRequest
ErrInvalidStorageClass
@ -1124,6 +1125,12 @@ var errorCodes = errorCodeMap{
Description: "A timeout occurred while trying to lock a resource, please reduce your request rate",
HTTPStatusCode: http.StatusServiceUnavailable,
},
ErrGatewayTimeout: {
ErrCode: ErrGatewayTimeout,
Code: "GatewayTimeout",
Description: "The server is acting as a gateway and cannot get a response in time",
HTTPStatusCode: http.StatusGatewayTimeout,
},
ErrOperationMaxedOut: {
ErrCode: ErrOperationMaxedOut,
Code: "SlowDown",

4
api/handler/util.go
View file

@ -52,6 +52,10 @@ func transformToS3Error(err error) error {
return errors.GetAPIError(errors.ErrAccessDenied)
}
if errorsStd.Is(err, layer.ErrGatewayTimeout) {
return errors.GetAPIError(errors.ErrGatewayTimeout)
}
return errors.GetAPIError(errors.ErrInternalError)
}

9
api/layer/frostfs.go
View file

@ -128,8 +128,13 @@ type PrmObjectDelete struct {
Object oid.ID
}
// ErrAccessDenied is returned from FrostFS in case of access violation.
var ErrAccessDenied = errors.New("access denied")
var (
// ErrAccessDenied is returned from FrostFS in case of access violation.
ErrAccessDenied = errors.New("access denied")
// ErrGatewayTimeout is returned from FrostFS in case of timeout, deadline exceeded etc.
ErrGatewayTimeout = errors.New("gateway timeout")
)
// FrostFS represents virtual connection to FrostFS network.
type FrostFS interface {

42
internal/frostfs/errors/errors.go Normal file
View file

@ -0,0 +1,42 @@
package errors
import (
"context"
"errors"
"strings"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
func UnwrapErr(err error) error {
unwrappedErr := errors.Unwrap(err)
for unwrappedErr != nil {
err = unwrappedErr
unwrappedErr = errors.Unwrap(err)
}
return err
}
func IsErrObjectAccessDenied(err error) (string, bool) {
err = UnwrapErr(err)
switch err := err.(type) {
default:
return "", false
case apistatus.ObjectAccessDenied:
return err.Reason(), true
case *apistatus.ObjectAccessDenied:
return err.Reason(), true
}
}
func IsTimeoutError(err error) bool {
if strings.Contains(err.Error(), "timeout") ||
errors.Is(err, context.DeadlineExceeded) {
return true
}
return status.Code(UnwrapErr(err)) == codes.DeadlineExceeded
}

123
internal/frostfs/frostfs.go
View file

@ -14,7 +14,7 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/authmate"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/creds/tokens"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
errorsFrost "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/frostfs/errors"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/acl"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
@ -61,7 +61,7 @@ func (x *FrostFS) TimeToEpoch(ctx context.Context, now, futureTime time.Time) (u
networkInfo, err := x.pool.NetworkInfo(ctx)
if err != nil {
return 0, 0, fmt.Errorf("get network info via client: %w", err)
return 0, 0, handleObjectError("get network info via client", err)
}
durEpoch := networkInfo.EpochDuration()
@ -94,7 +94,7 @@ func (x *FrostFS) Container(ctx context.Context, idCnr cid.ID) (*container.Conta
res, err := x.pool.GetContainer(ctx, prm)
if err != nil {
return nil, fmt.Errorf("read container via connection pool: %w", err)
return nil, handleObjectError("read container via connection pool", err)
}
return &res, nil
@ -136,7 +136,7 @@ func (x *FrostFS) CreateContainer(ctx context.Context, prm layer.PrmContainerCre
err := pool.SyncContainerWithNetwork(ctx, &cnr, x.pool)
if err != nil {
return cid.ID{}, fmt.Errorf("sync container with the network state: %w", err)
return cid.ID{}, handleObjectError("sync container with the network state", err)
}
var prmPut pool.PrmContainerPut
@ -149,11 +149,7 @@ func (x *FrostFS) CreateContainer(ctx context.Context, prm layer.PrmContainerCre
// send request to save the container
idCnr, err := x.pool.PutContainer(ctx, prmPut)
if err != nil {
return cid.ID{}, fmt.Errorf("save container via connection pool: %w", err)
}
return idCnr, nil
return idCnr, handleObjectError("save container via connection pool", err)
}
// UserContainers implements frostfs.FrostFS interface method.
@ -162,11 +158,7 @@ func (x *FrostFS) UserContainers(ctx context.Context, id user.ID) ([]cid.ID, err
prm.SetOwnerID(id)
r, err := x.pool.ListContainers(ctx, prm)
if err != nil {
return nil, fmt.Errorf("list user containers via connection pool: %w", err)
}
return r, nil
return r, handleObjectError("list user containers via connection pool", err)
}
// SetContainerEACL implements frostfs.FrostFS interface method.
@ -180,11 +172,7 @@ func (x *FrostFS) SetContainerEACL(ctx context.Context, table eacl.Table, sessio
}
err := x.pool.SetEACL(ctx, prm)
if err != nil {
return fmt.Errorf("save eACL via connection pool: %w", err)
}
return err
return handleObjectError("save eACL via connection pool", err)
}
// ContainerEACL implements frostfs.FrostFS interface method.
@ -194,7 +182,7 @@ func (x *FrostFS) ContainerEACL(ctx context.Context, id cid.ID) (*eacl.Table, er
res, err := x.pool.GetEACL(ctx, prm)
if err != nil {
return nil, fmt.Errorf("read eACL via connection pool: %w", err)
return nil, handleObjectError("read eACL via connection pool", err)
}
return &res, nil
@ -211,11 +199,7 @@ func (x *FrostFS) DeleteContainer(ctx context.Context, id cid.ID, token *session
}
err := x.pool.DeleteContainer(ctx, prm)
if err != nil {
return fmt.Errorf("delete container via connection pool: %w", err)
}
return nil
return handleObjectError("delete container via connection pool", err)
}
// CreateObject implements frostfs.FrostFS interface method.
@ -278,15 +262,7 @@ func (x *FrostFS) CreateObject(ctx context.Context, prm layer.PrmObjectCreate) (
}
idObj, err := x.pool.PutObject(ctx, prmPut)
if err != nil {
reason, ok := isErrAccessDenied(err)
if ok {
return oid.ID{}, fmt.Errorf("%w: %s", layer.ErrAccessDenied, reason)
}
return oid.ID{}, fmt.Errorf("save object via connection pool: %w", err)
}
return idObj, nil
return idObj, handleObjectError("save object via connection pool", err)
}
// wraps io.ReadCloser and transforms Read errors related to access violation
@ -297,13 +273,7 @@ type payloadReader struct {
func (x payloadReader) Read(p []byte) (int, error) {
n, err := x.ReadCloser.Read(p)
if err != nil {
if reason, ok := isErrAccessDenied(err); ok {
return n, fmt.Errorf("%w: %s", layer.ErrAccessDenied, reason)
}
}
return n, err
return n, handleObjectError("read payload", err)
}
// ReadObject implements frostfs.FrostFS interface method.
@ -325,18 +295,14 @@ func (x *FrostFS) ReadObject(ctx context.Context, prm layer.PrmObjectRead) (*lay
if prm.WithPayload {
res, err := x.pool.GetObject(ctx, prmGet)
if err != nil {
if reason, ok := isErrAccessDenied(err); ok {
return nil, fmt.Errorf("%w: %s", layer.ErrAccessDenied, reason)
}
return nil, fmt.Errorf("init full object reading via connection pool: %w", err)
return nil, handleObjectError("init full object reading via connection pool", err)
}
defer res.Payload.Close()
payload, err := io.ReadAll(res.Payload)
if err != nil {
return nil, fmt.Errorf("read full object payload: %w", err)
return nil, handleObjectError("read full object payload", err)
}
res.Header.SetPayload(payload)
@ -357,11 +323,7 @@ func (x *FrostFS) ReadObject(ctx context.Context, prm layer.PrmObjectRead) (*lay
hdr, err := x.pool.HeadObject(ctx, prmHead)
if err != nil {
if reason, ok := isErrAccessDenied(err); ok {
return nil, fmt.Errorf("%w: %s", layer.ErrAccessDenied, reason)
}
return nil, fmt.Errorf("read object header via connection pool: %w", err)
return nil, handleObjectError("read object header via connection pool", err)
}
return &layer.ObjectPart{
@ -370,11 +332,7 @@ func (x *FrostFS) ReadObject(ctx context.Context, prm layer.PrmObjectRead) (*lay
} else if prm.PayloadRange[0]+prm.PayloadRange[1] == 0 {
res, err := x.pool.GetObject(ctx, prmGet)
if err != nil {
if reason, ok := isErrAccessDenied(err); ok {
return nil, fmt.Errorf("%w: %s", layer.ErrAccessDenied, reason)
}
return nil, fmt.Errorf("init full payload range reading via connection pool: %w", err)
return nil, handleObjectError("init full payload range reading via connection pool", err)
}
return &layer.ObjectPart{
@ -395,11 +353,7 @@ func (x *FrostFS) ReadObject(ctx context.Context, prm layer.PrmObjectRead) (*lay
res, err := x.pool.ObjectRange(ctx, prmRange)
if err != nil {
if reason, ok := isErrAccessDenied(err); ok {
return nil, fmt.Errorf("%w: %s", layer.ErrAccessDenied, reason)
}
return nil, fmt.Errorf("init payload range reading via connection pool: %w", err)
return nil, handleObjectError("init payload range reading via connection pool", err)
}
return &layer.ObjectPart{
@ -423,32 +377,7 @@ func (x *FrostFS) DeleteObject(ctx context.Context, prm layer.PrmObjectDelete) e
}
err := x.pool.DeleteObject(ctx, prmDelete)
if err != nil {
if reason, ok := isErrAccessDenied(err); ok {
return fmt.Errorf("%w: %s", layer.ErrAccessDenied, reason)
}
return fmt.Errorf("mark object removal via connection pool: %w", err)
}
return nil
}
func isErrAccessDenied(err error) (string, bool) {
unwrappedErr := errors.Unwrap(err)
for unwrappedErr != nil {
err = unwrappedErr
unwrappedErr = errors.Unwrap(err)
}
switch err := err.(type) {
default:
return "", false
case apistatus.ObjectAccessDenied:
return err.Reason(), true
case *apistatus.ObjectAccessDenied:
return err.Reason(), true
}
return handleObjectError("mark object removal via connection pool", err)
}
// ResolverFrostFS represents virtual connection to the FrostFS network.
@ -466,7 +395,7 @@ func NewResolverFrostFS(p *pool.Pool) *ResolverFrostFS {
func (x *ResolverFrostFS) SystemDNS(ctx context.Context) (string, error) {
networkInfo, err := x.pool.NetworkInfo(ctx)
if err != nil {
return "", fmt.Errorf("read network info via client: %w", err)
return "", handleObjectError("read network info via client", err)
}
domain := networkInfo.RawNetworkParameter("SystemDNS")
@ -477,6 +406,22 @@ func (x *ResolverFrostFS) SystemDNS(ctx context.Context) (string, error) {
return string(domain), nil
}
func handleObjectError(msg string, err error) error {
if err == nil {
return nil
}
if reason, ok := errorsFrost.IsErrObjectAccessDenied(err); ok {
return fmt.Errorf("%s: %w: %s", msg, layer.ErrAccessDenied, reason)
}
if errorsFrost.IsTimeoutError(err) {
return fmt.Errorf("%s: %w: %s", msg, layer.ErrGatewayTimeout, err.Error())
}
return fmt.Errorf("%s: %w", msg, err)
}
// AuthmateFrostFS is a mediator which implements authmate.FrostFS through pool.Pool.
type AuthmateFrostFS struct {
frostFS *FrostFS

27
internal/frostfs/frostfs_test.go
View file

@ -1,12 +1,18 @@
package frostfs
import (
"context"
"errors"
"fmt"
"testing"
"time"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer"
errorsFrost "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/frostfs/errors"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
"github.com/stretchr/testify/require"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
func TestErrorChecking(t *testing.T) {
@ -16,10 +22,29 @@ func TestErrorChecking(t *testing.T) {
var wrappedError error
if fetchedReason, ok := isErrAccessDenied(err); ok {
if fetchedReason, ok := errorsFrost.IsErrObjectAccessDenied(err); ok {
wrappedError = fmt.Errorf("%w: %s", layer.ErrAccessDenied, fetchedReason)
}
require.ErrorIs(t, wrappedError, layer.ErrAccessDenied)
require.Contains(t, wrappedError.Error(), reason)
}
func TestErrorTimeoutChecking(t *testing.T) {
t.Run("simple timeout", func(t *testing.T) {
require.True(t, errorsFrost.IsTimeoutError(errors.New("timeout")))
})
t.Run("deadline exceeded", func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
defer cancel()
time.Sleep(50 * time.Millisecond)
require.True(t, errorsFrost.IsTimeoutError(ctx.Err()))
})
t.Run("grpc deadline exceeded", func(t *testing.T) {
err := fmt.Errorf("wrap grpc error: %w", status.Error(codes.DeadlineExceeded, "error"))
require.True(t, errorsFrost.IsTimeoutError(err))
})
}

4
internal/frostfs/services/pool_wrapper.go
View file

@ -9,6 +9,7 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/data"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/creds/accessbox"
errorsFrost "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/frostfs/errors"
"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/pkg/service/tree"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
treepool "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/pool/tree"
@ -188,6 +189,9 @@ func handleError(err error) error {
if errors.Is(err, treepool.ErrNodeAccessDenied) {
return fmt.Errorf("%w: %s", tree.ErrNodeAccessDenied, err.Error())
}
if errorsFrost.IsTimeoutError(err) {
return fmt.Errorf("%w: %s", tree.ErrGatewayTimeout, err.Error())
}
return err
}

3
pkg/service/tree/tree.go
View file

@ -62,6 +62,9 @@ var (
// ErrNodeAccessDenied is returned from ServiceClient service in case of access denied error.
ErrNodeAccessDenied = layer.ErrNodeAccessDenied
// ErrGatewayTimeout is returned from ServiceClient service in case of timeout error.
ErrGatewayTimeout = layer.ErrGatewayTimeout
)
const (