Skip to content

Swap querier /api/search and remaining endpoints to proto #3944

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mdisibio merged 3 commits into from
Aug 7, 2024
Merged

Swap querier /api/search and remaining endpoints to proto #3944

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ed58780
Improve performance of /api/search and remaining endpoints by switchi…
mdisibio Aug 6, 2024
6dbed3e
changelog
mdisibio Aug 6, 2024
8cac9f5
Revert metrics summary api back to json
mdisibio Aug 6, 2024
File filter

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
* [ENHANCEMENT] Implement arrays for traceql.Static with reused fields [#3827](https://github.com/grafana/tempo/pull/3827) (@stoewer)
* [ENHANCEMENT] Tag value lookup use protobuf internally for improved latency [#3731](https://github.com/grafana/tempo/pull/3731) (@mdisibio)
* [ENHANCEMENT] TraceQL metrics queries use protobuf internally for improved latency [#3745](https://github.com/grafana/tempo/pull/3745) (@mdisibio)
* [ENHANCEMENT] TraceQL search and other endpoints use protobuf internally for improved latency and resource usage [#3944](https://github.com/grafana/tempo/pull/3944) (@mdisibio)
* [ENHANCEMENT] Add local disk caching of metrics queries in local-blocks processor [#3799](https://github.com/grafana/tempo/pull/3799) (@mdisibio)
* [ENHANCEMENT] Improve use of OTEL semantic conventions on the service graph [#3711](https://github.com/grafana/tempo/pull/3711) (@zalegrala)
* [ENHANCEMENT] Performance improvement for `rate() by ()` queries [#3719](https://github.com/grafana/tempo/pull/3719) (@mapno)
Expand Down
8 changes: 8 additions & 0 deletions modules/frontend/frontend.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -221,6 +221,9 @@ func newMetricsSummaryHandler(next pipeline.AsyncRoundTripper[combiner.PipelineR
}, nil
}
prepareRequestForQueriers(req, tenant)
// This API is always json because it only ever has 1 job and this
// lets us return the response as-is.
req.Header.Set(api.HeaderAccept, api.HeaderAcceptJSON)

level.Info(logger).Log(
"msg", "metrics summary request",
Expand Down Expand Up @@ -251,6 +254,11 @@ func prepareRequestForQueriers(req *http.Request, tenant string) {
// set the tenant header
req.Header.Set(user.OrgIDHeaderName, tenant)

// All communication with the queriers should be proto for efficiency
// NOTE - This isn't strict and queriers may still return json if we missed
// an endpoint. But cache and response unmarshalling still work.
req.Header.Set(api.HeaderAccept, api.HeaderAcceptProtobuf)

@joe-elliott joe-elliott Aug 7, 2024

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 putting this here

// copy the url (which is correct) to the RequestURI
// we do this because dskit/common uses the RequestURI field to translate from http.Request to httpgrpc.Request
// https://github.com/grafana/dskit/blob/f5bd38371e1cfae5479b2c23b3893c1a97868bdf/httpgrpc/httpgrpc.go#L53
Expand Down
2 changes: 0 additions & 2 deletions modules/frontend/metrics_query_range_sharder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -426,7 +426,6 @@ func (s *queryRangeSharder) buildBackendRequests(ctx context.Context, tenantID s
}

subR = api.BuildQueryRangeRequest(subR, queryRangeReq)
subR.Header.Set(api.HeaderAccept, api.HeaderAcceptProtobuf)

prepareRequestForQueriers(subR, tenantID)
pipelineR := pipeline.NewHTTPRequest(subR)
Expand Down Expand Up @@ -465,7 +464,6 @@ func (s *queryRangeSharder) generatorRequest(searchReq tempopb.QueryRangeRequest
searchReq.Exemplars = uint32(s.cfg.MaxExemplars) // TODO: Review this

req := s.toUpstreamRequest(parent.Context(), searchReq, parent, tenantID)
req.Header.Set(api.HeaderAccept, api.HeaderAcceptProtobuf)

return req
}
Expand Down
2 changes: 0 additions & 2 deletions modules/frontend/tag_sharder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -300,7 +300,6 @@ func (s searchTagSharder) buildBackendRequests(ctx context.Context, tenantID str
errFn(err)
return
}
subR.Header.Set(api.HeaderAccept, api.HeaderAcceptProtobuf)
prepareRequestForQueriers(subR, tenantID)
pipelineR := pipeline.NewHTTPRequest(subR)

Expand Down Expand Up @@ -357,7 +356,6 @@ func (s searchTagSharder) buildIngesterRequest(ctx context.Context, tenantID str
if err != nil {
return nil, err
}
subR.Header.Set(api.HeaderAccept, api.HeaderAcceptProtobuf)
prepareRequestForQueriers(subR, tenantID)
return subR, nil
}
Expand Down
76 changes: 15 additions & 61 deletions modules/querier/http.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,13 +164,7 @@ func (q *Querier) SearchHandler(w http.ResponseWriter, r *http.Request) {
}
}

marshaller := &jsonpb.Marshaler{}
err := marshaller.Marshal(w, resp)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.Header().Set(api.HeaderContentType, api.HeaderAcceptJSON)
writeFormattedContentForRequest(w, r, resp, span)
}

func (q *Querier) SearchTagsHandler(w http.ResponseWriter, r *http.Request) {
Expand All @@ -183,43 +177,31 @@ func (q *Querier) SearchTagsHandler(w http.ResponseWriter, r *http.Request) {
span, ctx := opentracing.StartSpanFromContext(ctx, "Querier.SearchTagsHandler")
defer span.Finish()

var resp *tempopb.SearchTagsResponse
if !isSearchBlock {
req, err := api.ParseSearchTagsRequest(r)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
resp, err := q.SearchTags(ctx, req)
resp, err = q.SearchTags(ctx, req)
if err != nil {
handleError(w, err)
return
}

marshaller := &jsonpb.Marshaler{}
err = marshaller.Marshal(w, resp)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
} else {
req, err := api.ParseSearchTagsBlockRequest(r)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
resp, err := q.SearchTagsBlocks(ctx, req)
resp, err = q.SearchTagsBlocks(ctx, req)
if err != nil {
handleError(w, err)
return
}
marshaller := &jsonpb.Marshaler{}
err = marshaller.Marshal(w, resp)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
w.Header().Set(api.HeaderContentType, api.HeaderAcceptJSON)
writeFormattedContentForRequest(w, r, resp, span)
}

func (q *Querier) SearchTagsV2Handler(w http.ResponseWriter, r *http.Request) {
Expand All @@ -232,44 +214,33 @@ func (q *Querier) SearchTagsV2Handler(w http.ResponseWriter, r *http.Request) {
span, ctx := opentracing.StartSpanFromContext(ctx, "Querier.SearchTagsHandler")
defer span.Finish()

var resp *tempopb.SearchTagsV2Response
if !isSearchBlock {
req, err := api.ParseSearchTagsRequest(r)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}

resp, err := q.SearchTagsV2(ctx, req)
resp, err = q.SearchTagsV2(ctx, req)
if err != nil {
handleError(w, err)
return
}

marshaller := &jsonpb.Marshaler{}
err = marshaller.Marshal(w, resp)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
} else {
req, err := api.ParseSearchTagsBlockRequest(r)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
resp, err := q.SearchTagsBlocksV2(ctx, req)
resp, err = q.SearchTagsBlocksV2(ctx, req)
if err != nil {
handleError(w, err)
return
}
marshaller := &jsonpb.Marshaler{}
err = marshaller.Marshal(w, resp)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
w.Header().Set(api.HeaderContentType, api.HeaderAcceptJSON)

writeFormattedContentForRequest(w, r, resp, span)
}

func (q *Querier) SearchTagValuesHandler(w http.ResponseWriter, r *http.Request) {
Expand All @@ -282,44 +253,33 @@ func (q *Querier) SearchTagValuesHandler(w http.ResponseWriter, r *http.Request)
span, ctx := opentracing.StartSpanFromContext(ctx, "Querier.SearchTagValuesHandler")
defer span.Finish()

var resp *tempopb.SearchTagValuesResponse
if !isSearchBlock {
req, err := api.ParseSearchTagValuesRequest(r)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}

resp, err := q.SearchTagValues(ctx, req)
resp, err = q.SearchTagValues(ctx, req)
if err != nil {
handleError(w, err)
return
}
marshaller := &jsonpb.Marshaler{}
err = marshaller.Marshal(w, resp)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
} else {
req, err := api.ParseSearchTagValuesBlockRequest(r)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
resp, err := q.SearchTagValuesBlocks(ctx, req)
resp, err = q.SearchTagValuesBlocks(ctx, req)
if err != nil {
handleError(w, err)
return
}
marshaller := &jsonpb.Marshaler{}
err = marshaller.Marshal(w, resp)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}

w.Header().Set(api.HeaderContentType, api.HeaderAcceptJSON)
writeFormattedContentForRequest(w, r, resp, span)
}

func (q *Querier) SearchTagValuesV2Handler(w http.ResponseWriter, r *http.Request) {
Expand Down Expand Up @@ -381,13 +341,7 @@ func (q *Querier) SpanMetricsSummaryHandler(w http.ResponseWriter, r *http.Reque
return
}

marshaller := &jsonpb.Marshaler{}
err = marshaller.Marshal(w, resp)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.Header().Set(api.HeaderContentType, api.HeaderAcceptJSON)
writeFormattedContentForRequest(w, r, resp, span)
}

func (q *Querier) QueryRangeHandler(w http.ResponseWriter, r *http.Request) {
Expand Down