Tracing improvements (#2962)

* Tracing improvements

Create separate TracerProviders per service to improve tracing when
running multiple services in a single process (like e.g. ocis does).
Previously all service were referring to the same package-level
TracerProvider (which, among other things, caused the services to
overwrite each others services names).

We still create a "global" TracerProvider to cover the cases where
we don't have access to the per service Provider yet. This will be
improved in a follow up change.

* Attach TracerProvider to context

To generate more useful tracing (especially when running multiple
services in a single process) we'd like to generate Tracers from the
per service TracerProvider() instead of the global default provider.
Therefore we now pass the TracerProvider via context (similar to what
is already done for the logger.

changelog/unreleased/per-service-trace.md

@@ -0,0 +1,7 @@
+Enhancement: Per service TracerProvider
+
+To improve tracing we create separate TracerProviders per service now.
+This is especially helpful when running multiple reva services in a single
+process (like e.g. oCIS does).
+
+https://github.com/cs3org/reva/pull/2962

cmd/revad/runtime/runtime.go

@@ -39,6 +39,7 @@ import (
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
+	"go.opentelemetry.io/otel/trace"
 )
 
 // Run runs a reva server with the given config file and pid file.
@@ -87,12 +88,10 @@ func run(mainConf map[string]interface{}, coreConf *coreConf, logger *zerolog.Lo
 	host, _ := os.Hostname()
 	logger.Info().Msgf("host info: %s", host)
 
-	if coreConf.TracingEnabled {
-		initTracing(coreConf)
-	}
+	tp := initTracing(coreConf)
 	initCPUCount(coreConf, logger)
 
-	servers := initServers(mainConf, logger)
+	servers := initServers(mainConf, logger, tp)
 	watcher, err := initWatcher(logger, filename)
 	if err != nil {
 		log.Panic(err)
@@ -121,10 +120,10 @@ func initWatcher(log *zerolog.Logger, filename string) (*grace.Watcher, error) {
 	return watcher, err
 }
 
-func initServers(mainConf map[string]interface{}, log *zerolog.Logger) map[string]grace.Server {
+func initServers(mainConf map[string]interface{}, log *zerolog.Logger, tp trace.TracerProvider) map[string]grace.Server {
 	servers := map[string]grace.Server{}
 	if isEnabledHTTP(mainConf) {
-		s, err := getHTTPServer(mainConf["http"], log)
+		s, err := getHTTPServer(mainConf["http"], log, tp)
 		if err != nil {
 			log.Error().Err(err).Msg("error creating http server")
 			os.Exit(1)
@@ -133,7 +132,7 @@ func initServers(mainConf map[string]interface{}, log *zerolog.Logger) map[strin
 	}
 
 	if isEnabledGRPC(mainConf) {
-		s, err := getGRPCServer(mainConf["grpc"], log)
+		s, err := getGRPCServer(mainConf["grpc"], log, tp)
 		if err != nil {
 			log.Error().Err(err).Msg("error creating grpc server")
 			os.Exit(1)
@@ -148,8 +147,9 @@ func initServers(mainConf map[string]interface{}, log *zerolog.Logger) map[strin
 	return servers
 }
 
-func initTracing(conf *coreConf) {
-	rtrace.SetTraceProvider(conf.TracingCollector, conf.TracingEndpoint, conf.TracingServiceName)
+func initTracing(conf *coreConf) trace.TracerProvider {
+	rtrace.InitDefaultTracerProvider(conf.TracingCollector, conf.TracingEndpoint)
+	return rtrace.GetTracerProvider(conf.TracingEnabled, conf.TracingCollector, conf.TracingEndpoint, conf.TracingServiceName)
 }
 
 func initCPUCount(conf *coreConf, log *zerolog.Logger) {
@@ -244,19 +244,19 @@ func getWriter(out string) (io.Writer, error) {
 	return fd, nil
 }
 
-func getGRPCServer(conf interface{}, l *zerolog.Logger) (*rgrpc.Server, error) {
+func getGRPCServer(conf interface{}, l *zerolog.Logger, tp trace.TracerProvider) (*rgrpc.Server, error) {
 	sub := l.With().Str("pkg", "rgrpc").Logger()
-	s, err := rgrpc.NewServer(conf, sub)
+	s, err := rgrpc.NewServer(conf, sub, tp)
 	if err != nil {
 		err = errors.Wrap(err, "main: error creating grpc server")
 		return nil, err
 	}
 	return s, nil
 }
 
-func getHTTPServer(conf interface{}, l *zerolog.Logger) (*rhttp.Server, error) {
+func getHTTPServer(conf interface{}, l *zerolog.Logger, tp trace.TracerProvider) (*rhttp.Server, error) {
 	sub := l.With().Str("pkg", "rhttp").Logger()
-	s, err := rhttp.New(conf, sub)
+	s, err := rhttp.New(conf, sub, tp)
 	if err != nil {
 		err = errors.Wrap(err, "main: error creating http server")
 		return nil, err

internal/grpc/interceptors/appctx/appctx.go

@@ -23,7 +23,6 @@ import (
 	"runtime"
 
 	"github.com/cs3org/reva/v2/pkg/appctx"
-	rtrace "github.com/cs3org/reva/v2/pkg/trace"
 	"github.com/rs/zerolog"
 	semconv "go.opentelemetry.io/otel/semconv/v1.10.0"
 	"go.opentelemetry.io/otel/trace"
@@ -34,12 +33,12 @@ import (
 const tracerName = "appctx"
 
 // NewUnary returns a new unary interceptor that creates the application context.
-func NewUnary(log zerolog.Logger) grpc.UnaryServerInterceptor {
+func NewUnary(log zerolog.Logger, tp trace.TracerProvider) grpc.UnaryServerInterceptor {
 	interceptor := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
 		span := trace.SpanFromContext(ctx)
 		defer span.End()
 		if !span.SpanContext().HasTraceID() {
-			ctx, span = rtrace.Provider.Tracer(tracerName).Start(ctx, "grpc unary")
+			ctx, span = tp.Tracer(tracerName).Start(ctx, "grpc unary")
 		}
 		_, file, _, ok := runtime.Caller(1)
 		if ok {
@@ -48,6 +47,7 @@ func NewUnary(log zerolog.Logger) grpc.UnaryServerInterceptor {
 
 		sub := log.With().Str("traceid", span.SpanContext().TraceID().String()).Logger()
 		ctx = appctx.WithLogger(ctx, &sub)
+		ctx = appctx.WithTracerProvider(ctx, tp)
 		res, err := handler(ctx, req)
 		return res, err
 	}
@@ -56,14 +56,14 @@ func NewUnary(log zerolog.Logger) grpc.UnaryServerInterceptor {
 
 // NewStream returns a new server stream interceptor
 // that creates the application context.
-func NewStream(log zerolog.Logger) grpc.StreamServerInterceptor {
+func NewStream(log zerolog.Logger, tp trace.TracerProvider) grpc.StreamServerInterceptor {
 	interceptor := func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
 		ctx := ss.Context()
 		span := trace.SpanFromContext(ctx)
 		defer span.End()
 
 		if !span.SpanContext().HasTraceID() {
-			ctx, span = rtrace.Provider.Tracer(tracerName).Start(ctx, "grpc stream")
+			ctx, span = tp.Tracer(tracerName).Start(ctx, "grpc stream")
 		}
 		_, file, _, ok := runtime.Caller(1)
 		if ok {

internal/grpc/interceptors/auth/auth.go

@@ -34,7 +34,6 @@ import (
 	"github.com/cs3org/reva/v2/pkg/sharedconf"
 	"github.com/cs3org/reva/v2/pkg/token"
 	tokenmgr "github.com/cs3org/reva/v2/pkg/token/manager/registry"
-	rtrace "github.com/cs3org/reva/v2/pkg/trace"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
@@ -70,7 +69,7 @@ func parseConfig(m map[string]interface{}) (*config, error) {
 
 // NewUnary returns a new unary interceptor that adds
 // trace information for the request.
-func NewUnary(m map[string]interface{}, unprotected []string) (grpc.UnaryServerInterceptor, error) {
+func NewUnary(m map[string]interface{}, unprotected []string, tp trace.TracerProvider) (grpc.UnaryServerInterceptor, error) {
 	conf, err := parseConfig(m)
 	if err != nil {
 		err = errors.Wrap(err, "auth: error parsing config")
@@ -101,7 +100,7 @@ func NewUnary(m map[string]interface{}, unprotected []string) (grpc.UnaryServerI
 		span := trace.SpanFromContext(ctx)
 		defer span.End()
 		if !span.SpanContext().HasTraceID() {
-			ctx, span = rtrace.Provider.Tracer(tracerName).Start(ctx, "grpc auth unary")
+			ctx, span = tp.Tracer(tracerName).Start(ctx, "grpc auth unary")
 		}
 
 		if utils.Skip(info.FullMethod, unprotected) {
@@ -150,7 +149,7 @@ func NewUnary(m map[string]interface{}, unprotected []string) (grpc.UnaryServerI
 
 // NewStream returns a new server stream interceptor
 // that adds trace information to the request.
-func NewStream(m map[string]interface{}, unprotected []string) (grpc.StreamServerInterceptor, error) {
+func NewStream(m map[string]interface{}, unprotected []string, tp trace.TracerProvider) (grpc.StreamServerInterceptor, error) {
 	conf, err := parseConfig(m)
 	if err != nil {
 		return nil, err
@@ -180,7 +179,7 @@ func NewStream(m map[string]interface{}, unprotected []string) (grpc.StreamServe
 		span := trace.SpanFromContext(ctx)
 		defer span.End()
 		if !span.SpanContext().HasTraceID() {
-			ctx, span = rtrace.Provider.Tracer(tracerName).Start(ctx, "grpc auth new stream")
+			ctx, span = tp.Tracer(tracerName).Start(ctx, "grpc auth new stream")
 		}
 
 		if utils.Skip(info.FullMethod, unprotected) {

internal/grpc/services/gateway/usershareprovider.go

@@ -33,7 +33,6 @@ import (
 	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/grants"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
-	rtrace "github.com/cs3org/reva/v2/pkg/trace"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/pkg/errors"
 )
@@ -190,8 +189,7 @@ func (s *svc) GetReceivedShare(ctx context.Context, req *collaboration.GetReceiv
 //   1) if received share is mounted: we also do a rename in the storage
 //   2) if received share is not mounted: we only rename in user share provider.
 func (s *svc) UpdateReceivedShare(ctx context.Context, req *collaboration.UpdateReceivedShareRequest) (*collaboration.UpdateReceivedShareResponse, error) {
-	t := rtrace.Provider.Tracer("reva")
-	ctx, span := t.Start(ctx, "Gateway.UpdateReceivedShare")
+	ctx, span := appctx.GetTracerProvider(ctx).Tracer("gateway").Start(ctx, "Gateway.UpdateReceivedShare")
 	defer span.End()
 
 	// sanity checks

internal/grpc/services/publicstorageprovider/publicstorageprovider.go

@@ -38,7 +38,6 @@ import (
 	"github.com/cs3org/reva/v2/pkg/rgrpc/status"
 	"github.com/cs3org/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
-	rtrace "github.com/cs3org/reva/v2/pkg/trace"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
@@ -555,7 +554,7 @@ func (s *service) CreateContainer(ctx context.Context, req *provider.CreateConta
 		_, req.Ref.ResourceId.StorageId = storagespace.SplitStorageID(req.Ref.ResourceId.StorageId)
 	}
 
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(ctx, "CreateContainer")
+	ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "CreateContainer")
 	defer span.End()
 
 	span.SetAttributes(attribute.KeyValue{
@@ -616,7 +615,7 @@ func (s *service) Delete(ctx context.Context, req *provider.DeleteRequest) (*pro
 		_, req.Ref.ResourceId.StorageId = storagespace.SplitStorageID(req.Ref.ResourceId.StorageId)
 	}
 
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(ctx, "Delete")
+	ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Delete")
 	defer span.End()
 
 	span.SetAttributes(attribute.KeyValue{
@@ -663,7 +662,7 @@ func (s *service) Move(ctx context.Context, req *provider.MoveRequest) (*provide
 		_, req.Destination.ResourceId.StorageId = storagespace.SplitStorageID(req.Destination.ResourceId.StorageId)
 	}
 
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(ctx, "Move")
+	ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Move")
 	defer span.End()
 
 	span.SetAttributes(
@@ -730,7 +729,7 @@ func (s *service) Stat(ctx context.Context, req *provider.StatRequest) (*provide
 		_, req.Ref.ResourceId.StorageId = storagespace.SplitStorageID(req.Ref.ResourceId.StorageId)
 	}
 
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(ctx, "Stat")
+	ctx, span := appctx.GetTracerProvider(ctx).Tracer(tracerName).Start(ctx, "Stat")
 	defer span.End()
 
 	span.SetAttributes(

internal/grpc/services/storageprovider/storageprovider.go

@@ -714,7 +714,7 @@ func (s *service) Stat(ctx context.Context, req *provider.StatRequest) (*provide
 	providerID := unwrapProviderID(req.Ref.GetResourceId())
 	defer rewrapProviderID(req.Ref.GetResourceId(), providerID)
 
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(ctx, "stat")
+	ctx, span := rtrace.DefaultProvider().Tracer(tracerName).Start(ctx, "stat")
 	defer span.End()
 
 	span.SetAttributes(attribute.KeyValue{

internal/http/interceptors/appctx/appctx.go

@@ -25,7 +25,6 @@ import (
 	"net/http"
 
 	"github.com/cs3org/reva/v2/pkg/appctx"
-	rtrace "github.com/cs3org/reva/v2/pkg/trace"
 	"github.com/rs/zerolog"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -35,25 +34,26 @@ const tracerName = "appctx"
 
 // New returns a new HTTP middleware that stores the log
 // in the context with request ID information.
-func New(log zerolog.Logger) func(http.Handler) http.Handler {
+func New(log zerolog.Logger, tp trace.TracerProvider) func(http.Handler) http.Handler {
 	chain := func(h http.Handler) http.Handler {
-		return handler(log, h)
+		return handler(log, tp, h)
 	}
 	return chain
 }
 
-func handler(log zerolog.Logger, h http.Handler) http.Handler {
+func handler(log zerolog.Logger, tp trace.TracerProvider, h http.Handler) http.Handler {
 
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()
 		span := trace.SpanFromContext(ctx)
 		defer span.End()
 		if !span.SpanContext().HasTraceID() {
-			ctx, span = rtrace.Provider.Tracer(tracerName).Start(ctx, "http interceptor")
+			ctx, span = tp.Tracer(tracerName).Start(ctx, "http interceptor")
 		}
 
 		sub := log.With().Str("traceid", span.SpanContext().TraceID().String()).Logger()
 		ctx = appctx.WithLogger(ctx, &sub)
+		ctx = appctx.WithTracerProvider(ctx, tp)
 		r = r.WithContext(ctx)
 		h.ServeHTTP(w, r)
 	})

internal/http/interceptors/auth/auth.go

@@ -43,7 +43,6 @@ import (
 	"github.com/cs3org/reva/v2/pkg/sharedconf"
 	"github.com/cs3org/reva/v2/pkg/token"
 	tokenmgr "github.com/cs3org/reva/v2/pkg/token/manager/registry"
-	rtrace "github.com/cs3org/reva/v2/pkg/trace"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
@@ -84,7 +83,7 @@ func parseConfig(m map[string]interface{}) (*config, error) {
 }
 
 // New returns a new middleware with defined priority.
-func New(m map[string]interface{}, unprotected []string) (global.Middleware, error) {
+func New(m map[string]interface{}, unprotected []string, tp trace.TracerProvider) (global.Middleware, error) {
 	conf, err := parseConfig(m)
 	if err != nil {
 		return nil, err
@@ -168,7 +167,7 @@ func New(m map[string]interface{}, unprotected []string) (global.Middleware, err
 			span := trace.SpanFromContext(ctx)
 			defer span.End()
 			if !span.SpanContext().HasTraceID() {
-				_, span = rtrace.Provider.Tracer(tracerName).Start(ctx, "http auth interceptor")
+				_, span = tp.Tracer(tracerName).Start(ctx, "http auth interceptor")
 			}
 
 			if r.Method == "OPTIONS" {

internal/http/services/owncloud/ocdav/copy.go

@@ -37,7 +37,6 @@ import (
 	"github.com/cs3org/reva/v2/pkg/appctx"
 	"github.com/cs3org/reva/v2/pkg/rhttp"
 	"github.com/cs3org/reva/v2/pkg/rhttp/router"
-	rtrace "github.com/cs3org/reva/v2/pkg/trace"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/rs/zerolog"
 )
@@ -51,7 +50,7 @@ type copy struct {
 }
 
 func (s *svc) handlePathCopy(w http.ResponseWriter, r *http.Request, ns string) {
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(r.Context(), "copy")
+	ctx, span := s.tracerProvider.Tracer(tracerName).Start(r.Context(), "copy")
 	defer span.End()
 
 	if s.c.EnableHTTPTpc {
@@ -291,7 +290,7 @@ func (s *svc) executePathCopy(ctx context.Context, client gateway.GatewayAPIClie
 }
 
 func (s *svc) handleSpacesCopy(w http.ResponseWriter, r *http.Request, spaceID string) {
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(r.Context(), "spaces_copy")
+	ctx, span := s.tracerProvider.Tracer(tracerName).Start(r.Context(), "spaces_copy")
 	defer span.End()
 
 	dh := r.Header.Get(net.HeaderDestination)

internal/http/services/owncloud/ocdav/delete.go

@@ -31,7 +31,6 @@ import (
 	"github.com/cs3org/reva/v2/internal/http/services/owncloud/ocdav/spacelookup"
 	"github.com/cs3org/reva/v2/pkg/appctx"
 	"github.com/cs3org/reva/v2/pkg/rgrpc/status"
-	rtrace "github.com/cs3org/reva/v2/pkg/trace"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/rs/zerolog"
 )
@@ -61,7 +60,7 @@ func (s *svc) handlePathDelete(w http.ResponseWriter, r *http.Request, ns string
 }
 
 func (s *svc) handleDelete(ctx context.Context, w http.ResponseWriter, r *http.Request, ref *provider.Reference, log zerolog.Logger) {
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(ctx, "delete")
+	ctx, span := s.tracerProvider.Tracer(tracerName).Start(ctx, "delete")
 	defer span.End()
 
 	req := &provider.DeleteRequest{Ref: ref}
@@ -130,7 +129,7 @@ func (s *svc) handleDelete(ctx context.Context, w http.ResponseWriter, r *http.R
 
 func (s *svc) handleSpacesDelete(w http.ResponseWriter, r *http.Request, spaceID string) {
 	ctx := r.Context()
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(ctx, "spaces_delete")
+	ctx, span := s.tracerProvider.Tracer(tracerName).Start(ctx, "spaces_delete")
 	defer span.End()
 
 	sublog := appctx.GetLogger(ctx).With().Logger()

internal/http/services/owncloud/ocdav/get.go

@@ -38,13 +38,12 @@ import (
 	"github.com/cs3org/reva/v2/pkg/appctx"
 	"github.com/cs3org/reva/v2/pkg/rhttp"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
-	rtrace "github.com/cs3org/reva/v2/pkg/trace"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/rs/zerolog"
 )
 
 func (s *svc) handlePathGet(w http.ResponseWriter, r *http.Request, ns string) {
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(r.Context(), "get")
+	ctx, span := s.tracerProvider.Tracer(tracerName).Start(r.Context(), "get")
 	defer span.End()
 
 	fn := path.Join(ns, r.URL.Path)
@@ -179,7 +178,7 @@ func (s *svc) handleGet(ctx context.Context, w http.ResponseWriter, r *http.Requ
 }
 
 func (s *svc) handleSpacesGet(w http.ResponseWriter, r *http.Request, spaceID string) {
-	ctx, span := rtrace.Provider.Tracer(tracerName).Start(r.Context(), "spaces_get")
+	ctx, span := s.tracerProvider.Tracer(tracerName).Start(r.Context(), "spaces_get")
 	defer span.End()
 
 	sublog := appctx.GetLogger(ctx).With().Str("path", r.URL.Path).Str("spaceid", spaceID).Str("handler", "get").Logger()