feat: first iteration on datadog tracing with spans

server/main.go

@@ -62,15 +62,22 @@ func main() {
 		log.Fatal().Err(err).Msg("error initializing kv store")
 	}
 
-	searchStore, err := search.NewStore(&config.DefaultConfig.Search)
+	var searchStore search.Store
+	if config.DefaultConfig.Tracing.Enabled {
+		searchStore, err = search.NewStoreWithMetrics(&config.DefaultConfig.Search)
+	} else {
+		searchStore, err = search.NewStore(&config.DefaultConfig.Search)
+	}
 	if err != nil {
 		log.Fatal().Err(err).Msg("error initializing search store")
 	}
 
 	tenantMgr := metadata.NewTenantManager(kvStore)
 	log.Info().Msg("initialized tenant manager")
+
 	txMgr := transaction.NewManager(kvStore)
 	log.Info().Msg("initialized transaction manager")
+
 	mx := muxer.NewMuxer(&config.DefaultConfig, tenantMgr, txMgr)
 	mx.RegisterServices(kvStore, searchStore, tenantMgr, txMgr)
 	if err := mx.Start(config.DefaultConfig.Server.Host, config.DefaultConfig.Server.Port); err != nil {

server/metrics/requests.go

@@ -15,68 +15,96 @@
 package metrics
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
 	"github.com/tigrisdata/tigris/server/config"
+	"github.com/tigrisdata/tigris/server/request"
+	ulog "github.com/tigrisdata/tigris/util/log"
 	"google.golang.org/grpc"
 )
 
 const (
-	AdminServiceName       = "tigrisdata.admin.v1.Admin"
-	SystemTigrisTenantName = "system"
+	AdminServiceName            = "tigrisdata.admin.v1.Admin"
+	SystemTigrisTenantName      = "system"
+	DefaultReportedTigrisTenant = "unknown"
 )
 
 type RequestEndpointMetadata struct {
-	serviceName string
-	methodInfo  grpc.MethodInfo
+	serviceName   string
+	methodInfo    grpc.MethodInfo
+	namespaceName string
 }
 
