From df4229d362a0b227c28e89818fb79010469938d0 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Sat, 18 Jul 2026 13:42:32 -0700 Subject: [PATCH 1/3] Replace Datadog payload autosizing Size compressed metric payloads locally against both Datadog limits and submit the prepared bytes directly. Add Catalyst-shaped equivalence and benchmark coverage for the legacy and byte-budgeted paths. Co-authored-by: Codex Ai-assisted: true --- event/list.go | 1 + sink/datadog.go | 195 ++------------ sink/datadog_api.go | 400 +++++++++++++++++++++++++++++ sink/datadog_api_test.go | 97 +++++++ sink/datadog_catalyst_test.go | 466 ++++++++++++++++++++++++++++++++++ sink/datadog_test.go | 63 ++--- 6 files changed, 1013 insertions(+), 209 deletions(-) create mode 100644 sink/datadog_api.go create mode 100644 sink/datadog_api_test.go create mode 100644 sink/datadog_catalyst_test.go diff --git a/event/list.go b/event/list.go index 2219e161..7f191b8e 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 5623545c..b3f719d0 100644 --- a/sink/datadog.go +++ b/sink/datadog.go @@ -3,18 +3,13 @@ package sink import ( - "bytes" - "compress/zlib" "context" - "encoding/json" "fmt" - "math" "net/http" "os" "regexp" "strconv" - "strings" - "sync" + "sync/atomic" "time" "github.com/DataDog/datadog-api-client-go/v2/api/datadog" @@ -30,7 +25,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 +37,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 @@ -78,14 +71,14 @@ func NewDatadog(monitorId string, opts, tags map[string]string, httpClient *http } 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,8 +157,10 @@ 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 @@ -196,25 +191,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 +304,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 +319,7 @@ 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 -} - -// 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 + return s.sendAPI(ctx, dp[:n], m) } func (s *Datadog) Name() string { diff --git a/sink/datadog_api.go b/sink/datadog_api.go new file mode 100644 index 00000000..3f751345 --- /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 00000000..61e414f0 --- /dev/null +++ b/sink/datadog_api_test.go @@ -0,0 +1,97 @@ +package sink + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "errors" + "io" + "net/http" + "testing" + + "github.com/DataDog/datadog-api-client-go/v2/api/datadogV2" + "github.com/stretchr/testify/require" +) + +type rejectThenFailSubmitter struct { + calls int +} + +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 := catalystMetricSeriesSized(1_000, 0, 0) + 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 := catalystMetricSeriesSized(1_000, 0, 0) + intake := newCatalystIntake() + // Simulate an effective server limit lower than the documented limit. The + // locally valid first payload should receive 413 and trigger bounded halving. + intake.maxDecompressed = 150_000 + + sender := newCatalystDatadogSender(intake) + require.NoError(t, sender.sendAPI(context.Background(), series, catalystMetricsMetadata())) + + stats := intake.snapshot() + require.Greater(t, stats.rejected, 0) + require.Len(t, stats.acceptedSeries, len(series)) + require.Less(t, sender.maxSeriesPerRequest.Load(), int64(len(series))) + require.Equal(t, stats.requests, stats.rejected+stats.accepted) +} + +func TestDatadog413LimitSurvivesLaterFailure(t *testing.T) { + series := catalystMetricSeriesSized(10, 0, 0) + submitter := &rejectThenFailSubmitter{} + sender := &Datadog{ + compress: true, + payloadLimits: defaultDatadogPayloadLimits(), + submitter: submitter, + } + sender.maxSeriesPerRequest.Store(int64(len(series))) + + err := sender.sendAPI(context.Background(), series, catalystMetricsMetadata()) + 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 := catalystMetricSeriesSized(10, 0, 0) + _, _, err := prepareDatadogPayload(ctx, series, 0, 10, true, defaultDatadogPayloadLimits()) + require.ErrorIs(t, err, context.Canceled) +} diff --git a/sink/datadog_catalyst_test.go b/sink/datadog_catalyst_test.go new file mode 100644 index 00000000..50d669e2 --- /dev/null +++ b/sink/datadog_catalyst_test.go @@ -0,0 +1,466 @@ +package sink + +import ( + "bytes" + "compress/gzip" + "compress/zlib" + "context" + "encoding/json" + "fmt" + "io" + "math" + "net/http" + "os" + "sync" + "testing" + + "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" + "github.com/cashapp/blip/event" +) + +const ( + catalystSizeTableSeries = 9_155 + catalystWaitTableRows = 4_750 + catalystWaitMetrics = 4 + catalystFixedSeries = 100 + catalystTotalSeries = catalystSizeTableSeries + catalystWaitTableRows*catalystWaitMetrics + catalystFixedSeries +) + +type catalystIntakeStats struct { + requests int + rejected int + accepted int + totalWire int64 + totalRaw int64 + maxWire int + maxRaw int + acceptedSeries []datadogV2.MetricSeries +} + +// catalystIntake simulates the two independent Datadog v2 series limits and +// records only successfully accepted payloads. It is shared by the correctness +// test and benchmarks so both implementations see identical intake behavior. +type catalystIntake struct { + mu sync.Mutex + maxCompressed int + maxDecompressed int + stats catalystIntakeStats +} + +func newCatalystIntake() *catalystIntake { + return &catalystIntake{ + maxCompressed: datadogMaxCompressedPayloadSize, + maxDecompressed: datadogMaxDecompressedPayloadSize, + } +} + +func (i *catalystIntake) RoundTrip(req *http.Request) (*http.Response, error) { + wire, err := io.ReadAll(req.Body) + if err != nil { + return nil, err + } + + raw := wire + compressed := req.Header.Get("Content-Encoding") == "gzip" + if compressed { + reader, err := gzip.NewReader(bytes.NewReader(wire)) + if err != nil { + return nil, err + } + raw, err = io.ReadAll(reader) + closeErr := reader.Close() + if err != nil { + return nil, err + } + if closeErr != nil { + return nil, closeErr + } + } + + rejected := len(wire) > i.maxCompressed || (compressed && len(raw) >= i.maxDecompressed) + + i.mu.Lock() + i.stats.requests++ + i.stats.totalWire += int64(len(wire)) + i.stats.totalRaw += int64(len(raw)) + if len(wire) > i.stats.maxWire { + i.stats.maxWire = len(wire) + } + if len(raw) > i.stats.maxRaw { + i.stats.maxRaw = len(raw) + } + if rejected { + i.stats.rejected++ + i.mu.Unlock() + return catalystHTTPResponse(http.StatusRequestEntityTooLarge, `{"errors":["Payload too large"]}`), nil + } + + var payload datadogV2.MetricPayload + if err := json.Unmarshal(raw, &payload); err != nil { + i.mu.Unlock() + return nil, err + } + i.stats.accepted++ + i.stats.acceptedSeries = append(i.stats.acceptedSeries, payload.Series...) + i.mu.Unlock() + + return catalystHTTPResponse(http.StatusAccepted, `{"errors":[]}`), nil +} + +func catalystHTTPResponse(status int, body string) *http.Response { + return &http.Response{ + StatusCode: status, + Status: fmt.Sprintf("%d %s", status, http.StatusText(status)), + Header: http.Header{ + "Content-Type": []string{"application/json"}, + }, + Body: io.NopCloser(bytes.NewBufferString(body)), + } +} + +func (i *catalystIntake) snapshot() catalystIntakeStats { + i.mu.Lock() + defer i.mu.Unlock() + stats := i.stats + stats.acceptedSeries = append([]datadogV2.MetricSeries(nil), i.stats.acceptedSeries...) + return stats +} + +func catalystMetricSeries() []datadogV2.MetricSeries { + return catalystMetricSeriesSized(catalystSizeTableSeries, catalystWaitTableRows, catalystFixedSeries) +} + +func catalystMetricSeriesSized(sizeTableSeries, waitTableRows, fixedSeries int) []datadogV2.MetricSeries { + commonTags := []string{ + "infrastructure_type:aurora", + "org:database", + "team:aurora-mysql", + "cost_center:1234", + "service:doppler", + "env:production", + "blip_metric_version:2", + "metacluster:catalyst-production-aurora", + "cluster:catalyst-production-aurora-usw2-001", + "app:blippy-catalyst-production-aurora-001", + "dc:us-west-2", + "region:us-west-2", + "odsgroup:catalyst-production-aurora-001", + "hostname:catalyst-production-aurora-001.cluster-abcdefghijkl.us-west-2.rds.amazonaws.com", + "host:catalyst-production-aurora-001-instance-1", + "physical_host:unknown", + "dbtype:mysql", + "storage_type:aurora", + "channel:stable", + } + resources := []datadogV2.MetricResource{{ + Name: datadog.PtrString("catalyst-production-aurora-001-instance-1"), + Type: datadog.PtrString("host"), + }} + + series := make([]datadogV2.MetricSeries, 0, sizeTableSeries+waitTableRows*catalystWaitMetrics+fixedSeries) + appendTableSeries := func(metric string, metricType datadogV2.MetricIntakeType, table int) { + tags := make([]string, len(commonTags), len(commonTags)+2) + copy(tags, commonTags) + tags = append(tags, + fmt.Sprintf("db:database_%03d", table%160), + fmt.Sprintf("tbl:application_table_%05d", table), + ) + series = append(series, datadogV2.MetricSeries{ + Metric: metric, + Type: metricType.Ptr(), + Points: []datadogV2.MetricPoint{{ + Timestamp: datadog.PtrInt64(1_784_396_000), + Value: datadog.PtrFloat64(float64(table % 1000)), + }}, + Resources: resources, + Tags: tags, + }) + } + + for table := 0; table < sizeTableSeries; table++ { + appendTableSeries("mysql.size.table.bytes", datadogV2.METRICINTAKETYPE_GAUGE, table) + } + + waitMetrics := []string{ + "mysql.wait.io.table.count_fetch", + "mysql.wait.io.table.time_fetch", + "mysql.wait.io.table.count_insert", + "mysql.wait.io.table.time_insert", + } + for table := 0; table < waitTableRows; table++ { + for _, metric := range waitMetrics { + appendTableSeries(metric, datadogV2.METRICINTAKETYPE_COUNT, table) + } + } + + for metric := 0; metric < fixedSeries; metric++ { + series = append(series, datadogV2.MetricSeries{ + Metric: fmt.Sprintf("mysql.status.synthetic_%03d", metric), + Type: datadogV2.METRICINTAKETYPE_GAUGE.Ptr(), + Points: []datadogV2.MetricPoint{{ + Timestamp: datadog.PtrInt64(1_784_396_000), + Value: datadog.PtrFloat64(float64(metric)), + }}, + Resources: resources, + Tags: commonTags, + }) + } + + return series +} + +func catalystMetricsMetadata() *blip.Metrics { + return &blip.Metrics{ + MonitorId: "catalyst-production-aurora-001-instance-1", + Plan: "dd-mysql", + Level: "data-size", + Interval: 1, + } +} + +func catalystAPIClient(intake *catalystIntake) *datadog.APIClient { + cfg := datadog.NewConfiguration() + cfg.HTTPClient = &http.Client{Transport: intake} + cfg.Compress = true + return datadog.NewAPIClient(cfg) +} + +// legacyDatadogSender is the previous reactive autosizer, retained only in the +// test harness so old and new behavior can be measured against the same data. +type legacyDatadogSender struct { + api *datadogV2.MetricsApi + maxMetricsPerRequest int + maxPayloadSize int +} + +func newLegacyDatadogSender(intake *catalystIntake) *legacyDatadogSender { + return &legacyDatadogSender{ + api: datadogV2.NewMetricsApi(catalystAPIClient(intake)), + maxMetricsPerRequest: math.MaxInt32, + maxPayloadSize: datadogMaxCompressedPayloadSize, + } +} + +func (s *legacyDatadogSender) Send(ctx context.Context, series []datadogV2.MetricSeries) error { + ctx = context.WithValue(ctx, datadog.ContextAPIKeys, map[string]datadog.APIKey{ + "apiKeyAuth": {Key: "test-api-key"}, + }) + localMax := s.maxMetricsPerRequest + + for start := 0; start < len(series); { + end := start + localMax + if end > len(series) { + end = len(series) + } + + options := *datadogV2.NewSubmitMetricsOptionalParameters() + options.ContentEncoding = datadogV2.METRICCONTENTENCODING_GZIP.Ptr() + _, response, err := s.api.SubmitMetrics(ctx, *datadogV2.NewMetricPayload(series[start:end]), options) + if err != nil { + if response == nil || response.StatusCode != http.StatusRequestEntityTooLarge { + return err + } + if localMax == 1 { + return err + } + localMax, err = s.estimateMaxMetricsPerRequest(series[start:end], localMax) + if err != nil { + return err + } + continue + } + start = end + } + + if localMax < s.maxMetricsPerRequest { + s.maxMetricsPerRequest = localMax + } + return nil +} + +func (s *legacyDatadogSender) estimateMaxMetricsPerRequest(series []datadogV2.MetricSeries, currentMax int) (int, error) { + data, err := json.Marshal(series) + if err != nil { + return 0, err + } + + var compressed bytes.Buffer + w := zlib.NewWriter(&compressed) + if _, err := w.Write(data); err != nil { + return 0, err + } + if err := w.Close(); err != nil { + return 0, err + } + + estimatedMetricSize := compressed.Len() / len(series) + if estimatedMetricSize == 0 { + return 0, fmt.Errorf("legacy estimator produced a zero-byte metric estimate") + } + estimatedMax := (s.maxPayloadSize - 300) / estimatedMetricSize + if estimatedMax >= currentMax { + estimatedMax = int(float32(currentMax) * 0.9) + } + if estimatedMax < 1 { + estimatedMax = 1 + } + return estimatedMax, nil +} + +func newCatalystDatadogSender(intake *catalystIntake) *Datadog { + sink := &Datadog{ + monitorId: "catalyst-production-aurora-001-instance-1", + event: event.MonitorReceiver{MonitorId: "catalyst-production-aurora-001-instance-1"}, + compress: true, + payloadLimits: defaultDatadogPayloadLimits(), + submitter: &datadogAPISubmitter{ + client: catalystAPIClient(intake), + apiKey: "test-api-key", + }, + } + sink.maxSeriesPerRequest.Store(int64(sink.payloadLimits.maxSeries)) + return sink +} + +func TestDatadogCatalystHarnessEquivalent(t *testing.T) { + if os.Getenv("BLIP_CATALYST_HARNESS") != "1" { + t.Skip("set BLIP_CATALYST_HARNESS=1 to run the full 28,255-series comparison") + } + series := catalystMetricSeries() + require.Len(t, series, catalystTotalSeries) + + legacyIntake := newCatalystIntake() + legacy := newLegacyDatadogSender(legacyIntake) + require.NoError(t, legacy.Send(context.Background(), series)) + legacyStats := legacyIntake.snapshot() + + newIntake := newCatalystIntake() + modern := newCatalystDatadogSender(newIntake) + require.NoError(t, modern.sendAPI(context.Background(), series, catalystMetricsMetadata())) + newStats := newIntake.snapshot() + + legacyJSON, err := json.Marshal(legacyStats.acceptedSeries) + require.NoError(t, err) + newJSON, err := json.Marshal(newStats.acceptedSeries) + require.NoError(t, err) + require.Equal(t, legacyJSON, newJSON, "old and new senders must submit identical series and points") + + require.Equal(t, catalystTotalSeries, len(legacyStats.acceptedSeries)) + require.Equal(t, catalystTotalSeries, len(newStats.acceptedSeries)) + require.Greater(t, legacyStats.rejected, 0, "legacy harness should exercise reactive 413 autosizing") + require.Zero(t, newStats.rejected, "new sender should size payloads before submission") + require.Less(t, newStats.requests, legacyStats.requests) + require.Less(t, newStats.maxRaw, datadogMaxDecompressedPayloadSize) + require.LessOrEqual(t, newStats.maxWire, datadogMaxCompressedPayloadSize) + + t.Logf("legacy: requests=%d rejected=%d accepted=%d total-wire=%d total-raw=%d max-wire=%d max-raw=%d", + legacyStats.requests, legacyStats.rejected, legacyStats.accepted, + legacyStats.totalWire, legacyStats.totalRaw, legacyStats.maxWire, legacyStats.maxRaw) + t.Logf("new: requests=%d rejected=%d accepted=%d total-wire=%d total-raw=%d max-wire=%d max-raw=%d", + newStats.requests, newStats.rejected, newStats.accepted, + newStats.totalWire, newStats.totalRaw, newStats.maxWire, newStats.maxRaw) +} + +func TestDatadogPayloadAlgorithmsEquivalent(t *testing.T) { + series := catalystMetricSeriesSized(1_000, 0, 0) + + legacyIntake := newCatalystIntake() + legacyIntake.maxCompressed = 50_000 + legacyIntake.maxDecompressed = 150_000 + legacy := newLegacyDatadogSender(legacyIntake) + legacy.maxPayloadSize = legacyIntake.maxCompressed + require.NoError(t, legacy.Send(context.Background(), series)) + legacyStats := legacyIntake.snapshot() + + newIntake := newCatalystIntake() + newIntake.maxCompressed = legacyIntake.maxCompressed + newIntake.maxDecompressed = legacyIntake.maxDecompressed + modern := newCatalystDatadogSender(newIntake) + modern.payloadLimits = datadogPayloadLimits{ + maxCompressed: newIntake.maxCompressed, + maxDecompressed: newIntake.maxDecompressed, + targetCompressed: newIntake.maxCompressed * 9 / 10, + targetDecompressed: newIntake.maxDecompressed * 9 / 10, + maxSeries: len(series), + } + modern.maxSeriesPerRequest.Store(int64(len(series))) + require.NoError(t, modern.sendAPI(context.Background(), series, catalystMetricsMetadata())) + newStats := newIntake.snapshot() + + legacyJSON, err := json.Marshal(legacyStats.acceptedSeries) + require.NoError(t, err) + newJSON, err := json.Marshal(newStats.acceptedSeries) + require.NoError(t, err) + require.Equal(t, legacyJSON, newJSON) + require.Greater(t, legacyStats.rejected, 0) + require.Zero(t, newStats.rejected) +} + +func BenchmarkDatadogCatalystPayload(b *testing.B) { + series := catalystMetricSeries() + metadata := catalystMetricsMetadata() + + b.Run("legacy-reactive-autosizer", func(b *testing.B) { + b.ReportAllocs() + var totals catalystIntakeStats + for n := 0; n < b.N; n++ { + b.StopTimer() + intake := newCatalystIntake() + sender := newLegacyDatadogSender(intake) + b.StartTimer() + + if err := sender.Send(context.Background(), series); err != nil { + b.Fatal(err) + } + + b.StopTimer() + stats := intake.snapshot() + totals.requests += stats.requests + totals.rejected += stats.rejected + totals.accepted += stats.accepted + totals.totalWire += stats.totalWire + totals.totalRaw += stats.totalRaw + b.StartTimer() + } + b.ReportMetric(float64(totals.requests)/float64(b.N), "requests/op") + b.ReportMetric(float64(totals.rejected)/float64(b.N), "rejected/op") + b.ReportMetric(float64(totals.accepted)/float64(b.N), "accepted/op") + b.ReportMetric(float64(len(series)), "series/op") + b.ReportMetric(float64(totals.totalWire)/float64(b.N), "wire-bytes/op") + b.ReportMetric(float64(totals.totalRaw)/float64(b.N), "raw-bytes/op") + }) + + b.Run("byte-budgeted-prepared-payload", func(b *testing.B) { + b.ReportAllocs() + var totals catalystIntakeStats + for n := 0; n < b.N; n++ { + b.StopTimer() + intake := newCatalystIntake() + sender := newCatalystDatadogSender(intake) + b.StartTimer() + + if err := sender.sendAPI(context.Background(), series, metadata); err != nil { + b.Fatal(err) + } + + b.StopTimer() + stats := intake.snapshot() + totals.requests += stats.requests + totals.rejected += stats.rejected + totals.accepted += stats.accepted + totals.totalWire += stats.totalWire + totals.totalRaw += stats.totalRaw + b.StartTimer() + } + b.ReportMetric(float64(totals.requests)/float64(b.N), "requests/op") + b.ReportMetric(float64(totals.rejected)/float64(b.N), "rejected/op") + b.ReportMetric(float64(totals.accepted)/float64(b.N), "accepted/op") + b.ReportMetric(float64(len(series)), "series/op") + b.ReportMetric(float64(totals.totalWire)/float64(b.N), "wire-bytes/op") + b.ReportMetric(float64(totals.totalRaw)/float64(b.N), "raw-bytes/op") + }) +} diff --git a/sink/datadog_test.go b/sink/datadog_test.go index 8440399a..3c5b1be8 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.") } From 2590d88cad9827afbe8bf3001581acb62c141874 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Sat, 18 Jul 2026 14:04:08 -0700 Subject: [PATCH 2/3] Checkpoint Datadog streaming submissions Co-authored-by: Codex Ai-assisted: true --- sink/datadog.go | 14 ++ sink/datadog_api_test.go | 66 +++++++ sink/datadog_catalyst_test.go | 178 +++++++++++++++++-- sink/datadog_stream.go | 315 ++++++++++++++++++++++++++++++++++ sink/retry.go | 32 +++- sink/retry_test.go | 2 +- 6 files changed, 583 insertions(+), 24 deletions(-) create mode 100644 sink/datadog_stream.go diff --git a/sink/datadog.go b/sink/datadog.go index b3f719d0..7c1895cc 100644 --- a/sink/datadog.go +++ b/sink/datadog.go @@ -8,6 +8,7 @@ import ( "net/http" "os" "regexp" + "sort" "strconv" "sync/atomic" "time" @@ -69,6 +70,7 @@ func NewDatadog(monitorId string, opts, tags map[string]string, httpClient *http } } } + sort.Strings(tagList) d := &Datadog{ monitorId: monitorId, @@ -167,6 +169,10 @@ func NewDatadog(monitorId string, opts, tags map[string]string, httpClient *http } 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) @@ -181,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) } @@ -322,6 +332,10 @@ func (s *Datadog) Send(ctx context.Context, m *blip.Metrics) error { return s.sendAPI(ctx, dp[:n], m) } +func (s *Datadog) sendDogStatsD(ctx context.Context, m *blip.Metrics) error { + return s.send(ctx, m, false) +} + func (s *Datadog) Name() string { return "datadog" } diff --git a/sink/datadog_api_test.go b/sink/datadog_api_test.go index 61e414f0..584c7562 100644 --- a/sink/datadog_api_test.go +++ b/sink/datadog_api_test.go @@ -9,15 +9,45 @@ import ( "io" "net/http" "testing" + "time" "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 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") + } + + var decoded datadogV2.MetricPayload + if err := json.Unmarshal(payload.body, &decoded); 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 (s *rejectThenFailSubmitter) Submit(context.Context, preparedDatadogPayload) (datadogSubmitResult, error) { s.calls++ if s.calls == 1 { @@ -95,3 +125,39 @@ func TestPrepareDatadogPayloadStopsOnCancellation(t *testing.T) { _, _, 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_catalyst_test.go b/sink/datadog_catalyst_test.go index 50d669e2..feaf0385 100644 --- a/sink/datadog_catalyst_test.go +++ b/sink/datadog_catalyst_test.go @@ -11,8 +11,11 @@ import ( "math" "net/http" "os" + "sort" + "strings" "sync" "testing" + "time" "github.com/DataDog/datadog-api-client-go/v2/api/datadog" "github.com/DataDog/datadog-api-client-go/v2/api/datadogV2" @@ -134,8 +137,8 @@ func catalystMetricSeries() []datadogV2.MetricSeries { return catalystMetricSeriesSized(catalystSizeTableSeries, catalystWaitTableRows, catalystFixedSeries) } -func catalystMetricSeriesSized(sizeTableSeries, waitTableRows, fixedSeries int) []datadogV2.MetricSeries { - commonTags := []string{ +func catalystCommonTags() []string { + return []string{ "infrastructure_type:aurora", "org:database", "team:aurora-mysql", @@ -156,10 +159,18 @@ func catalystMetricSeriesSized(sizeTableSeries, waitTableRows, fixedSeries int) "storage_type:aurora", "channel:stable", } - resources := []datadogV2.MetricResource{{ +} + +func catalystResources() []datadogV2.MetricResource { + return []datadogV2.MetricResource{{ Name: datadog.PtrString("catalyst-production-aurora-001-instance-1"), Type: datadog.PtrString("host"), }} +} + +func catalystMetricSeriesSized(sizeTableSeries, waitTableRows, fixedSeries int) []datadogV2.MetricSeries { + commonTags := catalystCommonTags() + resources := catalystResources() series := make([]datadogV2.MetricSeries, 0, sizeTableSeries+waitTableRows*catalystWaitMetrics+fixedSeries) appendTableSeries := func(metric string, metricType datadogV2.MetricIntakeType, table int) { @@ -222,6 +233,81 @@ func catalystMetricsMetadata() *blip.Metrics { } } +func catalystBlipMetrics(series []datadogV2.MetricSeries) *blip.Metrics { + metrics := catalystMetricsMetadata() + metrics.Begin = time.Unix(1_784_396_000, 0) + metrics.End = metrics.Begin.Add(time.Second) + metrics.Values = map[string][]blip.MetricValue{} + commonTagCount := len(catalystCommonTags()) + + for _, item := range series { + separator := strings.LastIndexByte(item.Metric, '.') + if separator < 1 || separator == len(item.Metric)-1 { + panic("invalid Catalyst metric name: " + item.Metric) + } + metricType := blip.GAUGE + if item.Type != nil && *item.Type == datadogV2.METRICINTAKETYPE_COUNT { + metricType = blip.DELTA_COUNTER + } + value := blip.MetricValue{ + Name: item.Metric[separator+1:], + Value: *item.Points[0].Value, + Type: metricType, + } + if len(item.Tags) > commonTagCount { + value.Group = map[string]string{} + for _, tag := range item.Tags[commonTagCount:] { + key, val, ok := strings.Cut(tag, ":") + if !ok { + panic("invalid Catalyst tag: " + tag) + } + value.Group[key] = val + } + } + domain := item.Metric[:separator] + metrics.Values[domain] = append(metrics.Values[domain], value) + } + return metrics +} + +func canonicalCatalystSeries(series []datadogV2.MetricSeries) []string { + encoded := make([]string, len(series)) + for i := range series { + item := series[i] + item.Tags = append([]string(nil), item.Tags...) + sort.Strings(item.Tags) + data, err := json.Marshal(item) + if err != nil { + panic(err) + } + encoded[i] = string(data) + } + sort.Strings(encoded) + return encoded +} + +// materializeDatadogSeries models the pre-streaming conversion path: convert +// the complete Blip batch into one MetricSeries slice before payload sizing. +func materializeDatadogSeries(sender *Datadog, metrics *blip.Metrics) []datadogV2.MetricSeries { + domains := make([]string, 0, len(metrics.Values)) + total := 0 + for domain, values := range metrics.Values { + domains = append(domains, domain) + total += len(values) + } + sort.Strings(domains) + series := make([]datadogV2.MetricSeries, 0, total) + for _, domain := range domains { + for _, value := range metrics.Values[domain] { + converted, ok := sender.datadogMetricSeries(metrics, domain, value) + if ok { + series = append(series, converted) + } + } + } + return series +} + func catalystAPIClient(intake *catalystIntake) *datadog.APIClient { cfg := datadog.NewConfiguration() cfg.HTTPClient = &http.Client{Transport: intake} @@ -316,6 +402,8 @@ func newCatalystDatadogSender(intake *catalystIntake) *Datadog { monitorId: "catalyst-production-aurora-001-instance-1", event: event.MonitorReceiver{MonitorId: "catalyst-production-aurora-001-instance-1"}, compress: true, + tags: catalystCommonTags(), + resources: catalystResources(), payloadLimits: defaultDatadogPayloadLimits(), submitter: &datadogAPISubmitter{ client: catalystAPIClient(intake), @@ -343,26 +431,38 @@ func TestDatadogCatalystHarnessEquivalent(t *testing.T) { require.NoError(t, modern.sendAPI(context.Background(), series, catalystMetricsMetadata())) newStats := newIntake.snapshot() - legacyJSON, err := json.Marshal(legacyStats.acceptedSeries) - require.NoError(t, err) - newJSON, err := json.Marshal(newStats.acceptedSeries) - require.NoError(t, err) - require.Equal(t, legacyJSON, newJSON, "old and new senders must submit identical series and points") + finalIntake := newCatalystIntake() + final := newCatalystDatadogSender(finalIntake) + realMetrics := catalystBlipMetrics(series) + require.NoError(t, final.Send(context.Background(), realMetrics)) + finalStats := finalIntake.snapshot() + + legacyCanonical := canonicalCatalystSeries(legacyStats.acceptedSeries) + require.Equal(t, legacyCanonical, canonicalCatalystSeries(newStats.acceptedSeries), "pre and mid senders must submit identical series and points") + require.Equal(t, legacyCanonical, canonicalCatalystSeries(finalStats.acceptedSeries), "pre and final senders must submit identical series and points") require.Equal(t, catalystTotalSeries, len(legacyStats.acceptedSeries)) require.Equal(t, catalystTotalSeries, len(newStats.acceptedSeries)) + require.Equal(t, catalystTotalSeries, len(finalStats.acceptedSeries)) require.Greater(t, legacyStats.rejected, 0, "legacy harness should exercise reactive 413 autosizing") require.Zero(t, newStats.rejected, "new sender should size payloads before submission") + require.Zero(t, finalStats.rejected, "streaming sender should size payloads before submission") require.Less(t, newStats.requests, legacyStats.requests) + require.Equal(t, newStats.requests, finalStats.requests, "bounded conversion should not add payloads") require.Less(t, newStats.maxRaw, datadogMaxDecompressedPayloadSize) require.LessOrEqual(t, newStats.maxWire, datadogMaxCompressedPayloadSize) + require.Less(t, finalStats.maxRaw, datadogMaxDecompressedPayloadSize) + require.LessOrEqual(t, finalStats.maxWire, datadogMaxCompressedPayloadSize) t.Logf("legacy: requests=%d rejected=%d accepted=%d total-wire=%d total-raw=%d max-wire=%d max-raw=%d", legacyStats.requests, legacyStats.rejected, legacyStats.accepted, legacyStats.totalWire, legacyStats.totalRaw, legacyStats.maxWire, legacyStats.maxRaw) - t.Logf("new: requests=%d rejected=%d accepted=%d total-wire=%d total-raw=%d max-wire=%d max-raw=%d", + t.Logf("mid: requests=%d rejected=%d accepted=%d total-wire=%d total-raw=%d max-wire=%d max-raw=%d", newStats.requests, newStats.rejected, newStats.accepted, newStats.totalWire, newStats.totalRaw, newStats.maxWire, newStats.maxRaw) + t.Logf("final: requests=%d rejected=%d accepted=%d total-wire=%d total-raw=%d max-wire=%d max-raw=%d", + finalStats.requests, finalStats.rejected, finalStats.accepted, + finalStats.totalWire, finalStats.totalRaw, finalStats.maxWire, finalStats.maxRaw) } func TestDatadogPayloadAlgorithmsEquivalent(t *testing.T) { @@ -391,18 +491,28 @@ func TestDatadogPayloadAlgorithmsEquivalent(t *testing.T) { require.NoError(t, modern.sendAPI(context.Background(), series, catalystMetricsMetadata())) newStats := newIntake.snapshot() - legacyJSON, err := json.Marshal(legacyStats.acceptedSeries) - require.NoError(t, err) - newJSON, err := json.Marshal(newStats.acceptedSeries) - require.NoError(t, err) - require.Equal(t, legacyJSON, newJSON) + finalIntake := newCatalystIntake() + finalIntake.maxCompressed = legacyIntake.maxCompressed + finalIntake.maxDecompressed = legacyIntake.maxDecompressed + final := newCatalystDatadogSender(finalIntake) + final.payloadLimits = modern.payloadLimits + final.maxSeriesPerRequest.Store(int64(len(series))) + require.NoError(t, final.Send(context.Background(), catalystBlipMetrics(series))) + finalStats := finalIntake.snapshot() + + legacyCanonical := canonicalCatalystSeries(legacyStats.acceptedSeries) + require.Equal(t, legacyCanonical, canonicalCatalystSeries(newStats.acceptedSeries)) + require.Equal(t, legacyCanonical, canonicalCatalystSeries(finalStats.acceptedSeries)) require.Greater(t, legacyStats.rejected, 0) require.Zero(t, newStats.rejected) + require.Zero(t, finalStats.rejected) + require.Equal(t, newStats.requests, finalStats.requests) } func BenchmarkDatadogCatalystPayload(b *testing.B) { series := catalystMetricSeries() metadata := catalystMetricsMetadata() + realMetrics := catalystBlipMetrics(series) b.Run("legacy-reactive-autosizer", func(b *testing.B) { b.ReportAllocs() @@ -411,9 +521,11 @@ func BenchmarkDatadogCatalystPayload(b *testing.B) { b.StopTimer() intake := newCatalystIntake() sender := newLegacyDatadogSender(intake) + converter := newCatalystDatadogSender(intake) b.StartTimer() - if err := sender.Send(context.Background(), series); err != nil { + materialized := materializeDatadogSeries(converter, realMetrics) + if err := sender.Send(context.Background(), materialized); err != nil { b.Fatal(err) } @@ -430,6 +542,7 @@ func BenchmarkDatadogCatalystPayload(b *testing.B) { b.ReportMetric(float64(totals.rejected)/float64(b.N), "rejected/op") b.ReportMetric(float64(totals.accepted)/float64(b.N), "accepted/op") b.ReportMetric(float64(len(series)), "series/op") + b.ReportMetric(float64(len(series)), "max-converted-series/op") b.ReportMetric(float64(totals.totalWire)/float64(b.N), "wire-bytes/op") b.ReportMetric(float64(totals.totalRaw)/float64(b.N), "raw-bytes/op") }) @@ -443,7 +556,39 @@ func BenchmarkDatadogCatalystPayload(b *testing.B) { sender := newCatalystDatadogSender(intake) b.StartTimer() - if err := sender.sendAPI(context.Background(), series, metadata); err != nil { + materialized := materializeDatadogSeries(sender, realMetrics) + if err := sender.sendAPI(context.Background(), materialized, metadata); err != nil { + b.Fatal(err) + } + + b.StopTimer() + stats := intake.snapshot() + totals.requests += stats.requests + totals.rejected += stats.rejected + totals.accepted += stats.accepted + totals.totalWire += stats.totalWire + totals.totalRaw += stats.totalRaw + b.StartTimer() + } + b.ReportMetric(float64(totals.requests)/float64(b.N), "requests/op") + b.ReportMetric(float64(totals.rejected)/float64(b.N), "rejected/op") + b.ReportMetric(float64(totals.accepted)/float64(b.N), "accepted/op") + b.ReportMetric(float64(len(series)), "series/op") + b.ReportMetric(float64(len(series)), "max-converted-series/op") + b.ReportMetric(float64(totals.totalWire)/float64(b.N), "wire-bytes/op") + b.ReportMetric(float64(totals.totalRaw)/float64(b.N), "raw-bytes/op") + }) + + b.Run("checkpointed-streaming-conversion", func(b *testing.B) { + b.ReportAllocs() + var totals catalystIntakeStats + for n := 0; n < b.N; n++ { + b.StopTimer() + intake := newCatalystIntake() + sender := newCatalystDatadogSender(intake) + b.StartTimer() + + if err := sender.Send(context.Background(), realMetrics); err != nil { b.Fatal(err) } @@ -460,6 +605,7 @@ func BenchmarkDatadogCatalystPayload(b *testing.B) { b.ReportMetric(float64(totals.rejected)/float64(b.N), "rejected/op") b.ReportMetric(float64(totals.accepted)/float64(b.N), "accepted/op") b.ReportMetric(float64(len(series)), "series/op") + b.ReportMetric(float64(defaultDatadogPayloadLimits().maxSeries), "max-converted-series/op") b.ReportMetric(float64(totals.totalWire)/float64(b.N), "wire-bytes/op") b.ReportMetric(float64(totals.totalRaw)/float64(b.N), "raw-bytes/op") }) diff --git a/sink/datadog_stream.go b/sink/datadog_stream.go new file mode 100644 index 00000000..8e47c854 --- /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/retry.go b/sink/retry.go index 75646e76..31d80fb6 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 3f011d9f..4d9733e6 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 From 994e8b01ed39f5d578aaad6938b5ae2097b49f6b Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Sat, 18 Jul 2026 14:23:01 -0700 Subject: [PATCH 3/3] Remove Catalyst comparison harness Co-authored-by: Codex Ai-assisted: true --- sink/datadog_api_test.go | 115 ++++++- sink/datadog_catalyst_test.go | 612 ---------------------------------- 2 files changed, 99 insertions(+), 628 deletions(-) delete mode 100644 sink/datadog_catalyst_test.go diff --git a/sink/datadog_api_test.go b/sink/datadog_api_test.go index 584c7562..fc0e0d40 100644 --- a/sink/datadog_api_test.go +++ b/sink/datadog_api_test.go @@ -6,11 +6,14 @@ import ( "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" @@ -21,6 +24,29 @@ 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 @@ -33,8 +59,8 @@ func (s *checkpointFailureSubmitter) Submit(_ context.Context, payload preparedD return datadogSubmitResult{}, errors.New("injected checkpoint failure") } - var decoded datadogV2.MetricPayload - if err := json.Unmarshal(payload.body, &decoded); err != nil { + decoded, err := decodePreparedMetricPayload(payload) + if err != nil { return datadogSubmitResult{}, err } if s.acceptedNames == nil { @@ -48,6 +74,63 @@ func (s *checkpointFailureSubmitter) Submit(_ context.Context, payload preparedD 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 { @@ -57,7 +140,7 @@ func (s *rejectThenFailSubmitter) Submit(context.Context, preparedDatadogPayload } func TestPrepareDatadogPayloadHonorsCompressedLimit(t *testing.T) { - series := catalystMetricSeriesSized(1_000, 0, 0) + series := testMetricSeries(1_000) limits := datadogPayloadLimits{ maxCompressed: 1_000, maxDecompressed: 1_000_000, @@ -85,24 +168,24 @@ func TestPrepareDatadogPayloadHonorsCompressedLimit(t *testing.T) { } func TestDatadog413FallbackHalvesAndPersistsLimit(t *testing.T) { - series := catalystMetricSeriesSized(1_000, 0, 0) - intake := newCatalystIntake() + 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. - intake.maxDecompressed = 150_000 - - sender := newCatalystDatadogSender(intake) - require.NoError(t, sender.sendAPI(context.Background(), series, catalystMetricsMetadata())) + sender := newTestDatadogSender(intake) + require.NoError(t, sender.sendAPI(context.Background(), series, testMetricsMetadata())) - stats := intake.snapshot() - require.Greater(t, stats.rejected, 0) - require.Len(t, stats.acceptedSeries, len(series)) + require.Greater(t, intake.rejected, 0) + require.Len(t, intake.acceptedSeries, len(series)) require.Less(t, sender.maxSeriesPerRequest.Load(), int64(len(series))) - require.Equal(t, stats.requests, stats.rejected+stats.accepted) + require.Greater(t, intake.requests, intake.rejected) } func TestDatadog413LimitSurvivesLaterFailure(t *testing.T) { - series := catalystMetricSeriesSized(10, 0, 0) + series := testMetricSeries(10) submitter := &rejectThenFailSubmitter{} sender := &Datadog{ compress: true, @@ -111,7 +194,7 @@ func TestDatadog413LimitSurvivesLaterFailure(t *testing.T) { } sender.maxSeriesPerRequest.Store(int64(len(series))) - err := sender.sendAPI(context.Background(), series, catalystMetricsMetadata()) + 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()) @@ -121,7 +204,7 @@ func TestPrepareDatadogPayloadStopsOnCancellation(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - series := catalystMetricSeriesSized(10, 0, 0) + series := testMetricSeries(10) _, _, err := prepareDatadogPayload(ctx, series, 0, 10, true, defaultDatadogPayloadLimits()) require.ErrorIs(t, err, context.Canceled) } diff --git a/sink/datadog_catalyst_test.go b/sink/datadog_catalyst_test.go deleted file mode 100644 index feaf0385..00000000 --- a/sink/datadog_catalyst_test.go +++ /dev/null @@ -1,612 +0,0 @@ -package sink - -import ( - "bytes" - "compress/gzip" - "compress/zlib" - "context" - "encoding/json" - "fmt" - "io" - "math" - "net/http" - "os" - "sort" - "strings" - "sync" - "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" - "github.com/cashapp/blip/event" -) - -const ( - catalystSizeTableSeries = 9_155 - catalystWaitTableRows = 4_750 - catalystWaitMetrics = 4 - catalystFixedSeries = 100 - catalystTotalSeries = catalystSizeTableSeries + catalystWaitTableRows*catalystWaitMetrics + catalystFixedSeries -) - -type catalystIntakeStats struct { - requests int - rejected int - accepted int - totalWire int64 - totalRaw int64 - maxWire int - maxRaw int - acceptedSeries []datadogV2.MetricSeries -} - -// catalystIntake simulates the two independent Datadog v2 series limits and -// records only successfully accepted payloads. It is shared by the correctness -// test and benchmarks so both implementations see identical intake behavior. -type catalystIntake struct { - mu sync.Mutex - maxCompressed int - maxDecompressed int - stats catalystIntakeStats -} - -func newCatalystIntake() *catalystIntake { - return &catalystIntake{ - maxCompressed: datadogMaxCompressedPayloadSize, - maxDecompressed: datadogMaxDecompressedPayloadSize, - } -} - -func (i *catalystIntake) RoundTrip(req *http.Request) (*http.Response, error) { - wire, err := io.ReadAll(req.Body) - if err != nil { - return nil, err - } - - raw := wire - compressed := req.Header.Get("Content-Encoding") == "gzip" - if compressed { - reader, err := gzip.NewReader(bytes.NewReader(wire)) - if err != nil { - return nil, err - } - raw, err = io.ReadAll(reader) - closeErr := reader.Close() - if err != nil { - return nil, err - } - if closeErr != nil { - return nil, closeErr - } - } - - rejected := len(wire) > i.maxCompressed || (compressed && len(raw) >= i.maxDecompressed) - - i.mu.Lock() - i.stats.requests++ - i.stats.totalWire += int64(len(wire)) - i.stats.totalRaw += int64(len(raw)) - if len(wire) > i.stats.maxWire { - i.stats.maxWire = len(wire) - } - if len(raw) > i.stats.maxRaw { - i.stats.maxRaw = len(raw) - } - if rejected { - i.stats.rejected++ - i.mu.Unlock() - return catalystHTTPResponse(http.StatusRequestEntityTooLarge, `{"errors":["Payload too large"]}`), nil - } - - var payload datadogV2.MetricPayload - if err := json.Unmarshal(raw, &payload); err != nil { - i.mu.Unlock() - return nil, err - } - i.stats.accepted++ - i.stats.acceptedSeries = append(i.stats.acceptedSeries, payload.Series...) - i.mu.Unlock() - - return catalystHTTPResponse(http.StatusAccepted, `{"errors":[]}`), nil -} - -func catalystHTTPResponse(status int, body string) *http.Response { - return &http.Response{ - StatusCode: status, - Status: fmt.Sprintf("%d %s", status, http.StatusText(status)), - Header: http.Header{ - "Content-Type": []string{"application/json"}, - }, - Body: io.NopCloser(bytes.NewBufferString(body)), - } -} - -func (i *catalystIntake) snapshot() catalystIntakeStats { - i.mu.Lock() - defer i.mu.Unlock() - stats := i.stats - stats.acceptedSeries = append([]datadogV2.MetricSeries(nil), i.stats.acceptedSeries...) - return stats -} - -func catalystMetricSeries() []datadogV2.MetricSeries { - return catalystMetricSeriesSized(catalystSizeTableSeries, catalystWaitTableRows, catalystFixedSeries) -} - -func catalystCommonTags() []string { - return []string{ - "infrastructure_type:aurora", - "org:database", - "team:aurora-mysql", - "cost_center:1234", - "service:doppler", - "env:production", - "blip_metric_version:2", - "metacluster:catalyst-production-aurora", - "cluster:catalyst-production-aurora-usw2-001", - "app:blippy-catalyst-production-aurora-001", - "dc:us-west-2", - "region:us-west-2", - "odsgroup:catalyst-production-aurora-001", - "hostname:catalyst-production-aurora-001.cluster-abcdefghijkl.us-west-2.rds.amazonaws.com", - "host:catalyst-production-aurora-001-instance-1", - "physical_host:unknown", - "dbtype:mysql", - "storage_type:aurora", - "channel:stable", - } -} - -func catalystResources() []datadogV2.MetricResource { - return []datadogV2.MetricResource{{ - Name: datadog.PtrString("catalyst-production-aurora-001-instance-1"), - Type: datadog.PtrString("host"), - }} -} - -func catalystMetricSeriesSized(sizeTableSeries, waitTableRows, fixedSeries int) []datadogV2.MetricSeries { - commonTags := catalystCommonTags() - resources := catalystResources() - - series := make([]datadogV2.MetricSeries, 0, sizeTableSeries+waitTableRows*catalystWaitMetrics+fixedSeries) - appendTableSeries := func(metric string, metricType datadogV2.MetricIntakeType, table int) { - tags := make([]string, len(commonTags), len(commonTags)+2) - copy(tags, commonTags) - tags = append(tags, - fmt.Sprintf("db:database_%03d", table%160), - fmt.Sprintf("tbl:application_table_%05d", table), - ) - series = append(series, datadogV2.MetricSeries{ - Metric: metric, - Type: metricType.Ptr(), - Points: []datadogV2.MetricPoint{{ - Timestamp: datadog.PtrInt64(1_784_396_000), - Value: datadog.PtrFloat64(float64(table % 1000)), - }}, - Resources: resources, - Tags: tags, - }) - } - - for table := 0; table < sizeTableSeries; table++ { - appendTableSeries("mysql.size.table.bytes", datadogV2.METRICINTAKETYPE_GAUGE, table) - } - - waitMetrics := []string{ - "mysql.wait.io.table.count_fetch", - "mysql.wait.io.table.time_fetch", - "mysql.wait.io.table.count_insert", - "mysql.wait.io.table.time_insert", - } - for table := 0; table < waitTableRows; table++ { - for _, metric := range waitMetrics { - appendTableSeries(metric, datadogV2.METRICINTAKETYPE_COUNT, table) - } - } - - for metric := 0; metric < fixedSeries; metric++ { - series = append(series, datadogV2.MetricSeries{ - Metric: fmt.Sprintf("mysql.status.synthetic_%03d", metric), - Type: datadogV2.METRICINTAKETYPE_GAUGE.Ptr(), - Points: []datadogV2.MetricPoint{{ - Timestamp: datadog.PtrInt64(1_784_396_000), - Value: datadog.PtrFloat64(float64(metric)), - }}, - Resources: resources, - Tags: commonTags, - }) - } - - return series -} - -func catalystMetricsMetadata() *blip.Metrics { - return &blip.Metrics{ - MonitorId: "catalyst-production-aurora-001-instance-1", - Plan: "dd-mysql", - Level: "data-size", - Interval: 1, - } -} - -func catalystBlipMetrics(series []datadogV2.MetricSeries) *blip.Metrics { - metrics := catalystMetricsMetadata() - metrics.Begin = time.Unix(1_784_396_000, 0) - metrics.End = metrics.Begin.Add(time.Second) - metrics.Values = map[string][]blip.MetricValue{} - commonTagCount := len(catalystCommonTags()) - - for _, item := range series { - separator := strings.LastIndexByte(item.Metric, '.') - if separator < 1 || separator == len(item.Metric)-1 { - panic("invalid Catalyst metric name: " + item.Metric) - } - metricType := blip.GAUGE - if item.Type != nil && *item.Type == datadogV2.METRICINTAKETYPE_COUNT { - metricType = blip.DELTA_COUNTER - } - value := blip.MetricValue{ - Name: item.Metric[separator+1:], - Value: *item.Points[0].Value, - Type: metricType, - } - if len(item.Tags) > commonTagCount { - value.Group = map[string]string{} - for _, tag := range item.Tags[commonTagCount:] { - key, val, ok := strings.Cut(tag, ":") - if !ok { - panic("invalid Catalyst tag: " + tag) - } - value.Group[key] = val - } - } - domain := item.Metric[:separator] - metrics.Values[domain] = append(metrics.Values[domain], value) - } - return metrics -} - -func canonicalCatalystSeries(series []datadogV2.MetricSeries) []string { - encoded := make([]string, len(series)) - for i := range series { - item := series[i] - item.Tags = append([]string(nil), item.Tags...) - sort.Strings(item.Tags) - data, err := json.Marshal(item) - if err != nil { - panic(err) - } - encoded[i] = string(data) - } - sort.Strings(encoded) - return encoded -} - -// materializeDatadogSeries models the pre-streaming conversion path: convert -// the complete Blip batch into one MetricSeries slice before payload sizing. -func materializeDatadogSeries(sender *Datadog, metrics *blip.Metrics) []datadogV2.MetricSeries { - domains := make([]string, 0, len(metrics.Values)) - total := 0 - for domain, values := range metrics.Values { - domains = append(domains, domain) - total += len(values) - } - sort.Strings(domains) - series := make([]datadogV2.MetricSeries, 0, total) - for _, domain := range domains { - for _, value := range metrics.Values[domain] { - converted, ok := sender.datadogMetricSeries(metrics, domain, value) - if ok { - series = append(series, converted) - } - } - } - return series -} - -func catalystAPIClient(intake *catalystIntake) *datadog.APIClient { - cfg := datadog.NewConfiguration() - cfg.HTTPClient = &http.Client{Transport: intake} - cfg.Compress = true - return datadog.NewAPIClient(cfg) -} - -// legacyDatadogSender is the previous reactive autosizer, retained only in the -// test harness so old and new behavior can be measured against the same data. -type legacyDatadogSender struct { - api *datadogV2.MetricsApi - maxMetricsPerRequest int - maxPayloadSize int -} - -func newLegacyDatadogSender(intake *catalystIntake) *legacyDatadogSender { - return &legacyDatadogSender{ - api: datadogV2.NewMetricsApi(catalystAPIClient(intake)), - maxMetricsPerRequest: math.MaxInt32, - maxPayloadSize: datadogMaxCompressedPayloadSize, - } -} - -func (s *legacyDatadogSender) Send(ctx context.Context, series []datadogV2.MetricSeries) error { - ctx = context.WithValue(ctx, datadog.ContextAPIKeys, map[string]datadog.APIKey{ - "apiKeyAuth": {Key: "test-api-key"}, - }) - localMax := s.maxMetricsPerRequest - - for start := 0; start < len(series); { - end := start + localMax - if end > len(series) { - end = len(series) - } - - options := *datadogV2.NewSubmitMetricsOptionalParameters() - options.ContentEncoding = datadogV2.METRICCONTENTENCODING_GZIP.Ptr() - _, response, err := s.api.SubmitMetrics(ctx, *datadogV2.NewMetricPayload(series[start:end]), options) - if err != nil { - if response == nil || response.StatusCode != http.StatusRequestEntityTooLarge { - return err - } - if localMax == 1 { - return err - } - localMax, err = s.estimateMaxMetricsPerRequest(series[start:end], localMax) - if err != nil { - return err - } - continue - } - start = end - } - - if localMax < s.maxMetricsPerRequest { - s.maxMetricsPerRequest = localMax - } - return nil -} - -func (s *legacyDatadogSender) estimateMaxMetricsPerRequest(series []datadogV2.MetricSeries, currentMax int) (int, error) { - data, err := json.Marshal(series) - if err != nil { - return 0, err - } - - var compressed bytes.Buffer - w := zlib.NewWriter(&compressed) - if _, err := w.Write(data); err != nil { - return 0, err - } - if err := w.Close(); err != nil { - return 0, err - } - - estimatedMetricSize := compressed.Len() / len(series) - if estimatedMetricSize == 0 { - return 0, fmt.Errorf("legacy estimator produced a zero-byte metric estimate") - } - estimatedMax := (s.maxPayloadSize - 300) / estimatedMetricSize - if estimatedMax >= currentMax { - estimatedMax = int(float32(currentMax) * 0.9) - } - if estimatedMax < 1 { - estimatedMax = 1 - } - return estimatedMax, nil -} - -func newCatalystDatadogSender(intake *catalystIntake) *Datadog { - sink := &Datadog{ - monitorId: "catalyst-production-aurora-001-instance-1", - event: event.MonitorReceiver{MonitorId: "catalyst-production-aurora-001-instance-1"}, - compress: true, - tags: catalystCommonTags(), - resources: catalystResources(), - payloadLimits: defaultDatadogPayloadLimits(), - submitter: &datadogAPISubmitter{ - client: catalystAPIClient(intake), - apiKey: "test-api-key", - }, - } - sink.maxSeriesPerRequest.Store(int64(sink.payloadLimits.maxSeries)) - return sink -} - -func TestDatadogCatalystHarnessEquivalent(t *testing.T) { - if os.Getenv("BLIP_CATALYST_HARNESS") != "1" { - t.Skip("set BLIP_CATALYST_HARNESS=1 to run the full 28,255-series comparison") - } - series := catalystMetricSeries() - require.Len(t, series, catalystTotalSeries) - - legacyIntake := newCatalystIntake() - legacy := newLegacyDatadogSender(legacyIntake) - require.NoError(t, legacy.Send(context.Background(), series)) - legacyStats := legacyIntake.snapshot() - - newIntake := newCatalystIntake() - modern := newCatalystDatadogSender(newIntake) - require.NoError(t, modern.sendAPI(context.Background(), series, catalystMetricsMetadata())) - newStats := newIntake.snapshot() - - finalIntake := newCatalystIntake() - final := newCatalystDatadogSender(finalIntake) - realMetrics := catalystBlipMetrics(series) - require.NoError(t, final.Send(context.Background(), realMetrics)) - finalStats := finalIntake.snapshot() - - legacyCanonical := canonicalCatalystSeries(legacyStats.acceptedSeries) - require.Equal(t, legacyCanonical, canonicalCatalystSeries(newStats.acceptedSeries), "pre and mid senders must submit identical series and points") - require.Equal(t, legacyCanonical, canonicalCatalystSeries(finalStats.acceptedSeries), "pre and final senders must submit identical series and points") - - require.Equal(t, catalystTotalSeries, len(legacyStats.acceptedSeries)) - require.Equal(t, catalystTotalSeries, len(newStats.acceptedSeries)) - require.Equal(t, catalystTotalSeries, len(finalStats.acceptedSeries)) - require.Greater(t, legacyStats.rejected, 0, "legacy harness should exercise reactive 413 autosizing") - require.Zero(t, newStats.rejected, "new sender should size payloads before submission") - require.Zero(t, finalStats.rejected, "streaming sender should size payloads before submission") - require.Less(t, newStats.requests, legacyStats.requests) - require.Equal(t, newStats.requests, finalStats.requests, "bounded conversion should not add payloads") - require.Less(t, newStats.maxRaw, datadogMaxDecompressedPayloadSize) - require.LessOrEqual(t, newStats.maxWire, datadogMaxCompressedPayloadSize) - require.Less(t, finalStats.maxRaw, datadogMaxDecompressedPayloadSize) - require.LessOrEqual(t, finalStats.maxWire, datadogMaxCompressedPayloadSize) - - t.Logf("legacy: requests=%d rejected=%d accepted=%d total-wire=%d total-raw=%d max-wire=%d max-raw=%d", - legacyStats.requests, legacyStats.rejected, legacyStats.accepted, - legacyStats.totalWire, legacyStats.totalRaw, legacyStats.maxWire, legacyStats.maxRaw) - t.Logf("mid: requests=%d rejected=%d accepted=%d total-wire=%d total-raw=%d max-wire=%d max-raw=%d", - newStats.requests, newStats.rejected, newStats.accepted, - newStats.totalWire, newStats.totalRaw, newStats.maxWire, newStats.maxRaw) - t.Logf("final: requests=%d rejected=%d accepted=%d total-wire=%d total-raw=%d max-wire=%d max-raw=%d", - finalStats.requests, finalStats.rejected, finalStats.accepted, - finalStats.totalWire, finalStats.totalRaw, finalStats.maxWire, finalStats.maxRaw) -} - -func TestDatadogPayloadAlgorithmsEquivalent(t *testing.T) { - series := catalystMetricSeriesSized(1_000, 0, 0) - - legacyIntake := newCatalystIntake() - legacyIntake.maxCompressed = 50_000 - legacyIntake.maxDecompressed = 150_000 - legacy := newLegacyDatadogSender(legacyIntake) - legacy.maxPayloadSize = legacyIntake.maxCompressed - require.NoError(t, legacy.Send(context.Background(), series)) - legacyStats := legacyIntake.snapshot() - - newIntake := newCatalystIntake() - newIntake.maxCompressed = legacyIntake.maxCompressed - newIntake.maxDecompressed = legacyIntake.maxDecompressed - modern := newCatalystDatadogSender(newIntake) - modern.payloadLimits = datadogPayloadLimits{ - maxCompressed: newIntake.maxCompressed, - maxDecompressed: newIntake.maxDecompressed, - targetCompressed: newIntake.maxCompressed * 9 / 10, - targetDecompressed: newIntake.maxDecompressed * 9 / 10, - maxSeries: len(series), - } - modern.maxSeriesPerRequest.Store(int64(len(series))) - require.NoError(t, modern.sendAPI(context.Background(), series, catalystMetricsMetadata())) - newStats := newIntake.snapshot() - - finalIntake := newCatalystIntake() - finalIntake.maxCompressed = legacyIntake.maxCompressed - finalIntake.maxDecompressed = legacyIntake.maxDecompressed - final := newCatalystDatadogSender(finalIntake) - final.payloadLimits = modern.payloadLimits - final.maxSeriesPerRequest.Store(int64(len(series))) - require.NoError(t, final.Send(context.Background(), catalystBlipMetrics(series))) - finalStats := finalIntake.snapshot() - - legacyCanonical := canonicalCatalystSeries(legacyStats.acceptedSeries) - require.Equal(t, legacyCanonical, canonicalCatalystSeries(newStats.acceptedSeries)) - require.Equal(t, legacyCanonical, canonicalCatalystSeries(finalStats.acceptedSeries)) - require.Greater(t, legacyStats.rejected, 0) - require.Zero(t, newStats.rejected) - require.Zero(t, finalStats.rejected) - require.Equal(t, newStats.requests, finalStats.requests) -} - -func BenchmarkDatadogCatalystPayload(b *testing.B) { - series := catalystMetricSeries() - metadata := catalystMetricsMetadata() - realMetrics := catalystBlipMetrics(series) - - b.Run("legacy-reactive-autosizer", func(b *testing.B) { - b.ReportAllocs() - var totals catalystIntakeStats - for n := 0; n < b.N; n++ { - b.StopTimer() - intake := newCatalystIntake() - sender := newLegacyDatadogSender(intake) - converter := newCatalystDatadogSender(intake) - b.StartTimer() - - materialized := materializeDatadogSeries(converter, realMetrics) - if err := sender.Send(context.Background(), materialized); err != nil { - b.Fatal(err) - } - - b.StopTimer() - stats := intake.snapshot() - totals.requests += stats.requests - totals.rejected += stats.rejected - totals.accepted += stats.accepted - totals.totalWire += stats.totalWire - totals.totalRaw += stats.totalRaw - b.StartTimer() - } - b.ReportMetric(float64(totals.requests)/float64(b.N), "requests/op") - b.ReportMetric(float64(totals.rejected)/float64(b.N), "rejected/op") - b.ReportMetric(float64(totals.accepted)/float64(b.N), "accepted/op") - b.ReportMetric(float64(len(series)), "series/op") - b.ReportMetric(float64(len(series)), "max-converted-series/op") - b.ReportMetric(float64(totals.totalWire)/float64(b.N), "wire-bytes/op") - b.ReportMetric(float64(totals.totalRaw)/float64(b.N), "raw-bytes/op") - }) - - b.Run("byte-budgeted-prepared-payload", func(b *testing.B) { - b.ReportAllocs() - var totals catalystIntakeStats - for n := 0; n < b.N; n++ { - b.StopTimer() - intake := newCatalystIntake() - sender := newCatalystDatadogSender(intake) - b.StartTimer() - - materialized := materializeDatadogSeries(sender, realMetrics) - if err := sender.sendAPI(context.Background(), materialized, metadata); err != nil { - b.Fatal(err) - } - - b.StopTimer() - stats := intake.snapshot() - totals.requests += stats.requests - totals.rejected += stats.rejected - totals.accepted += stats.accepted - totals.totalWire += stats.totalWire - totals.totalRaw += stats.totalRaw - b.StartTimer() - } - b.ReportMetric(float64(totals.requests)/float64(b.N), "requests/op") - b.ReportMetric(float64(totals.rejected)/float64(b.N), "rejected/op") - b.ReportMetric(float64(totals.accepted)/float64(b.N), "accepted/op") - b.ReportMetric(float64(len(series)), "series/op") - b.ReportMetric(float64(len(series)), "max-converted-series/op") - b.ReportMetric(float64(totals.totalWire)/float64(b.N), "wire-bytes/op") - b.ReportMetric(float64(totals.totalRaw)/float64(b.N), "raw-bytes/op") - }) - - b.Run("checkpointed-streaming-conversion", func(b *testing.B) { - b.ReportAllocs() - var totals catalystIntakeStats - for n := 0; n < b.N; n++ { - b.StopTimer() - intake := newCatalystIntake() - sender := newCatalystDatadogSender(intake) - b.StartTimer() - - if err := sender.Send(context.Background(), realMetrics); err != nil { - b.Fatal(err) - } - - b.StopTimer() - stats := intake.snapshot() - totals.requests += stats.requests - totals.rejected += stats.rejected - totals.accepted += stats.accepted - totals.totalWire += stats.totalWire - totals.totalRaw += stats.totalRaw - b.StartTimer() - } - b.ReportMetric(float64(totals.requests)/float64(b.N), "requests/op") - b.ReportMetric(float64(totals.rejected)/float64(b.N), "rejected/op") - b.ReportMetric(float64(totals.accepted)/float64(b.N), "accepted/op") - b.ReportMetric(float64(len(series)), "series/op") - b.ReportMetric(float64(defaultDatadogPayloadLimits().maxSeries), "max-converted-series/op") - b.ReportMetric(float64(totals.totalWire)/float64(b.N), "wire-bytes/op") - b.ReportMetric(float64(totals.totalRaw)/float64(b.N), "raw-bytes/op") - }) -}