Merge pull request dapr#8649 from cicoyle/feat-baggage

Feat add baggage support

Author: JoshVanL

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/argoproj/argo-rollouts v1.4.1
 	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/cloudevents/sdk-go/v2 v2.15.2
-	github.com/dapr/components-contrib v1.15.1-0.20250409220637-70c99725fd12
+	github.com/dapr/components-contrib v1.15.1-0.20250423123324-b2c31ceba20b
 	github.com/dapr/durabletask-go v0.6.5
 	github.com/dapr/kit v0.15.2
 	github.com/diagridio/go-etcd-cron v0.6.2
@@ -286,7 +286,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 // indirect
 	github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c // indirect
 	github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed // indirect
-	github.com/hamba/avro/v2 v2.25.1 // indirect
+	github.com/hamba/avro/v2 v2.28.0 // indirect
 	github.com/hashicorp/consul/api v1.25.1 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect

go.sum

@@ -502,8 +502,8 @@ github.com/dancannon/gorethink v4.0.0+incompatible h1:KFV7Gha3AuqT+gr0B/eKvGhbjm
 github.com/dancannon/gorethink v4.0.0+incompatible/go.mod h1:BLvkat9KmZc1efyYwhz3WnybhRZtgF1K929FD8z1avU=
 github.com/danieljoos/wincred v1.1.2 h1:QLdCxFs1/Yl4zduvBdcHB8goaYk9RARS2SgLLRuAyr0=
 github.com/danieljoos/wincred v1.1.2/go.mod h1:GijpziifJoIBfYh+S7BbkdUTU4LfM+QnGqR5Vl2tAx0=
-github.com/dapr/components-contrib v1.15.1-0.20250409220637-70c99725fd12 h1:VETULsWb8KIjzuqmEiXCv6BHw0QVgkjMkP7tztIv9NI=
-github.com/dapr/components-contrib v1.15.1-0.20250409220637-70c99725fd12/go.mod h1:0yFCUv1c6TMWvC6sQYdv0FMWmdCeEpWGE8AYp9cb6Ic=
+github.com/dapr/components-contrib v1.15.1-0.20250423123324-b2c31ceba20b h1:u1xUF4CuNsTlv8pZsrX6XH1FolmMCjJ0Kyfef0wJkzY=
+github.com/dapr/components-contrib v1.15.1-0.20250423123324-b2c31ceba20b/go.mod h1:mDA9c8w/KGAOx18yCTaMXHkfWS6mFHR/o3FjI6a14dA=
 github.com/dapr/durabletask-go v0.6.5 h1:aWcxMfYudojpgRjJRdUr7yyZ7rGcvLtWXUuA4cGHBR0=
 github.com/dapr/durabletask-go v0.6.5/go.mod h1:nTZ5fCbJLnZbVdi6Z2YxdDF1OgQZL3LroogGuetrwuA=
 github.com/dapr/kit v0.15.2 h1:5H9IhKScU/SpE2Hxvr5vUlmYN1e2MJN15RoT8/KSziU=
@@ -946,8 +946,8 @@ github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c h1:6rhixN/i8
 github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c/go.mod h1:NMPJylDgVpX0MLRlPy15sqSwOFv/U1GZ2m21JhFfek0=
 github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed h1:5upAirOpQc1Q53c0bnx2ufif5kANL7bfZWcc6VJWJd8=
 github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed/go.mod h1:tMWxXQ9wFIaZeTI9F+hmhFiGpFmhOHzyShyFUhRm0H4=
-github.com/hamba/avro/v2 v2.25.1 h1:t8cOyv0wkNAPF6/khArMtR0nK9HtGa+WKbp9q+KdFZQ=
-github.com/hamba/avro/v2 v2.25.1/go.mod h1:I8glyswHnpED3Nlx2ZdUe+4LJnCOOyiCzLMno9i/Uu0=
+github.com/hamba/avro/v2 v2.28.0 h1:E8J5D27biyAulWKNiEBhV85QPc9xRMCUCGJewS0KYCE=
+github.com/hamba/avro/v2 v2.28.0/go.mod h1:9TVrlt1cG1kkTUtm9u2eO5Qb7rZXlYzoKqPt8TSH+TA=
 github.com/hashicorp/consul/api v1.1.0/go.mod h1:VmuI/Lkw1nC05EYQWNKwWGbkg+FbDBtguAZLlVdkD9Q=
 github.com/hashicorp/consul/api v1.3.0/go.mod h1:MmDNSzIMUjNpY/mQ398R4bk2FnqQLoPndWW5VkKPlCE=
 github.com/hashicorp/consul/api v1.25.1 h1:CqrdhYzc8XZuPnhIYZWH45toM0LB9ZeYr/gvpLVI3PE=