-func newRequestEndpointMetadata(serviceName string, methodInfo grpc.MethodInfo) RequestEndpointMetadata {
-	return RequestEndpointMetadata{serviceName: serviceName, methodInfo: methodInfo}
+func getNamespaceName(ctx context.Context) string {
+	namespace, err := request.GetNamespace(ctx)
+	if ulog.E(err) && err == request.ErrNamespaceNotFound {
+		return DefaultReportedTigrisTenant
+	} else {
+		return namespace
+	}
+}
+
+func newRequestEndpointMetadata(ctx context.Context, serviceName string, methodInfo grpc.MethodInfo) RequestEndpointMetadata {
+	return RequestEndpointMetadata{serviceName: serviceName, methodInfo: methodInfo, namespaceName: getNamespaceName(ctx)}
+}
+
+func (r *RequestEndpointMetadata) GetMethodName() string {
+	return strings.Split(r.methodInfo.Name, "/")[2]
+}
+
+func (r *RequestEndpointMetadata) GetServiceType() string {
+	if r.methodInfo.IsServerStream {
+		return "stream"
+	} else {
+		return "unary"
+	}
 }
 
-func (g *RequestEndpointMetadata) GetPreInitializedTags() map[string]string {
-	if g.serviceName == AdminServiceName {
+func (r *RequestEndpointMetadata) GetTags() map[string]string {
+	if r.serviceName == AdminServiceName {
 		return map[string]string{
-			"method":        g.methodInfo.Name,
-			"grpc_service":  g.serviceName,
-			"tigris_tenant": SystemTigrisTenantName,
+			"grpc_method":       r.methodInfo.Name,
+			"grpc_service":      r.serviceName,
+			"tigris_tenant":     r.namespaceName,
+			"grpc_service_type": r.GetServiceType(),
 		}
 	} else {
 		return map[string]string{
-			"method":        g.methodInfo.Name,
-			"grpc_service":  g.serviceName,
-			"tigris_tenant": DefaultReportedTigrisTenant,
+			"grpc_method":       r.methodInfo.Name,
+			"grpc_service":      r.serviceName,
+			"tigris_tenant":     r.namespaceName,
+			"grpc_service_type": r.GetServiceType(),
 		}
 	}
 }
 
-func (g *RequestEndpointMetadata) GetSpecificErrorTags(source string, code string) map[string]string {
-	if g.serviceName == AdminServiceName {
+func (r *RequestEndpointMetadata) GetSpecificErrorTags(source string, code string) map[string]string {
+	if r.serviceName == AdminServiceName {
 		return map[string]string{
-			"method":        g.methodInfo.Name,
-			"grpc_service":  g.serviceName,
-			"source":        source,
-			"code":          code,
+			"grpc_method":   r.methodInfo.Name,
+			"grpc_service":  r.serviceName,
+			"error_source":  source,
+			"error_code":    code,
 			"tigris_tenant": SystemTigrisTenantName,
 		}
 	} else {
 		return map[string]string{
-			"method":        g.methodInfo.Name,
-			"grpc_service":  g.serviceName,
-			"source":        source,
-			"code":          code,
+			"grpc_method":   r.methodInfo.Name,
+			"grpc_service":  r.serviceName,
+			"error_source":  source,
+			"error_code":    code,
 			"tigris_tenant": DefaultReportedTigrisTenant,
 		}
 	}
 }
 
-func (g *RequestEndpointMetadata) getFullMethod() string {
-	return fmt.Sprintf("/%s/%s", g.serviceName, g.methodInfo.Name)
+func (r *RequestEndpointMetadata) getFullMethod() string {
+	return fmt.Sprintf("/%s/%s", r.serviceName, r.methodInfo.Name)
 }
 
-func GetGrpcEndPointMetadataFromFullMethod(fullMethod string, methodType string) RequestEndpointMetadata {
+func GetGrpcEndPointMetadataFromFullMethod(ctx context.Context, fullMethod string, methodType string) RequestEndpointMetadata {
 	var methodInfo grpc.MethodInfo
 	methodList := strings.Split(fullMethod, "/")
 	svcName := methodList[1]
@@ -94,12 +122,7 @@ func GetGrpcEndPointMetadataFromFullMethod(fullMethod string, methodType string)
 			IsServerStream: true,
 		}
 	}
-	return newRequestEndpointMetadata(svcName, methodInfo)
-}
-
-func GetPreinitializedTagsFromFullMethod(fullMethod string, methodType string) map[string]string {
-	metaData := GetGrpcEndPointMetadataFromFullMethod(fullMethod, methodType)
-	return metaData.GetPreInitializedTags()
+	return newRequestEndpointMetadata(ctx, svcName, methodInfo)
 }
 
 func InitializeRequestScopes() {

server/metrics/requests_test.go

@@ -15,6 +15,7 @@
 package metrics
 
 import (
+	"context"
 	"fmt"
 	"testing"
 
@@ -44,29 +45,33 @@ func TestGRPCMetrics(t *testing.T) {
 		IsClientStream: false,
 	}
 
-	unaryEndPointMetadata := newRequestEndpointMetadata(svcName, unaryMethodInfo)
-	streamingEndpointMetadata := newRequestEndpointMetadata(svcName, streamingMethodInfo)
+	ctx := context.Background()
+
+	unaryEndPointMetadata := newRequestEndpointMetadata(ctx, svcName, unaryMethodInfo)
+	streamingEndpointMetadata := newRequestEndpointMetadata(ctx, svcName, streamingMethodInfo)
 
 	t.Run("Test GetGrpcEndPointMetadataFromFullMethod", func(t *testing.T) {
-		unaryMetadata := GetGrpcEndPointMetadataFromFullMethod(unaryEndPointMetadata.getFullMethod(), "unary")
+		unaryMetadata := GetGrpcEndPointMetadataFromFullMethod(ctx, unaryEndPointMetadata.getFullMethod(), "unary")
 		assert.Equal(t, unaryMetadata, unaryEndPointMetadata)
 
-		streamMetadata := GetGrpcEndPointMetadataFromFullMethod(streamingEndpointMetadata.getFullMethod(), "stream")
+		streamMetadata := GetGrpcEndPointMetadataFromFullMethod(ctx, streamingEndpointMetadata.getFullMethod(), "stream")
 		assert.Equal(t, streamMetadata, streamingEndpointMetadata)
 	})
 
 	t.Run("Test GetPreinitializedTagsFromFullMethod", func(t *testing.T) {
-		unaryTags := unaryEndPointMetadata.GetPreInitializedTags()
+		unaryTags := unaryEndPointMetadata.GetTags()
 		assert.Equal(t, unaryTags, map[string]string{
-			"method":        unaryMethodInfo.Name,
-			"grpc_service":  "tigrisdata.v1.Tigris",
-			"tigris_tenant": DefaultReportedTigrisTenant,
+			"grpc_method":       unaryMethodInfo.Name,
+			"grpc_service":      "tigrisdata.v1.Tigris",
+			"grpc_service_type": "unary",
+			"tigris_tenant":     DefaultReportedTigrisTenant,
 		})
-		streamTags := streamingEndpointMetadata.GetPreInitializedTags()
+		streamTags := streamingEndpointMetadata.GetTags()
 		assert.Equal(t, streamTags, map[string]string{
-			"method":        streamingMethodInfo.Name,
-			"grpc_service":  "tigrisdata.v1.Tigris",
-			"tigris_tenant": DefaultReportedTigrisTenant,
+			"grpc_method":       streamingMethodInfo.Name,
+			"grpc_service":      "tigrisdata.v1.Tigris",
+			"grpc_service_type": "stream",
+			"tigris_tenant":     DefaultReportedTigrisTenant,
 		})
 	})
 
@@ -88,10 +93,10 @@ func TestGRPCMetrics(t *testing.T) {
 	})
 
 	t.Run("Test unary method preinitialized tags", func(t *testing.T) {
-		tags := unaryEndPointMetadata.GetPreInitializedTags()
+		tags := unaryEndPointMetadata.GetTags()
 		for tagName, tagValue := range tags {
 			switch tagName {
-			case "method":
+			case "grpc_method":
 				assert.Equal(t, tagValue, "TestUnaryMethod")
 			case "grpc_service":
 				assert.Equal(t, tagValue, "tigrisdata.v1.Tigris")
@@ -100,10 +105,10 @@ func TestGRPCMetrics(t *testing.T) {
 	})
 
 	t.Run("Test streaming method preinitialized tags", func(t *testing.T) {
-		tags := streamingEndpointMetadata.GetPreInitializedTags()
+		tags := streamingEndpointMetadata.GetTags()
 		for tagName, tagValue := range tags {
 			switch tagName {
-			case "method":
+			case "grpc_method":
 				assert.Equal(t, tagValue, "TestStreamingMethod")
 			case "grpc_service":
 				assert.Equal(t, tagValue, "tigrisdata.v1.Tigris")

server/metrics/tenant.go

@@ -16,14 +16,11 @@ package metrics
 
 import (
 	"context"
+
 	"github.com/tigrisdata/tigris/server/request"
 	ulog "github.com/tigrisdata/tigris/util/log"
 )
 
-const (
-	DefaultReportedTigrisTenant string = "unknown"
-)
-
 func addTigrisTenantToTags(ctx context.Context, tags map[string]string) map[string]string {
 	namespace, err := request.GetNamespace(ctx)
 	if ulog.E(err) {

server/metrics/tracing.go

@@ -0,0 +1,72 @@
+// Copyright 2022 Tigris Data, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metrics
+
+import (
+	"context"
+
+	"github.com/tigrisdata/tigris/server/config"
+	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
+)
+
+const (
+	GrpcTracingServiceName      = "tigris.grpc"
+	KvTracingServiceName        = "tigris.fdb.kv"
+	TxManagerTracingServiceName = "tigris.tx.manager"
+)
+
+type SpanMeta struct {
+	serviceName  string
+	resourceName string
+	spanType     string
+	tags         map[string]string
+}
+
+func NewSpanMeta(serviceName string, resourceName string, spanType string, tags map[string]string) *SpanMeta {
+	return &SpanMeta{serviceName: serviceName, resourceName: resourceName, spanType: spanType, tags: tags}
+}
+
+func (s *SpanMeta) GetSpanOptions() []tracer.StartSpanOption {
+	return []tracer.StartSpanOption{
+		tracer.ServiceName(s.serviceName),
+		tracer.ResourceName(s.resourceName),
+		tracer.SpanType(s.spanType),
+		tracer.Measured(),
+	}
+}
+
+func (s *SpanMeta) StartTracing(ctx context.Context, childOnly bool) (context.Context, func()) {
+	if !config.DefaultConfig.Tracing.Enabled {
+		return ctx, func() {}
+	}
+	spanOpts := s.GetSpanOptions()
+	parentSpan, exists := tracer.SpanFromContext(ctx)
+	if exists {
+		// This is a child span, parents need to be marked
+		spanOpts = append(spanOpts, tracer.ChildOf(parentSpan.Context()))
+	}
+	if childOnly && !exists {
+		// There is no parent span, no need to start tracing here
+		return ctx, func() {}
+	}
+	span := tracer.StartSpan(s.resourceName, spanOpts...)
+	for k, v := range s.tags {
+		span.SetTag(k, v)
+	}
+	ctx = tracer.ContextWithSpan(ctx, span)
+	return ctx, func() {
+		span.Finish()
+	}
+}