Breadcrumbs for generator to serve a span summary

Signed-off-by: Zach Leslie <[email protected]>

modules/generator/config.go

@@ -29,6 +29,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"`
+	SummaryTimeout        time.Duration `yaml:"summary_timeout"`
 }
 
 // RegisterFlagsAndApplyDefaults registers the flags.

modules/generator/generator.go

@@ -283,3 +283,17 @@ 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 (q *Generator) SpanSummary(ctx context.Context, req *tempopb.SpanSummaryRequest) (*tempopb.SpanSummaryResponse, error) {
+	instanceID, err := user.ExtractOrgID(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	instance, err := q.getOrCreateInstance(instanceID)
+	if err != nil {
+		return nil, err
+	}
+
+	return instance.spanSummary(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) SpanSummaryHandler(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := context.WithDeadline(r.Context(), time.Now().Add(g.cfg.SummaryTimeout))
+	defer cancel()
+
+	span, ctx := opentracing.StartSpanFromContext(ctx, "Generator.SpanSummaryHandler")
+	defer span.Finish()
+
+	span.SetTag("requestURI", r.RequestURI)
+
+	req, err := api.ParseSummaryRequest(r)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	var resp *tempopb.SpanSummaryResponse
+	resp, err = g.SpanSummary(ctx, req)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+	}
+
+	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

@@ -349,6 +349,22 @@ func (i *instance) preprocessSpans(req *tempopb.PushSpansRequest) {
 	i.updatePushMetrics(size, spanCount, expiredSpanCount)
 }
 
+func (i *instance) spanSummary(ctx context.Context, req *tempopb.SpanSummaryRequest) (resp *tempopb.SpanSummaryResponse, err error) {
+	select {
+	case <-ctx.Done():
+		return nil, nil
+		// case i.spanSummaryCh <- req:
+	default:
+		// for _, processor := range i.processors {
+		// switch p := processor.(type) {
+		// case *localblocks.Processor:
+		// processor.GetMetrics(ctx, req)
+		// }
+		// }
+		return nil, fmt.Errorf("not implemented yet")
+	}
+}
+
 func (i *instance) updatePushMetrics(bytesIngested int, spanCount int, expiredSpanCount int) {
 	metricBytesIngested.WithLabelValues(i.instanceID).Add(float64(bytesIngested))
 	metricSpansIngested.WithLabelValues(i.instanceID).Add(float64(spanCount))