Skip to content

Commit

Permalink
Tracing improvements
Browse files Browse the repository at this point in the history 
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.
  • Loading branch information
@rhafer
rhafer committed Jun 14, 2022
1 parent 03c87b6 commit be028fa
Show file tree
Hide file tree
Showing 30 changed files with 157 additions and 139 deletions.
26 changes: 13 additions & 13 deletions cmd/revad/runtime/runtime.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand All @@ -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) {
Expand Down Expand Up @@ -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
Expand Down
9 changes: 4 additions & 5 deletions internal/grpc/interceptors/appctx/appctx.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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 {
Expand All @@ -56,14 +55,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 {
Expand Down
9 changes: 4 additions & 5 deletions internal/grpc/interceptors/auth/auth.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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) {
Expand Down
2 changes: 1 addition & 1 deletion internal/grpc/services/gateway/usershareprovider.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -190,7 +190,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")
t := rtrace.DefaultProvider().Tracer("reva")
ctx, span := t.Start(ctx, "Gateway.UpdateReceivedShare")
defer span.End()

Expand Down
8 changes: 4 additions & 4 deletions internal/grpc/services/publicstorageprovider/publicstorageprovider.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -555,7 +555,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 := rtrace.DefaultProvider().Tracer(tracerName).Start(ctx, "CreateContainer")
defer span.End()

span.SetAttributes(attribute.KeyValue{
Expand Down Expand Up @@ -616,7 +616,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 := rtrace.DefaultProvider().Tracer(tracerName).Start(ctx, "Delete")
defer span.End()

span.SetAttributes(attribute.KeyValue{
Expand Down Expand Up @@ -663,7 +663,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 := rtrace.DefaultProvider().Tracer(tracerName).Start(ctx, "Move")
defer span.End()

span.SetAttributes(
Expand Down Expand Up @@ -730,7 +730,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 := rtrace.DefaultProvider().Tracer(tracerName).Start(ctx, "Stat")
defer span.End()

span.SetAttributes(
Expand Down
2 changes: 1 addition & 1 deletion internal/grpc/services/storageprovider/storageprovider.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand Down
9 changes: 4 additions & 5 deletions internal/http/interceptors/appctx/appctx.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand All @@ -35,21 +34,21 @@ 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()
Expand Down
5 changes: 2 additions & 3 deletions internal/http/interceptors/auth/auth.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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" {
Expand Down
5 changes: 2 additions & 3 deletions internal/http/services/owncloud/ocdav/copy.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand All @@ -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 {
Expand Down Expand Up @@ -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)
Expand Down
5 changes: 2 additions & 3 deletions internal/http/services/owncloud/ocdav/delete.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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}
Expand Down Expand Up @@ -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()
Expand Down
5 changes: 2 additions & 3 deletions internal/http/services/owncloud/ocdav/get.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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()
Expand Down
5 changes: 2 additions & 3 deletions internal/http/services/owncloud/ocdav/head.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@ import (
"time"

"github.com/cs3org/reva/v2/pkg/storagespace"
rtrace "github.com/cs3org/reva/v2/pkg/trace"

rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
Expand All @@ -42,7 +41,7 @@ import (
)

func (s *svc) handlePathHead(w http.ResponseWriter, r *http.Request, ns string) {
ctx, span := rtrace.Provider.Tracer(tracerName).Start(r.Context(), "head")
ctx, span := s.tracerProvider.Tracer(tracerName).Start(r.Context(), "head")
defer span.End()

fn := path.Join(ns, r.URL.Path)
Expand Down Expand Up @@ -109,7 +108,7 @@ func (s *svc) handleHead(ctx context.Context, w http.ResponseWriter, r *http.Req
}

func (s *svc) handleSpacesHead(w http.ResponseWriter, r *http.Request, spaceID string) {
ctx, span := rtrace.Provider.Tracer(tracerName).Start(r.Context(), "spaces_head")
ctx, span := s.tracerProvider.Tracer(tracerName).Start(r.Context(), "spaces_head")
defer span.End()

sublog := appctx.GetLogger(ctx).With().Str("spaceid", spaceID).Str("path", r.URL.Path).Logger()
Expand Down
Loading

0 comments on commit be028fa

Please sign in to comment.