New querier API to summarize the generator metrics

cmd/tempo/app/modules.go

@@ -101,7 +101,6 @@ func (t *App) initServer() (services.Service, error) {
 }
 
 func (t *App) initInternalServer() (services.Service, error) {
-
 	if !t.cfg.InternalServer.Enable {
 		return services.NewIdleService(nil, nil), nil
 	}
@@ -239,7 +238,15 @@ func (t *App) initQuerier() (services.Service, error) {
 	}
 
 	// todo: make ingester client a module instead of passing config everywhere
-	querier, err := querier.New(t.cfg.Querier, t.cfg.IngesterClient, t.ring, t.store, t.Overrides)
+	querier, err := querier.New(
+		t.cfg.Querier,
+		t.cfg.IngesterClient,
+		t.ring,
+		t.cfg.GeneratorClient,
+		t.generatorRing,
+		t.store,
+		t.Overrides,
+	)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create querier %w", err)
 	}
@@ -267,6 +274,9 @@ func (t *App) initQuerier() (services.Service, error) {
 	searchTagValuesV2Handler := t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.querier.SearchTagValuesV2Handler))
 	t.Server.HTTP.Handle(path.Join(api.PathPrefixQuerier, addHTTPAPIPrefix(&t.cfg, api.PathSearchTagValuesV2)), searchTagValuesV2Handler)
 
+	spanMetricsSummaryHandler := t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.querier.SpanMetricsSummaryHandler))
+	t.Server.HTTP.Handle(path.Join(api.PathPrefixQuerier, addHTTPAPIPrefix(&t.cfg, api.PathSpanMetricsSummary)), spanMetricsSummaryHandler)
+
 	return t.querier, t.querier.CreateAndRegisterWorker(t.Server.HTTPServer.Handler)
 }
 
