aws/defaults: Add recording clock skew metadata per request attempt (#500)

Updates the SDK's default request handlers to record metadata about the
clock skew from a request attempt. This metadata will be used by the
SDK's request retry metadata header.

This update does NOT add clock skew correction to the SDK's request
signatures.

Author: jasdel

aws/defaults/defaults.go

@@ -76,6 +76,7 @@ func Handlers() aws.Handlers {
 	handlers.Send.PushFrontNamed(RetryMetricHeaderHandler)
 	handlers.Send.PushBackNamed(ValidateReqSigHandler)
 	handlers.Send.PushBackNamed(SendHandler)
+	handlers.Send.PushBackNamed(AttemptClockSkewHandler)
 	handlers.ShouldRetry.PushBackNamed(RetryableCheckHandler)
 	handlers.ValidateResponse.PushBackNamed(ValidateResponseHandler)
 

aws/defaults/handlers.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/aws/awserr"
+	"github.com/aws/aws-sdk-go-v2/internal/sdk"
 )
 
 // Interface for matching types which also have a Len method.
@@ -122,6 +123,7 @@ var SendHandler = aws.NamedHandler{
 
 		var err error
 		r.HTTPResponse, err = r.Config.HTTPClient.Do(r.HTTPRequest)
+		r.ResponseAt = sdk.NowTime()
 		if err != nil {
 			handleSendError(r, err)
 		}
@@ -204,12 +206,32 @@ var RetryMetricHeaderHandler = aws.NamedHandler{
 		if max := r.Retryer.MaxAttempts(); max != 0 {
 			parts = append(parts, fmt.Sprintf("max=%d", max))
 		}
-		// TODO "ttl=YYYYmmddTHHMMSSZ"
-		//    ttl = current_time + socket_read_timeout + estimated_skew
-		// SDK doesn't implement clock skew yet and not obvious how to obtain
-		// read deadlines.
 
-		r.HTTPRequest.Header.Set("amz-sdk-request", strings.Join(parts, "; "))
+		type timeoutGetter interface {
+			GetTimeout() time.Duration
+		}
+
+		var ttl time.Time
+		// Attempt extract the TTL from context deadline, or timeout on the client.
+		if v, ok := r.Config.HTTPClient.(timeoutGetter); ok {
+			if t := v.GetTimeout(); t > 0 {
+				ttl = sdk.NowTime().Add(t)
+			}
+		}
+		if ttl.IsZero() {
+			if deadline, ok := r.Context().Deadline(); ok {
+				ttl = deadline
+			}
+		}
+
+		// Only append the TTL if it can be determined.
+		if !ttl.IsZero() && len(r.AttemptClockSkews) > 0 {
+			const unixTimeFormat = "20060102T150405Z"
+			ttl = ttl.Add(r.AttemptClockSkews[len(r.AttemptClockSkews)-1])
+			parts = append(parts, fmt.Sprintf("ttl=%s", ttl.Format(unixTimeFormat)))
+		}
+
+		r.HTTPRequest.Header.Set(retryMetricHeader, strings.Join(parts, "; "))
 	}}
 
 // RetryableCheckHandler performs final checks to determine if the request should
@@ -252,3 +274,34 @@ var ValidateEndpointHandler = aws.NamedHandler{Name: "core.ValidateEndpointHandl
 		r.Error = aws.ErrMissingEndpoint
 	}
 }}
+
+// AttemptClockSkewHandler records the estimated clock skew between the client
+// and service response clocks. This estimation will be no more granular than
+// one second. It will not be populated until after at least the first
+// attempt's response is received.
+var AttemptClockSkewHandler = aws.NamedHandler{
+	Name: "core.AttemptClockSkewHandler",
+	Fn: func(r *aws.Request) {
+		if r.ResponseAt.IsZero() || r.HTTPResponse == nil || r.HTTPResponse.StatusCode == 0 {
+			return
+		}
+
+		respDateHeader := r.HTTPResponse.Header.Get("Date")
+		if len(respDateHeader) == 0 {
+			return
+		}
+
+		respDate, err := http.ParseTime(respDateHeader)
+		if err != nil {
+			if r.Config.Logger != nil {
+				r.Config.Logger.Log(fmt.Sprintf("ERROR: unable to determine clock skew for %s/%s API response, invalid Date header value, %v",
+					r.Metadata.ServiceName, r.Operation.Name, respDateHeader))
+			}
+			return
+		}
+
+		r.AttemptClockSkews = append(r.AttemptClockSkews,
+			respDate.Sub(r.ResponseAt),
+		)
+	},
+}

aws/defaults/handlers_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/aws/retry"
 	"github.com/aws/aws-sdk-go-v2/internal/awstesting"
 	"github.com/aws/aws-sdk-go-v2/internal/awstesting/unit"
+	"github.com/aws/aws-sdk-go-v2/internal/sdk"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 )
 
