trace proxie middlewares (#6313)

* trace proxie middlewares Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de> * Update ocis-pkg/service/grpc/client.go Co-authored-by: Christian Richter <1058116+dragonchaser@users.noreply.github.com> * default tls is off Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de> --------- Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de> Co-authored-by: Christian Richter <1058116+dragonchaser@users.noreply.github.com>
2023-05-27 10:18:24 +02:00
parent 12dff34442
commit 632b206675
95 changed files with 2409 additions and 3195 deletions
@@ -1,31 +0,0 @@
-# OpenCensus wrappers
-
-OpenCensus wrappers propagate traces (spans) accross services.
-
-## Usage
-
-```go
-service := micro.NewService(
-    micro.Name("go.micro.srv.greeter"),
-    micro.WrapClient(opencensus.NewClientWrapper()),
-    micro.WrapHandler(opencensus.NewHandlerWrapper()),
-    micro.WrapSubscriber(opencensus.NewSubscriberWrapper()),
-)
-```
-
-### Views
-
-The OpenCensus package exposes some convenience views.
-Don't forget to register these views:
-
-```go
-// Register to all RPC server views.
-if err := view.Register(opencensus.DefaultServerViews...); err != nil {
-    log.Fatal(err)
-}
-
-// Register to all RPC client views.
-if err := view.Register(opencensus.DefaultClientViews...); err != nil {
-    log.Fatal(err)
-}
-```
@@ -1,150 +0,0 @@
-// Package opencensus provides wrappers for OpenCensus tracing.
-package opencensus
-
-import (
-	"context"
-	"encoding/base64"
-	"fmt"
-
-	"go-micro.dev/v4/client"
-	log "go-micro.dev/v4/logger"
-	"go-micro.dev/v4/metadata"
-	"go-micro.dev/v4/server"
-	"go.opencensus.io/trace"
-	"go.opencensus.io/trace/propagation"
-)
-
-const (
-	// TracePropagationField is the key for the tracing context
-	// that will be injected in go-micro's metadata.
-	TracePropagationField = "X-Trace-Context"
-)
-
-// clientWrapper wraps an RPC client and adds tracing.
-type clientWrapper struct {
-	client.Client
-}
-
-func injectTraceIntoCtx(ctx context.Context, span *trace.Span) context.Context {
-	spanCtx := propagation.Binary(span.SpanContext())
-	return metadata.Set(ctx, TracePropagationField, base64.RawStdEncoding.EncodeToString(spanCtx))
-}
-
-// Call implements client.Client.Call.
-func (w *clientWrapper) Call(
-	ctx context.Context,
-	req client.Request,
-	rsp interface{},
-	opts ...client.CallOption) (err error) {
-	t := newRequestTracker(req, ClientProfile)
-	ctx = t.start(ctx, true)
-
-	defer func() { t.end(ctx, err) }()
-
-	ctx = injectTraceIntoCtx(ctx, t.span)
-
-	err = w.Client.Call(ctx, req, rsp, opts...)
-	return
-}
-
-// Publish implements client.Client.Publish.
-func (w *clientWrapper) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) (err error) {
-	t := newEventTracker(p, ClientProfile)
-	ctx = t.start(ctx, true)
-
-	defer func() { t.end(ctx, err) }()
-
-	ctx = injectTraceIntoCtx(ctx, t.span)
-
-	err = w.Client.Publish(ctx, p, opts...)
-	return
-}
-
-// NewClientWrapper returns a client.Wrapper
-// that adds monitoring to outgoing requests.
-func NewClientWrapper() client.Wrapper {
-	return func(c client.Client) client.Client {
-		return &clientWrapper{c}
-	}
-}
-
-func getTraceFromCtx(ctx context.Context) *trace.SpanContext {
-	encodedTraceCtx, ok := metadata.Get(ctx, TracePropagationField)
-	if !ok {
-		return nil
-	}
-
-	traceCtxBytes, err := base64.RawStdEncoding.DecodeString(encodedTraceCtx)
-	if err != nil {
-		log.Errorf("Could not decode trace context: %s", err.Error())
-		return nil
-	}
-
-	spanCtx, ok := propagation.FromBinary(traceCtxBytes)
-	if !ok {
-		log.Errorf("Could not decode trace context from binary")
-		return nil
-	}
-
-	return &spanCtx
-}
-
-// NewHandlerWrapper returns a server.HandlerWrapper
-// that adds tracing to incoming requests.
-func NewHandlerWrapper() server.HandlerWrapper {
-	return func(fn server.HandlerFunc) server.HandlerFunc {
-		return func(ctx context.Context, req server.Request, rsp interface{}) (err error) {
-			t := newRequestTracker(req, ServerProfile)
-			ctx = t.start(ctx, false)
-
-			defer func() { t.end(ctx, err) }()
-
-			spanCtx := getTraceFromCtx(ctx)
-			if spanCtx != nil {
-				ctx, t.span = trace.StartSpanWithRemoteParent(
-					ctx,
-					fmt.Sprintf("rpc/%s/%s/%s", ServerProfile.Role, req.Service(), req.Endpoint()),
-					*spanCtx,
-				)
-			} else {
-				ctx, t.span = trace.StartSpan(
-					ctx,
-					fmt.Sprintf("rpc/%s/%s/%s", ServerProfile.Role, req.Service(), req.Endpoint()),
-				)
-			}
-
-			err = fn(ctx, req, rsp)
-			return
-		}
-	}
-}
-
-// NewSubscriberWrapper returns a server.SubscriberWrapper
-// that adds tracing to subscription requests.
-func NewSubscriberWrapper() server.SubscriberWrapper {
-	return func(fn server.SubscriberFunc) server.SubscriberFunc {
-		return func(ctx context.Context, p server.Message) (err error) {
-			t := newEventTracker(p, ServerProfile)
-			ctx = t.start(ctx, false)
-
-			defer func() { t.end(ctx, err) }()
-
-			spanCtx := getTraceFromCtx(ctx)
-			if spanCtx != nil {
-				ctx, t.span = trace.StartSpanWithRemoteParent(
-					ctx,
-					fmt.Sprintf("rpc/%s/pubsub/%s", ServerProfile.Role, p.Topic()),
-					*spanCtx,
-				)
-			} else {
-				ctx, t.span = trace.StartSpan(
-					ctx,
-					fmt.Sprintf("rpc/%s/pubsub/%s", ServerProfile.Role, p.Topic()),
-				)
-			}
-
-			err = fn(ctx, p)
-			return
-		}
-	}
-}
@@ -1,142 +0,0 @@
-package opencensus
-
-import (
-	"go.opencensus.io/stats"
-	"go.opencensus.io/stats/view"
-	"go.opencensus.io/tag"
-)
-
-// The following client RPC measures are supported for use in custom views.
-var (
-	ClientRequestCount = stats.Int64("opencensus.io/rpc/client/request_count", "Number of RPC requests started", stats.UnitNone)
-	ClientLatency      = stats.Float64("opencensus.io/rpc/client/latency", "End-to-end latency", stats.UnitMilliseconds)
-)
-
-// The following server RPC measures are supported for use in custom views.
-var (
-	ServerRequestCount = stats.Int64("opencensus.io/rpc/server/request_count", "Number of RPC requests received", stats.UnitNone)
-	ServerLatency      = stats.Float64("opencensus.io/rpc/server/latency", "End-to-end latency", stats.UnitMilliseconds)
-)
-
-// The following tags are applied to stats recorded by this package.
-// Service and Method are applied to all measures.
-// StatusCode is not applied to ClientRequestCount or ServerRequestCount,
-// since it is recorded before the status is known.
-var (
-	// StatusCode is the RPC status code.
-	StatusCode, _ = tag.NewKey("rpc.status")
-
-	// Service is the name of the micro-service.
-	Service, _ = tag.NewKey("rpc.service")
-
-	// Method is the service method called.
-	Endpoint, _ = tag.NewKey("rpc.endpoint")
-)
-
-// Default distributions used by views in this package.
-var (
-	DefaultLatencyDistribution = view.Distribution(0, 1, 2, 3, 4, 5, 6, 8, 10, 13, 16, 20, 25, 30, 40, 50, 65, 80, 100, 130, 160, 200, 250, 300, 400, 500, 650, 800, 1000, 2000, 5000, 10000, 20000, 50000, 100000)
-)
-
-// This package provides some convenience views.
-// You need to subscribe to the views for data to actually be collected.
-var (
-	ClientRequestCountView = &view.View{
-		Name:        "opencensus.io/rpc/client/request_count",
-		Description: "Count of RPC requests started",
-		Measure:     ClientRequestCount,
-		Aggregation: view.Count(),
-	}
-
-	ClientLatencyView = &view.View{
-		Name:        "opencensus.io/rpc/client/latency",
-		Description: "Latency distribution of RPC requests",
-		Measure:     ClientLatency,
-		Aggregation: DefaultLatencyDistribution,
-	}
-
-	ClientRequestCountByMethod = &view.View{
-		Name:        "opencensus.io/rpc/client/request_count_by_method",
-		Description: "Client request count by RPC method",
-		TagKeys:     []tag.Key{Endpoint},
-		Measure:     ClientRequestCount,
-		Aggregation: view.Count(),
-	}
-
-	ClientResponseCountByStatusCode = &view.View{
-		Name:        "opencensus.io/rpc/client/response_count_by_status_code",
-		Description: "Client response count by RPC status code",
-		TagKeys:     []tag.Key{StatusCode},
-		Measure:     ClientLatency,
-		Aggregation: view.Count(),
-	}
-
-	ServerRequestCountView = &view.View{
-		Name:        "opencensus.io/rpc/server/request_count",
-		Description: "Count of RPC requests received",
-		Measure:     ServerRequestCount,
-		Aggregation: view.Count(),
-	}
-
-	ServerLatencyView = &view.View{
-		Name:        "opencensus.io/rpc/server/latency",
-		Description: "Latency distribution of RPC requests",
-		Measure:     ServerLatency,
-		Aggregation: DefaultLatencyDistribution,
-	}
-
-	ServerRequestCountByMethod = &view.View{
-		Name:        "opencensus.io/rpc/server/request_count_by_method",
-		Description: "Server request count by RPC method",
-		TagKeys:     []tag.Key{Endpoint},
-		Measure:     ServerRequestCount,
-		Aggregation: view.Count(),
-	}
-
-	ServerResponseCountByStatusCode = &view.View{
-		Name:        "opencensus.io/rpc/server/response_count_by_status_code",
-		Description: "Server response count by RPC status code",
-		TagKeys:     []tag.Key{StatusCode},
-		Measure:     ServerLatency,
-		Aggregation: view.Count(),
-	}
-)
-
-// DefaultClientViews are the default client views provided by this package.
-var DefaultClientViews = []*view.View{
-	ClientRequestCountView,
-	ClientLatencyView,
-	ClientRequestCountByMethod,
-	ClientResponseCountByStatusCode,
-}
-
-// DefaultServerViews are the default server views provided by this package.
-var DefaultServerViews = []*view.View{
-	ServerRequestCountView,
-	ServerLatencyView,
-	ServerRequestCountByMethod,
-	ServerResponseCountByStatusCode,
-}
-
-// StatsProfile groups metrics-related data.
-type StatsProfile struct {
-	Role           string
-	CountMeasure   *stats.Int64Measure
-	LatencyMeasure *stats.Float64Measure
-}
-
-var (
-	// ClientProfile is used for RPC clients.
-	ClientProfile = &StatsProfile{
-		Role:           "client",
-		CountMeasure:   ClientRequestCount,
-		LatencyMeasure: ClientLatency,
-	}
-
-	// ServerProfile is used for RPC servers.
-	ServerProfile = &StatsProfile{
-		Role:           "server",
-		CountMeasure:   ServerRequestCount,
-		LatencyMeasure: ServerLatency,
-	}
-)
@@ -1,45 +0,0 @@
-package opencensus
-
-import (
-	"fmt"
-
-	microerr "go-micro.dev/v4/errors"
-
-	"go.opencensus.io/trace"
-
-	"google.golang.org/genproto/googleapis/rpc/code"
-)
-
-var microCodeToStatusCode = map[int32]code.Code{
-	400: code.Code_INVALID_ARGUMENT,
-	401: code.Code_UNAUTHENTICATED,
-	403: code.Code_PERMISSION_DENIED,
-	404: code.Code_NOT_FOUND,
-	409: code.Code_ABORTED,
-	500: code.Code_INTERNAL,
-}
-
-func getResponseStatus(err error) trace.Status {
-	if err != nil {
-		microErr, ok := err.(*microerr.Error)
-		if ok {
-			statusCode := microErr.Code
-			code, ok := microCodeToStatusCode[microErr.Code]
-			if ok {
-				statusCode = int32(code)
-			}
-
-			return trace.Status{
-				Code:    statusCode,
-				Message: fmt.Sprintf("%s: %s", microErr.Id, microErr.Detail),
-			}
-		}
-
-		return trace.Status{
-			Code:    int32(code.Code_UNKNOWN),
-			Message: err.Error(),
-		}
-	}
-
-	return trace.Status{}
-}
@@ -1,81 +0,0 @@
-package opencensus
-
-import (
-	"context"
-	"fmt"
-	"strconv"
-	"time"
-
-	"go.opencensus.io/stats"
-	"go.opencensus.io/tag"
-	"go.opencensus.io/trace"
-)
-
-type tracker struct {
-	startedAt time.Time
-
-	profile *StatsProfile
-	span    *trace.Span
-
-	method  string
-	service string
-}
-
-type requestDescriptor interface {
-	Service() string
-	Endpoint() string
-}
-
-type publicationDescriptor interface {
-	Topic() string
-}
-
-// newRequestTracker creates a new tracker for an RPC request (client or server).
-func newRequestTracker(req requestDescriptor, profile *StatsProfile) *tracker {
-	return &tracker{
-		profile: profile,
-		method:  req.Endpoint(),
-		service: req.Service(),
-	}
-}
-
-// newEventTracker creates a new tracker for a publication (client or server).
-func newEventTracker(pub publicationDescriptor, profile *StatsProfile) *tracker {
-	return &tracker{
-		profile: profile,
-		method:  pub.Topic(),
-		service: "pubsub",
-	}
-}
-
-// start monitoring a request. You can choose to let this method
-// start a span for the request or attach one later.
-func (t *tracker) start(ctx context.Context, startSpan bool) context.Context {
-	t.startedAt = time.Now()
-
-	ctx, _ = tag.New(ctx, tag.Upsert(Service, t.service), tag.Upsert(Endpoint, t.method))
-	stats.Record(ctx, t.profile.CountMeasure.M(1))
-
-	if startSpan {
-		ctx, t.span = trace.StartSpan(
-			ctx,
-			fmt.Sprintf("rpc/%s/%s/%s", t.profile.Role, t.service, t.method),
-		)
-	}
-
-	return ctx
-}
-
-// end a request's monitoring session. If there is a span ongoing, it will
-// be ended and metrics will be recorded.
-func (t *tracker) end(ctx context.Context, err error) {
-	status := getResponseStatus(err)
-
-	ctx, _ = tag.New(ctx, tag.Upsert(StatusCode, strconv.Itoa(int(status.Code))))
-	stats.Record(ctx, t.profile.LatencyMeasure.M(float64(time.Since(t.startedAt))/float64(time.Millisecond)))
-
-	if t.span != nil {
-		t.span.SetStatus(status)
-		t.span.End()
-	}
-}
@@ -0,0 +1,14 @@
+# OpenTelemetry wrappers
+
+OpenTelemetry wrappers propagate traces (spans) accross services.
+
+## Usage
+
+```go
+service := micro.NewService(
+    micro.Name("go.micro.srv.greeter"),
+    micro.WrapClient(opentelemetry.NewClientWrapper()),
+    micro.WrapHandler(open.NewHandlerWrapper()),
+    micro.WrapSubscriber(opentelemetry.NewSubscriberWrapper()),
+)
+```
@@ -0,0 +1,55 @@
+package opentelemetry
+
+import (
+	"context"
+	"strings"
+
+	"go-micro.dev/v4/metadata"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/baggage"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/trace"
+)
+
+const (
+	instrumentationName = "github.com/go-micro/plugins/v4/wrapper/trace/opentelemetry"
+)
+
+// StartSpanFromContext returns a new span with the given operation name and options. If a span
+// is found in the context, it will be used as the parent of the resulting span.
+func StartSpanFromContext(ctx context.Context, tp trace.TracerProvider, name string, opts ...trace.SpanStartOption) (context.Context, trace.Span) {
+	md, ok := metadata.FromContext(ctx)
+	if !ok {
+		md = make(metadata.Metadata)
+	}
+	propagator, carrier := otel.GetTextMapPropagator(), make(propagation.MapCarrier)
+	for k, v := range md {
+		for _, f := range propagator.Fields() {
+			if strings.EqualFold(k, f) {
+				carrier[f] = v
+			}
+		}
+	}
+	ctx = propagator.Extract(ctx, carrier)
+	spanCtx := trace.SpanContextFromContext(ctx)
+	ctx = baggage.ContextWithBaggage(ctx, baggage.FromContext(ctx))
+
+	var tracer trace.Tracer
+	var span trace.Span
+	if tp != nil {
+		tracer = tp.Tracer(instrumentationName)
+	} else {
+		tracer = otel.Tracer(instrumentationName)
+	}
+	ctx, span = tracer.Start(trace.ContextWithRemoteSpanContext(ctx, spanCtx), name, opts...)
+
+	carrier = make(propagation.MapCarrier)
+	propagator.Inject(ctx, carrier)
+	for k, v := range carrier {
+		//lint:ignore SA1019 no unicode punctution handle needed
+		md.Set(strings.Title(k), v)
+	}
+	ctx = metadata.NewContext(ctx, md)
+
+	return ctx, span
+}
@@ -0,0 +1,72 @@
+package opentelemetry
+
+import (
+	"context"
+
+	"go-micro.dev/v4/client"
+	"go-micro.dev/v4/server"
+	"go.opentelemetry.io/otel/trace"
+)
+
+type Options struct {
+	TraceProvider trace.TracerProvider
+
+	CallFilter       CallFilter
+	StreamFilter     StreamFilter
+	PublishFilter    PublishFilter
+	SubscriberFilter SubscriberFilter
+	HandlerFilter    HandlerFilter
+}
+
+// CallFilter used to filter client.Call, return true to skip call trace.
+type CallFilter func(context.Context, client.Request) bool
+
+// StreamFilter used to filter client.Stream, return true to skip stream trace.
+type StreamFilter func(context.Context, client.Request) bool
+
+// PublishFilter used to filter client.Publish, return true to skip publish trace.
+type PublishFilter func(context.Context, client.Message) bool
+
+// SubscriberFilter used to filter server.Subscribe, return true to skip subcribe trace.
+type SubscriberFilter func(context.Context, server.Message) bool
+
+// HandlerFilter used to filter server.Handle, return true to skip handle trace.
+type HandlerFilter func(context.Context, server.Request) bool
+
+type Option func(*Options)
+
+func WithTraceProvider(tp trace.TracerProvider) Option {
+	return func(o *Options) {
+		o.TraceProvider = tp
+	}
+}
+
+func WithCallFilter(filter CallFilter) Option {
+	return func(o *Options) {
+		o.CallFilter = filter
+	}
+}
+
+func WithStreamFilter(filter StreamFilter) Option {
+	return func(o *Options) {
+		o.StreamFilter = filter
+	}
+}
+
+func WithPublishFilter(filter PublishFilter) Option {
+	return func(o *Options) {
+		o.PublishFilter = filter
+	}
+}
+
+func WithSubscribeFilter(filter SubscriberFilter) Option {
+	return func(o *Options) {
+		o.SubscriberFilter = filter
+	}
+}
+
+func WithHandleFilter(filter HandlerFilter) Option {
+	return func(o *Options) {
+		o.HandlerFilter = filter
+	}
+}
@@ -0,0 +1,175 @@
+package opentelemetry
+
+import (
+	"context"
+	"fmt"
+
+	"go-micro.dev/v4/client"
+	"go-micro.dev/v4/registry"
+	"go-micro.dev/v4/server"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// NewCallWrapper accepts an opentracing Tracer and returns a Call Wrapper.
+func NewCallWrapper(opts ...Option) client.CallWrapper {
+	options := Options{}
+	for _, o := range opts {
+		o(&options)
+	}
+	return func(cf client.CallFunc) client.CallFunc {
+		return func(ctx context.Context, node *registry.Node, req client.Request, rsp interface{}, opts client.CallOptions) error {
+			if options.CallFilter != nil && options.CallFilter(ctx, req) {
+				return cf(ctx, node, req, rsp, opts)
+			}
+			name := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
+			spanOpts := []trace.SpanStartOption{
+				trace.WithSpanKind(trace.SpanKindClient),
+			}
+			ctx, span := StartSpanFromContext(ctx, options.TraceProvider, name, spanOpts...)
+			defer span.End()
+			if err := cf(ctx, node, req, rsp, opts); err != nil {
+				span.SetStatus(codes.Error, err.Error())
+				span.RecordError(err)
+				return err
+			}
+			return nil
+		}
+	}
+}
+
+// NewHandlerWrapper accepts an opentracing Tracer and returns a Handler Wrapper.
+func NewHandlerWrapper(opts ...Option) server.HandlerWrapper {
+	options := Options{}
+	for _, o := range opts {
+		o(&options)
+	}
+	return func(h server.HandlerFunc) server.HandlerFunc {
+		return func(ctx context.Context, req server.Request, rsp interface{}) error {
+			if options.HandlerFilter != nil && options.HandlerFilter(ctx, req) {
+				return h(ctx, req, rsp)
+			}
+			name := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
+			spanOpts := []trace.SpanStartOption{
+				trace.WithSpanKind(trace.SpanKindServer),
+			}
+			ctx, span := StartSpanFromContext(ctx, options.TraceProvider, name, spanOpts...)
+			defer span.End()
+			if err := h(ctx, req, rsp); err != nil {
+				span.SetStatus(codes.Error, err.Error())
+				span.RecordError(err)
+				return err
+			}
+			return nil
+		}
+	}
+}
+
+// NewSubscriberWrapper accepts an opentracing Tracer and returns a Subscriber Wrapper.
+func NewSubscriberWrapper(opts ...Option) server.SubscriberWrapper {
+	options := Options{}
+	for _, o := range opts {
+		o(&options)
+	}
+	return func(next server.SubscriberFunc) server.SubscriberFunc {
+		return func(ctx context.Context, msg server.Message) error {
+			if options.SubscriberFilter != nil && options.SubscriberFilter(ctx, msg) {
+				return next(ctx, msg)
+			}
+			name := "Sub from " + msg.Topic()
+			spanOpts := []trace.SpanStartOption{
+				trace.WithSpanKind(trace.SpanKindServer),
+			}
+			ctx, span := StartSpanFromContext(ctx, options.TraceProvider, name, spanOpts...)
+			defer span.End()
+			if err := next(ctx, msg); err != nil {
+				span.SetStatus(codes.Error, err.Error())
+				span.RecordError(err)
+				return err
+			}
+			return nil
+		}
+	}
+}
+
+// NewClientWrapper returns a client.Wrapper
+// that adds monitoring to outgoing requests.
+func NewClientWrapper(opts ...Option) client.Wrapper {
+	options := Options{}
+	for _, o := range opts {
+		o(&options)
+	}
+	return func(c client.Client) client.Client {
+		w := &clientWrapper{
+			Client:        c,
+			tp:            options.TraceProvider,
+			callFilter:    options.CallFilter,
+			streamFilter:  options.StreamFilter,
+			publishFilter: options.PublishFilter,
+		}
+		return w
+	}
+}
+
+type clientWrapper struct {
+	client.Client
+
+	tp            trace.TracerProvider
+	callFilter    CallFilter
+	streamFilter  StreamFilter
+	publishFilter PublishFilter
+}
+
+func (w *clientWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
+	if w.callFilter != nil && w.callFilter(ctx, req) {
+		return w.Client.Call(ctx, req, rsp, opts...)
+	}
+	name := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
+	spanOpts := []trace.SpanStartOption{
+		trace.WithSpanKind(trace.SpanKindClient),
+	}
+	ctx, span := StartSpanFromContext(ctx, w.tp, name, spanOpts...)
+	defer span.End()
+	if err := w.Client.Call(ctx, req, rsp, opts...); err != nil {
+		span.SetStatus(codes.Error, err.Error())
+		span.RecordError(err)
+		return err
+	}
+	return nil
+}
+
+func (w *clientWrapper) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
+	if w.streamFilter != nil && w.streamFilter(ctx, req) {
+		return w.Client.Stream(ctx, req, opts...)
+	}
+	name := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
+	spanOpts := []trace.SpanStartOption{
+		trace.WithSpanKind(trace.SpanKindClient),
+	}
+	ctx, span := StartSpanFromContext(ctx, w.tp, name, spanOpts...)
+	defer span.End()
+	stream, err := w.Client.Stream(ctx, req, opts...)
+	if err != nil {
+		span.SetStatus(codes.Error, err.Error())
+		span.RecordError(err)
+	}
+	return stream, err
+}
+
+func (w *clientWrapper) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
+	if w.publishFilter != nil && w.publishFilter(ctx, p) {
+		return w.Client.Publish(ctx, p, opts...)
+	}
+	name := fmt.Sprintf("Pub to %s", p.Topic())
+	spanOpts := []trace.SpanStartOption{
+		trace.WithSpanKind(trace.SpanKindClient),
+	}
+	ctx, span := StartSpanFromContext(ctx, w.tp, name, spanOpts...)
+	defer span.End()
+	if err := w.Client.Publish(ctx, p, opts...); err != nil {
+		span.SetStatus(codes.Error, err.Error())
+		span.RecordError(err)
+		return err
+	}
+	return nil
+}