[#60] Use periodic white space XML writer in Complete Multipart Upload

This mechanism is used by Amazon S3 to keep client's
connection alive while object is being constructed from
the upload parts.

Signed-off-by: Alex Vanin <a.vanin@yadro.com>
This commit is contained in:
Alexey Vanin 2023-03-20 11:44:35 +03:00
parent 2282c32822
commit 8151753eeb
7 changed files with 59 additions and 26 deletions

View file

@ -34,6 +34,7 @@ type (
CopiesNumber uint32 CopiesNumber uint32
ResolveZoneList []string ResolveZoneList []string
IsResolveListAllow bool // True if ResolveZoneList contains allowed zones IsResolveListAllow bool // True if ResolveZoneList contains allowed zones
CompleteMultipartKeepalive time.Duration
} }
PlacementPolicy interface { PlacementPolicy interface {

View file

@ -400,9 +400,15 @@ func (h *handler) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.
Parts: reqBody.Parts, Parts: reqBody.Parts,
} }
// Next operations might take some time, so we want to keep client's
// connection alive. To do so, gateway sends periodic white spaces
// back to the client the same way as Amazon S3 service does.
stopPeriodicResponseWriter := periodicXMLWriter(w, h.cfg.CompleteMultipartKeepalive)
uploadData, extendedObjInfo, err := h.obj.CompleteMultipartUpload(r.Context(), c) uploadData, extendedObjInfo, err := h.obj.CompleteMultipartUpload(r.Context(), c)
if err != nil { if err != nil {
h.logAndSendError(w, "could not complete multipart upload", reqInfo, err, additional...) logAndSendError := h.periodicWriterErrorSender(stopPeriodicResponseWriter())
logAndSendError(w, "could not complete multipart upload", reqInfo, err, additional...)
return return
} }
objInfo := extendedObjInfo.ObjectInfo objInfo := extendedObjInfo.ObjectInfo
@ -418,7 +424,8 @@ func (h *handler) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.
NodeVersion: extendedObjInfo.NodeVersion, NodeVersion: extendedObjInfo.NodeVersion,
} }
if _, err = h.obj.PutObjectTagging(r.Context(), tagPrm); err != nil { if _, err = h.obj.PutObjectTagging(r.Context(), tagPrm); err != nil {
h.logAndSendError(w, "could not put tagging file of completed multipart upload", reqInfo, err, additional...) logAndSendError := h.periodicWriterErrorSender(stopPeriodicResponseWriter())
logAndSendError(w, "could not put tagging file of completed multipart upload", reqInfo, err, additional...)
return return
} }
} }
@ -426,12 +433,14 @@ func (h *handler) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.
if len(uploadData.ACLHeaders) != 0 { if len(uploadData.ACLHeaders) != 0 {
key, err := h.bearerTokenIssuerKey(r.Context()) key, err := h.bearerTokenIssuerKey(r.Context())
if err != nil { if err != nil {
h.logAndSendError(w, "couldn't get gate key", reqInfo, err) logAndSendError := h.periodicWriterErrorSender(stopPeriodicResponseWriter())
logAndSendError(w, "couldn't get gate key", reqInfo, err)
return return
} }
acl, err := parseACLHeaders(r.Header, key) acl, err := parseACLHeaders(r.Header, key)
if err != nil { if err != nil {
h.logAndSendError(w, "could not parse acl", reqInfo, err) logAndSendError := h.periodicWriterErrorSender(stopPeriodicResponseWriter())
logAndSendError(w, "could not parse acl", reqInfo, err)
return return
} }
@ -441,11 +450,13 @@ func (h *handler) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.
} }
astObject, err := aclToAst(acl, resInfo) astObject, err := aclToAst(acl, resInfo)
if err != nil { if err != nil {
h.logAndSendError(w, "could not translate acl of completed multipart upload to ast", reqInfo, err, additional...) logAndSendError := h.periodicWriterErrorSender(stopPeriodicResponseWriter())
logAndSendError(w, "could not translate acl of completed multipart upload to ast", reqInfo, err, additional...)
return return
} }
if _, err = h.updateBucketACL(r, astObject, bktInfo, sessionTokenSetEACL); err != nil { if _, err = h.updateBucketACL(r, astObject, bktInfo, sessionTokenSetEACL); err != nil {
h.logAndSendError(w, "could not update bucket acl while completing multipart upload", reqInfo, err, additional...) logAndSendError := h.periodicWriterErrorSender(stopPeriodicResponseWriter())
logAndSendError(w, "could not update bucket acl while completing multipart upload", reqInfo, err, additional...)
return return
} }
} }
@ -460,25 +471,27 @@ func (h *handler) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.
h.log.Error("couldn't send notification: %w", zap.Error(err)) h.log.Error("couldn't send notification: %w", zap.Error(err))
} }
bktSettings, err := h.obj.GetBucketSettings(r.Context(), bktInfo)
if err != nil {
h.logAndSendError(w, "could not get bucket settings", reqInfo, err)
}
response := CompleteMultipartUploadResponse{ response := CompleteMultipartUploadResponse{
Bucket: objInfo.Bucket, Bucket: objInfo.Bucket,
ETag: objInfo.HashSum, ETag: objInfo.HashSum,
Key: objInfo.Name, Key: objInfo.Name,
} }
if bktSettings.VersioningEnabled() { // Here we previously set api.AmzVersionID header for versioned bucket.
w.Header().Set(api.AmzVersionID, objInfo.VersionID()) // It is not possible after #60, because we introduced periodic white
} // space XML writer to keep connection with the client.
headerIsWritten := stopPeriodicResponseWriter()
if headerIsWritten {
if err = api.EncodeToResponseNoHeader(w, response); err != nil {
h.logAndSendErrorNoHeader(w, "something went wrong", reqInfo, err)
}
} else {
if err = api.EncodeToResponse(w, response); err != nil { if err = api.EncodeToResponse(w, response); err != nil {
h.logAndSendError(w, "something went wrong", reqInfo, err) h.logAndSendError(w, "something went wrong", reqInfo, err)
} }
} }
}
func (h *handler) ListMultipartUploadsHandler(w http.ResponseWriter, r *http.Request) { func (h *handler) ListMultipartUploadsHandler(w http.ResponseWriter, r *http.Request) {
reqInfo := api.GetReqInfo(r.Context()) reqInfo := api.GetReqInfo(r.Context())
@ -732,3 +745,12 @@ func periodicXMLWriter(w io.Writer, dur time.Duration) (stop func() bool) {
return stop return stop
} }
// periodicWriterErrorSender returns handler function to send error. If header is
// alreay written by periodic XML writer, do not send HTTP and XML headers.
func (h *handler) periodicWriterErrorSender(headerWritten bool) func(http.ResponseWriter, string, *api.ReqInfo, error, ...zap.Field) {
if headerWritten {
return h.logAndSendErrorNoHeader
}
return h.logAndSendError
}

View file

@ -654,6 +654,8 @@ func (a *App) initHandler() {
cfg.ResolveZoneList = a.cfg.GetStringSlice(cfgResolveBucketDeny) cfg.ResolveZoneList = a.cfg.GetStringSlice(cfgResolveBucketDeny)
} }
cfg.CompleteMultipartKeepalive = a.cfg.GetDuration(cfgKludgeCompleteMultipartUploadKeepalive)
var err error var err error
a.api, err = handler.New(a.log, a.obj, a.nc, cfg) a.api, err = handler.New(a.log, a.obj, a.nc, cfg)
if err != nil { if err != nil {

View file

@ -115,6 +115,7 @@ const ( // Settings.
// Kludge. // Kludge.
cfgKludgeUseDefaultXMLNSForCompleteMultipartUpload = "kludge.use_default_xmlns_for_complete_multipart" cfgKludgeUseDefaultXMLNSForCompleteMultipartUpload = "kludge.use_default_xmlns_for_complete_multipart"
cfgKludgeCompleteMultipartUploadKeepalive = "kludge.complete_multipart_keepalive"
// Command line args. // Command line args.
cmdHelp = "help" cmdHelp = "help"
@ -258,6 +259,7 @@ func newSettings() *viper.Viper {
// kludge // kludge
v.SetDefault(cfgKludgeUseDefaultXMLNSForCompleteMultipartUpload, false) v.SetDefault(cfgKludgeUseDefaultXMLNSForCompleteMultipartUpload, false)
v.SetDefault(cfgKludgeCompleteMultipartUploadKeepalive, 10*time.Second)
// Bind flags // Bind flags
if err := bindFlags(v, flags); err != nil { if err := bindFlags(v, flags); err != nil {

View file

@ -130,3 +130,5 @@ S3_GW_RESOLVE_BUCKET_ALLOW=container
# Enable using default xml namespace `http://s3.amazonaws.com/doc/2006-03-01/` when parse`CompleteMultipartUpload` xml body. # Enable using default xml namespace `http://s3.amazonaws.com/doc/2006-03-01/` when parse`CompleteMultipartUpload` xml body.
S3_GW_KLUDGE_USE_DEFAULT_XMLNS_FOR_COMPLETE_MULTIPART=false S3_GW_KLUDGE_USE_DEFAULT_XMLNS_FOR_COMPLETE_MULTIPART=false
# Set timeout between whitespace transmissions during CompleteMultipartUpload processing.
S3_GW_KLUDGE_COMPLETE_MULTIPART_KEEPALIVE=10s

View file

@ -153,3 +153,5 @@ resolve_bucket:
kludge: kludge:
# Enable using default xml namespace `http://s3.amazonaws.com/doc/2006-03-01/` when parse`CompleteMultipartUpload` xml body. # Enable using default xml namespace `http://s3.amazonaws.com/doc/2006-03-01/` when parse`CompleteMultipartUpload` xml body.
use_default_xmlns_for_complete_multipart: false use_default_xmlns_for_complete_multipart: false
# Set timeout between whitespace transmissions during CompleteMultipartUpload processing.
complete_multipart_keepalive: 10s

View file

@ -504,8 +504,10 @@ Workarounds for non-standard use cases.
```yaml ```yaml
kludge: kludge:
use_default_xmlns_for_complete_multipart: false use_default_xmlns_for_complete_multipart: false
complete_multipart_keepalive: 10s
``` ```
| Parameter | Type | SIGHUP reload | Default value | Description | | Parameter | Type | SIGHUP reload | Default value | Description |
|--------------------------------------------|--------|---------------|---------------|-----------------------------------------------------------------------------------------------------------------------------| |--------------------------------------------|------------|---------------|---------------|-----------------------------------------------------------------------------------------------------------------------------|
| `use_default_xmlns_for_complete_multipart` | `bool` | yes | false | Enable using default xml namespace `http://s3.amazonaws.com/doc/2006-03-01/` when parse `CompleteMultipartUpload` xml body. | | `use_default_xmlns_for_complete_multipart` | `bool` | yes | false | Enable using default xml namespace `http://s3.amazonaws.com/doc/2006-03-01/` when parse `CompleteMultipartUpload` xml body. |
| `complete_multipart_keepalive` | `duration` | no | 10s | Set timeout between whitespace transmissions during CompleteMultipartUpload processing. |