@@ -349,10 +350,21 @@ func TestRequestInvocationIDHeaderHandler(t *testing.T) {
 }
 
 func TestRetryMetricHeaderHandler(t *testing.T) {
+	nowTime := sdk.NowTime
+	defer func() {
+		sdk.NowTime = nowTime
+	}()
+	sdk.NowTime = func() time.Time {
+		return time.Date(2020, 2, 2, 0, 0, 0, 0, time.UTC)
+	}
+
 	cases := map[string]struct {
-		Attempt     int
-		MaxAttempts int
-		Expect      string
+		Attempt           int
+		MaxAttempts       int
+		Client            aws.HTTPClient
+		ContextDeadline   time.Time
+		AttemptClockSkews []time.Duration
+		Expect            string
 	}{
 		"first attempt": {
 			Attempt: 1, MaxAttempts: 3,
@@ -366,15 +378,43 @@ func TestRetryMetricHeaderHandler(t *testing.T) {
 			Attempt: 10,
 			Expect:  "attempt=10",
 		},
+		"with ttl client timeout": {
+			Attempt: 2, MaxAttempts: 3,
+			AttemptClockSkews: []time.Duration{
+				10 * time.Second,
+			},
+			Client: func() aws.HTTPClient {
+				c := &aws.BuildableHTTPClient{}
+				return c.WithTimeout(10 * time.Second)
+			}(),
+			Expect: "attempt=2; max=3; ttl=20200202T000020Z",
+		},
+		"with ttl context deadline": {
+			Attempt: 1, MaxAttempts: 3,
+			AttemptClockSkews: []time.Duration{
+				10 * time.Second,
+			},
+			ContextDeadline: sdk.NowTime().Add(10 * time.Second),
+			Expect:          "attempt=1; max=3; ttl=20200202T000020Z",
+		},
 	}
 
 	for name, c := range cases {
 		t.Run(name, func(t *testing.T) {
 			cfg := unit.Config()
+			if c.Client != nil {
+				cfg.HTTPClient = c.Client
+			}
 			r := aws.New(cfg, aws.Metadata{}, cfg.Handlers, aws.NoOpRetryer{},
 				&aws.Operation{}, &struct{}{}, struct{}{})
+			if !c.ContextDeadline.IsZero() {
+				ctx, cancel := context.WithDeadline(r.Context(), c.ContextDeadline)
+				defer cancel()
+				r.SetContext(ctx)
+			}
 
 			r.AttemptNum = c.Attempt
+			r.AttemptClockSkews = c.AttemptClockSkews
 			r.Retryer = retry.AddWithMaxAttempts(r.Retryer, c.MaxAttempts)
 
 			defaults.RetryMetricHeaderHandler.Fn(r)
@@ -388,3 +428,92 @@ func TestRetryMetricHeaderHandler(t *testing.T) {
 		})
 	}
 }
+
+func TestAttemptClockSkewHandler(t *testing.T) {
+	cases := map[string]struct {
+		Req    *aws.Request
+		Expect []time.Duration
+	}{
+		"no response": {Req: &aws.Request{}},
+		"failed response": {
+			Req: &aws.Request{
+				HTTPResponse: &http.Response{StatusCode: 0, Header: http.Header{}},
+			},
+		},
+		"no date header response": {
+			Req: &aws.Request{
+				HTTPResponse: &http.Response{StatusCode: 200, Header: http.Header{}},
+			},
+		},
+		"invalid date header response": {
+			Req: &aws.Request{
+				HTTPResponse: &http.Response{
+					StatusCode: 200,
+					Header:     http.Header{"Date": []string{"abc123"}},
+				},
+			},
+		},
+		"response at unset": {
+			Req: &aws.Request{
+				HTTPResponse: &http.Response{
+					StatusCode: 200,
+					Header: http.Header{
+						"Date": []string{"Thu, 05 Mar 2020 22:25:15 GMT"},
+					},
+				},
+			},
+		},
+		"first date response": {
+			Req: &aws.Request{
+				HTTPResponse: &http.Response{
+					StatusCode: 200,
+					Header: http.Header{
+						"Date": []string{"Thu, 05 Mar 2020 22:25:15 GMT"},
+					},
+				},
+				ResponseAt: time.Date(2020, 3, 5, 22, 25, 17, 0, time.UTC),
+			},
+			Expect: []time.Duration{-2 * time.Second},
+		},
+		"subsequent date response": {
+			Req: &aws.Request{
+				HTTPResponse: &http.Response{
+					StatusCode: 200,
+					Header: http.Header{
+						"Date": []string{"Thu, 05 Mar 2020 22:25:15 GMT"},
+					},
+				},
+				ResponseAt: time.Date(2020, 3, 5, 22, 25, 14, 0, time.UTC),
+				AttemptClockSkews: []time.Duration{
+					-2 * time.Second,
+				},
+			},
+			Expect: []time.Duration{
+				-2 * time.Second,
+				1 * time.Second,
+			},
+		},
+	}
+
+	for name, c := range cases {
+		t.Run(name, func(t *testing.T) {
+			r := new(aws.Request)
+			*r = *c.Req
+
+			defaults.AttemptClockSkewHandler.Fn(r)
+			if err := r.Error; err != nil {
+				t.Fatalf("expect no error, got %v", err)
+			}
+
+			if e, a := len(c.Expect), len(r.AttemptClockSkews); e != a {
+				t.Errorf("expect %v skews, got %v", e, a)
+			}
+
+			for i, s := range r.AttemptClockSkews {
+				if e, a := c.Expect[i], s; e != a {
+					t.Errorf("%d, expect %v skew, got %v", i, e, a)
+				}
+			}
+		})
+	}
+}

aws/http_client.go

@@ -44,7 +44,9 @@ type BuildableHTTPClient struct {
 	dialer    *net.Dialer
 
 	initOnce *sync.Once
-	client   *http.Client
+
+	clientTimeout time.Duration
+	client        *http.Client
 }
 
 // NewBuildableHTTPClient returns an initialized client for invoking HTTP
@@ -91,11 +93,12 @@ func (b BuildableHTTPClient) build() *http.Client {
 	http2.ConfigureTransport(tr)
 
 	return wrapWithoutRedirect(&http.Client{
+		Timeout:   b.clientTimeout,
 		Transport: tr,
 	})
 }
 
-func (b BuildableHTTPClient) reset() BuildableHTTPClient {
+func (b BuildableHTTPClient) initReset() BuildableHTTPClient {
 	b.initOnce = new(sync.Once)
 	b.client = nil
 	return b
@@ -108,7 +111,7 @@ func (b BuildableHTTPClient) reset() BuildableHTTPClient {
 // will be replaced with a default Transport value before invoking the option
 // functions.
 func (b BuildableHTTPClient) WithTransportOptions(opts ...func(*http.Transport)) HTTPClient {
-	b = b.reset()
+	b = b.initReset()
 
 	tr := b.GetTransport()
 	for _, opt := range opts {
@@ -123,7 +126,7 @@ func (b BuildableHTTPClient) WithTransportOptions(opts ...func(*http.Transport))
 // net.Dialer options applied. Will set the client's http.Transport DialContext
 // member.
 func (b BuildableHTTPClient) WithDialerOptions(opts ...func(*net.Dialer)) HTTPClient {
-	b = b.reset()
+	b = b.initReset()
 
 	dialer := b.GetDialer()
 	for _, opt := range opts {
@@ -138,6 +141,15 @@ func (b BuildableHTTPClient) WithDialerOptions(opts ...func(*net.Dialer)) HTTPCl
 	return &b
 }
 
+// WithTimeout Sets the timeout used by the client for all requests.
+func (b BuildableHTTPClient) WithTimeout(timeout time.Duration) HTTPClient {
+	b = b.initReset()
+
+	b.clientTimeout = timeout
+
+	return &b
+}
+
 // GetTransport returns a copy of the client's HTTP Transport.
 func (b BuildableHTTPClient) GetTransport() *http.Transport {
 	var tr *http.Transport
@@ -162,6 +174,11 @@ func (b BuildableHTTPClient) GetDialer() *net.Dialer {
 	return dialer
 }
 
+// GetTimeout returns a copy of the client's timeout to cancel requests with.
+func (b BuildableHTTPClient) GetTimeout() time.Duration {
+	return b.clientTimeout
+}
+
 func defaultDialer() *net.Dialer {
 	return &net.Dialer{
 		Timeout:   DefaultDialConnectTimeout,

aws/http_client_test.go

@@ -4,6 +4,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 )
@@ -27,3 +28,18 @@ func TestBuildableHTTPClient_NoFollowRedirect(t *testing.T) {
 		t.Errorf("expect %v code, got %v", e, a)
 	}
 }
+
+func TestBuildableHTTPClient_WithTimeout(t *testing.T) {
+	client := &aws.BuildableHTTPClient{}
+
+	expect := 10 * time.Millisecond
+	client2 := client.WithTimeout(expect).(*aws.BuildableHTTPClient)
+
+	if e, a := time.Duration(0), client.GetTimeout(); e != a {
+		t.Errorf("expect %v initial timeout, got %v", e, a)
+	}
+
+	if e, a := expect, client2.GetTimeout(); e != a {
+		t.Errorf("expect %v timeout, got %v", e, a)
+	}
+}

aws/request.go

@@ -118,6 +118,15 @@ type Request struct {
 	SignedHeaderVals http.Header
 	LastSignedAt     time.Time
 
+	// The time the response headers were received from the API call.
+	ResponseAt time.Time
+
+	// AttemptClockSkews are the estimated clock skew between the service
+	// response Date header, and when the SDK received the response per
+	// attempt. This estimate will include the latency of receiving the
+	// service's response headers.
+	AttemptClockSkews []time.Duration
+
 	// ID for this operation's request that is shared across attempts.
 	InvocationID string