Add error handling for errors in Datadog API response

event/list.go

@@ -53,3 +53,9 @@ const (
 	STATE_CHANGE_END         = "state-change-end"
 	REPL_SOURCE_CHANGE       = "repl-soruce-change"
 )
+
+// Sink Events
+
+const (
+	SINK_ERROR = "sink-error"
+)

sink/datadog.go

@@ -17,6 +17,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cashapp/blip/event"
+
 	"github.com/DataDog/datadog-go/v5/statsd"
 
 	"github.com/DataDog/datadog-api-client-go/v2/api/datadog"
@@ -373,16 +375,17 @@ func (s *Datadog) sendApi(ddCtx context.Context, dp []datadogV2.MetricSeries) er
 			optParams.ContentEncoding = datadogV2.METRICCONTENTENCODING_GZIP.Ptr()
 		}
 
-		if _, r, err := s.metricsApi.SubmitMetrics(ddCtx, *datadogV2.NewMetricPayload(dp[rangeStart:rangeEnd]), optParams); err != nil {
+		apiResponse, r, err := s.metricsApi.SubmitMetrics(ddCtx, *datadogV2.NewMetricPayload(dp[rangeStart:rangeEnd]), optParams)
+		if err != nil {
 			if r != nil && r.StatusCode == http.StatusRequestEntityTooLarge {
 				// Is the number of metrics sent already the smallest possible?
 				if localMaxMetricsPerRequest == 1 {
-					return fmt.Errorf("Unable to send metrics: %v", err)
+					return fmt.Errorf("unable to send metrics: %v", err)
 				}
 
 				// The payload was too large, so we need to recalculate it and try with a smaller size
 				if localMaxMetricsPerRequest, err = s.estimateMaxMetricsPerRequest(dp[rangeStart:rangeEnd], localMaxMetricsPerRequest); err != nil {
-					return fmt.Errorf("Unable to determine proper number of metrics per request: %v", err)
+					return fmt.Errorf("unable to determine proper number of metrics per request: %v", err)
 				}
 
 				// Retry the metrics with the new payload size
@@ -393,6 +396,16 @@ func (s *Datadog) sendApi(ddCtx context.Context, dp []datadogV2.MetricSeries) er
 			return err
 		}
 
+		if len(apiResponse.Errors) > 0 {
+			// datadog can return a 202 Accepted response code but errors in response payload
+			// this can be partial success, log it, raise an event and continue
+			errMsg := fmt.Sprintf("%s: error(s) returned from datadog: %s", s.monitorId, strings.Join(apiResponse.Errors, ","))
+			blip.Debug(errMsg)
+
+			// Send an event as well
+			event.Errorf(event.SINK_ERROR, errMsg) // log by default
+		}
+
 		rangeStart = rangeEnd
 	}
 

sink/datadog_test.go

@@ -1,6 +1,7 @@
 package sink
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -10,6 +11,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+
 	"github.com/DataDog/datadog-api-client-go/v2/api/datadogV2"
 	"github.com/cashapp/blip"
 	"github.com/cashapp/blip/test/mock"
@@ -284,3 +287,31 @@ func TestDatadogMetricsPerRequestMultipleFail(t *testing.T) {
 		t.Fatal(diff)
 	}
 }
+
+func TestDatadogMetricsErrorResponseFromAPI(t *testing.T) {
+	errors := []string{"validation error 1", "validation error 2"}
+	resp := map[string][]string{
+		"errors": errors,
+	}
+	respJSON, err := json.Marshal(resp)
+	require.NoError(t, err)
+
+	httpClient := &http.Client{
+		Transport: &mock.Transport{
+			RoundTripFunc: func(r *http.Request) (*http.Response, error) {
+				return &http.Response{
+					StatusCode: http.StatusAccepted,
+					Body:       ioutil.NopCloser(bytes.NewReader(respJSON)),
+				}, nil
+			},
+		},
+	}
+
+	ddSink, err := NewDatadog("testmonitor", defaultOps(), map[string]string{}, httpClient)
+	require.NoError(t, err)
+
+	err = ddSink.Send(context.Background(), getBlipMetrics(10))
+
+	// validation errors should be logged and the sink should continue
+	require.NoError(t, err)
+}