[#644] Support keepalive during listing

Send whitespaces every time as new object in list is ready
to prevent client from context cancelling.

Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>

api/handler/api.go

@@ -42,6 +42,7 @@ type (
 		RetryMaxBackoff() time.Duration
 		RetryStrategy() RetryStrategy
 		TLSTerminationHeader() string
+		ListingKeepaliveThrottle() time.Duration
 	}
 
 	FrostFSID interface {

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
 }

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 {
+	headerIsWritten := stopPeriodicResponseWriter()
-		h.logAndSendError(ctx, w, "something went wrong", reqInfo, err)
+	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 {
+	headerIsWritten := stopPeriodicResponseWriter()
-		h.logAndSendError(ctx, w, "something went wrong", reqInfo, err)
+	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())
+	response := encodeListObjectVersionsToResponse(p, list, p.BktInfo.Name, h.cfg.MD5Enabled())
-	if err = middleware.EncodeToResponse(w, response); err != nil {
+	headerIsWritten := stopPeriodicResponseWriter()
-		h.logAndSendError(ctx, w, "something went wrong", reqInfo, err)
+	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
+}

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)

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),

api/layer/layer.go

@@ -194,6 +194,7 @@ type (
 		Prefix          string
 		VersionIDMarker string
 		Encode          string
+		Chan            chan<- struct{}
 	}
 
 	ListBucketsParams struct {

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)

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)

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

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"

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

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

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                                                                                                                                                                                      |
+| 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.                                                                                              |
+| `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. |
+| `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.                                                                                                                                                    |
+| `default_namespaces`                      | `[]string`                       | yes           | `["","root"]`        | Namespaces that should be handled as default.                                                                                                                                                                                                                                          |
-| `profile`                                 | [[]Profile](#profile-subsection) | yes           |               | An array of configurable profiles.                                                                                                                                                               |
+| `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