aws: Adds sdk error checking when seeking readers (#379)

Adds missing sdk error checking when seeking readers. Also adds support for nonseekable io.Reader and support for streamed payloads for unsigned body request.

Fixes #371

Author: skotambkar

CHANGELOG_PENDING.md

@@ -1,18 +1,20 @@
 ### SDK Features
 
 ### SDK Enhancements
-* `aws/endpoints`: Expose DNSSuffix for partitions ([#368](https://github.com/aws/aws-sdk-go/pull/368))
+* `aws/endpoints`: Expose DNSSuffix for partitions ([#369](https://github.com/aws/aws-sdk-go-v2/pull/369))
   * Exposes the underlying partition metadata's DNSSuffix value via the `DNSSuffix` method on the endpoint's `Partition` type. This allows access to the partition's DNS suffix, e.g. "amazon.com".
-  * Fixes [#347](https://github.com/aws/aws-sdk-go/issues/347)
+  * Fixes [#347](https://github.com/aws/aws-sdk-go-v2/issues/347)
 * `private/protocol`: Add support for parsing fractional timestamp ([#367](https://github.com/aws/aws-sdk-go-v2/pull/367))
   * Fixes the SDK's ability to parse fractional unix timestamp values and adds tests.
   * Fixes [#365](https://github.com/aws/aws-sdk-go-v2/issues/365)
-* `aws/ec2metadata`: Add marketplaceProductCodes to EC2 Instance Identity Document
+* `aws/ec2metadata`: Add marketplaceProductCodes to EC2 Instance Identity Document ([#374](https://github.com/aws/aws-sdk-go-v2/pull/374))
   * Adds `MarketplaceProductCodes` to the EC2 Instance Metadata's Identity Document. The ec2metadata client will now retrieve these values if they are available.
   * Related to: [aws/aws-sdk-go#2781](https://github.com/aws/aws-sdk-go/issues/2781)
 
 ### SDK Bugs
 * `aws`: Fixes bug in calculating throttled retry delay ([#373](https://github.com/aws/aws-sdk-go-v2/pull/373))
   * The `Retry-After` duration specified in the request is now added to the Retry delay for throttled exception. Adds test for retry delays for throttled exceptions. Fixes bug where the throttled retry's math was off.
   * Fixes [#45](https://github.com/aws/aws-sdk-go-v2/issues/45)
-
+* `aws` : Adds missing sdk error checking when seeking readers [#379](https://github.com/aws/aws-sdk-go-v2/pull/379).
+  * Adds support for nonseekable io.Reader. Adds support for streamed payloads for unsigned body request. 
+  * Fixes [#371](https://github.com/aws/aws-sdk-go-v2/issues/371)

aws/client_logger.go

@@ -43,17 +43,26 @@ func (reader *teeReaderCloser) Close() error {
 
 func logRequest(r *Request) {
 	logBody := r.Config.LogLevel.Matches(LogDebugWithHTTPBody)
+	bodySeekable := IsReaderSeekable(r.Body)
+
 	dumpedBody, err := httputil.DumpRequestOut(r.HTTPRequest, logBody)
 	if err != nil {
 		r.Config.Logger.Log(fmt.Sprintf(logReqErrMsg, r.Metadata.ServiceName, r.Operation.Name, err))
 		return
 	}
 
 	if logBody {
-		// 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.ResetBody()
+		if !bodySeekable {
+			r.SetReaderBody(ReadSeekCloser(r.HTTPRequest.Body))
+		}
+
+		// 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.
+		if err := r.Error; err != nil {
+			r.Config.Logger.Log(fmt.Sprintf(logReqErrMsg, r.Metadata.ServiceName, r.Operation.Name, err))
+			return
+		}
 	}
 
 	r.Config.Logger.Log(fmt.Sprintf(logReqMsg, r.Metadata.ServiceName, r.Operation.Name, string(dumpedBody)))

aws/client_logger_test.go

@@ -2,7 +2,11 @@ package aws
 
 import (
 	"bytes"
+	"fmt"
 	"io"
+	"io/ioutil"
+	"reflect"
+	"runtime"
 	"testing"
 )
 
@@ -55,3 +59,87 @@ func TestLogWriter(t *testing.T) {
 		t.Errorf("Expected %q, but received %q", expected, lw.buf.String())
 	}
 }
+
+func TestLogRequest(t *testing.T) {
+	cases := []struct {
+		Body       io.ReadSeeker
+		ExpectBody []byte
+		LogLevel   LogLevel
+	}{
+		{
+			Body:       ReadSeekCloser(bytes.NewBuffer([]byte("body content"))),
+			ExpectBody: []byte("body content"),
+		},
+		{
+			Body:       ReadSeekCloser(bytes.NewBuffer([]byte("body content"))),
+			LogLevel:   LogDebugWithHTTPBody,
+			ExpectBody: []byte("body content"),
+		},
+		{
+			Body:       bytes.NewReader([]byte("body content")),
+			ExpectBody: []byte("body content"),
+		},
+		{
+			Body:       bytes.NewReader([]byte("body content")),
+			LogLevel:   LogDebugWithHTTPBody,
+			ExpectBody: []byte("body content"),
+		},
+	}
+
+	for i, c := range cases {
+		var logW bytes.Buffer
+		req := New(
+			Config{
+				EndpointResolver: ResolveWithEndpointURL("https://endpoint"),
+				Credentials:      AnonymousCredentials,
+				Logger:           &bufLogger{w: &logW},
+				LogLevel:         c.LogLevel,
+				Region:           "mock-region",
+			},
+			Metadata{
+				EndpointsID: "https://mock-service.mock-region.amazonaws.com",
+			},
+			testHandlers(),
+			nil,
+			&Operation{
+				Name:       "APIName",
+				HTTPMethod: "POST",
+				HTTPPath:   "/",
+			},
+			struct{}{}, nil,
+		)
+		req.SetReaderBody(c.Body)
+		req.Build()
+
+		logRequest(req)
+
+		b, err := ioutil.ReadAll(req.HTTPRequest.Body)
+		if err != nil {
+			t.Fatalf("%d, expect to read SDK request Body", i)
+		}
+
+		if e, a := c.ExpectBody, b; !reflect.DeepEqual(e, a) {
+			t.Errorf("%d, expect %v body, got %v", i, e, a)
+		}
+	}
+}
+
+type bufLogger struct {
+	w *bytes.Buffer
+}
+
+func (l *bufLogger) Log(args ...interface{}) {
+	fmt.Fprintln(l.w, args...)
+}
+
+func testHandlers() Handlers {
+	var handlers Handlers
+	handler := NamedHandler{
+		Name: "core.SDKVersionUserAgentHandler",
+		Fn: MakeAddToUserAgentHandler(SDKName, SDKVersion,
+			runtime.Version(), runtime.GOOS, runtime.GOARCH),
+	}
+	handlers.Build.PushBackNamed(handler)
+
+	return handlers
+}

aws/defaults/handlers.go

@@ -40,9 +40,19 @@ var BuildContentLengthHandler = aws.NamedHandler{Name: "core.BuildContentLengthH
 		case lener:
 			length = int64(body.Len())
 		case io.Seeker:
-			r.BodyStart, _ = body.Seek(0, 1)
-			end, _ := body.Seek(0, 2)
-			body.Seek(r.BodyStart, 0) // make sure to seek back to original location
+			var err error
+			r.BodyStart, err = body.Seek(0, io.SeekCurrent)
+			if err != nil {
+				r.Error = awserr.New(aws.ErrCodeSerialization, "failed to determine start of the request body", err)
+			}
+			end, err := body.Seek(0, io.SeekEnd)
+			if err != nil {
+				r.Error = awserr.New(aws.ErrCodeSerialization, "failed to determine end of the request body", err)
+			}
+			_, err = body.Seek(r.BodyStart, io.SeekStart) // make sure to seek back to original location
+			if err != nil {
+				r.Error = awserr.New(aws.ErrCodeSerialization, "failed to seek back to the original location", err)
+			}
 			length = end - r.BodyStart
 		default:
 			panic("Cannot get length of body, must provide `ContentLength`")

aws/offset_reader.go

@@ -13,12 +13,14 @@ type offsetReader struct {
 	closed bool
 }
 
-func newOffsetReader(buf io.ReadSeeker, offset int64) *offsetReader {
+func newOffsetReader(buf io.ReadSeeker, offset int64) (*offsetReader, error) {
 	reader := &offsetReader{}
-	buf.Seek(offset, 0)
-
+	_, err := buf.Seek(offset, io.SeekStart)
+	if err != nil {
+		return nil, err
+	}
 	reader.buf = buf
-	return reader
+	return reader, nil
 }
 
 // Close will close the instance of the offset reader's access to
@@ -52,7 +54,9 @@ func (o *offsetReader) Seek(offset int64, whence int) (int64, error) {
 
 // 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 {
-	o.Close()
+func (o *offsetReader) CloseAndCopy(offset int64) (*offsetReader, error) {
+	if err := o.Close(); err != nil {
+		return nil, err
+	}
 	return newOffsetReader(o.buf, offset)
 }

aws/offset_reader_test.go

@@ -21,7 +21,7 @@ func TestOffsetReaderRead(t *testing.T) {
 		t.Errorf("expect %v, got %v", e, a)
 	}
 	if err != nil {
-		t.Errorf("expect no error, got %v", err)
+		t.Fatalf("expect no error, got %v", err)
 	}
 	if e, a := buf, tempBuf; !bytes.Equal(e, a) {
 		t.Errorf("expect %v, got %v", e, a)
@@ -30,27 +30,30 @@ func TestOffsetReaderRead(t *testing.T) {
 
 func TestOffsetReaderSeek(t *testing.T) {
 	buf := []byte("testData")
-	reader := newOffsetReader(bytes.NewReader(buf), 0)
+	reader, err := newOffsetReader(bytes.NewReader(buf), 0)
+	if err != nil {
+		t.Fatalf("expect no error, got %v", err)
+	}
 
-	orig, err := reader.Seek(0, 1)
+	orig, err := reader.Seek(0, io.SeekCurrent)
 	if err != nil {
-		t.Errorf("expect no error, got %v", err)
+		t.Fatalf("expect no error, got %v", err)
 	}
 	if e, a := int64(0), orig; e != a {
 		t.Errorf("expect %v, got %v", e, a)
 	}
 
-	n, err := reader.Seek(0, 2)
+	n, err := reader.Seek(0, io.SeekEnd)
 	if err != nil {
-		t.Errorf("expect no error, got %v", err)
+		t.Fatalf("expect no error, got %v", err)
 	}
 	if e, a := int64(len(buf)), n; e != a {
 		t.Errorf("expect %v, got %v", e, a)
 	}
 
-	n, err = reader.Seek(orig, 0)
+	n, err = reader.Seek(orig, io.SeekStart)
 	if err != nil {
-		t.Errorf("expect no error, got %v", err)
+		t.Fatalf("expect no error, got %v", err)
 	}
 	if e, a := int64(0), n; e != a {
 		t.Errorf("expect %v, got %v", e, a)
@@ -81,8 +84,10 @@ func TestOffsetReaderCloseAndCopy(t *testing.T) {
 	tempBuf := make([]byte, len(buf))
 	reader := &offsetReader{buf: bytes.NewReader(buf)}
 
-	newReader := reader.CloseAndCopy(0)
-
+	newReader, err := reader.CloseAndCopy(0)
+	if err != nil {
+		t.Fatalf("expect no error, got %v", err)
+	}
 	n, err := reader.Read(tempBuf)
 	if e, a := 0, n; e != a {
 		t.Errorf("expect %v, got %v", e, a)
@@ -96,7 +101,7 @@ func TestOffsetReaderCloseAndCopy(t *testing.T) {
 		t.Errorf("expect %v, got %v", e, a)
 	}
 	if err != nil {
-		t.Errorf("expect no error, got %v", err)
+		t.Fatalf("expect no error, got %v", err)
 	}
 	if e, a := buf, tempBuf; !bytes.Equal(e, a) {
 		t.Errorf("expect %v, got %v", e, a)
@@ -108,13 +113,16 @@ func TestOffsetReaderCloseAndCopyOffset(t *testing.T) {
 	tempBuf := make([]byte, len(buf))
 	reader := &offsetReader{buf: bytes.NewReader(buf)}
 
-	newReader := reader.CloseAndCopy(4)
+	newReader, err := reader.CloseAndCopy(4)
+	if err != nil {
+		t.Fatalf("expect no error, got %v", err)
+	}
 	n, err := newReader.Read(tempBuf)
 	if e, a := n, len(buf)-4; e != a {
 		t.Errorf("expect %v, got %v", e, a)
 	}
 	if err != nil {
-		t.Errorf("expect no error, got %v", err)
+		t.Fatalf("expect no error, got %v", err)
 	}
 
 	expected := []byte{'D', 'a', 't', 'a', 0, 0, 0, 0}