[Traceql Metrics] PR 2 - API

cmd/tempo/app/modules.go

@@ -279,6 +279,9 @@ func (t *App) initGenerator() (services.Service, error) {
 	spanStatsHandler := t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.generator.SpanMetricsHandler))
 	t.Server.HTTP.Handle(path.Join(api.PathPrefixGenerator, addHTTPAPIPrefix(&t.cfg, api.PathSpanMetrics)), spanStatsHandler)
 
+	queryRangeHandler := t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.generator.QueryRangeHandler))
+	t.Server.HTTP.Handle(path.Join(api.PathPrefixGenerator, addHTTPAPIPrefix(&t.cfg, api.PathMetricsQueryRange)), queryRangeHandler)
+
 	tempopb.RegisterMetricsGeneratorServer(t.Server.GRPC, t.generator)
 
 	return t.generator, nil
@@ -346,6 +349,9 @@ func (t *App) initQuerier() (services.Service, error) {
 	spanMetricsSummaryHandler := t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.querier.SpanMetricsSummaryHandler))
 	t.Server.HTTP.Handle(path.Join(api.PathPrefixQuerier, addHTTPAPIPrefix(&t.cfg, api.PathSpanMetricsSummary)), spanMetricsSummaryHandler)
 
+	queryRangeHandler := t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.querier.QueryRangeHandler))
+	t.Server.HTTP.Handle(path.Join(api.PathPrefixQuerier, addHTTPAPIPrefix(&t.cfg, api.PathMetricsQueryRange)), queryRangeHandler)
+
 	return t.querier, t.querier.CreateAndRegisterWorker(t.Server.HTTPServer.Handler)
 }
 
