diff --git a/api/handler/api.go b/api/handler/api.go index 19e2c4e8..fa5c57b8 100644 --- a/api/handler/api.go +++ b/api/handler/api.go @@ -42,6 +42,7 @@ type ( RetryMaxBackoff() time.Duration RetryStrategy() RetryStrategy TLSTerminationHeader() string + ListingKeepaliveThrottle() time.Duration } FrostFSID interface { diff --git a/api/handler/handlers_test.go b/api/handler/handlers_test.go index dc71d1c2..2c903b38 100644 --- a/api/handler/handlers_test.go +++ b/api/handler/handlers_test.go @@ -154,6 +154,10 @@ func (c *configMock) TLSTerminationHeader() string { return c.tlsTerminationHeader } +func (c *configMock) ListingKeepaliveThrottle() time.Duration { + return 0 +} + func (c *configMock) putLocationConstraint(constraint string) { c.placementPolicies[constraint] = c.defaultPolicy } diff --git a/api/handler/object_list.go b/api/handler/object_list.go index 6d68501d..56df4468 100644 --- a/api/handler/object_list.go +++ b/api/handler/object_list.go @@ -1,6 +1,9 @@ package handler import ( + "context" + "encoding/xml" + "io" "net/http" "net/url" "strconv" @@ -14,6 +17,7 @@ import ( "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer" "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/middleware" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" + "go.uber.org/zap" ) const maxObjectList = 1000 // Limit number of objects in a listObjectsResponse/listObjectsVersionsResponse. @@ -35,14 +39,28 @@ func (h *handler) ListObjectsV1Handler(w http.ResponseWriter, r *http.Request) { return } + ch := make(chan struct{}) + defer close(ch) + params.Chan = ch + + stopPeriodicResponseWriter := periodicXMLWriter(w, h.cfg.ListingKeepaliveThrottle(), ch) + list, err := h.obj.ListObjectsV1(ctx, params) if err != nil { - h.logAndSendError(ctx, w, "something went wrong", reqInfo, err) + logAndSendError := h.periodicWriterErrorSender(stopPeriodicResponseWriter()) + logAndSendError(ctx, w, "could not list objects v1", reqInfo, err) return } - if err = middleware.EncodeToResponse(w, h.encodeV1(params, list)); err != nil { - h.logAndSendError(ctx, w, "something went wrong", reqInfo, err) + headerIsWritten := stopPeriodicResponseWriter() + if headerIsWritten { + if err = middleware.EncodeToResponseNoHeader(w, h.encodeV1(params, list)); err != nil { + h.logAndSendErrorNoHeader(ctx, w, "could not encode listing v1 response", reqInfo, err) + } + } else { + if err = middleware.EncodeToResponse(w, h.encodeV1(params, list)); err != nil { + h.logAndSendError(ctx, w, "could not encode listing v1 response", reqInfo, err) + } } } @@ -82,14 +100,28 @@ func (h *handler) ListObjectsV2Handler(w http.ResponseWriter, r *http.Request) { return } + ch := make(chan struct{}) + defer close(ch) + params.Chan = ch + + stopPeriodicResponseWriter := periodicXMLWriter(w, h.cfg.ListingKeepaliveThrottle(), ch) + list, err := h.obj.ListObjectsV2(ctx, params) if err != nil { - h.logAndSendError(ctx, w, "something went wrong", reqInfo, err) + logAndSendError := h.periodicWriterErrorSender(stopPeriodicResponseWriter()) + logAndSendError(ctx, w, "could not list objects v2", reqInfo, err) return } - if err = middleware.EncodeToResponse(w, h.encodeV2(params, list)); err != nil { - h.logAndSendError(ctx, w, "something went wrong", reqInfo, err) + headerIsWritten := stopPeriodicResponseWriter() + if headerIsWritten { + if err = middleware.EncodeToResponseNoHeader(w, h.encodeV2(params, list)); err != nil { + h.logAndSendErrorNoHeader(ctx, w, "could not encode listing v2 response", reqInfo, err) + } + } else { + if err = middleware.EncodeToResponse(w, h.encodeV2(params, list)); err != nil { + h.logAndSendError(ctx, w, "could not encode listing v2 response", reqInfo, err) + } } } @@ -243,15 +275,29 @@ func (h *handler) ListBucketObjectVersionsHandler(w http.ResponseWriter, r *http return } - info, err := h.obj.ListObjectVersions(ctx, p) + ch := make(chan struct{}) + defer close(ch) + p.Chan = ch + + stopPeriodicResponseWriter := periodicXMLWriter(w, h.cfg.ListingKeepaliveThrottle(), ch) + + list, err := h.obj.ListObjectVersions(ctx, p) if err != nil { - h.logAndSendError(ctx, w, "something went wrong", reqInfo, err) + logAndSendError := h.periodicWriterErrorSender(stopPeriodicResponseWriter()) + logAndSendError(ctx, w, "could not list objects versions", reqInfo, err) return } - response := encodeListObjectVersionsToResponse(p, info, p.BktInfo.Name, h.cfg.MD5Enabled()) - if err = middleware.EncodeToResponse(w, response); err != nil { - h.logAndSendError(ctx, w, "something went wrong", reqInfo, err) + response := encodeListObjectVersionsToResponse(p, list, p.BktInfo.Name, h.cfg.MD5Enabled()) + headerIsWritten := stopPeriodicResponseWriter() + if headerIsWritten { + if err = middleware.EncodeToResponseNoHeader(w, response); err != nil { + h.logAndSendErrorNoHeader(ctx, w, "could not encode listing versions response", reqInfo, err) + } + } else { + if err = middleware.EncodeToResponse(w, response); err != nil { + h.logAndSendError(ctx, w, "could not encode listing versions response", reqInfo, err) + } } } @@ -334,3 +380,70 @@ func encodeListObjectVersionsToResponse(p *layer.ListObjectVersionsParams, info return &res } + +// periodicWriterErrorSender returns handler function to send error. If header is +// already written by periodic XML writer, do not send HTTP and XML headers. +func (h *handler) periodicWriterErrorSender(headerWritten bool) func(context.Context, http.ResponseWriter, string, *middleware.ReqInfo, error, ...zap.Field) { + if headerWritten { + return h.logAndSendErrorNoHeader + } + return h.logAndSendError +} + +// periodicXMLWriter creates go routine to write xml header and whitespaces +// over time to avoid connection drop from the client. To work properly, +// pass `http.ResponseWriter` with implemented `http.Flusher` interface. +// Returns stop function which returns boolean if writer has been used +// during goroutine execution. Zero or negative duration disable writing. +func periodicXMLWriter(w io.Writer, dur time.Duration, ch <-chan struct{}) (stop func() bool) { + if dur <= 0 { + return func() bool { return false } + } + + whitespaceChar := []byte(" ") + closer := make(chan struct{}) + done := make(chan struct{}) + headerWritten := false + + go func() { + defer close(done) + + var lastEvent time.Time + + for { + select { + case _, ok := <-ch: + if !ok { + return + } + if time.Since(lastEvent) < dur { + continue + } + + lastEvent = time.Now() + + if !headerWritten { + _, err := w.Write([]byte(xml.Header)) + headerWritten = err == nil + } + _, err := w.Write(whitespaceChar) + if err != nil { + return // is there anything we can do better than ignore error? + } + if buffered, ok := w.(http.Flusher); ok { + buffered.Flush() + } + case <-closer: + return + } + } + }() + + stop = func() bool { + close(closer) + <-done // wait for goroutine to stop + return headerWritten + } + + return stop +} diff --git a/api/handler/object_list_test.go b/api/handler/object_list_test.go index 7a6664b7..3c6345b1 100644 --- a/api/handler/object_list_test.go +++ b/api/handler/object_list_test.go @@ -1,7 +1,9 @@ package handler import ( + "bytes" "context" + "encoding/xml" "fmt" "net/http" "net/http/httptest" @@ -841,6 +843,101 @@ func TestListingsWithInvalidEncodingType(t *testing.T) { listObjectsV1Err(hc, bktName, "invalid", apierr.GetAPIError(apierr.ErrInvalidEncodingMethod)) } +func TestPeriodicWriter(t *testing.T) { + const dur = 100 * time.Millisecond + const whitespaces = 8 + expected := []byte(xml.Header) + for i := 0; i < whitespaces; i++ { + expected = append(expected, []byte(" ")...) + } + + t.Run("writes data", func(t *testing.T) { + ch := make(chan struct{}) + go func() { + defer close(ch) + + for range whitespaces { + ch <- struct{}{} + time.Sleep(dur) + } + }() + + buf := bytes.NewBuffer(nil) + stop := periodicXMLWriter(buf, time.Nanosecond, ch) + + // N number of whitespaces + half durations to guarantee at least N writes in buffer + time.Sleep(whitespaces*dur + dur/2) + require.True(t, stop()) + require.Equal(t, string(expected), buf.String()) + + t.Run("no additional data after stop", func(t *testing.T) { + time.Sleep(2 * dur) + require.Equal(t, string(expected), buf.String()) + }) + }) + + t.Run("does not write data", func(t *testing.T) { + buf := bytes.NewBuffer(nil) + + ch := make(chan struct{}) + go func() { + defer close(ch) + + for range whitespaces { + time.Sleep(2 * dur) + ch <- struct{}{} + } + }() + + stop := periodicXMLWriter(buf, time.Nanosecond, ch) + require.False(t, stop()) + require.Empty(t, buf.Bytes()) + }) + + t.Run("throttling works", func(t *testing.T) { + buf := bytes.NewBuffer(nil) + + ch := make(chan struct{}) + go func() { + defer close(ch) + + for range whitespaces { + ch <- struct{}{} + time.Sleep(dur / 2) + } + }() + + stop := periodicXMLWriter(buf, dur, ch) + // N number of whitespaces + half durations to guarantee at least N writes in buffer + time.Sleep(whitespaces*dur + dur/2) + require.True(t, stop()) + require.Equal(t, string(expected[:len(expected)-4]), buf.String()) + }) + + t.Run("disabled", func(t *testing.T) { + buf := bytes.NewBuffer(nil) + + ch := make(chan struct{}) + go func() { + defer close(ch) + + for range whitespaces { + select { + case ch <- struct{}{}: + default: + } + time.Sleep(dur / 2) + } + }() + + stop := periodicXMLWriter(buf, 0, ch) + // N number of whitespaces + half durations to guarantee at least N writes in buffer + time.Sleep(whitespaces*dur + dur/2) + require.False(t, stop()) + require.Empty(t, buf.String()) + }) +} + func checkVersionsNames(t *testing.T, versions *ListObjectsVersionsResponse, names []string) { for i, v := range versions.Version { require.Equal(t, names[i], v.Key) diff --git a/api/handler/util.go b/api/handler/util.go index f5520b31..0e86dd67 100644 --- a/api/handler/util.go +++ b/api/handler/util.go @@ -45,6 +45,23 @@ func (h *handler) logAndSendError(ctx context.Context, w http.ResponseWriter, lo h.reqLogger(ctx).Error(logs.RequestFailed, append(fields, logs.TagField(logs.TagDatapath))...) } +func (h *handler) logAndSendErrorNoHeader(ctx context.Context, w http.ResponseWriter, logText string, reqInfo *middleware.ReqInfo, err error, additional ...zap.Field) { + err = handleDeleteMarker(w, err) + if wrErr := middleware.WriteErrorResponseNoHeader(w, reqInfo, apierr.TransformToS3Error(err)); wrErr != nil { + additional = append(additional, zap.NamedError("write_response_error", wrErr)) + } + fields := []zap.Field{ + zap.String("method", reqInfo.API), + zap.String("bucket", reqInfo.BucketName), + zap.String("object", reqInfo.ObjectName), + zap.String("description", logText), + zap.String("user", reqInfo.User), + zap.Error(err), + } + fields = append(fields, additional...) + h.reqLogger(ctx).Error(logs.RequestFailed, append(fields, logs.TagField(logs.TagDatapath))...) +} + func (h *handler) logError(ctx context.Context, logText string, reqInfo *middleware.ReqInfo, err error, additional ...zap.Field) { fields := []zap.Field{ zap.String("method", reqInfo.API), diff --git a/api/layer/layer.go b/api/layer/layer.go index 2dc2f993..e7f9cfa2 100644 --- a/api/layer/layer.go +++ b/api/layer/layer.go @@ -194,6 +194,7 @@ type ( Prefix string VersionIDMarker string Encode string + Chan chan<- struct{} } ListBucketsParams struct { diff --git a/api/layer/listing.go b/api/layer/listing.go index 03460600..50088ff5 100644 --- a/api/layer/listing.go +++ b/api/layer/listing.go @@ -27,6 +27,7 @@ type ( Encode string MaxKeys int Prefix string + Chan chan<- struct{} } // ListObjectsParamsV1 contains params for ListObjectsV1. @@ -81,6 +82,8 @@ type ( MaxKeys int Marker string Bookmark string + // Chan is a channel to prevent client from context canceling during long listing. + Chan chan<- struct{} } commonLatestVersionsListingParams struct { @@ -111,6 +114,7 @@ func (n *Layer) ListObjectsV1(ctx context.Context, p *ListObjectsParamsV1) (*Lis MaxKeys: p.MaxKeys, Marker: p.Marker, Bookmark: p.Marker, + Chan: p.Chan, }, ListType: ListObjectsV1Type, } @@ -145,6 +149,7 @@ func (n *Layer) ListObjectsV2(ctx context.Context, p *ListObjectsParamsV2) (*Lis MaxKeys: p.MaxKeys, Marker: p.StartAfter, Bookmark: p.ContinuationToken, + Chan: p.Chan, }, ListType: ListObjectsV2Type, } @@ -175,6 +180,7 @@ func (n *Layer) ListObjectVersions(ctx context.Context, p *ListObjectVersionsPar MaxKeys: p.MaxKeys, Marker: p.KeyMarker, Bookmark: p.VersionIDMarker, + Chan: p.Chan, } objects, isTruncated, err := n.getAllObjectsVersions(ctx, prm) @@ -218,6 +224,10 @@ func (n *Layer) getLatestObjectsVersions(ctx context.Context, p commonLatestVers objects = append(objects, session.Next...) for obj := range objOutCh { objects = append(objects, obj) + select { + case p.Chan <- struct{}{}: + default: + } } if err = <-errorCh; err != nil { @@ -287,6 +297,11 @@ func handleGeneratedVersions(objOutCh <-chan *data.ExtendedNodeVersion, p common allObjects = append(allObjects, eoi) } lastName = name + + select { + case p.Chan <- struct{}{}: + default: + } } formVersionsListRow(allObjects, listRowStartIndex, session) diff --git a/api/middleware/response.go b/api/middleware/response.go index fded5b33..02497699 100644 --- a/api/middleware/response.go +++ b/api/middleware/response.go @@ -144,6 +144,17 @@ func WriteErrorResponse(w http.ResponseWriter, reqInfo *ReqInfo, err error) (int return code, nil } +// WriteErrorResponseNoHeader writes XML encoded error to the response body. +func WriteErrorResponseNoHeader(w http.ResponseWriter, reqInfo *ReqInfo, err error) error { + errorResponse := getAPIErrorResponse(reqInfo, err) + encodedErrorResponse, err := EncodeResponseNoHeader(errorResponse) + if err != nil { + return err + } + + return WriteResponseBody(w, encodedErrorResponse) +} + // Write http common headers. func setCommonHeaders(w http.ResponseWriter) { w.Header().Set(hdrServerInfo, version.Server) @@ -200,6 +211,18 @@ func EncodeResponse(response interface{}) ([]byte, error) { return bytesBuffer.Bytes(), nil } +// EncodeResponseNoHeader encodes response without setting xml.Header. +// Should be used with periodicXMLWriter which sends xml.Header to the client +// with whitespaces to keep connection alive. +func EncodeResponseNoHeader(response interface{}) ([]byte, error) { + var bytesBuffer bytes.Buffer + if err := xml.NewEncoder(&bytesBuffer).Encode(response); err != nil { + return nil, err + } + + return bytesBuffer.Bytes(), nil +} + // EncodeToResponse encodes the response into ResponseWriter. func EncodeToResponse(w http.ResponseWriter, response interface{}) error { w.WriteHeader(http.StatusOK) diff --git a/cmd/s3-gw/app.go b/cmd/s3-gw/app.go index 023e5d54..807c453b 100644 --- a/cmd/s3-gw/app.go +++ b/cmd/s3-gw/app.go @@ -139,6 +139,7 @@ type ( tombstoneMembersSize int tombstoneLifetime uint64 tlsTerminationHeader string + listingKeepaliveThrottle time.Duration } maxClientsConfig struct { @@ -374,6 +375,7 @@ func (s *appSettings) update(v *viper.Viper, log *zap.Logger) { tombstoneMembersSize := fetchTombstoneMembersSize(v) tombstoneLifetime := fetchTombstoneLifetime(v) tlsTerminationHeader := v.GetString(cfgEncryptionTLSTerminationHeader) + listingKeepaliveThrottle := v.GetDuration(cfgKludgeListingKeepAliveThrottle) s.mu.Lock() defer s.mu.Unlock() @@ -407,6 +409,7 @@ func (s *appSettings) update(v *viper.Viper, log *zap.Logger) { s.tombstoneMembersSize = tombstoneMembersSize s.tombstoneLifetime = tombstoneLifetime s.tlsTerminationHeader = tlsTerminationHeader + s.listingKeepaliveThrottle = listingKeepaliveThrottle } func (s *appSettings) prepareVHSNamespaces(v *viper.Viper, log *zap.Logger, defaultNamespaces []string) map[string]bool { @@ -644,6 +647,12 @@ func (s *appSettings) TombstoneLifetime() uint64 { return s.tombstoneLifetime } +func (s *appSettings) ListingKeepaliveThrottle() time.Duration { + s.mu.RLock() + defer s.mu.RUnlock() + return s.listingKeepaliveThrottle +} + func (a *App) initAPI(ctx context.Context) { a.initLayer(ctx) a.initHandler() diff --git a/cmd/s3-gw/app_settings.go b/cmd/s3-gw/app_settings.go index 884a116a..3c02c554 100644 --- a/cmd/s3-gw/app_settings.go +++ b/cmd/s3-gw/app_settings.go @@ -202,6 +202,8 @@ const ( cfgKludgeBypassContentEncodingCheckInChunks = "kludge.bypass_content_encoding_check_in_chunks" cfgKludgeDefaultNamespaces = "kludge.default_namespaces" cfgKludgeProfile = "kludge.profile" + cfgKludgeListingKeepAliveThrottle = "kludge.listing_keepalive_throttle" + // Web. cfgWebReadTimeout = "web.read_timeout" cfgWebReadHeaderTimeout = "web.read_header_timeout" diff --git a/config/config.env b/config/config.env index 28b1c7b7..53171673 100644 --- a/config/config.env +++ b/config/config.env @@ -193,6 +193,10 @@ S3_GW_KLUDGE_USE_DEFAULT_XMLNS=false S3_GW_KLUDGE_BYPASS_CONTENT_ENCODING_CHECK_IN_CHUNKS=false # Namespaces that should be handled as default S3_GW_KLUDGE_DEFAULT_NAMESPACES="" "root" +# During listing the s3 gate may send whitespaces to client to prevent it from cancelling request. +# The gate is going to send whitespace every time it receives chunk of data from FrostFS storage. +# This parameter enables this feature and limits frequency of whitespace transmissions. +S3_GW_KLUDGE_LISTING_KEEPALIVE_THROTTLE=10s # Kludge profiles S3_GW_KLUDGE_PROFILE_0_USER_AGENT=aws-cli S3_GW_KLUDGE_PROFILE_0_USE_DEFAULT_XMLNS=true diff --git a/config/config.yaml b/config/config.yaml index d9630e09..28e83654 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -234,6 +234,10 @@ kludge: bypass_content_encoding_check_in_chunks: false # Namespaces that should be handled as default default_namespaces: [ "", "root" ] + # During listing the s3 gate may send whitespaces to client to prevent it from cancelling request. + # The gate is going to send whitespace every time it receives chunk of data from FrostFS storage. + # This parameter enables this feature and limits frequency of whitespace transmissions. + listing_keepalive_throttle: 10s # new profile section override defaults based on user agent profile: - user_agent: aws-cli diff --git a/docs/configuration.md b/docs/configuration.md index 42d5e021..5c923b0c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -670,6 +670,7 @@ kludge: use_default_xmlns: false bypass_content_encoding_check_in_chunks: false default_namespaces: [ "", "root" ] + listing_keepalive_throttle: 10s profile: - user_agent: aws-cli use_default_xmlns: false @@ -678,12 +679,13 @@ kludge: bypass_content_encoding_check_in_chunks: false ``` -| Parameter | Type | SIGHUP reload | Default value | Description | -|-------------------------------------------|----------------------------------|---------------|---------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| -| `use_default_xmlns` | `bool` | yes | `false` | Enable using default xml namespace `http://s3.amazonaws.com/doc/2006-03-01/` when parse xml bodies. | -| `bypass_content_encoding_check_in_chunks` | `bool` | yes | `false` | Use this flag to be able to use [chunked upload approach](https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html) without having `aws-chunked` value in `Content-Encoding` header. | -| `default_namespaces` | `[]string` | yes | `["","root"]` | Namespaces that should be handled as default. | -| `profile` | [[]Profile](#profile-subsection) | yes | | An array of configurable profiles. | +| Parameter | Type | SIGHUP reload | Default value | Description | +|-------------------------------------------|----------------------------------|---------------|----------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `use_default_xmlns` | `bool` | yes | `false` | Enable using default xml namespace `http://s3.amazonaws.com/doc/2006-03-01/` when parse xml bodies. | +| `bypass_content_encoding_check_in_chunks` | `bool` | yes | `false` | Use this flag to be able to use [chunked upload approach](https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html) without having `aws-chunked` value in `Content-Encoding` header. | +| `default_namespaces` | `[]string` | yes | `["","root"]` | Namespaces that should be handled as default. | +| `listing_keepalive_throttle` | `duration` | yes | `0` (means disabled) | During listing the s3 gate may send whitespaces to client to prevent it from cancelling request. The gate is going to send whitespace every time it receives chunk of data from FrostFS storage. This parameter enables this feature and limits frequency of whitespace transmissions. | +| `profile` | [[]Profile](#profile-subsection) | yes | | An array of configurable profiles. | #### `profile` subsection