From ec03d80fe13838494ff9f96988920dab851bff9a Mon Sep 17 00:00:00 2001 From: Allen Yan Date: Wed, 22 Jul 2026 18:36:21 -0700 Subject: [PATCH 1/2] Distributor: return HTTP 499 instead of 500 on client push cancellation Maps client-canceled remote-write push requests to HTTP 499 (Client Closed Request) instead of 500, so ordinary client-side cancellations (disconnects, or a fronting proxy/LB timeout) are no longer counted as server-side errors. ring.DoBatch returns a bare context.Canceled when the client cancels the request context. Unlike the other error paths in the distributor's Push (which wrap via httpgrpc.Errorf), this error carries no gRPC status, so httpgrpc.HTTPResponseFromError returns ok == false and the handler falls through to the generic 500. The query-frontend already handles this correctly (StatusClientClosedRequest = 499); this ports the same convention to the write path. Changes: - pkg/util/push/push.go: in the PRW1 and PRW2 handlers, map context.Canceled (via errors.Is, so wrapped cancellations are also caught) to httpgrpc.Errorf(util_api.StatusClientClosedRequest, ...) before status extraction, and exclude 499 from the push refused warn log. - pkg/util/push/otlp.go: same mapping for the OTLP write path. - Tests for bare and wrapped cancellation across the PRW1, PRW2, and OTLP handlers. Signed-off-by: Allen Yan --- pkg/util/push/otlp.go | 7 +++++- pkg/util/push/otlp_test.go | 35 ++++++++++++++++++++++++++++ pkg/util/push/push.go | 12 ++++++++-- pkg/util/push/push_test.go | 47 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 98 insertions(+), 3 deletions(-) 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) { From eb8ba6f2222fb0c1e219744c5a96a1ce66a67496 Mon Sep 17 00:00:00 2001 From: Allen Yan Date: Wed, 22 Jul 2026 18:39:34 -0700 Subject: [PATCH 2/2] Add CHANGELOG entry for HTTP 499 on client push cancellation Signed-off-by: Allen Yan --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) 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