Merge pull request #180 from beanliu/master

fix build

.travis.yml

@@ -18,7 +18,7 @@ script:
   - export GO111MODULE=on
   - go mod download
   - golangci-lint run
-  - go test -v -covermode atomic -race -timeout 60s -cpu 4 -parallel 8 -coverprofile ./coverage.out $(go list ./...)
+  - go test -v -covermode atomic -race -timeout 70s -cpu 4 -parallel 8 -coverprofile ./coverage.out $(go list ./...)
   - "! go tool cover -func coverage.out | grep -v 100.0%"
 
 after_script:

sfxclient/httpsink.go

@@ -102,15 +102,9 @@ type TooManyRequestError struct {
 	Err error
 }
 
-func (e *TooManyRequestError) Error() string {
-	if e.ThrottleType == "" {
-		return fmt.Sprintf("too many requests, retry after %.3f seconds", e.RetryAfter.Seconds())
-	}
-	return fmt.Sprintf("too many %s requests, retry after %.3f seconds", e.ThrottleType, e.RetryAfter.Seconds())
-}
-
-func (e *TooManyRequestError) Unwrap() error {
-	return e.Err
+func (e TooManyRequestError) Error() string {
+	return fmt.Sprintf("[%s] too many requests, retry after %.3f seconds",
+		e.ThrottleType, e.RetryAfter.Seconds())
 }
 
 type responseValidator func(respBody []byte) error
@@ -491,22 +485,18 @@ func sapmMarshal(v []*trace.Span) ([]byte, error) {
 }
 
 func parseRetryAfterHeader(v string) (time.Duration, error) {
-	if v == "" {
-		return 0, errors.New("empty header value")
-	}
-
 	// Retry-After: <http-date>
 	// https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Date
 	if t, err := http.ParseTime(v); err == nil {
-		return t.Sub(time.Now()), nil
+		return time.Until(t), nil
 	}
 
 	// Retry-After: <delay-seconds>
-	if i, err := strconv.Atoi(v); err != nil {
+	i, err := strconv.Atoi(v)
+	if err != nil {
 		return 0, err
-	} else {
-		return time.Duration(i) * time.Second, nil
 	}
+	return time.Duration(i) * time.Second, nil
 }
 
 // NewHTTPSink creates a default NewHTTPSink using package level constants as

sfxclient/httpsink_test.go

@@ -165,6 +165,7 @@ func TestHTTPDatapointSink(t *testing.T) {
 		Convey("with a test endpoint", func() {
 			retString := `"OK"`
 			retCode := http.StatusOK
+			retHeaders := map[string]string{}
 			var blockResponse chan struct{}
 			var cancelCallback func()
 			seenBodyPoints := &com_signalfx_metrics_protobuf.DataPointUploadMessage{}
@@ -174,6 +175,9 @@ func TestHTTPDatapointSink(t *testing.T) {
 				log.IfErr(log.Panic, err)
 				log.IfErr(log.Panic, req.Body.Close())
 				log.IfErr(log.Panic, proto.Unmarshal(bodyBytes.Bytes(), seenBodyPoints))
+				for k, v := range retHeaders {
+					rw.Header().Add(k, v)
+				}
 				rw.WriteHeader(retCode)
 				errors.PanicIfErrWrite(io.WriteString(rw, retString))
 				if blockResponse != nil {
@@ -250,6 +254,30 @@ func TestHTTPDatapointSink(t *testing.T) {
 				retCode = http.StatusNotAcceptable
 				So(errors.Details(s.AddDatapoints(ctx, dps)), ShouldContainSubstring, "invalid status code")
 			})
+			Convey("HTTP 429 should be checked", func() {
+				retCode = http.StatusTooManyRequests
+				retHeaders = map[string]string{
+					"Throttle-Type": "x",
+				}
+				Convey("retry after seconds", func() {
+					retHeaders["Retry-After"] = "1"
+					err, ok := s.AddDatapoints(ctx, dps).(*TooManyRequestError)
+					So(ok, ShouldBeTrue)
+					So(err.RetryAfter, ShouldEqual, time.Second)
+					So(err.Error(), ShouldContainSubstring, "[x] too many requests, retry after")
+				})
+				Convey("retry until date", func() {
+					retHeaders["Retry-After"] = time.Now().Add(time.Second).Format(time.RFC850)
+					err, ok := s.AddDatapoints(ctx, dps).(*TooManyRequestError)
+					So(ok, ShouldBeTrue)
+					So(err.RetryAfter, ShouldBeLessThanOrEqualTo, time.Second)
+					So(err.Error(), ShouldContainSubstring, "[x] too many requests, retry after")
+				})
+				Convey("error fallback", func() {
+					retHeaders["Retry-After"] = ""
+					So(errors.Details(s.AddDatapoints(ctx, dps)), ShouldContainSubstring, "invalid status code")
+				})
+			})
 			Convey("return string should be checked", func() {
 				retString = `"nope"`
 				So(errors.Details(s.AddDatapoints(ctx, dps)), ShouldContainSubstring, "invalid response body")

sfxclient/multitokensink.go

@@ -1,7 +1,6 @@
 package sfxclient
 
 import (
-	"errors"
 	"fmt"
 	"hash"
 	"hash/fnv"
@@ -55,9 +54,12 @@ func getHTTPStatusCode(status *tokenStatus, err error) *tokenStatus {
 	if err == nil {
 		status.status = http.StatusOK
 	} else {
-		var apiErr *SFXAPIError
-		if errors.As(err, &apiErr) {
-			status.status = apiErr.StatusCode
+		// refactor: use `errors.As` to simplify type assertion on error chain once we get rid of go 1.12 build.
+		if obj, ok := err.(*TooManyRequestError); ok {
+			err = obj.Err
+		}
+		if obj, ok := err.(*SFXAPIError); ok {
+			status.status = obj.StatusCode
 		}
 	}
 	return status

sfxclient/multitokensink_test.go

@@ -104,6 +104,28 @@ func AddSpansGetSuccess(ctx context.Context, evs []*trace.Span) (err error) {
 	return
 }
 
+func TestGetHTTPStatusCode(t *testing.T) {
+	ts := &tokenStatus{}
+	Convey("should handle different type of errors", t, func() {
+		So(getHTTPStatusCode(ts, nil).status, ShouldEqual, http.StatusOK)
+
+		So(getHTTPStatusCode(ts, &SFXAPIError{
+			StatusCode:   http.StatusBadRequest,
+			ResponseBody: "",
+		}).status, ShouldEqual, http.StatusBadRequest)
+
+		err := &TooManyRequestError{
+			ThrottleType: "",
+			RetryAfter:   0,
+			Err: &SFXAPIError{
+				StatusCode:   http.StatusTooManyRequests,
+				ResponseBody: "",
+			},
+		}
+		So(getHTTPStatusCode(ts, err).status, ShouldEqual, http.StatusTooManyRequests)
+	})
+}
+
 func TestWorkerErrorHandlerDps(t *testing.T) {
 	Convey("An AsyncMultiTokeSink Worker Dps", t, func() {
 		Convey("should handle errors while emitting datapoints", func() {