Search v0.1.0 (ingester only)

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## main / unreleased
 
+* [FEATURE] Add ability to search ingesters for traces [#806](https://github.com/grafana/tempo/pull/806) (@mdisibio)
 * [BUGFIX] Update port spec for GCS docker-compose example [#869](https://github.com/grafana/tempo/pull/869) (@zalegrala)
 * [BUGFIX] Cortex upgrade to fix an issue where unhealthy compactors can't be forgotten [#878](https://github.com/grafana/tempo/pull/878) (@joe-elliott)
 * [ENHANCEMENT] Added "query blocks" cli option. [#876](https://github.com/grafana/tempo/pull/876) (@joe-elliott)

Makefile

@@ -167,13 +167,18 @@ gen-proto:
 
 	rm -rf $(PROTO_INTERMEDIATE_DIR)
 
+.PHONY: gen-flat
+gen-flat:
+	# -o /pkg generates into same folder as tempo.fbs for simpler imports.
+	docker run -v${PWD}:/opt/src neomantra/flatbuffers flatc --go -o /opt/src/pkg /opt/src/pkg/tempofb/tempo.fbs
 
 ### Check vendored files and generated proto
 .PHONY: vendor-check
-vendor-check: gen-proto
+vendor-check: gen-proto gen-flat
 	go mod vendor
 	go mod tidy -e
-	git diff --exit-code -- go.sum go.mod vendor/ pkg/tempopb/
+	git diff --exit-code -- go.sum go.mod vendor/ pkg/tempopb/ pkg/tempofb/
+
 
 ### Release (intended to be used in the .github/workflows/images.yml)
 $(GORELEASER):

cmd/tempo-cli/cmd-gen-index.go

@@ -95,7 +95,7 @@ func VerifyIndex(indexReader common.IndexReader, dataReader common.DataReader) e
 		}
 
 		// read data file at record position
-		_, _, err = dataReader.Read(context.TODO(), []common.Record{*record}, nil)
+		_, _, err = dataReader.Read(context.TODO(), []common.Record{*record}, nil, nil)
 		if err != nil {
 			fmt.Println("index/data is corrupt, record/data mismatch")
 			return err

cmd/tempo-query/main.go

@@ -5,15 +5,19 @@ import (
 	"io"
 	"strings"
 
-	"github.com/spf13/viper"
-
-	"github.com/grafana/tempo/cmd/tempo-query/tempo"
 	"github.com/hashicorp/go-hclog"
+	hcplugin "github.com/hashicorp/go-plugin"
 	"github.com/jaegertracing/jaeger/plugin/storage/grpc"
 	"github.com/jaegertracing/jaeger/plugin/storage/grpc/shared"
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
+	otgrpc "github.com/opentracing-contrib/go-grpc"
+	"github.com/opentracing/opentracing-go"
+	"github.com/spf13/viper"
 	jaeger_config "github.com/uber/jaeger-client-go/config"
+	google_grpc "google.golang.org/grpc"
+
+	"github.com/grafana/tempo/cmd/tempo-query/tempo"
 )
 
 func main() {
@@ -51,8 +55,13 @@ func main() {
 
 	backend := tempo.New(cfg)
 	plugin := &plugin{backend: backend}
-	grpc.Serve(&shared.PluginServices{
+	grpc.ServeWithGRPCServer(&shared.PluginServices{
 		Store: plugin,
+	}, func(options []google_grpc.ServerOption) *google_grpc.Server {
+		return hcplugin.DefaultGRPCServer([]google_grpc.ServerOption{
+			google_grpc.UnaryInterceptor(otgrpc.OpenTracingServerInterceptor(opentracing.GlobalTracer())),
+			google_grpc.StreamInterceptor(otgrpc.OpenTracingStreamServerInterceptor(opentracing.GlobalTracer())),
+		})
 	})
 }
 

cmd/tempo-query/tempo/plugin.go

@@ -5,8 +5,12 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"net/url"
+	"strconv"
 	"time"
 
+	"github.com/gogo/protobuf/jsonpb"
+	"github.com/grafana/tempo/pkg/tempopb"
 	"github.com/opentracing/opentracing-go"
 	ot_log "github.com/opentracing/opentracing-go/log"
 	"github.com/weaveworks/common/user"
@@ -25,48 +29,45 @@ const (
 	ProtobufTypeHeaderValue = "application/protobuf"
 )
 
+const (
+	serviceSearchTag     = "service.name"
+	operationSearchTag   = "name"
+	minDurationSearchTag = "minDuration"
+	maxDurationSearchTag = "maxDuration"
+	numTracesSearchTag   = "limit"
+)
+
 type Backend struct {
-	tempoEndpoint string
+	tempoBackend string
 }
 
 func New(cfg *Config) *Backend {
 	return &Backend{
-		tempoEndpoint: "http://" + cfg.Backend + "/api/traces/",
+		tempoBackend: cfg.Backend,
 	}
 }
 
 func (b *Backend) GetDependencies(ctx context.Context, endTs time.Time, lookback time.Duration) ([]jaeger.DependencyLink, error) {
 	return nil, nil
 }
+
 func (b *Backend) GetTrace(ctx context.Context, traceID jaeger.TraceID) (*jaeger.Trace, error) {
-	hexID := fmt.Sprintf("%016x%016x", traceID.High, traceID.Low)
+	url := fmt.Sprintf("http://%s/api/traces/%s", b.tempoBackend, traceID)
 
-	span, _ := opentracing.StartSpanFromContext(ctx, "GetTrace")
+	span, ctx := opentracing.StartSpanFromContext(ctx, "tempo-query.GetTrace")
 	defer span.Finish()
 
-	req, err := http.NewRequestWithContext(ctx, "GET", b.tempoEndpoint+hexID, nil)
+	req, err := b.newGetRequest(ctx, url, span)
 	if err != nil {
 		return nil, err
 	}
 
-	if tracer := opentracing.GlobalTracer(); tracer != nil {
-		// this is not really loggable or anything we can react to.  just ignoring this error
-		_ = tracer.Inject(span.Context(), opentracing.HTTPHeaders, opentracing.HTTPHeadersCarrier(req.Header))
-	}
-
-	// currently Jaeger Query will only propagate bearer token to the grpc backend and no other headers
-	// so we are going to extract the tenant id from the header, if it exists and use it
-	tenantID, found := extractBearerToken(ctx)
-	if found {
-		req.Header.Set(user.OrgIDHeaderName, tenantID)
-	}
-
-	// Set content type to grpc
+	// Set content type to GRPC
 	req.Header.Set(AcceptHeaderKey, ProtobufTypeHeaderValue)
 
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
-		return nil, fmt.Errorf("failed get to tempo %w", err)
+		return nil, fmt.Errorf("failed GET to tempo %w", err)
 	}
 	defer resp.Body.Close()
 
@@ -91,7 +92,7 @@ func (b *Backend) GetTrace(ctx context.Context, traceID jaeger.TraceID) (*jaeger
 
 	jaegerBatches, err := ot_jaeger.InternalTracesToJaegerProto(otTrace)
 	if err != nil {
-		return nil, fmt.Errorf("error translating to jaegerBatches %v: %w", hexID, err)
+		return nil, fmt.Errorf("error translating to jaegerBatches %v: %w", traceID, err)
 	}
 
 	jaegerTrace := &jaeger.Trace{
@@ -117,21 +118,186 @@ func (b *Backend) GetTrace(ctx context.Context, traceID jaeger.TraceID) (*jaeger
 }
 
 func (b *Backend) GetServices(ctx context.Context) ([]string, error) {
-	return nil, nil
+	span, ctx := opentracing.StartSpanFromContext(ctx, "tempo-query.GetOperations")
+	defer span.Finish()
+
+	return b.lookupTagValues(ctx, span, serviceSearchTag)
 }
+
 func (b *Backend) GetOperations(ctx context.Context, query jaeger_spanstore.OperationQueryParameters) ([]jaeger_spanstore.Operation, error) {
-	return nil, nil
+	span, ctx := opentracing.StartSpanFromContext(ctx, "tempo-query.GetOperations")
+	defer span.Finish()
+
+	tagValues, err := b.lookupTagValues(ctx, span, operationSearchTag)
+	if err != nil {
+		return nil, err
+	}
+
+	var operations []jaeger_spanstore.Operation
+	for _, value := range tagValues {
+		operations = append(operations, jaeger_spanstore.Operation{
+			Name:     value,
+			SpanKind: "",
+		})
+	}
+
+	return operations, nil
+
 }
+
 func (b *Backend) FindTraces(ctx context.Context, query *jaeger_spanstore.TraceQueryParameters) ([]*jaeger.Trace, error) {
-	return nil, nil
+	span, ctx := opentracing.StartSpanFromContext(ctx, "tempo-query.FindTraces")
+	defer span.Finish()
+
+	traceIDs, err := b.FindTraceIDs(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+
+	span.LogFields(ot_log.String("msg", fmt.Sprintf("Found %d trace IDs", len(traceIDs))))
+
+	// for every traceID, get the full trace
+	var jaegerTraces []*jaeger.Trace
+	for _, traceID := range traceIDs {
+		trace, err := b.GetTrace(ctx, traceID)
+		if err != nil {
+			// TODO this seems to be an internal inconsistency error, ignore so we can still show the rest
+			span.LogFields(ot_log.Error(fmt.Errorf("could not get trace for traceID %v: %w", traceID, err)))
+			continue
+		}
+
+		jaegerTraces = append(jaegerTraces, trace)
+	}
+
+	span.LogFields(ot_log.String("msg", fmt.Sprintf("Returning %d traces", len(jaegerTraces))))
+
+	return jaegerTraces, nil
 }
+
 func (b *Backend) FindTraceIDs(ctx context.Context, query *jaeger_spanstore.TraceQueryParameters) ([]jaeger.TraceID, error) {
-	return nil, nil
+	span, ctx := opentracing.StartSpanFromContext(ctx, "tempo-query.FindTraceIDs")
+	defer span.Finish()
+
+	url := url.URL{
+		Scheme: "http",
+		Host:   b.tempoBackend,
+		Path:   "api/search",
+	}
+	urlQuery := url.Query()
+	urlQuery.Set(serviceSearchTag, query.ServiceName)
+	urlQuery.Set(operationSearchTag, query.OperationName)
+	urlQuery.Set(minDurationSearchTag, query.DurationMin.String())
+	urlQuery.Set(maxDurationSearchTag, query.DurationMax.String())
+	urlQuery.Set(numTracesSearchTag, strconv.Itoa(query.NumTraces))
+	for k, v := range query.Tags {
+		urlQuery.Set(k, v)
+	}
+	url.RawQuery = urlQuery.Encode()
+
+	req, err := b.newGetRequest(ctx, url.String(), span)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed GET to tempo %w", err)
+	}
+	defer resp.Body.Close()
+
+	// if search endpoint returns 404, search is most likely not enabled
+	if resp.StatusCode == http.StatusNotFound {
+		return nil, nil
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return nil, fmt.Errorf("error reading response from Tempo: got %s", resp.Status)
+		}
+		return nil, fmt.Errorf("%s", body)
+	}
+
+	var searchResponse tempopb.SearchResponse
+	err = jsonpb.Unmarshal(resp.Body, &searchResponse)
+	if err != nil {
+		return nil, fmt.Errorf("error unmarshaling Tempo response: %w", err)
+	}
+
+	jaegerTraceIDs := make([]jaeger.TraceID, len(searchResponse.Traces))
+
+	for i, traceMetadata := range searchResponse.Traces {
+		jaegerTraceID, err := jaeger.TraceIDFromString(traceMetadata.TraceID)
+		if err != nil {
+			return nil, fmt.Errorf("could not convert traceID into Jaeger's traceID %w", err)
+		}
+		jaegerTraceIDs[i] = jaegerTraceID
+	}
+
+	return jaegerTraceIDs, nil
 }
+
+func (b *Backend) lookupTagValues(ctx context.Context, span opentracing.Span, tagName string) ([]string, error) {
+	url := fmt.Sprintf("http://%s/api/search/tag/%s/values", b.tempoBackend, tagName)
+
+	req, err := b.newGetRequest(ctx, url, span)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed GET to tempo %w", err)
+	}
+	defer resp.Body.Close()
+
+	// if search endpoint returns 404, search is most likely not enabled
+	if resp.StatusCode == http.StatusNotFound {
+		return nil, nil
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		body, err := io.ReadAll(resp.Body)
+		if err != nil {
+			return nil, fmt.Errorf("error reading response from Tempo: got %s", resp.Status)
+		}
+		return nil, fmt.Errorf("%s", body)
+	}
+
+	var searchLookupResponse tempopb.SearchTagValuesResponse
+	err = jsonpb.Unmarshal(resp.Body, &searchLookupResponse)
+	if err != nil {
+		return nil, fmt.Errorf("error unmarshaling Tempo response: %w", err)
+	}
+
+	return searchLookupResponse.TagValues, nil
+}
+
 func (b *Backend) WriteSpan(ctx context.Context, span *jaeger.Span) error {
 	return nil
 }
 
+func (b *Backend) newGetRequest(ctx context.Context, url string, span opentracing.Span) (*http.Request, error) {
+	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	if tracer := opentracing.GlobalTracer(); tracer != nil {
+		// this is not really loggable or anything we can react to.  just ignoring this error
+		_ = tracer.Inject(span.Context(), opentracing.HTTPHeaders, opentracing.HTTPHeadersCarrier(req.Header))
+	}
+
+	// currently Jaeger Query will only propagate bearer token to the grpc backend and no other headers
+	// so we are going to extract the tenant id from the header, if it exists and use it
+	tenantID, found := extractBearerToken(ctx)
+	if found {
+		req.Header.Set(user.OrgIDHeaderName, tenantID)
+	}
+
+	return req, nil
+}
+
 func extractBearerToken(ctx context.Context) (string, bool) {
 	if md, ok := metadata.FromIncomingContext(ctx); ok {
 		values := md.Get(spanstore.BearerTokenKey)

cmd/tempo/app/app.go

@@ -46,6 +46,7 @@ type Config struct {
 	Target              string `yaml:"target,omitempty"`
 	AuthEnabled         bool   `yaml:"auth_enabled,omitempty"`
 	MultitenancyEnabled bool   `yaml:"multitenancy_enabled,omitempty"`
+	SearchEnabled       bool   `yaml:"search_enabled,omitempty"`
 	HTTPAPIPrefix       string `yaml:"http_api_prefix"`
 	UseOTelTracer       bool   `yaml:"use_otel_tracer,omitempty"`
 
@@ -68,6 +69,7 @@ func (c *Config) RegisterFlagsAndApplyDefaults(prefix string, f *flag.FlagSet) {
 	f.StringVar(&c.Target, "target", All, "target module")
 	f.BoolVar(&c.AuthEnabled, "auth.enabled", false, "Set to true to enable auth (deprecated: use multitenancy.enabled)")
 	f.BoolVar(&c.MultitenancyEnabled, "multitenancy.enabled", false, "Set to true to enable multitenancy.")
+	f.BoolVar(&c.SearchEnabled, "search.enabled", false, "Set to true to enable search (unstable).")
 	f.StringVar(&c.HTTPAPIPrefix, "http-api-prefix", "", "String prefix for all http api endpoints.")
 	f.BoolVar(&c.UseOTelTracer, "use-otel-tracer", false, "Set to true to replace the OpenTracing tracer with the OpenTelemetry tracer")