pkg/api/grpc/grpc.go

@@ -22,10 +22,12 @@ import (
 	"net/http"
 	"slices"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	otelbaggage "go.opentelemetry.io/otel/baggage"
 	otelTrace "go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -47,6 +49,7 @@ import (
 	stateLoader "github.com/dapr/dapr/pkg/components/state"
 	"github.com/dapr/dapr/pkg/config"
 	diag "github.com/dapr/dapr/pkg/diagnostics"
+	diagConsts "github.com/dapr/dapr/pkg/diagnostics/consts"
 	diagUtils "github.com/dapr/dapr/pkg/diagnostics/utils"
 	"github.com/dapr/dapr/pkg/encryption"
 	"github.com/dapr/dapr/pkg/messages"
@@ -505,23 +508,33 @@ func (a *api) InvokeBinding(ctx context.Context, in *runtimev1pb.InvokeBindingRe
 	sc := span.SpanContext()
 	tp := diag.SpanContextToW3CString(sc)
 	if span != nil {
-		if _, ok := req.Metadata[diag.TraceparentHeader]; !ok {
-			req.Metadata[diag.TraceparentHeader] = tp
+		if _, ok := req.Metadata[diagConsts.TraceparentHeader]; !ok {
+			req.Metadata[diagConsts.TraceparentHeader] = tp
 		}
-		if _, ok := req.Metadata[diag.TracestateHeader]; !ok {
+		if _, ok := req.Metadata[diagConsts.TracestateHeader]; !ok {
 			if sc.TraceState().Len() > 0 {
-				req.Metadata[diag.TracestateHeader] = diag.TraceStateToW3CString(sc)
+				req.Metadata[diagConsts.TracestateHeader] = diag.TraceStateToW3CString(sc)
 			}
 		}
 	}
 
 	// Allow for distributed tracing by passing context metadata.
 	if incomingMD, ok := metadata.FromIncomingContext(ctx); ok {
+		if baggageValues := incomingMD[diagConsts.BaggageHeader]; len(baggageValues) > 0 {
+			baggageString := strings.Join(baggageValues, ",")
+			baggage, err := otelbaggage.Parse(baggageString)
+			if err != nil {
+				return nil, err
+			}
+			ctx = otelbaggage.ContextWithBaggage(ctx, baggage)
+			req.Metadata[diagConsts.BaggageHeader] = baggageString
+		}
+
 		for key, val := range incomingMD {
 			sanitizedKey := invokev1.ReservedGRPCMetadataToDaprPrefixHeader(key)
 			// Not to overwrite the existing metadata
 			// But if the key is traceparent or tracestate, we allow overwrite the existing metadata.
-			if _, exist := req.Metadata[sanitizedKey]; !exist || (key == diag.TraceparentHeader || key == diag.TracestateHeader) {
+			if _, exist := req.Metadata[sanitizedKey]; !exist || (key == diagConsts.TraceparentHeader || key == diagConsts.TracestateHeader) {
 				req.Metadata[sanitizedKey] = val[0]
 			}
 		}

pkg/api/grpc/proxy/handler.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/dapr/dapr/pkg/api/grpc/proxy/codec"
 	"github.com/dapr/dapr/pkg/diagnostics"
+	diagConsts "github.com/dapr/dapr/pkg/diagnostics/consts"
 	"github.com/dapr/dapr/pkg/resiliency"
 	"github.com/dapr/kit/utils"
 )
@@ -96,7 +97,7 @@ func (s *handler) handler(srv any, serverStream grpc.ServerStream) error {
 	// Fetch the AppId so we can reference it for resiliency.
 	ctx := serverStream.Context()
 	md, _ := metadata.FromIncomingContext(ctx)
-	v := md[diagnostics.GRPCProxyAppIDKey]
+	v := md[diagConsts.GRPCProxyAppIDKey]
 
 	// The app id check is handled in the StreamDirector. If we don't have it here, we just use a NoOp policy since we know the request is impossible.
 	var policyDef *resiliency.PolicyDefinition

pkg/api/grpc/proxy/handler_test.go

@@ -44,6 +44,7 @@ import (
 	pb "github.com/dapr/dapr/pkg/api/grpc/proxy/testservice"
 	"github.com/dapr/dapr/pkg/config"
 	diag "github.com/dapr/dapr/pkg/diagnostics"
+	diagConsts "github.com/dapr/dapr/pkg/diagnostics/consts"
 	"github.com/dapr/dapr/pkg/resiliency"
 	"github.com/dapr/kit/logger"
 	"github.com/dapr/kit/retry"
@@ -440,7 +441,7 @@ func (s *proxyTestSuite) TestResiliencyUnary() {
 
 		ctx, cancel := s.ctx()
 		defer cancel()
-		ctx = metadata.NewOutgoingContext(ctx, metadata.Pairs(diag.GRPCProxyAppIDKey, testAppID))
+		ctx = metadata.NewOutgoingContext(ctx, metadata.Pairs(diagConsts.GRPCProxyAppIDKey, testAppID))
 
 		// Reset callCount before this test
 		s.service.pingCallCount.Store(0)
@@ -476,7 +477,7 @@ func (s *proxyTestSuite) TestResiliencyUnary() {
 
 		setupMetrics(s)
 
-		ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(diag.GRPCProxyAppIDKey, testAppID))
+		ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(diagConsts.GRPCProxyAppIDKey, testAppID))
 
 		_, err := s.testClient.Ping(ctx, &pb.PingRequest{Value: message})
 		require.Error(t, err, "Ping should fail due to timeouts")
@@ -515,7 +516,7 @@ func (s *proxyTestSuite) TestResiliencyUnary() {
 			go func(i int) {
 				for j := range numOperations {
 					pingMsg := fmt.Sprintf("%d:%d", i, j)
-					ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(diag.GRPCProxyAppIDKey, testAppID))
+					ctx := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(diagConsts.GRPCProxyAppIDKey, testAppID))
 					res, err := s.testClient.Ping(ctx, &pb.PingRequest{Value: pingMsg})
 					require.NoErrorf(t, err, "Ping should succeed for operation %d:%d", i, j)
 					require.NotNilf(t, res, "Response should not be nil for operation %d:%d", i, j)
@@ -584,7 +585,7 @@ func (s *proxyTestSuite) TestResiliencyStreaming() {
 		ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
 		defer cancel()
 		ctx = metadata.NewOutgoingContext(ctx, metadata.Pairs(
-			diag.GRPCProxyAppIDKey, "test",
+			diagConsts.GRPCProxyAppIDKey, "test",
 			"dapr-test", t.Name(),
 		))
 
@@ -622,7 +623,7 @@ func (s *proxyTestSuite) TestResiliencyStreaming() {
 		setupMetrics(s)
 
 		ctx = metadata.NewOutgoingContext(ctx, metadata.Pairs(
-			diag.GRPCProxyAppIDKey, testAppID,
+			diagConsts.GRPCProxyAppIDKey, testAppID,
 			StreamMetadataKey, "1",
 			"dapr-test", t.Name(),
 		))
@@ -673,7 +674,7 @@ func (s *proxyTestSuite) TestResiliencyStreaming() {
 		setupMetrics(s)
 
 		ctx = metadata.NewOutgoingContext(ctx, metadata.Pairs(
-			diag.GRPCProxyAppIDKey, testAppID,
+			diagConsts.GRPCProxyAppIDKey, testAppID,
 			StreamMetadataKey, "1",
 			"dapr-test", t.Name(),
 		))

pkg/api/http/http.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/go-chi/chi/v5"
 	"github.com/mitchellh/mapstructure"
+	otelBaggage "go.opentelemetry.io/otel/baggage"
 	"go.opentelemetry.io/otel/trace"
 
 	"github.com/dapr/components-contrib/bindings"
@@ -410,7 +411,8 @@ func (a *api) onOutputBindingMessage(w nethttp.ResponseWriter, r *nethttp.Reques
 		return
 	}
 
-	b, err := json.Marshal(req.Data)
+	var b []byte
+	b, err = json.Marshal(req.Data)
 	if err != nil {
 		resp := messages.NewAPIErrorHTTP(fmt.Sprintf(messages.ErrMalformedRequestData, err), errorcodes.CommonMalformedRequestData, nethttp.StatusInternalServerError)
 		respondWithError(w, resp)
@@ -433,6 +435,19 @@ func (a *api) onOutputBindingMessage(w nethttp.ResponseWriter, r *nethttp.Reques
 		}
 	}
 
+	if baggageHeaders := r.Header.Values(diagConsts.BaggageHeader); len(baggageHeaders) > 0 {
+		baggageString := strings.Join(baggageHeaders, ",")
+		if _, err = otelBaggage.Parse(baggageString); err != nil {
+			resp := messages.NewAPIErrorHTTP(fmt.Sprintf("invalid baggage header: %v", err), errorcodes.CommonMalformedRequest, nethttp.StatusBadRequest)
+			respondWithError(w, resp)
+			return
+		}
+		if req.Metadata == nil {
+			req.Metadata = map[string]string{}
+		}
+		req.Metadata[diagConsts.BaggageHeader] = baggageString
+	}
+
 	start := time.Now()
 	resp, err := a.sendToOutputBindingFn(r.Context(), name, &bindings.InvokeRequest{
 		Metadata:  req.Metadata,

pkg/diagnostics/consts/consts.go

@@ -61,6 +61,21 @@ const (
 	// Keys used in the context's metadata for streaming calls
 	// Note: these keys must always be all-lowercase
 	DaprCallLocalStreamMethodKey = "__dapr_calllocalstream_method"
+
+	// We have leveraged the code from opencensus-go plugin to adhere the w3c trace context.
+	// Reference : https://github.com/census-instrumentation/opencensus-go/blob/master/plugin/ochttp/propagation/tracecontext/propagation.go
+	// Trace context headers
+	TraceparentHeader = "traceparent"
+	TracestateHeader  = "tracestate"
+	BaggageHeader     = "baggage"
+
+	GRPCTraceContextKey = "grpc-trace-bin"
+	GRPCProxyAppIDKey   = "dapr-app-id"
+
+	// Trace sampling constants
+	SupportedVersion = 0
+	MaxVersion       = 254
+	MaxTracestateLen = 512
 )
 
 // GrpcAppendSpanAttributesFn is the interface that applies to gRPC requests that add span attributes.

pkg/diagnostics/grpc_monitoring.go

@@ -25,6 +25,7 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/dapr/dapr/pkg/api/grpc/metadata"
+	diagConsts "github.com/dapr/dapr/pkg/diagnostics/consts"
 	diagUtils "github.com/dapr/dapr/pkg/diagnostics/utils"
 )
 
@@ -264,7 +265,7 @@ func (g *grpcMetrics) StreamingServerInterceptor() grpc.StreamServerInterceptor
 	return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
 		ctx := ss.Context()
 		md, _ := metadata.FromIncomingContext(ctx)
-		vals, ok := md[GRPCProxyAppIDKey]
+		vals, ok := md[diagConsts.GRPCProxyAppIDKey]
 		if !ok || len(vals) == 0 {
 			return handler(srv, ss)
 		}
@@ -285,7 +286,7 @@ func (g *grpcMetrics) StreamingClientInterceptor() grpc.StreamServerInterceptor
 	return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
 		ctx := ss.Context()
 		md, _ := metadata.FromIncomingContext(ctx)
-		vals, ok := md[GRPCProxyAppIDKey]
+		vals, ok := md[diagConsts.GRPCProxyAppIDKey]
 		if !ok || len(vals) == 0 {
 			return handler(srv, ss)
 		}

pkg/diagnostics/grpc_monitoring_test.go

@@ -41,7 +41,7 @@ func (f *fakeProxyStream) Context() context.Context {
 	}
 
 	ctx := context.Background()
-	ctx = grpcMetadata.NewIncomingContext(ctx, grpcMetadata.New(map[string]string{GRPCProxyAppIDKey: f.appID}))
+	ctx = grpcMetadata.NewIncomingContext(ctx, grpcMetadata.New(map[string]string{"dapr-app-id": f.appID}))
 	ctx, _ = metadata.SetMetadataInTapHandle(ctx, nil)
 	return ctx
 }