@@ -390,6 +396,8 @@ func (t *App) initQueryFrontend() (services.Service, error) {
 
 	// http metrics endpoints
 	t.Server.HTTP.Handle(addHTTPAPIPrefix(&t.cfg, api.PathSpanMetricsSummary), base.Wrap(queryFrontend.SpanMetricsSummaryHandler))
+	t.Server.HTTP.Handle(addHTTPAPIPrefix(&t.cfg, api.PathMetricsQueryRange), base.Wrap(queryFrontend.QueryRangeHandler))
+	t.Server.HTTP.Handle(addHTTPAPIPrefix(&t.cfg, api.PathPromQueryRange), base.Wrap(queryFrontend.QueryRangeHandler))
 
 	// the query frontend needs to have knowledge of the blocks so it can shard search jobs
 	t.store.EnablePolling(context.Background(), nil)

modules/frontend/config.go

@@ -20,6 +20,7 @@ type Config struct {
 	MaxRetries                int             `yaml:"max_retries,omitempty"`
 	Search                    SearchConfig    `yaml:"search"`
 	TraceByID                 TraceByIDConfig `yaml:"trace_by_id"`
+	Metrics                   MetricsConfig   `yaml:"metrics"`
 	MultiTenantQueriesEnabled bool            `yaml:"multi_tenant_queries_enabled"`
 }
 
@@ -35,6 +36,10 @@ type TraceByIDConfig struct {
 	SLO              SLOConfig     `yaml:",inline"`
 }
 
+type MetricsConfig struct {
+	Sharder QueryRangeSharderConfig `yaml:",inline"`
+}
+
 type HedgingConfig struct {
 	HedgeRequestsAt   time.Duration `yaml:"hedge_requests_at"`
 	HedgeRequestsUpTo int           `yaml:"hedge_requests_up_to"`
@@ -74,6 +79,14 @@ func (cfg *Config) RegisterFlagsAndApplyDefaults(string, *flag.FlagSet) {
 			HedgeRequestsUpTo: 2,
 		},
 	}
+	cfg.Metrics = MetricsConfig{
+		Sharder: QueryRangeSharderConfig{
+			QueryBackendAfter:     time.Hour,
+			ConcurrentRequests:    defaultConcurrentRequests,
+			TargetBytesPerRequest: 100 * 1024 * 1024,
+			Interval:              5 * time.Minute,
+		},
+	}
 
 	// enable multi tenant queries by default
 	cfg.MultiTenantQueriesEnabled = true

modules/frontend/frontend.go

@@ -14,10 +14,10 @@ import (
 	"github.com/golang/protobuf/jsonpb" //nolint:all //deprecated
 	"github.com/golang/protobuf/proto"  //nolint:all //deprecated
 	"github.com/grafana/dskit/user"
-	"github.com/grafana/tempo/modules/frontend/combiner"
 	"github.com/opentracing/opentracing-go"
 	"github.com/prometheus/client_golang/prometheus"
 
+	"github.com/grafana/tempo/modules/frontend/combiner"
 	"github.com/grafana/tempo/modules/overrides"
 	"github.com/grafana/tempo/pkg/api"
 	"github.com/grafana/tempo/pkg/cache"
@@ -28,11 +28,11 @@ import (
 type streamingSearchHandler func(req *tempopb.SearchRequest, srv tempopb.StreamingQuerier_SearchServer) error
 
 type QueryFrontend struct {
-	TraceByIDHandler, SearchHandler, SpanMetricsSummaryHandler, SearchWSHandler                http.Handler
-	SearchTagsHandler, SearchTagsV2Handler, SearchTagsValuesHandler, SearchTagsValuesV2Handler http.Handler
-	cacheProvider                                                                              cache.Provider
-	streamingSearch                                                                            streamingSearchHandler
-	logger                                                                                     log.Logger
+	TraceByIDHandler, SearchHandler, SpanMetricsSummaryHandler, SearchWSHandler, QueryRangeHandler http.Handler
+	SearchTagsHandler, SearchTagsV2Handler, SearchTagsValuesHandler, SearchTagsValuesV2Handler     http.Handler
+	cacheProvider                                                                                  cache.Provider
+	streamingSearch                                                                                streamingSearchHandler
+	logger                                                                                         log.Logger
 }
 
 // New returns a new QueryFrontend
@@ -84,10 +84,13 @@ func New(cfg Config, next http.RoundTripper, o overrides.Interface, reader tempo
 	searchTagsValuesV2Middleware := MergeMiddlewares(
 		newMultiTenantMiddleware(cfg, combiner.NewSearchTagValuesV2, logger),
 		newSearchTagsMiddleware(), retryWare)
+	metricsMiddleware := MergeMiddlewares(
+		newMultiTenantUnsupportedMiddleware(cfg, logger),
+		newMetricsMiddleware(), retryWare)
 
-	spanMetricsMiddleware := MergeMiddlewares(
+	queryRangeMiddleware := MergeMiddlewares(
 		newMultiTenantUnsupportedMiddleware(cfg, logger),
-		newSpanMetricsMiddleware(), retryWare)
+		newQueryRangeMiddleware(cfg, o, reader, logger), retryWare)
 
 	traces := traceByIDMiddleware.Wrap(next)
 	search := searchMiddleware.Wrap(next)
@@ -96,7 +99,8 @@ func New(cfg Config, next http.RoundTripper, o overrides.Interface, reader tempo
 	searchTagValues := searchTagsValuesMiddleware.Wrap(next)
 	searchTagValuesV2 := searchTagsValuesV2Middleware.Wrap(next)
 
-	metrics := spanMetricsMiddleware.Wrap(next)
+	metrics := metricsMiddleware.Wrap(next)
+	queryrange := queryRangeMiddleware.Wrap(next)
 
 	return &QueryFrontend{
 		TraceByIDHandler:          newHandler(traces, traceByIDSLOPostHook(cfg.TraceByID.SLO), nil, logger),
@@ -105,8 +109,8 @@ func New(cfg Config, next http.RoundTripper, o overrides.Interface, reader tempo
 		SearchTagsV2Handler:       newHandler(searchTagsV2, nil, nil, logger),
 		SearchTagsValuesHandler:   newHandler(searchTagValues, nil, nil, logger),
 		SearchTagsValuesV2Handler: newHandler(searchTagValuesV2, nil, nil, logger),
-
 		SpanMetricsSummaryHandler: newHandler(metrics, nil, nil, logger),
+		QueryRangeHandler:         newHandler(queryrange, nil, nil, logger),
 		SearchWSHandler:           newSearchStreamingWSHandler(cfg, o, retryWare.Wrap(next), reader, searchCache, apiPrefix, logger),
 		cacheProvider:             cacheProvider,
 		streamingSearch:           newSearchStreamingGRPCHandler(cfg, o, retryWare.Wrap(next), reader, searchCache, apiPrefix, logger),
@@ -238,8 +242,8 @@ func newSearchTagsMiddleware() Middleware {
 	})
 }
 
-// newSpanMetricsMiddleware creates a new frontend middleware to handle search and search tags requests.
-func newSpanMetricsMiddleware() Middleware {
+// newSpanMetricsMiddleware creates a new frontend middleware to handle metrics-generator requests.
+func newMetricsMiddleware() Middleware {
 	return MiddlewareFunc(func(next http.RoundTripper) http.RoundTripper {
 		generatorRT := next
 
@@ -268,3 +272,13 @@ func buildUpstreamRequestURI(originalURI string, params url.Values) string {
 
 	return uri
 }
+
+func newQueryRangeMiddleware(cfg Config, o overrides.Interface, reader tempodb.Reader, logger log.Logger) Middleware {
+	return MiddlewareFunc(func(next http.RoundTripper) http.RoundTripper {
+		searchRT := NewRoundTripper(next, newQueryRangeSharder(reader, o, cfg.Metrics.Sharder, newSearchProgress, logger))
+
+		return RoundTripperFunc(func(r *http.Request) (*http.Response, error) {
+			return searchRT.RoundTrip(r)
+		})
+	})
+}