diff --git a/CHANGELOG.md b/CHANGELOG.md index 541cc6856e3..c20a99efde7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -78,6 +78,7 @@ * [BUGFIX] Store Gateway: Fix misleading "no index cache backend addresses" validation error being reported for chunks-cache, metadata-cache, and parquet caches when their memcached or redis backend is configured without addresses. The message is now the cache-type-agnostic "no cache backend addresses". #7675 * [BUGFIX] Querier/Query Frontend: Fix DNS watcher dropping all query-frontend/scheduler worker connections on a transient DNS lookup failure. #7698 * [BUGFIX] Ring: Fix DynamoDB KV CAS not retrying on transactional conditional check failures. `TransactWriteItems` reports condition failures as `TransactionCanceledException` with a `ConditionalCheckFailed` cancellation reason, which was not recognized as retryable, so any concurrent ring update conflict (e.g. many ingesters joining during a rolling update) failed immediately instead of re-reading and retrying. `TransactionConflict` cancellation reasons are also treated as retryable. #7706 +* [BUGFIX] Distributor: Return HTTP 499 (Client Closed Request) instead of 500 when a remote-write or OTLP push is canceled by the client, so client-side cancellations are no longer counted as server-side errors. #7717 ## 1.21.1 2026-06-04 diff --git a/pkg/util/push/otlp.go b/pkg/util/push/otlp.go index 132df49018e..910da6d1783 100644 --- a/pkg/util/push/otlp.go +++ b/pkg/util/push/otlp.go @@ -4,6 +4,7 @@ import ( "bytes" "compress/gzip" "context" + "errors" "fmt" "io" "net/http" @@ -25,6 +26,7 @@ import ( "github.com/cortexproject/cortex/pkg/cortexpb" "github.com/cortexproject/cortex/pkg/distributor" "github.com/cortexproject/cortex/pkg/util" + util_api "github.com/cortexproject/cortex/pkg/util/api" util_log "github.com/cortexproject/cortex/pkg/util/log" "github.com/cortexproject/cortex/pkg/util/users" "github.com/cortexproject/cortex/pkg/util/validation" @@ -94,6 +96,9 @@ func OTLPHandler(maxRecvMsgSize int, overrides *validation.Overrides, cfg distri prwReq.Metadata = metadata if _, err := push(ctx, &prwReq); err != nil { + if errors.Is(err, context.Canceled) { + err = httpgrpc.Errorf(util_api.StatusClientClosedRequest, "%s", err.Error()) + } resp, ok := httpgrpc.HTTPResponseFromError(err) if !ok { http.Error(w, err.Error(), http.StatusInternalServerError) @@ -101,7 +106,7 @@ func OTLPHandler(maxRecvMsgSize int, overrides *validation.Overrides, cfg distri } if resp.GetCode()/100 == 5 { level.Error(logger).Log("msg", "push error", "err", err) - } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests { + } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests && resp.GetCode() != util_api.StatusClientClosedRequest { level.Warn(logger).Log("msg", "push refused", "err", err) } http.Error(w, string(resp.Body), int(resp.Code)) diff --git a/pkg/util/push/otlp_test.go b/pkg/util/push/otlp_test.go index da528ee5f5d..8cd86fd37a8 100644 --- a/pkg/util/push/otlp_test.go +++ b/pkg/util/push/otlp_test.go @@ -1048,6 +1048,41 @@ func TestOTLPHandler_MetricCollection(t *testing.T) { assert.Equal(t, 1.0, val) } +func TestOTLPHandler_ClientCancellation(t *testing.T) { + cfg := distributor.OTLPConfig{ + ConvertAllAttributes: false, + DisableTargetInfo: false, + } + overrides := validation.NewOverrides(querier.DefaultLimitsConfig(), nil) + exportRequest := generateOTLPWriteRequest() + + // A client-canceled push (bare or wrapped context.Canceled) should return 499 + // (Client Closed Request) instead of 500, matching the remote-write handlers. + for _, tc := range []struct { + name string + err error + }{ + {name: "bare cancellation", err: context.Canceled}, + {name: "wrapped cancellation", err: fmt.Errorf("send failed: %w", context.Canceled)}, + } { + t.Run(tc.name, func(t *testing.T) { + pushErr := tc.err + pushFunc := func(ctx context.Context, req *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error) { + return nil, pushErr + } + + req, err := getOTLPHttpRequest(&exportRequest, pbContentType, "") + require.NoError(t, err) + + handler := OTLPHandler(100000, overrides, cfg, nil, pushFunc, nil) + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, req) + + assert.Equal(t, 499, recorder.Result().StatusCode) + }) + } +} + func TestOTLPWriteHandler(t *testing.T) { cfg := distributor.OTLPConfig{ ConvertAllAttributes: false, diff --git a/pkg/util/push/push.go b/pkg/util/push/push.go index fe8efe53e94..7f2ae9dd167 100644 --- a/pkg/util/push/push.go +++ b/pkg/util/push/push.go @@ -2,6 +2,7 @@ package push import ( "context" + "errors" "fmt" "net/http" "strconv" @@ -18,6 +19,7 @@ import ( "github.com/cortexproject/cortex/pkg/cortexpb" "github.com/cortexproject/cortex/pkg/util" + util_api "github.com/cortexproject/cortex/pkg/util/api" "github.com/cortexproject/cortex/pkg/util/extract" "github.com/cortexproject/cortex/pkg/util/log" "github.com/cortexproject/cortex/pkg/util/users" @@ -73,6 +75,9 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool, } if _, err := push(ctx, &req.WriteRequest); err != nil { + if errors.Is(err, context.Canceled) { + err = httpgrpc.Errorf(util_api.StatusClientClosedRequest, "%s", err.Error()) + } resp, ok := httpgrpc.HTTPResponseFromError(err) if !ok { http.Error(w, err.Error(), http.StatusInternalServerError) @@ -80,7 +85,7 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool, } if resp.GetCode()/100 == 5 { level.Error(logger).Log("msg", "push error", "err", err) - } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests { + } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests && resp.GetCode() != util_api.StatusClientClosedRequest { level.Warn(logger).Log("msg", "push refused", "err", err) } http.Error(w, string(resp.Body), int(resp.Code)) @@ -123,6 +128,9 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool, } if writeResp, err := push(ctx, &v1Req.WriteRequest); err != nil { + if errors.Is(err, context.Canceled) { + err = httpgrpc.Errorf(util_api.StatusClientClosedRequest, "%s", err.Error()) + } resp, ok := httpgrpc.HTTPResponseFromError(err) if !ok { setPRW2RespHeader(w, 0, 0, 0) @@ -137,7 +145,7 @@ func Handler(remoteWrite2Enabled bool, acceptUnknownRemoteWriteContentType bool, } if resp.GetCode()/100 == 5 { level.Error(logger).Log("msg", "push error", "err", err) - } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests { + } else if resp.GetCode() != http.StatusAccepted && resp.GetCode() != http.StatusTooManyRequests && resp.GetCode() != util_api.StatusClientClosedRequest { level.Warn(logger).Log("msg", "push refused", "err", err) } http.Error(w, string(resp.Body), int(resp.Code)) diff --git a/pkg/util/push/push_test.go b/pkg/util/push/push_test.go index 28070a51559..6ad88c9589e 100644 --- a/pkg/util/push/push_test.go +++ b/pkg/util/push/push_test.go @@ -855,6 +855,53 @@ func TestHandler_remoteWrite(t *testing.T) { assert.Equal(t, "1", respHeader[rw20WrittenExemplarsHeader][0]) assert.Contains(t, resp.Body.String(), "HA deduplication") }) + + // A client-canceled push (bare or wrapped context.Canceled) should return 499 + // (Client Closed Request) instead of 500, matching the query-frontend convention. + cancellationTests := []struct { + name string + err error + isV2 bool + }{ + {name: "remote write v1 client cancellation", err: context.Canceled, isV2: false}, + {name: "remote write v2 client cancellation", err: context.Canceled, isV2: true}, + {name: "remote write v1 wrapped client cancellation", err: fmt.Errorf("send failed: %w", context.Canceled), isV2: false}, + {name: "remote write v2 wrapped client cancellation", err: fmt.Errorf("send failed: %w", context.Canceled), isV2: true}, + } + + for _, test := range cancellationTests { + t.Run(test.name, func(t *testing.T) { + pushErr := test.err + pushFunc := func(ctx context.Context, req *cortexpb.WriteRequest) (*cortexpb.WriteResponse, error) { + return nil, pushErr + } + + ctx := context.Background() + ctx = user.InjectOrgID(ctx, "user-1") + handler := Handler(true, false, 100000, overrides, nil, pushFunc, nil) + + var body []byte + if test.isV2 { + body = createPrometheusRemoteWriteV2Protobuf(t) + } else { + body = createPrometheusRemoteWriteProtobuf(t) + } + req := createRequest(t, body, test.isV2) + req = req.WithContext(ctx) + resp := httptest.NewRecorder() + handler.ServeHTTP(resp, req) + + assert.Equal(t, 499, resp.Code) + + if test.isV2 { + // Written-stats headers should still be set (to zero) on the error path. + respHeader := resp.Header() + assert.Equal(t, "0", respHeader[rw20WrittenSamplesHeader][0]) + assert.Equal(t, "0", respHeader[rw20WrittenHistogramsHeader][0]) + assert.Equal(t, "0", respHeader[rw20WrittenExemplarsHeader][0]) + } + }) + } } func TestHandler_ContentTypeAndEncoding(t *testing.T) {