@@ -293,6 +303,7 @@ func (t *App) initQueryFrontend() (services.Service, error) {
 
 	traceByIDHandler := middleware.Wrap(queryFrontend.TraceByIDHandler)
 	searchHandler := middleware.Wrap(queryFrontend.SearchHandler)
+	spanMetricsSummaryHandler := middleware.Wrap(queryFrontend.SpanMetricsSummaryHandler)
 
 	// register grpc server for queriers to connect to
 	frontend_v1pb.RegisterFrontendServer(t.Server.GRPC, t.frontend)
@@ -308,6 +319,9 @@ func (t *App) initQueryFrontend() (services.Service, error) {
 	t.Server.HTTP.Handle(addHTTPAPIPrefix(&t.cfg, api.PathSearchTagValues), searchHandler)
 	t.Server.HTTP.Handle(addHTTPAPIPrefix(&t.cfg, api.PathSearchTagValuesV2), searchHandler)
 
+	// http metrics endpoints
+	t.Server.HTTP.Handle(addHTTPAPIPrefix(&t.cfg, api.PathSpanMetricsSummary), spanMetricsSummaryHandler)
+
 	// the query frontend needs to have knowledge of the blocks so it can shard search jobs
 	t.store.EnablePolling(nil)
 
@@ -452,7 +466,7 @@ func (t *App) setupModuleManager() error {
 		Distributor:          {Ring, Server, Overrides, UsageReport, MetricsGeneratorRing},
 		Ingester:             {Store, Server, Overrides, MemberlistKV, UsageReport},
 		MetricsGenerator:     {Server, Overrides, MemberlistKV, UsageReport},
-		Querier:              {Store, Ring, Overrides, UsageReport},
+		Querier:              {Store, Ring, MetricsGeneratorRing, Overrides, UsageReport},
 		Compactor:            {Store, Server, Overrides, MemberlistKV, UsageReport},
 		SingleBinary:         {Compactor, QueryFrontend, Querier, Ingester, Distributor, MetricsGenerator},
 		ScalableSingleBinary: {SingleBinary},

modules/frontend/frontend.go

@@ -28,14 +28,15 @@ import (
 const (
 	traceByIDOp = "traces"
 	searchOp    = "search"
+	metricsOp   = "metrics"
 )
 
 type streamingSearchHandler func(req *tempopb.SearchRequest, srv tempopb.StreamingQuerier_SearchServer) error
 
 type QueryFrontend struct {
-	TraceByIDHandler, SearchHandler http.Handler
-	streamingSearch                 streamingSearchHandler
-	logger                          log.Logger
+	TraceByIDHandler, SearchHandler, SpanMetricsSummaryHandler http.Handler
+	streamingSearch                                            streamingSearchHandler
+	logger                                                     log.Logger
 }
 
 // New returns a new QueryFrontend
@@ -69,17 +70,21 @@ func New(cfg Config, next http.RoundTripper, o overrides.Interface, reader tempo
 	// tracebyid middleware
 	traceByIDMiddleware := MergeMiddlewares(newTraceByIDMiddleware(cfg, logger), retryWare)
 	searchMiddleware := MergeMiddlewares(newSearchMiddleware(cfg, o, reader, logger), retryWare)
+	spanMetricsMiddleware := MergeMiddlewares(newSpanMetricsMiddleware(cfg, o, reader, logger), retryWare)
 
 	traceByIDCounter := queriesPerTenant.MustCurryWith(prometheus.Labels{"op": traceByIDOp})
 	searchCounter := queriesPerTenant.MustCurryWith(prometheus.Labels{"op": searchOp})
+	spanMetricsCounter := queriesPerTenant.MustCurryWith(prometheus.Labels{"op": metricsOp})
 
 	traces := traceByIDMiddleware.Wrap(next)
 	search := searchMiddleware.Wrap(next)
+	metrics := spanMetricsMiddleware.Wrap(next)
 	return &QueryFrontend{
-		TraceByIDHandler: newHandler(traces, traceByIDCounter, logger),
-		SearchHandler:    newHandler(search, searchCounter, logger),
-		streamingSearch:  newSearchStreamingHandler(cfg, o, retryWare.Wrap(next), reader, apiPrefix, logger),
-		logger:           logger,
+		TraceByIDHandler:          newHandler(traces, traceByIDCounter, logger),
+		SearchHandler:             newHandler(search, searchCounter, logger),
+		SpanMetricsSummaryHandler: newHandler(metrics, spanMetricsCounter, logger),
+		streamingSearch:           newSearchStreamingHandler(cfg, o, retryWare.Wrap(next), reader, apiPrefix, logger),
+		logger:                    logger,
 	}, nil
 }
 
@@ -199,6 +204,23 @@ func newSearchMiddleware(cfg Config, o overrides.Interface, reader tempodb.Reade
 	})
 }
 
+// newSpanMetricsMiddleware creates a new frontend middleware to handle search and search tags requests.
+func newSpanMetricsMiddleware(cfg Config, o overrides.Interface, reader tempodb.Reader, logger log.Logger) Middleware {
+	return MiddlewareFunc(func(next http.RoundTripper) http.RoundTripper {
+		generatorRT := next
+
+		return RoundTripperFunc(func(r *http.Request) (*http.Response, error) {
+			// ingester search queries only need to be proxied to a single querier
+			orgID, _ := user.ExtractOrgID(r.Context())
+
+			r.Header.Set(user.OrgIDHeaderName, orgID)
+			r.RequestURI = buildUpstreamRequestURI(r.RequestURI, nil)
+
+			return generatorRT.RoundTrip(r)
+		})
+	})
+}
+
 // buildUpstreamRequestURI returns a uri based on the passed parameters
 // we do this because weaveworks/common uses the RequestURI field to translate from http.Request to httpgrpc.Request
 // https://github.com/weaveworks/common/blob/47e357f4e1badb7da17ad74bae63e228bdd76e8f/httpgrpc/server/server.go#L48

modules/querier/http.go

@@ -266,6 +266,35 @@ func (q *Querier) SearchTagValuesV2Handler(w http.ResponseWriter, r *http.Reques
 	w.Header().Set(api.HeaderContentType, api.HeaderAcceptJSON)
 }
 
+func (q *Querier) SpanMetricsSummaryHandler(w http.ResponseWriter, r *http.Request) {
+	// Enforce the query timeout while querying backends
+	ctx, cancel := context.WithDeadline(r.Context(), time.Now().Add(q.cfg.Search.QueryTimeout))
+	defer cancel()
+
+	span, ctx := opentracing.StartSpanFromContext(ctx, "Querier.SpanMetricsSummaryHandler")
+	defer span.Finish()
+
+	req, err := api.ParseSpanMetricsSummaryRequest(r)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	resp, err := q.SpanMetricsSummary(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)
+}
+
 func handleError(w http.ResponseWriter, err error) {
 	if errors.Is(err, context.Canceled) {
 		// ignore this error. we regularly cancel context once queries are complete