api: fix race between Respond() and query/queryRange

CHANGELOG.md

@@ -15,6 +15,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#5502](https://github.com/thanos-io/thanos/pull/5502) Receive: Handle exemplar storage errors as conflict error.
 - [#5534](https://github.com/thanos-io/thanos/pull/5534) Query: Set struct return by query api alerts same as prometheus api.
 - [#5554](https://github.com/thanos-io/thanos/pull/5554) Query/Receiver: Fix querying exemplars from multi-tenant receivers.
+- [#5583](https://github.com/thanos-io/thanos/pull/5583) Query: fix data race between Respond() and query/queryRange functions. Fixes [#5410](https://github.com/thanos-io/thanos/pull/5410).
 
 ### Added
 

pkg/api/api.go

@@ -121,7 +121,7 @@ func SetCORS(w http.ResponseWriter) {
 	}
 }
 
-type ApiFunc func(r *http.Request) (interface{}, []error, *ApiError)
+type ApiFunc func(r *http.Request) (interface{}, []error, *ApiError, func())
 
 type BaseAPI struct {
 	logger      log.Logger
@@ -156,20 +156,20 @@ func (api *BaseAPI) Register(r *route.Router, tracer opentracing.Tracer, logger
 	r.Get("/status/buildinfo", instr("status_build", api.serveBuildInfo))
 }
 
-func (api *BaseAPI) options(r *http.Request) (interface{}, []error, *ApiError) {
-	return nil, nil, nil
+func (api *BaseAPI) options(r *http.Request) (interface{}, []error, *ApiError, func()) {
+	return nil, nil, nil, func() {}
 }
 
-func (api *BaseAPI) flags(r *http.Request) (interface{}, []error, *ApiError) {
-	return api.flagsMap, nil, nil
+func (api *BaseAPI) flags(r *http.Request) (interface{}, []error, *ApiError, func()) {
+	return api.flagsMap, nil, nil, func() {}
 }
 
-func (api *BaseAPI) serveRuntimeInfo(r *http.Request) (interface{}, []error, *ApiError) {
-	return api.runtimeInfo(), nil, nil
+func (api *BaseAPI) serveRuntimeInfo(r *http.Request) (interface{}, []error, *ApiError, func()) {
+	return api.runtimeInfo(), nil, nil, func() {}
 }
 
-func (api *BaseAPI) serveBuildInfo(r *http.Request) (interface{}, []error, *ApiError) {
-	return api.buildInfo, nil, nil
+func (api *BaseAPI) serveBuildInfo(r *http.Request) (interface{}, []error, *ApiError, func()) {
+	return api.buildInfo, nil, nil, func() {}
 }
 
 func GetRuntimeInfoFunc(logger log.Logger) RuntimeInfoFn {
@@ -208,12 +208,15 @@ func GetInstr(
 			if !disableCORS {
 				SetCORS(w)
 			}
-			if data, warnings, err := f(r); err != nil {
+			if data, warnings, err, releaseResources := f(r); err != nil {
 				RespondError(w, err, data)
+				releaseResources()
 			} else if data != nil {
 				Respond(w, data, warnings)
+				releaseResources()
 			} else {
 				w.WriteHeader(http.StatusNoContent)
+				releaseResources()
 			}
 		})
 

pkg/api/blocks/v1.go

@@ -86,48 +86,48 @@ func (bapi *BlocksAPI) Register(r *route.Router, tracer opentracing.Tracer, logg
 	r.Post("/blocks/mark", instr("blocks_mark", bapi.markBlock))
 }
 
-func (bapi *BlocksAPI) markBlock(r *http.Request) (interface{}, []error, *api.ApiError) {
+func (bapi *BlocksAPI) markBlock(r *http.Request) (interface{}, []error, *api.ApiError, func()) {
 	idParam := r.FormValue("id")
 	actionParam := r.FormValue("action")
 	detailParam := r.FormValue("detail")
 
 	if idParam == "" {
-		return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: errors.New("ID cannot be empty")}
+		return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: errors.New("ID cannot be empty")}, func() {}
 	}
 
 	if actionParam == "" {
-		return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: errors.New("Action cannot be empty")}
+		return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: errors.New("Action cannot be empty")}, func() {}
 	}
 
 	id, err := ulid.Parse(idParam)
 	if err != nil {
-		return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: errors.Errorf("ULID %q is not valid: %v", idParam, err)}
+		return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: errors.Errorf("ULID %q is not valid: %v", idParam, err)}, func() {}
 	}
 
 	actionType := parse(actionParam)
 	switch actionType {
 	case Deletion:
 		err := block.MarkForDeletion(r.Context(), bapi.logger, bapi.bkt, id, detailParam, promauto.With(nil).NewCounter(prometheus.CounterOpts{}))
 		if err != nil {
-			return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: err}
+			return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: err}, func() {}
 		}
 	case NoCompaction:
 		err := block.MarkForNoCompact(r.Context(), bapi.logger, bapi.bkt, id, metadata.ManualNoCompactReason, detailParam, promauto.With(nil).NewCounter(prometheus.CounterOpts{}))
 		if err != nil {
-			return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: err}
+			return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: err}, func() {}
 		}
 	default:
-		return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: errors.Errorf("not supported marker %v", actionParam)}
+		return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: errors.Errorf("not supported marker %v", actionParam)}, func() {}
 	}
-	return nil, nil, nil
+	return nil, nil, nil, func() {}
 }
 
-func (bapi *BlocksAPI) blocks(r *http.Request) (interface{}, []error, *api.ApiError) {
+func (bapi *BlocksAPI) blocks(r *http.Request) (interface{}, []error, *api.ApiError, func()) {
 	viewParam := r.URL.Query().Get("view")
 	if viewParam == "loaded" {
-		return bapi.loadedBlocksInfo, nil, nil
+		return bapi.loadedBlocksInfo, nil, nil, func() {}
 	}
-	return bapi.globalBlocksInfo, nil, nil
+	return bapi.globalBlocksInfo, nil, nil, func() {}
 }
 
 func (b *BlocksInfo) set(blocks []metadata.Meta, err error) {

pkg/api/blocks/v1_test.go

@@ -71,7 +71,8 @@ func testEndpoint(t *testing.T, test endpointTestCase, name string, responseComp
 			req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
 		}
 
-		resp, _, apiErr := test.endpoint(req.WithContext(ctx))
+		resp, _, apiErr, releaseResources := test.endpoint(req.WithContext(ctx))
+		defer releaseResources()
 		if apiErr != nil {
 			if test.errType == baseAPI.ErrorNone {
 				t.Fatalf("Unexpected error: %s", apiErr)