diff --git a/event/list.go b/event/list.go index 2219e16..7f191b8 100644 --- a/event/list.go +++ b/event/list.go @@ -66,6 +66,7 @@ const ( // Sink Events const ( SINK_INVALID_METRICS = "sink-invalid-metrics" // invalid metrics, drop + SINK_PAYLOAD = "sink-payload" // payload sizing and submission diagnostics SINK_SERVER_ERROR = "sink-server-error" // send ok but remote server returned an error SINK_SEND_ERROR = "sink-send-error" // e.g. network timeout ) diff --git a/sink/datadog.go b/sink/datadog.go index 5623545..7c1895c 100644 --- a/sink/datadog.go +++ b/sink/datadog.go @@ -3,18 +3,14 @@ package sink import ( - "bytes" - "compress/zlib" "context" - "encoding/json" "fmt" - "math" "net/http" "os" "regexp" + "sort" "strconv" - "strings" - "sync" + "sync/atomic" "time" "github.com/DataDog/datadog-api-client-go/v2/api/datadog" @@ -30,7 +26,7 @@ import ( var portRe = regexp.MustCompile(`:\d+$`) const ( - MAX_PAYLOAD_SIZE int = 512000 + MAX_PAYLOAD_SIZE int = datadogMaxCompressedPayloadSize ) // Datadog sends metrics to Datadog. @@ -42,15 +38,13 @@ type Datadog struct { event event.MonitorReceiver // -- Api - metricsApi *datadogV2.MetricsApi - apiKeyAuth string - appKeyAuth string - resources []datadogV2.MetricResource - compress bool - - maxMetricsPerRequest int // Limit the number of metrics we send per request. Only used with the API - maxMetricsPerRequestLock sync.Mutex - maxPayloadSize int + apiKeyAuth string + appKeyAuth string + resources []datadogV2.MetricResource + compress bool + submitter datadogMetricSubmitter + payloadLimits datadogPayloadLimits + maxSeriesPerRequest atomic.Int64 // -- DogStatsD dogstatsd bool @@ -76,16 +70,17 @@ func NewDatadog(monitorId string, opts, tags map[string]string, httpClient *http } } } + sort.Strings(tagList) d := &Datadog{ - monitorId: monitorId, - event: event.MonitorReceiver{MonitorId: monitorId}, - tags: tagList, - resources: resources, - maxMetricsPerRequest: math.MaxInt32, // By default, don't limit the number of metrics per request. - compress: true, - maxPayloadSize: MAX_PAYLOAD_SIZE, + monitorId: monitorId, + event: event.MonitorReceiver{MonitorId: monitorId}, + tags: tagList, + resources: resources, + compress: true, + payloadLimits: defaultDatadogPayloadLimits(), } + d.maxSeriesPerRequest.Store(int64(d.payloadLimits.maxSeries)) for k, v := range opts { switch k { @@ -164,14 +159,20 @@ func NewDatadog(monitorId string, opts, tags map[string]string, httpClient *http c := datadog.NewConfiguration() c.HTTPClient = httpClient c.Compress = d.compress - metricsApi := datadogV2.NewMetricsApi(datadog.NewAPIClient(c)) - d.metricsApi = metricsApi + d.submitter = &datadogAPISubmitter{ + client: datadog.NewAPIClient(c), + apiKey: d.apiKeyAuth, + } } return d, nil } func (s *Datadog) Send(ctx context.Context, m *blip.Metrics) error { + return s.send(ctx, m, true) +} + +func (s *Datadog) send(ctx context.Context, m *blip.Metrics, allowCheckpoint bool) error { status.Monitor(s.monitorId, s.Name(), "sending metrics") // Pre-alloc data points if using Datadog API (not DogStatsD) @@ -186,6 +187,10 @@ func (s *Datadog) Send(ctx context.Context, m *blip.Metrics) error { blip.Debug("%s: zero metric values collect: %s", m) return nil } + if !s.dogstatsd && allowCheckpoint { + _, err := s.SendWithCheckpoint(ctx, m, nil) + return err + } if !s.dogstatsd { dp = make([]datadogV2.MetricSeries, n) } @@ -196,25 +201,6 @@ func (s *Datadog) Send(ctx context.Context, m *blip.Metrics) error { status.Monitor(s.monitorId, s.Name(), "last sent %d metrics at %s", n, time.Now()) }() - // Make a copy of maxMetricsPerRequest in case it gets updated by other threads - localMaxMetricsPerRequest := s.maxMetricsPerRequest - rangeStart := 0 - var apiErrors []string - - // Setup our context for API calls - ddCtx := context.WithValue( - ctx, - datadog.ContextAPIKeys, - map[string]datadog.APIKey{ - "apiKeyAuth": { - Key: s.apiKeyAuth, - }, - "appKeyAuth": { - Key: s.apiKeyAuth, - }, - }, - ) - // Convert Blip metric values to Datadog data points for domain := range m.Values { // each domain metrics := m.Values[domain] @@ -328,14 +314,6 @@ func (s *Datadog) Send(ctx context.Context, m *blip.Metrics) error { } n++ - - // Check if we have reached the maximum number of metrics per request - if !s.dogstatsd && n%localMaxMetricsPerRequest == 0 { - if err := s.sendApi(ddCtx, dp[rangeStart:n]); err != nil { - apiErrors = append(apiErrors, err.Error()) - } - rangeStart = n - } } // metric } // domain @@ -351,128 +329,11 @@ func (s *Datadog) Send(ctx context.Context, m *blip.Metrics) error { return nil // success (dogstatsd) } - if n-rangeStart > 0 { - if err := s.sendApi(ddCtx, dp[rangeStart:n]); err != nil { - apiErrors = append(apiErrors, err.Error()) - } - } - - if len(apiErrors) > 0 { - return fmt.Errorf("%s", strings.Join(apiErrors, "\n")) - } - - return nil // success (API) -} - -// Send metrics to the API taking into consideration the number of metrics sent per request. -func (s *Datadog) sendApi(ddCtx context.Context, dp []datadogV2.MetricSeries) error { - localMaxMetricsPerRequest := s.maxMetricsPerRequest - - for rangeStart := 0; rangeStart < len(dp); { - // Determine the subetset of metrics to send based on our - // max per request - rangeEnd := rangeStart + localMaxMetricsPerRequest - if rangeEnd > len(dp) { - rangeEnd = len(dp) - } - - optParams := *datadogV2.NewSubmitMetricsOptionalParameters() - if s.compress { - optParams.ContentEncoding = datadogV2.METRICCONTENTENCODING_GZIP.Ptr() - } - - apiResponse, r, err := s.metricsApi.SubmitMetrics(ddCtx, *datadogV2.NewMetricPayload(dp[rangeStart:rangeEnd]), optParams) - if err != nil { - if r != nil { - if r.StatusCode == http.StatusRequestEntityTooLarge { - // Is the number of metrics sent already the smallest possible? - if localMaxMetricsPerRequest == 1 { - return fmt.Errorf("HTTP status %d (request too large) but dynamic request size at minimum: 1 metric per request; send err: %v; response body: %v", r.StatusCode, err, r.Body) - } - - // The payload was too large, so we need to recalculate it and try with a smaller size - var err2 error - if localMaxMetricsPerRequest, err2 = s.estimateMaxMetricsPerRequest(dp[rangeStart:rangeEnd], localMaxMetricsPerRequest); err2 != nil { - return fmt.Errorf("HTTP status %d (request too large) and error estimating new dynamic request size: %v; send err: %v; response body: %v", r.StatusCode, err2, err, r.Body) - } - - continue // Retry the metrics with the new payload size - } - return fmt.Errorf("%s (HTTP status %d: %v)", err, r.StatusCode, r.Body) - } - - return fmt.Errorf("network error (nil response): %v", err) - } - - // Datadog can return a 202 Accepted response _and_ errors in the response. - // This probably means partial success, so keep sending but log the error. - if len(apiResponse.Errors) > 0 { - errMsg := fmt.Sprintf("Datadog returned success and %d errors: %s", len(apiResponse.Errors), strings.Join(apiResponse.Errors, ", ")) - s.event.Errorf(event.SINK_SERVER_ERROR, "%s", errMsg) - } - - rangeStart = rangeEnd - } - - // Update the maxMetricsPerRequest for the sink - if localMaxMetricsPerRequest < s.maxMetricsPerRequest { - s.maxMetricsPerRequestLock.Lock() - // Check the value again in case it changed after getting the lock - if localMaxMetricsPerRequest < s.maxMetricsPerRequest { - s.maxMetricsPerRequest = localMaxMetricsPerRequest - } - s.maxMetricsPerRequestLock.Unlock() - } - - return nil -} - -// Estimate the number of metrics we can send in a payload based on a sample metric -func (s *Datadog) estimateMaxMetricsPerRequest(metrics []datadogV2.MetricSeries, currentMaxMetricsPerRequest int) (int, error) { - // Estimate the size of a single metric - estMetricSize, err := s.estimateSize(metrics) - if err != nil { - return 0, err - } - - // Using our estimated metric size determine out how many metrics can fit inside the max payload, but pad it slightly to control for headers, etc. - estMaxMetricsPerRequest := (s.maxPayloadSize - 300) / estMetricSize - - if estMaxMetricsPerRequest >= currentMaxMetricsPerRequest { - // If the estimated maximum is greater than what we currently have set as the maximum then - // reduce the current maximum by 10% as a guess for finding a maximnum number of metrics - // to send that will not be rejected by the API. - estMaxMetricsPerRequest = int(float32(currentMaxMetricsPerRequest) * .9) - } - - // Ensure we send at least one metric per request - if estMaxMetricsPerRequest <= 0 { - estMaxMetricsPerRequest = 1 - } - - return estMaxMetricsPerRequest, nil + return s.sendAPI(ctx, dp[:n], m) } -// Estimate the size of a metric payload for use in determining the maximum number of -// metrics per request. We take the total size of the payload and divide by the number -// of metrics. -func (s *Datadog) estimateSize(metrics []datadogV2.MetricSeries) (int, error) { - data, err := json.Marshal(metrics) - if err != nil { - return 0, err - } - - size := len(data) - - if s.compress { - var b bytes.Buffer - w := zlib.NewWriter(&b) - w.Write(data) - w.Close() - size = len(b.Bytes()) - } - - return size / len(metrics), nil +func (s *Datadog) sendDogStatsD(ctx context.Context, m *blip.Metrics) error { + return s.send(ctx, m, false) } func (s *Datadog) Name() string { diff --git a/sink/datadog_api.go b/sink/datadog_api.go new file mode 100644 index 0000000..3f75134 --- /dev/null +++ b/sink/datadog_api.go @@ -0,0 +1,400 @@ +// Copyright 2024 Block, Inc. + +package sink + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/DataDog/datadog-api-client-go/v2/api/datadog" + "github.com/DataDog/datadog-api-client-go/v2/api/datadogV2" + + "github.com/cashapp/blip" + "github.com/cashapp/blip/event" + "github.com/cashapp/blip/status" +) + +const ( + datadogMaxCompressedPayloadSize = 512_000 + datadogMaxDecompressedPayloadSize = 5 * 1024 * 1024 + + // Keep payloads below the documented hard limits. Exact byte measurement + // makes the margin defensive rather than part of the sizing algorithm. + datadogTargetCompressedPayloadSize = datadogMaxCompressedPayloadSize * 9 / 10 + datadogTargetDecompressedPayloadSize = datadogMaxDecompressedPayloadSize * 9 / 10 + + // This is a latency and CPU guard, not a payload-size estimate. Exact raw + // and compressed byte counts determine whether a payload can be submitted. + datadogMaxSeriesPerPayload = 10_000 + datadogMax413Retries = 4 +) + +var ( + datadogPayloadPrefix = []byte(`{"series":[`) + datadogPayloadSuffix = []byte(`]}`) +) + +type datadogPayloadLimits struct { + maxCompressed int + maxDecompressed int + targetCompressed int + targetDecompressed int + maxSeries int +} + +func defaultDatadogPayloadLimits() datadogPayloadLimits { + return datadogPayloadLimits{ + maxCompressed: datadogMaxCompressedPayloadSize, + maxDecompressed: datadogMaxDecompressedPayloadSize, + targetCompressed: datadogTargetCompressedPayloadSize, + targetDecompressed: datadogTargetDecompressedPayloadSize, + maxSeries: datadogMaxSeriesPerPayload, + } +} + +func (l datadogPayloadLimits) validate() error { + if l.maxCompressed <= 0 || l.maxDecompressed <= 0 || l.maxSeries <= 0 { + return fmt.Errorf("invalid Datadog payload limits: %+v", l) + } + if l.targetCompressed <= 0 || l.targetCompressed > l.maxCompressed { + return fmt.Errorf("invalid Datadog compressed payload target: %d", l.targetCompressed) + } + if l.targetDecompressed <= 0 || l.targetDecompressed > l.maxDecompressed { + return fmt.Errorf("invalid Datadog decompressed payload target: %d", l.targetDecompressed) + } + return nil +} + +type preparedDatadogPayload struct { + body []byte + seriesCount int + uncompressedBytes int + compressedBytes int + compressed bool +} + +type datadogSubmitResult struct { + statusCode int + errors []string + body string +} + +type datadogMetricSubmitter interface { + Submit(context.Context, preparedDatadogPayload) (datadogSubmitResult, error) +} + +// datadogAPISubmitter submits a body that has already been encoded and sized. +// The generated Datadog SubmitMetrics method cannot accept a prepared body: it +// always marshals and compresses the MetricPayload itself. +type datadogAPISubmitter struct { + client *datadog.APIClient + apiKey string +} + +func (s *datadogAPISubmitter) Submit(ctx context.Context, payload preparedDatadogPayload) (datadogSubmitResult, error) { + var result datadogSubmitResult + requestCtx := context.WithValue(ctx, datadog.ContextAPIKeys, map[string]datadog.APIKey{ + "apiKeyAuth": {Key: s.apiKey}, + }) + + baseURL, err := s.client.GetConfig().ServerURLWithContext(requestCtx, "v2.MetricsApi.SubmitMetrics") + if err != nil { + return result, err + } + + req, err := http.NewRequestWithContext(requestCtx, http.MethodPost, strings.TrimRight(baseURL, "/")+"/api/v2/series", bytes.NewReader(payload.body)) + if err != nil { + return result, err + } + req.Header.Set("Accept", "application/json") + req.Header.Set("Content-Type", "application/json") + req.Header.Set("DD-API-KEY", s.apiKey) + if payload.compressed { + req.Header.Set("Content-Encoding", "gzip") + } + + cfg := s.client.GetConfig() + for name, value := range cfg.DefaultHeader { + req.Header.Set(name, value) + } + if cfg.UserAgent != "" { + req.Header.Set("User-Agent", cfg.UserAgent) + } + + resp, err := s.client.CallAPI(req) + if resp != nil { + result.statusCode = resp.StatusCode + } + if err != nil { + return result, err + } + defer resp.Body.Close() + + responseBody, err := io.ReadAll(resp.Body) + if err != nil { + return result, err + } + resp.Body = io.NopCloser(bytes.NewReader(responseBody)) + result.body = string(responseBody) + + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return result, fmt.Errorf("HTTP status %d: %s", resp.StatusCode, strings.TrimSpace(result.body)) + } + + if len(bytes.TrimSpace(responseBody)) > 0 { + var accepted datadogV2.IntakePayloadAccepted + if err := json.Unmarshal(responseBody, &accepted); err != nil { + return result, fmt.Errorf("decode Datadog response: %w", err) + } + result.errors = accepted.Errors + } + + return result, nil +} + +// prepareDatadogPayload constructs the exact JSON body used on the wire. It +// stops at both a decompressed byte budget and a defensive series-count guard, +// then validates the actual gzip size. The returned end index is exclusive. +func prepareDatadogPayload(ctx context.Context, series []datadogV2.MetricSeries, start, maxSeries int, compress bool, limits datadogPayloadLimits) (preparedDatadogPayload, int, error) { + var prepared preparedDatadogPayload + if err := limits.validate(); err != nil { + return prepared, start, err + } + if start < 0 || start >= len(series) { + return prepared, start, fmt.Errorf("invalid Datadog payload start index %d for %d series", start, len(series)) + } + if maxSeries <= 0 || maxSeries > limits.maxSeries { + maxSeries = limits.maxSeries + } + + rawTarget := limits.targetCompressed + rawHardLimit := limits.maxCompressed + if compress { + rawTarget = limits.targetDecompressed + rawHardLimit = limits.maxDecompressed + } + + var raw bytes.Buffer + grow := rawTarget + if estimated := maxSeries * 512; estimated < grow { + grow = estimated + } + if grow > 0 { + raw.Grow(grow) + } + raw.Write(datadogPayloadPrefix) + + offsets := make([]int, 0, maxSeries) + end := start + for end < len(series) && len(offsets) < maxSeries { + select { + case <-ctx.Done(): + return prepared, start, ctx.Err() + default: + } + + encoded, err := json.Marshal(series[end]) + if err != nil { + return prepared, start, fmt.Errorf("marshal Datadog series %d: %w", end, err) + } + + separatorBytes := 0 + if len(offsets) > 0 { + separatorBytes = 1 + } + projected := raw.Len() + separatorBytes + len(encoded) + len(datadogPayloadSuffix) + if len(offsets) > 0 && projected > rawTarget { + break + } + if projected > rawHardLimit { + return prepared, start, fmt.Errorf("Datadog series %d requires a %d-byte payload, exceeding the %d-byte limit", end, projected, rawHardLimit) + } + + if separatorBytes != 0 { + raw.WriteByte(',') + } + raw.Write(encoded) + offsets = append(offsets, raw.Len()) + end++ + } + + if len(offsets) == 0 { + return prepared, start, fmt.Errorf("could not fit Datadog series %d in a payload", start) + } + raw.Write(datadogPayloadSuffix) + + if !compress { + prepared = preparedDatadogPayload{ + body: raw.Bytes(), + seriesCount: len(offsets), + uncompressedBytes: raw.Len(), + compressedBytes: raw.Len(), + } + return prepared, end, nil + } + + rawBytes := raw.Bytes() + compressed, err := gzipDatadogPayload(rawBytes) + if err != nil { + return prepared, start, err + } + + count := len(offsets) + for len(compressed) > limits.targetCompressed && count > 1 { + keep := int(float64(count) * float64(limits.targetCompressed) / float64(len(compressed)) * 0.90) + if keep >= count { + keep = count - 1 + } + if keep < 1 { + keep = 1 + } + + count = keep + end = start + count + rawBytes = payloadPrefixForSeries(rawBytes, offsets[count-1]) + compressed, err = gzipDatadogPayload(rawBytes) + if err != nil { + return prepared, start, err + } + } + + if len(rawBytes) > limits.maxDecompressed { + return prepared, start, fmt.Errorf("Datadog payload is %d bytes decompressed, exceeding the %d-byte limit", len(rawBytes), limits.maxDecompressed) + } + if len(compressed) > limits.maxCompressed { + return prepared, start, fmt.Errorf("Datadog payload is %d bytes compressed, exceeding the %d-byte limit", len(compressed), limits.maxCompressed) + } + + prepared = preparedDatadogPayload{ + body: compressed, + seriesCount: count, + uncompressedBytes: len(rawBytes), + compressedBytes: len(compressed), + compressed: true, + } + return prepared, end, nil +} + +func payloadPrefixForSeries(raw []byte, end int) []byte { + prefix := make([]byte, end+len(datadogPayloadSuffix)) + copy(prefix, raw[:end]) + copy(prefix[end:], datadogPayloadSuffix) + return prefix +} + +func gzipDatadogPayload(raw []byte) ([]byte, error) { + var compressed bytes.Buffer + w := gzip.NewWriter(&compressed) + if _, err := w.Write(raw); err != nil { + return nil, err + } + if err := w.Close(); err != nil { + return nil, err + } + return compressed.Bytes(), nil +} + +func (s *Datadog) sendAPI(ctx context.Context, series []datadogV2.MetricSeries, metrics *blip.Metrics) error { + if s.submitter == nil { + return fmt.Errorf("Datadog API submitter is not configured") + } + + batchStart := time.Now() + rangeStart := 0 + chunk := 0 + for rangeStart < len(series) { + chunk++ + maxSeries := int(s.maxSeriesPerRequest.Load()) + if maxSeries <= 0 { + maxSeries = s.payloadLimits.maxSeries + } + + prepareStart := time.Now() + payload, rangeEnd, err := prepareDatadogPayload(ctx, series, rangeStart, maxSeries, s.compress, s.payloadLimits) + if err != nil { + return err + } + prepareDuration := time.Since(prepareStart) + + attempt := 0 + for { + s.event.Sendf(event.SINK_PAYLOAD, + "plan=%s level=%s interval=%d chunk=%d attempt=%d series=%d raw-bytes=%d wire-bytes=%d prepare=%s", + metrics.Plan, metrics.Level, metrics.Interval, chunk, attempt+1, payload.seriesCount, + payload.uncompressedBytes, payload.compressedBytes, prepareDuration) + status.Monitor(s.monitorId, s.Name(), + "%s/%s/%d: chunk %d sending %d series (%d raw bytes, %d wire bytes)", + metrics.Plan, metrics.Level, metrics.Interval, chunk, payload.seriesCount, + payload.uncompressedBytes, payload.compressedBytes) + + sendStart := time.Now() + result, err := s.submitter.Submit(ctx, payload) + sendDuration := time.Since(sendStart) + if err == nil { + s.event.Sendf(event.SINK_PAYLOAD, + "plan=%s level=%s interval=%d chunk=%d attempt=%d status=%d series=%d raw-bytes=%d wire-bytes=%d send=%s", + metrics.Plan, metrics.Level, metrics.Interval, chunk, attempt+1, result.statusCode, + payload.seriesCount, payload.uncompressedBytes, payload.compressedBytes, sendDuration) + if len(result.errors) > 0 { + s.event.Errorf(event.SINK_SERVER_ERROR, "Datadog returned success and %d errors: %s", len(result.errors), strings.Join(result.errors, ", ")) + } + rangeStart = rangeEnd + break + } + + if result.statusCode != http.StatusRequestEntityTooLarge { + if result.statusCode == 0 { + return fmt.Errorf("network error (nil response): %w", err) + } + return err + } + + attempt++ + s.event.Sendf(event.SINK_PAYLOAD, + "plan=%s level=%s interval=%d chunk=%d attempt=%d status=413 series=%d raw-bytes=%d wire-bytes=%d send=%s", + metrics.Plan, metrics.Level, metrics.Interval, chunk, attempt, + payload.seriesCount, payload.uncompressedBytes, payload.compressedBytes, sendDuration) + if payload.seriesCount == 1 || attempt > datadogMax413Retries { + return fmt.Errorf("Datadog rejected %d locally-sized series with HTTP 413 after %d attempts: raw-bytes=%d wire-bytes=%d: %w", + payload.seriesCount, attempt, payload.uncompressedBytes, payload.compressedBytes, err) + } + + maxSeries = payload.seriesCount / 2 + if maxSeries < 1 { + maxSeries = 1 + } + s.reduceMaxSeriesPerRequest(maxSeries) + + prepareStart = time.Now() + payload, rangeEnd, err = prepareDatadogPayload(ctx, series, rangeStart, maxSeries, s.compress, s.payloadLimits) + if err != nil { + return err + } + prepareDuration = time.Since(prepareStart) + } + } + + s.event.Sendf(event.SINK_PAYLOAD, + "plan=%s level=%s interval=%d chunks=%d series=%d total=%s", + metrics.Plan, metrics.Level, metrics.Interval, chunk, len(series), time.Since(batchStart)) + return nil +} + +func (s *Datadog) reduceMaxSeriesPerRequest(limit int) { + for { + current := s.maxSeriesPerRequest.Load() + if current > 0 && int64(limit) >= current { + return + } + if s.maxSeriesPerRequest.CompareAndSwap(current, int64(limit)) { + return + } + } +} diff --git a/sink/datadog_api_test.go b/sink/datadog_api_test.go new file mode 100644 index 0000000..fc0e0d4 --- /dev/null +++ b/sink/datadog_api_test.go @@ -0,0 +1,246 @@ +package sink + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "testing" + "time" + + "github.com/DataDog/datadog-api-client-go/v2/api/datadog" + "github.com/DataDog/datadog-api-client-go/v2/api/datadogV2" + "github.com/stretchr/testify/require" + + "github.com/cashapp/blip" +) + +type rejectThenFailSubmitter struct { + calls int +} + +type payloadLimitSubmitter struct { + maxCompressed int + maxDecompressed int + requests int + rejected int + acceptedSeries []datadogV2.MetricSeries +} + +func (s *payloadLimitSubmitter) Submit(_ context.Context, payload preparedDatadogPayload) (datadogSubmitResult, error) { + s.requests++ + if payload.compressedBytes > s.maxCompressed || payload.uncompressedBytes > s.maxDecompressed { + s.rejected++ + return datadogSubmitResult{statusCode: http.StatusRequestEntityTooLarge}, errors.New("payload too large") + } + + decoded, err := decodePreparedMetricPayload(payload) + if err != nil { + return datadogSubmitResult{}, err + } + s.acceptedSeries = append(s.acceptedSeries, decoded.Series...) + return datadogSubmitResult{statusCode: http.StatusAccepted}, nil +} + +type checkpointFailureSubmitter struct { + calls int + acceptedNames map[string]int + acceptedBody map[string]int +} + +func (s *checkpointFailureSubmitter) Submit(_ context.Context, payload preparedDatadogPayload) (datadogSubmitResult, error) { + s.calls++ + if s.calls == 2 { + return datadogSubmitResult{}, errors.New("injected checkpoint failure") + } + + decoded, err := decodePreparedMetricPayload(payload) + if err != nil { + return datadogSubmitResult{}, err + } + if s.acceptedNames == nil { + s.acceptedNames = map[string]int{} + s.acceptedBody = map[string]int{} + } + for _, series := range decoded.Series { + s.acceptedNames[series.Metric]++ + } + s.acceptedBody[string(payload.body)]++ + return datadogSubmitResult{statusCode: http.StatusAccepted}, nil +} + +func decodePreparedMetricPayload(payload preparedDatadogPayload) (datadogV2.MetricPayload, error) { + body := payload.body + if payload.compressed { + reader, err := gzip.NewReader(bytes.NewReader(body)) + if err != nil { + return datadogV2.MetricPayload{}, err + } + body, err = io.ReadAll(reader) + if err != nil { + return datadogV2.MetricPayload{}, err + } + if err := reader.Close(); err != nil { + return datadogV2.MetricPayload{}, err + } + } + + var decoded datadogV2.MetricPayload + if err := json.Unmarshal(body, &decoded); err != nil { + return datadogV2.MetricPayload{}, err + } + return decoded, nil +} + +func testMetricSeries(count int) []datadogV2.MetricSeries { + series := make([]datadogV2.MetricSeries, count) + for i := range series { + series[i] = datadogV2.MetricSeries{ + Metric: fmt.Sprintf("mysql.test.metric_%04d", i), + Type: datadogV2.METRICINTAKETYPE_GAUGE.Ptr(), + Points: []datadogV2.MetricPoint{{ + Timestamp: datadog.PtrInt64(1_700_000_000), + Value: datadog.PtrFloat64(float64(i)), + }}, + Tags: []string{ + fmt.Sprintf("table:table_%04d", i), + "description:" + strings.Repeat("x", 100), + }, + } + } + return series +} + +func testMetricsMetadata() *blip.Metrics { + return &blip.Metrics{MonitorId: "test", Plan: "test", Level: "test", Interval: 1} +} + +func newTestDatadogSender(submitter datadogMetricSubmitter) *Datadog { + sender := &Datadog{ + monitorId: "test", + compress: true, + payloadLimits: defaultDatadogPayloadLimits(), + submitter: submitter, + } + sender.maxSeriesPerRequest.Store(int64(sender.payloadLimits.maxSeries)) + return sender +} + +func (s *rejectThenFailSubmitter) Submit(context.Context, preparedDatadogPayload) (datadogSubmitResult, error) { + s.calls++ + if s.calls == 1 { + return datadogSubmitResult{statusCode: http.StatusRequestEntityTooLarge}, errors.New("payload too large") + } + return datadogSubmitResult{}, errors.New("network failure") +} + +func TestPrepareDatadogPayloadHonorsCompressedLimit(t *testing.T) { + series := testMetricSeries(1_000) + limits := datadogPayloadLimits{ + maxCompressed: 1_000, + maxDecompressed: 1_000_000, + targetCompressed: 900, + targetDecompressed: 900_000, + maxSeries: len(series), + } + + payload, end, err := prepareDatadogPayload(context.Background(), series, 0, len(series), true, limits) + require.NoError(t, err) + require.Less(t, end, len(series), "compressed byte budget should split the input") + require.LessOrEqual(t, payload.compressedBytes, limits.targetCompressed) + require.LessOrEqual(t, payload.uncompressedBytes, limits.targetDecompressed) + + reader, err := gzip.NewReader(bytes.NewReader(payload.body)) + require.NoError(t, err) + raw, err := io.ReadAll(reader) + require.NoError(t, err) + require.NoError(t, reader.Close()) + + var decoded datadogV2.MetricPayload + require.NoError(t, json.Unmarshal(raw, &decoded)) + require.Len(t, decoded.Series, payload.seriesCount) + require.Equal(t, series[:end], decoded.Series) +} + +func TestDatadog413FallbackHalvesAndPersistsLimit(t *testing.T) { + series := testMetricSeries(1_000) + intake := &payloadLimitSubmitter{ + maxCompressed: datadogMaxCompressedPayloadSize, + maxDecompressed: 80_000, + } + // Simulate an effective server limit lower than the documented limit. The + // locally valid first payload should receive 413 and trigger bounded halving. + sender := newTestDatadogSender(intake) + require.NoError(t, sender.sendAPI(context.Background(), series, testMetricsMetadata())) + + require.Greater(t, intake.rejected, 0) + require.Len(t, intake.acceptedSeries, len(series)) + require.Less(t, sender.maxSeriesPerRequest.Load(), int64(len(series))) + require.Greater(t, intake.requests, intake.rejected) +} + +func TestDatadog413LimitSurvivesLaterFailure(t *testing.T) { + series := testMetricSeries(10) + submitter := &rejectThenFailSubmitter{} + sender := &Datadog{ + compress: true, + payloadLimits: defaultDatadogPayloadLimits(), + submitter: submitter, + } + sender.maxSeriesPerRequest.Store(int64(len(series))) + + err := sender.sendAPI(context.Background(), series, testMetricsMetadata()) + require.ErrorContains(t, err, "network failure") + require.Equal(t, 2, submitter.calls) + require.Equal(t, int64(len(series)/2), sender.maxSeriesPerRequest.Load()) +} + +func TestPrepareDatadogPayloadStopsOnCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + series := testMetricSeries(10) + _, _, err := prepareDatadogPayload(ctx, series, 0, 10, true, defaultDatadogPayloadLimits()) + require.ErrorIs(t, err, context.Canceled) +} + +func TestDatadogRetryCheckpointDoesNotResendAcknowledgedPayload(t *testing.T) { + const metricCount = 40 + submitter := &checkpointFailureSubmitter{} + sender := &Datadog{ + monitorId: "checkpoint-test", + compress: false, + payloadLimits: datadogPayloadLimits{ + maxCompressed: 1_500, + maxDecompressed: 10_000, + targetCompressed: 1_200, + targetDecompressed: 9_000, + maxSeries: metricCount, + }, + submitter: submitter, + } + sender.maxSeriesPerRequest.Store(metricCount) + retry := NewRetry(RetryArgs{ + MonitorId: "checkpoint-test", + Sink: sender, + BufferSize: 2, + SendTimeout: 5 * time.Second, + SendRetryWait: time.Millisecond, + }) + + require.NoError(t, retry.Send(context.Background(), getBlipMetrics(metricCount, blip.GAUGE, 1, false))) + require.Greater(t, submitter.calls, 3, "test data must span at least three chunks") + require.Equal(t, metricCount, len(submitter.acceptedNames)) + for name, count := range submitter.acceptedNames { + require.Equalf(t, 1, count, "acknowledged metric %s was submitted more than once", name) + } + for body, count := range submitter.acceptedBody { + require.Equalf(t, 1, count, "acknowledged payload was submitted more than once: %s", body) + } + require.Equal(t, -1, retry.top, "successful resume should remove the checkpointed queue entry") +} diff --git a/sink/datadog_stream.go b/sink/datadog_stream.go new file mode 100644 index 0000000..8e47c85 --- /dev/null +++ b/sink/datadog_stream.go @@ -0,0 +1,315 @@ +// Copyright 2024 Block, Inc. + +package sink + +import ( + "context" + "fmt" + "net/http" + "sort" + "strconv" + "strings" + "time" + + "github.com/DataDog/datadog-api-client-go/v2/api/datadog" + "github.com/DataDog/datadog-api-client-go/v2/api/datadogV2" + + "github.com/cashapp/blip" + "github.com/cashapp/blip/event" + "github.com/cashapp/blip/status" +) + +// datadogMetricCursor identifies the next Blip metric to convert. It is kept +// in the retry queue as part of an opaque checkpoint, so a retry starts after +// the last chunk acknowledged by Datadog. +type datadogMetricCursor struct { + domain int + metric int +} + +type datadogSendCheckpoint struct { + domains []string + cursor datadogMetricCursor + chunks int + sentSeries int + maxPending int + started time.Time + + // pending is bounded by payloadLimits.maxSeries. Keeping an already + // converted window avoids translating the unused suffix again when the byte + // budget selects only a prefix, and makes retries reproduce that suffix. + pending []datadogV2.MetricSeries + pendingCursors []datadogMetricCursor + pendingFinal datadogMetricCursor + pendingDone bool +} + +// SendWithCheckpoint converts only a bounded window of Blip metrics at a time. +// The returned checkpoint advances only after Datadog acknowledges a chunk. +func (s *Datadog) SendWithCheckpoint(ctx context.Context, metrics *blip.Metrics, checkpoint any) (any, error) { + if s.dogstatsd { + return nil, s.sendDogStatsD(ctx, metrics) + } + if s.submitter == nil { + return checkpoint, fmt.Errorf("Datadog API submitter is not configured") + } + if err := s.payloadLimits.validate(); err != nil { + return checkpoint, err + } + totalValues := 0 + for _, values := range metrics.Values { + totalValues += len(values) + } + if totalValues == 0 { + blip.Debug("%s: zero metric values collect: %s", metrics) + return nil, nil + } + + state, err := newDatadogSendCheckpoint(metrics, checkpoint) + if err != nil { + return checkpoint, err + } + status.Monitor(s.monitorId, s.Name(), "sending metrics") + + for { + maxSeries := int(s.maxSeriesPerRequest.Load()) + if maxSeries <= 0 || maxSeries > s.payloadLimits.maxSeries { + maxSeries = s.payloadLimits.maxSeries + } + + if len(state.pending) < maxSeries && !state.pendingDone { + collectFrom := state.cursor + if len(state.pendingCursors) > 0 { + collectFrom = state.pendingCursors[len(state.pendingCursors)-1] + } + var converted []datadogV2.MetricSeries + var cursors []datadogMetricCursor + converted, cursors, state.pendingFinal, state.pendingDone, err = s.collectDatadogSeries(ctx, metrics, state.domains, collectFrom, maxSeries-len(state.pending)) + if err != nil { + return state, err + } + if len(state.pending) == 0 { + state.pending = converted + state.pendingCursors = cursors + } else { + state.pending = append(state.pending, converted...) + state.pendingCursors = append(state.pendingCursors, cursors...) + } + if len(state.pending) > state.maxPending { + state.maxPending = len(state.pending) + } + } + if len(state.pending) == 0 { + if state.sentSeries == 0 { + errMsg := fmt.Sprintf("zero data points created after processing Blip metrics: %s", metrics) + s.event.Errorf(event.SINK_INVALID_METRICS, "%s", errMsg) + } + status.Monitor(s.monitorId, s.Name(), "last sent %d metrics at %s", state.sentSeries, time.Now()) + s.event.Sendf(event.SINK_PAYLOAD, + "plan=%s level=%s interval=%d chunks=%d series=%d max-window-series=%d total=%s conversion=streaming", + metrics.Plan, metrics.Level, metrics.Interval, state.chunks, state.sentSeries, state.maxPending, time.Since(state.started)) + return nil, nil + } + + chunk := state.chunks + 1 + prepareStart := time.Now() + payload, rangeEnd, err := prepareDatadogPayload(ctx, state.pending, 0, maxSeries, s.compress, s.payloadLimits) + if err != nil { + return state, err + } + prepareDuration := time.Since(prepareStart) + + attempt := 0 + for { + s.event.Sendf(event.SINK_PAYLOAD, + "plan=%s level=%s interval=%d chunk=%d attempt=%d series=%d raw-bytes=%d wire-bytes=%d prepare=%s conversion=streaming", + metrics.Plan, metrics.Level, metrics.Interval, chunk, attempt+1, payload.seriesCount, + payload.uncompressedBytes, payload.compressedBytes, prepareDuration) + status.Monitor(s.monitorId, s.Name(), + "%s/%s/%d: chunk %d sending %d series (%d raw bytes, %d wire bytes)", + metrics.Plan, metrics.Level, metrics.Interval, chunk, payload.seriesCount, + payload.uncompressedBytes, payload.compressedBytes) + + sendStart := time.Now() + result, submitErr := s.submitter.Submit(ctx, payload) + sendDuration := time.Since(sendStart) + if submitErr == nil { + s.event.Sendf(event.SINK_PAYLOAD, + "plan=%s level=%s interval=%d chunk=%d attempt=%d status=%d series=%d raw-bytes=%d wire-bytes=%d send=%s conversion=streaming", + metrics.Plan, metrics.Level, metrics.Interval, chunk, attempt+1, result.statusCode, + payload.seriesCount, payload.uncompressedBytes, payload.compressedBytes, sendDuration) + if len(result.errors) > 0 { + s.event.Errorf(event.SINK_SERVER_ERROR, "Datadog returned success and %d errors: %s", len(result.errors), strings.Join(result.errors, ", ")) + } + + state.cursor = state.pendingCursors[rangeEnd-1] + if state.pendingDone && rangeEnd == len(state.pending) { + state.cursor = state.pendingFinal + } + remaining := copy(state.pending, state.pending[rangeEnd:]) + for i := remaining; i < len(state.pending); i++ { + state.pending[i] = datadogV2.MetricSeries{} + } + state.pending = state.pending[:remaining] + remaining = copy(state.pendingCursors, state.pendingCursors[rangeEnd:]) + state.pendingCursors = state.pendingCursors[:remaining] + state.chunks++ + state.sentSeries += payload.seriesCount + break + } + + if result.statusCode != http.StatusRequestEntityTooLarge { + if result.statusCode == 0 { + return state, fmt.Errorf("network error (nil response): %w", submitErr) + } + return state, submitErr + } + + attempt++ + s.event.Sendf(event.SINK_PAYLOAD, + "plan=%s level=%s interval=%d chunk=%d attempt=%d status=413 series=%d raw-bytes=%d wire-bytes=%d send=%s conversion=streaming", + metrics.Plan, metrics.Level, metrics.Interval, chunk, attempt, + payload.seriesCount, payload.uncompressedBytes, payload.compressedBytes, sendDuration) + if payload.seriesCount == 1 || attempt > datadogMax413Retries { + return state, fmt.Errorf("Datadog rejected %d locally-sized series with HTTP 413 after %d attempts: raw-bytes=%d wire-bytes=%d: %w", + payload.seriesCount, attempt, payload.uncompressedBytes, payload.compressedBytes, submitErr) + } + + maxSeries = payload.seriesCount / 2 + if maxSeries < 1 { + maxSeries = 1 + } + s.reduceMaxSeriesPerRequest(maxSeries) + + prepareStart = time.Now() + payload, rangeEnd, err = prepareDatadogPayload(ctx, state.pending, 0, maxSeries, s.compress, s.payloadLimits) + if err != nil { + return state, err + } + prepareDuration = time.Since(prepareStart) + } + + if state.pendingDone && len(state.pending) == 0 && state.cursor == state.pendingFinal { + status.Monitor(s.monitorId, s.Name(), "last sent %d metrics at %s", state.sentSeries, time.Now()) + s.event.Sendf(event.SINK_PAYLOAD, + "plan=%s level=%s interval=%d chunks=%d series=%d max-window-series=%d total=%s conversion=streaming", + metrics.Plan, metrics.Level, metrics.Interval, state.chunks, state.sentSeries, state.maxPending, time.Since(state.started)) + return nil, nil + } + } +} + +func newDatadogSendCheckpoint(metrics *blip.Metrics, checkpoint any) (*datadogSendCheckpoint, error) { + if checkpoint != nil { + state, ok := checkpoint.(*datadogSendCheckpoint) + if !ok { + return nil, fmt.Errorf("invalid Datadog send checkpoint type %T", checkpoint) + } + return state, nil + } + + domains := make([]string, 0, len(metrics.Values)) + for domain := range metrics.Values { + domains = append(domains, domain) + } + sort.Strings(domains) + return &datadogSendCheckpoint{domains: domains, started: time.Now()}, nil +} + +func (s *Datadog) collectDatadogSeries(ctx context.Context, metrics *blip.Metrics, domains []string, start datadogMetricCursor, limit int) ([]datadogV2.MetricSeries, []datadogMetricCursor, datadogMetricCursor, bool, error) { + series := make([]datadogV2.MetricSeries, 0, limit) + cursors := make([]datadogMetricCursor, 0, limit) + cursor := start + + for cursor.domain < len(domains) { + values := metrics.Values[domains[cursor.domain]] + for cursor.metric < len(values) { + select { + case <-ctx.Done(): + return nil, nil, start, false, ctx.Err() + default: + } + + value := values[cursor.metric] + cursor.metric++ + converted, ok := s.datadogMetricSeries(metrics, domains[cursor.domain], value) + if !ok { + continue + } + series = append(series, converted) + cursors = append(cursors, cursor) + if len(series) == limit { + return series, cursors, cursor, false, nil + } + } + cursor.domain++ + cursor.metric = 0 + } + + return series, cursors, cursor, true, nil +} + +func (s *Datadog) datadogMetricSeries(metrics *blip.Metrics, domain string, value blip.MetricValue) (datadogV2.MetricSeries, bool) { + name := domain + "." + value.Name + if s.tr != nil { + name = s.tr.Translate(domain, value.Name) + } + if s.prefix != "" { + name = s.prefix + name + } + + timestamp := metrics.Begin.Unix() + if tsStr, ok := value.Meta["ts"]; ok { + msTs, err := strconv.ParseInt(tsStr, 10, 64) + if err != nil { + blip.Debug("invalid timestamp for %s %s: %s: %s", domain, value.Name, tsStr, err) + return datadogV2.MetricSeries{}, false + } + timestamp = msTs / 1000 + } + + var metricType *datadogV2.MetricIntakeType + switch value.Type { + case blip.CUMULATIVE_COUNTER, blip.DELTA_COUNTER: + metricType = datadogV2.METRICINTAKETYPE_COUNT.Ptr() + case blip.GAUGE, blip.BOOL: + metricType = datadogV2.METRICINTAKETYPE_GAUGE.Ptr() + default: + return datadogV2.MetricSeries{}, false + } + + tags := s.tags + if len(value.Meta) != 0 || len(value.Group) != 0 { + tags = make([]string, 0, len(s.tags)+len(value.Meta)+len(value.Group)) + tags = append(tags, s.tags...) + keys := make([]string, 0, len(value.Meta)) + for key := range value.Meta { + if key != "ts" { + keys = append(keys, key) + } + } + sort.Strings(keys) + for _, key := range keys { + tags = append(tags, fmt.Sprintf("%s:%s", key, value.Meta[key])) + } + keys = keys[:0] + for key := range value.Group { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + tags = append(tags, fmt.Sprintf("%s:%s", key, value.Group[key])) + } + } + + return datadogV2.MetricSeries{ + Metric: name, + Type: metricType, + Points: []datadogV2.MetricPoint{{ + Value: datadog.PtrFloat64(value.Value), + Timestamp: datadog.PtrInt64(timestamp), + }}, + Tags: tags, + Resources: s.resources, + }, true +} diff --git a/sink/datadog_test.go b/sink/datadog_test.go index 8440399..3c5b1be 100644 --- a/sink/datadog_test.go +++ b/sink/datadog_test.go @@ -6,7 +6,6 @@ import ( "encoding/json" "fmt" "io" - "math" "net/http" "sync" "testing" @@ -151,6 +150,8 @@ func TestDatadogSendBoundedByClientTimeout(t *testing.T) { func TestDatadogMetricsPerRequest(t *testing.T) { callCount := 0 + rejectedCount := 0 + maxBodySize := 0 testPayloadSize := 5000 metricCount := 100 @@ -163,8 +164,12 @@ func TestDatadogMetricsPerRequest(t *testing.T) { if err != nil { return nil, err } + if bodySize > maxBodySize { + maxBodySize = bodySize + } if bodySize > testPayloadSize { + rejectedCount++ return &http.Response{ StatusCode: http.StatusRequestEntityTooLarge, }, nil @@ -180,11 +185,9 @@ func TestDatadogMetricsPerRequest(t *testing.T) { ops := defaultOps() ops["api-compress"] = "false" // Turn off compression so that we get easier calculations for sizing ddSink, err := NewDatadog("testmonitor", ops, map[string]string{}, httpClient) - ddSink.maxPayloadSize = testPayloadSize // Set the payload size for testing - - if err != nil { - t.Fatalf("Expected no error but got %v", err) - } + require.NoError(t, err) + ddSink.payloadLimits.maxCompressed = testPayloadSize + ddSink.payloadLimits.targetCompressed = testPayloadSize * 9 / 10 err = ddSink.Send(context.Background(), getBlipMetrics(metricCount, blip.GAUGE, 1.0, false)) @@ -192,17 +195,15 @@ func TestDatadogMetricsPerRequest(t *testing.T) { t.Fatalf("Expected no error but got %v", err) } - if ddSink.maxMetricsPerRequest == math.MaxInt { - t.Error("Expected maxMetricsPerRequest to be adjusted but got MaxInt") - } - - if callCount != 4 { - t.Errorf("Expected 4 calls but got %d", callCount) - } + require.Greater(t, callCount, 1) + require.Zero(t, rejectedCount, "locally-sized requests should not receive HTTP 413") + require.LessOrEqual(t, maxBodySize, testPayloadSize) } func TestDatadogMetricsPerRequestWithCompression(t *testing.T) { callCount := 0 + rejectedCount := 0 + maxBodySize := 0 testPayloadSize := 1500 metricCount := 600 @@ -216,7 +217,11 @@ func TestDatadogMetricsPerRequestWithCompression(t *testing.T) { return nil, err } + if bodySize > maxBodySize { + maxBodySize = bodySize + } if bodySize > testPayloadSize { + rejectedCount++ return &http.Response{ StatusCode: http.StatusRequestEntityTooLarge, }, nil @@ -231,11 +236,9 @@ func TestDatadogMetricsPerRequestWithCompression(t *testing.T) { ops := defaultOps() ddSink, err := NewDatadog("testmonitor", ops, map[string]string{}, httpClient) - ddSink.maxPayloadSize = testPayloadSize // Set the payload size for testing - - if err != nil { - t.Fatalf("Expected no error but got %v", err) - } + require.NoError(t, err) + ddSink.payloadLimits.maxCompressed = testPayloadSize + ddSink.payloadLimits.targetCompressed = testPayloadSize * 9 / 10 err = ddSink.Send(context.Background(), getBlipMetrics(metricCount, blip.GAUGE, 1.0, false)) @@ -243,16 +246,12 @@ func TestDatadogMetricsPerRequestWithCompression(t *testing.T) { t.Fatalf("Expected no error but got %v", err) } - if ddSink.maxMetricsPerRequest == math.MaxInt { - t.Error("Expected maxMetricsPerRequest to be adjusted but got MaxInt") - } - - if callCount == 1 { - t.Error("Expected more than 1 call but got only 1") - } + require.Greater(t, callCount, 1) + require.Zero(t, rejectedCount, "locally-sized requests should not receive HTTP 413") + require.LessOrEqual(t, maxBodySize, testPayloadSize) } -func TestDatadogMetricsPerRequestMultipleFail(t *testing.T) { +func TestDatadogMetricsPerRequestHeterogeneousSeries(t *testing.T) { callCount := 0 testPayloadSize := 5000 metricCount := 500 @@ -295,7 +294,9 @@ func TestDatadogMetricsPerRequestMultipleFail(t *testing.T) { ops := defaultOps() ops["api-compress"] = "false" // Turn off compression so that we get easier calculations for sizing ddSink, err := NewDatadog("testmonitor", ops, map[string]string{}, httpClient) - ddSink.maxPayloadSize = testPayloadSize // Set the payload size for testing + require.NoError(t, err) + ddSink.payloadLimits.maxCompressed = testPayloadSize + ddSink.payloadLimits.targetCompressed = testPayloadSize * 9 / 10 ddSink.tr = &mock.Tr{ TranslateFunc: func(domain, metric string) string { trCount++ @@ -309,10 +310,6 @@ func TestDatadogMetricsPerRequestMultipleFail(t *testing.T) { }, } - if err != nil { - t.Fatalf("Expected no error but got %v", err) - } - blipMetrics := getBlipMetrics(metricCount, blip.GAUGE, 1.0, false) err = ddSink.Send(context.Background(), blipMetrics) @@ -320,10 +317,6 @@ func TestDatadogMetricsPerRequestMultipleFail(t *testing.T) { t.Fatalf("Expected no error but got %v", err) } - if ddSink.maxMetricsPerRequest == math.MaxInt { - t.Error("Expected maxMetricsPerRequest to be adjusted but got MaxInt") - } - if callCount == 1 { t.Error("Expected more than 1 call but only got 1.") } diff --git a/sink/retry.go b/sink/retry.go index 75646e7..31d80fb 100644 --- a/sink/retry.go +++ b/sink/retry.go @@ -37,11 +37,23 @@ type Retry struct { event event.MonitorReceiver stackMux *sync.Mutex - stack []*blip.Metrics // LIFO + stack []*retryItem // LIFO max int top int } +// CheckpointSink can resume a partially acknowledged metrics batch. Retry +// stores the opaque checkpoint with the same queue entry as the metrics. A +// sink must advance the returned checkpoint only after remote acknowledgement. +type CheckpointSink interface { + SendWithCheckpoint(context.Context, *blip.Metrics, any) (any, error) +} + +type retryItem struct { + metrics *blip.Metrics + checkpoint any +} + type RetryArgs struct { MonitorId string // required Sink blip.Sink // required @@ -84,7 +96,7 @@ func NewRetry(args RetryArgs) *Retry { retryWait: args.SendRetryWait, stackMux: &sync.Mutex{}, - stack: make([]*blip.Metrics, args.BufferSize), + stack: make([]*retryItem, args.BufferSize), max: int(args.BufferSize) - 1, top: -1, } @@ -100,7 +112,7 @@ func (rb *Retry) Name() string { // Send buffers, sends, and retries sending metrics on failure. It is safe to call // from multiple goroutines. func (rb *Retry) Send(ctx context.Context, m *blip.Metrics) error { - rb.push(m) // top of stack + rb.push(&retryItem{metrics: m}) // top of stack rb.sendMux.Lock() if rb.sending { @@ -141,7 +153,13 @@ func (rb *Retry) Send(ctx context.Context, m *blip.Metrics) error { n += 1 // Send next oldest metrics - if err := rb.sink.Send(ctx2, next); err != nil { + var err error + if sink, ok := rb.sink.(CheckpointSink); ok { + next.checkpoint, err = sink.SendWithCheckpoint(ctx2, next.metrics, next.checkpoint) + } else { + err = rb.sink.Send(ctx2, next.metrics) + } + if err != nil { rb.event.Errorf(event.SINK_SEND_ERROR, "%s", err.Error()) next = nil // don't pop metrics; retry stack from top down } @@ -150,7 +168,7 @@ func (rb *Retry) Send(ctx context.Context, m *blip.Metrics) error { return nil } -func (rb *Retry) push(m *blip.Metrics) { +func (rb *Retry) push(item *retryItem) { rb.stackMux.Lock() defer rb.stackMux.Unlock() if rb.top < rb.max { @@ -159,10 +177,10 @@ func (rb *Retry) push(m *blip.Metrics) { // Push down stack (push off oldest metrics) copy(rb.stack, rb.stack[1:]) } - rb.stack[rb.top] = m + rb.stack[rb.top] = item } -func (rb *Retry) pop(sent *blip.Metrics) *blip.Metrics { +func (rb *Retry) pop(sent *retryItem) *retryItem { rb.stackMux.Lock() defer rb.stackMux.Unlock() diff --git a/sink/retry_test.go b/sink/retry_test.go index 3f011d9..4d9733e 100644 --- a/sink/retry_test.go +++ b/sink/retry_test.go @@ -22,7 +22,7 @@ func stack(rb *Retry) []string { if rb.stack[i] == nil { stack = append(stack, "") } else { - stack = append(stack, rb.stack[i].Level) + stack = append(stack, rb.stack[i].metrics.Level) } } return stack