Add new metrics API to generator and protobuf definition

cmd/tempo/app/modules.go

@@ -212,6 +212,9 @@ func (t *App) initGenerator() (services.Service, error) {
 	}
 	t.generator = genSvc
 
+	spanStatsHandler := t.HTTPAuthMiddleware.Wrap(http.HandlerFunc(t.generator.SpanMetricsHandler))
+	t.Server.HTTP.Handle(path.Join(api.PathPrefixGenerator, addHTTPAPIPrefix(&t.cfg, api.PathSpanMetrics)), spanStatsHandler)
+
 	tempopb.RegisterMetricsGeneratorServer(t.Server.GRPC, t.generator)
 
 	return t.generator, nil

modules/generator/config.go

@@ -33,6 +33,7 @@ type Config struct {
 	// MetricsIngestionSlack is the max amount of time passed since a span's start time
 	// for the span to be considered in metrics generation
 	MetricsIngestionSlack time.Duration `yaml:"metrics_ingestion_time_range_slack"`
+	QueryTimeout          time.Duration `yaml:"query_timeout"`
 }
 
 // RegisterFlagsAndApplyDefaults registers the flags.
@@ -45,6 +46,7 @@ func (cfg *Config) RegisterFlagsAndApplyDefaults(prefix string, f *flag.FlagSet)
 
 	// setting default for max span age before discarding to 30s
 	cfg.MetricsIngestionSlack = 30 * time.Second
