diff --git a/aws/client/client.go b/aws/client/client.go index 4003c04b012..7c0e7d9dd1e 100644 --- a/aws/client/client.go +++ b/aws/client/client.go @@ -2,7 +2,6 @@ package client import ( "fmt" - "io/ioutil" "net/http/httputil" "github.com/aws/aws-sdk-go/aws" @@ -104,8 +103,7 @@ func logRequest(r *request.Request) { // Reset the request body because dumpRequest will re-wrap the r.HTTPRequest's // Body as a NoOpCloser and will not be reset after read by the HTTP // client reader. - r.Body.Seek(r.BodyStart, 0) - r.HTTPRequest.Body = ioutil.NopCloser(r.Body) + r.ResetBody() } r.Config.Logger.Log(fmt.Sprintf(logReqMsg, r.ClientInfo.ServiceName, r.Operation.Name, string(dumpedBody))) diff --git a/aws/request/offset_reader.go b/aws/request/offset_reader.go index da6396d2d93..02f07f4a462 100644 --- a/aws/request/offset_reader.go +++ b/aws/request/offset_reader.go @@ -9,7 +9,7 @@ import ( // with retrying requests type offsetReader struct { buf io.ReadSeeker - lock sync.RWMutex + lock sync.Mutex closed bool } @@ -21,7 +21,8 @@ func newOffsetReader(buf io.ReadSeeker, offset int64) *offsetReader { return reader } -// Close is a thread-safe close. Uses the write lock. +// Close will close the instance of the offset reader's access to +// the underlying io.ReadSeeker. func (o *offsetReader) Close() error { o.lock.Lock() defer o.lock.Unlock() @@ -29,10 +30,10 @@ func (o *offsetReader) Close() error { return nil } -// Read is a thread-safe read using a read lock. +// Read is a thread-safe read of the underlying io.ReadSeeker func (o *offsetReader) Read(p []byte) (int, error) { - o.lock.RLock() - defer o.lock.RUnlock() + o.lock.Lock() + defer o.lock.Unlock() if o.closed { return 0, io.EOF @@ -41,6 +42,14 @@ func (o *offsetReader) Read(p []byte) (int, error) { return o.buf.Read(p) } +// Seek is a thread-safe seeking operation. +func (o *offsetReader) Seek(offset int64, whence int) (int64, error) { + o.lock.Lock() + defer o.lock.Unlock() + + return o.buf.Seek(offset, whence) +} + // CloseAndCopy will return a new offsetReader with a copy of the old buffer // and close the old buffer. func (o *offsetReader) CloseAndCopy(offset int64) *offsetReader { diff --git a/aws/request/offset_reader_test.go b/aws/request/offset_reader_test.go index 8472258c23d..01856e316d3 100644 --- a/aws/request/offset_reader_test.go +++ b/aws/request/offset_reader_test.go @@ -24,6 +24,23 @@ func TestOffsetReaderRead(t *testing.T) { assert.Equal(t, buf, tempBuf) } +func TestOffsetReaderSeek(t *testing.T) { + buf := []byte("testData") + reader := newOffsetReader(bytes.NewReader(buf), 0) + + orig, err := reader.Seek(0, 1) + assert.NoError(t, err) + assert.Equal(t, int64(0), orig) + + n, err := reader.Seek(0, 2) + assert.NoError(t, err) + assert.Equal(t, int64(len(buf)), n) + + n, err = reader.Seek(orig, 0) + assert.NoError(t, err) + assert.Equal(t, int64(0), n) +} + func TestOffsetReaderClose(t *testing.T) { buf := []byte("testData") reader := &offsetReader{buf: bytes.NewReader(buf)} diff --git a/aws/request/request.go b/aws/request/request.go index 2832aaa4368..8ef9715c6b2 100644 --- a/aws/request/request.go +++ b/aws/request/request.go @@ -4,7 +4,6 @@ import ( "bytes" "fmt" "io" - "io/ioutil" "net/http" "net/url" "reflect" @@ -42,6 +41,12 @@ type Request struct { LastSignedAt time.Time built bool + + // Need to persist an intermideant body betweend the input Body and HTTP + // request body because the HTTP Client's transport can maintain a reference + // to the HTTP request's body after the client has returned. This value is + // safe to use concurrently and rewraps the input Body for each HTTP request. + safeBody *offsetReader } // An Operation is the service API operation to be made. @@ -135,8 +140,8 @@ func (r *Request) SetStringBody(s string) { // SetReaderBody will set the request's body reader. func (r *Request) SetReaderBody(reader io.ReadSeeker) { - r.HTTPRequest.Body = newOffsetReader(reader, 0) r.Body = reader + r.ResetBody() } // Presign returns the request's signed URL. Error will be returned @@ -220,6 +225,24 @@ func (r *Request) Sign() error { return r.Error } +// ResetBody rewinds the request body backto its starting position, and +// set's the HTTP Request body reference. When the body is read prior +// to being sent in the HTTP request it will need to be rewound. +func (r *Request) ResetBody() { + if r.safeBody != nil { + r.safeBody.Close() + } + + r.safeBody = newOffsetReader(r.Body, r.BodyStart) + r.HTTPRequest.Body = r.safeBody +} + +// GetBody will return an io.ReadSeeker of the Request's underlying +// input body with a concurrency safe wrapper. +func (r *Request) GetBody() io.ReadSeeker { + return r.safeBody +} + // Send will send the request returning error if errors are encountered. // // Send will sign the request prior to sending. All Send Handlers will @@ -231,6 +254,8 @@ func (r *Request) Sign() error { // // readLoop() and getConn(req *Request, cm connectMethod) // https://github.com/golang/go/blob/master/src/net/http/transport.go +// +// Send will not close the request.Request's body. func (r *Request) Send() error { for { if aws.BoolValue(r.Retryable) { @@ -239,21 +264,15 @@ func (r *Request) Send() error { r.ClientInfo.ServiceName, r.Operation.Name, r.RetryCount)) } - var body io.ReadCloser - if reader, ok := r.HTTPRequest.Body.(*offsetReader); ok { - body = reader.CloseAndCopy(r.BodyStart) - } else { - if r.Config.Logger != nil { - r.Config.Logger.Log("Request body type has been overwritten. May cause race conditions") - } - r.Body.Seek(r.BodyStart, 0) - body = ioutil.NopCloser(r.Body) - } + // The previous http.Request will have a reference to the r.Body + // and the HTTP Client's Transport may still be reading from + // the request's body even though the Client's Do returned. + r.HTTPRequest = copyHTTPRequest(r.HTTPRequest, nil) + r.ResetBody() - r.HTTPRequest = copyHTTPRequest(r.HTTPRequest, body) + // Closing response body to ensure that no response body is leaked + // between retry attempts. if r.HTTPResponse != nil && r.HTTPResponse.Body != nil { - // Closing response body. Since we are setting a new request to send off, this - // response will get squashed and leaked. r.HTTPResponse.Body.Close() } } @@ -281,7 +300,6 @@ func (r *Request) Send() error { debugLogReqError(r, "Send Request", true, err) continue } - r.Handlers.UnmarshalMeta.Run(r) r.Handlers.ValidateResponse.Run(r) if r.Error != nil { diff --git a/aws/signer/v4/v4.go b/aws/signer/v4/v4.go index dbc8ea88aa4..77ef84abe62 100644 --- a/aws/signer/v4/v4.go +++ b/aws/signer/v4/v4.go @@ -366,7 +366,9 @@ func signSDKRequestWithCurrTime(req *request.Request, curTimeFn func() time.Time signingTime = req.LastSignedAt } - signedHeaders, err := v4.signWithBody(req.HTTPRequest, req.Body, name, region, req.ExpireTime, signingTime) + signedHeaders, err := v4.signWithBody(req.HTTPRequest, req.GetBody(), + name, region, req.ExpireTime, signingTime, + ) if err != nil { req.Error = err req.SignedHeaderVals = nil