+	cfg.QueryTimeout = 30 * time.Second
 }
 
 type ProcessorConfig struct {

modules/generator/generator.go

@@ -322,3 +322,18 @@ func (g *Generator) OnRingInstanceStopping(lifecycler *ring.BasicLifecycler) {
 // OnRingInstanceHeartbeat implements ring.BasicLifecyclerDelegate
 func (g *Generator) OnRingInstanceHeartbeat(lifecycler *ring.BasicLifecycler, ringDesc *ring.Desc, instanceDesc *ring.InstanceDesc) {
 }
+
+func (g *Generator) GetMetrics(ctx context.Context, req *tempopb.SpanMetricsRequest) (*tempopb.SpanMetricsResponse, error) {
+	instanceID, err := user.ExtractOrgID(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	// return empty if we don't have an instance
+	instance, ok := g.getInstanceByID(instanceID)
+	if !ok || instance == nil {
+		return &tempopb.SpanMetricsResponse{}, nil
+	}
+
+	return instance.GetMetrics(ctx, req)
+}

modules/generator/http.go

@@ -0,0 +1,43 @@
+package generator
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"github.com/gogo/protobuf/jsonpb"
+	"github.com/grafana/tempo/pkg/api"
+	"github.com/grafana/tempo/pkg/tempopb"
+	"github.com/opentracing/opentracing-go"
+)
+
+func (g *Generator) SpanMetricsHandler(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := context.WithDeadline(r.Context(), time.Now().Add(g.cfg.QueryTimeout))
+	defer cancel()
+
+	span, ctx := opentracing.StartSpanFromContext(ctx, "Generator.SpanMetricsHandler")
+	defer span.Finish()
+
+	span.SetTag("requestURI", r.RequestURI)
+
+	req, err := api.ParseSpanMetricsRequest(r)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	var resp *tempopb.SpanMetricsResponse
+	resp, err = g.GetMetrics(ctx, req)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		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)
+}

modules/generator/instance.go

@@ -364,6 +364,18 @@ func (i *instance) preprocessSpans(req *tempopb.PushSpansRequest) {
 	i.updatePushMetrics(size, spanCount, expiredSpanCount)
 }
 
+func (i *instance) GetMetrics(ctx context.Context, req *tempopb.SpanMetricsRequest) (resp *tempopb.SpanMetricsResponse, err error) {
+	for _, processor := range i.processors {
+		switch p := processor.(type) {
+		case *localblocks.Processor:
+			return p.GetMetrics(ctx, req)
+		default:
+		}
+	}
+
+	return nil, fmt.Errorf("localblocks processor not found")
+}
+
 func (i *instance) updatePushMetrics(bytesIngested int, spanCount int, expiredSpanCount int) {
 	metricBytesIngested.WithLabelValues(i.instanceID).Add(float64(bytesIngested))
 	metricSpansIngested.WithLabelValues(i.instanceID).Add(float64(spanCount))

modules/generator/processor/localblocks/processor.go

@@ -14,6 +14,8 @@ import (
 	"github.com/grafana/tempo/pkg/model"
 	"github.com/grafana/tempo/pkg/tempopb"
 	v1 "github.com/grafana/tempo/pkg/tempopb/trace/v1"
+	"github.com/grafana/tempo/pkg/traceql"
+	"github.com/grafana/tempo/pkg/traceqlmetrics"
 	"github.com/grafana/tempo/pkg/util/log"
 	"github.com/grafana/tempo/tempodb/backend"
 	"github.com/grafana/tempo/tempodb/encoding"
@@ -242,6 +244,57 @@ func (p *Processor) completeBlock() error {
 	return nil
 }
 
+func (p *Processor) GetMetrics(ctx context.Context, req *tempopb.SpanMetricsRequest) (*tempopb.SpanMetricsResponse, error) {
+
+	fetcher := traceql.NewSpansetFetcherWrapper(func(ctx context.Context, req traceql.FetchSpansRequest) (traceql.FetchSpansResponse, error) {
+		if p.headBlock == nil {
+			return traceql.FetchSpansResponse{}, fmt.Errorf("head block is nil")
+		}
+		return p.headBlock.Fetch(ctx, req, common.DefaultSearchOptions())
+	})
+
+	m, err := traceqlmetrics.GetMetrics(ctx, req.Query, req.GroupBy, int(req.Limit), fetcher)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get metrics")
+	}
+
+	resp := &tempopb.SpanMetricsResponse{
+		SpanCount: uint64(m.SpanCount),
+		Estimated: m.Estimated,
+		Metrics:   make([]*tempopb.SpanMetrics, 0, len(m.Series)),
+	}
+
+	var rawHistorgram *tempopb.RawHistogram
+	var errCount int
+	for static, series := range m.Series {
+		h := []*tempopb.RawHistogram{}
+
+		for bucket, count := range series.Buckets() {
+			if count != 0 {
+				rawHistorgram = &tempopb.RawHistogram{
+					Bucket: uint64(bucket),
+					Count:  uint64(count),
+				}
+
+				h = append(h, rawHistorgram)
+			}
+		}
+
+		errCount = 0
+		if errs, ok := m.Errors[static]; ok {
+			errCount = errs
+		}
+
+		resp.Metrics = append(resp.Metrics, &tempopb.SpanMetrics{
+			LatencyHistogram: h,
+			Static:           toStaticProto(static),
+			Errors:           uint64(errCount),
+		})
+	}
+
+	return resp, nil
+}
+
 func (p *Processor) deleteOldBlocks() (err error) {
 	p.blocksMtx.Lock()
 	defer p.blocksMtx.Unlock()
@@ -490,3 +543,16 @@ func filterBatch(batch *v1.ResourceSpans) *v1.ResourceSpans {
 
 	return nil
 }
+
+func toStaticProto(static traceql.Static) *tempopb.TraceQLStatic {
+	return &tempopb.TraceQLStatic{
+		Type:   int32(static.Type),
+		N:      int64(static.N),
+		F:      static.F,
+		S:      static.S,
+		B:      static.B,
+		D:      uint64(static.D),
+		Status: int32(static.Status),
+		Kind:   int32(static.Kind),
+	}
+}

pkg/api/http.go

@@ -49,19 +49,24 @@ const (
 	// search tags
 	urlParamScope = "scope"
 
+	// generator summary
+	urlParamGroupBy = "groupBy"
+
 	HeaderAccept         = "Accept"
 	HeaderContentType    = "Content-Type"
 	HeaderAcceptProtobuf = "application/protobuf"
 	HeaderAcceptJSON     = "application/json"
 
-	PathPrefixQuerier = "/querier"
+	PathPrefixQuerier   = "/querier"
+	PathPrefixGenerator = "/generator"
 
 	PathTraces          = "/api/traces/{traceID}"
 	PathSearch          = "/api/search"
 	PathSearchTags      = "/api/search/tags"
 	PathSearchTagValues = "/api/search/tag/{" + muxVarTagName + "}/values"
 	PathEcho            = "/api/echo"
 	PathUsageStats      = "/status/usage-stats"
+	PathSpanMetrics     = "/api/metrics"
 
 	PathSearchTagValuesV2 = "/api/v2/search/tag/{" + muxVarTagName + "}/values"
 	PathSearchTagsV2      = "/api/v2/search/tags"
@@ -313,6 +318,27 @@ func ParseSearchBlockRequest(r *http.Request) (*tempopb.SearchBlockRequest, erro
 	return req, nil
 }
 
+func ParseSpanMetricsRequest(r *http.Request) (*tempopb.SpanMetricsRequest, error) {
+	req := &tempopb.SpanMetricsRequest{}
+
+	groupBy := r.URL.Query().Get(urlParamGroupBy)
+	req.GroupBy = groupBy
+
+	query := r.URL.Query().Get(urlParamQuery)
+	req.Query = query
+
+	l := r.URL.Query().Get(urlParamLimit)
+	if l != "" {
+		limit, err := strconv.Atoi(l)
+		if err != nil {
+			return nil, fmt.Errorf("invalid limit: %w", err)
+		}
+		req.Limit = uint64(limit)
+	}
+
+	return req, nil
+}
+
 // BuildSearchRequest takes a tempopb.SearchRequest and populates the passed http.Request
 // with the appropriate params. If no http.Request is provided a new one is created.
 func BuildSearchRequest(req *http.Request, searchReq *tempopb.SearchRequest) (*http.Request, error) {