Merge branch 'master' into test-integration-stream-order

Author: cicoyle

.github/workflows/dapr-bot-schedule.yml

@@ -50,26 +50,26 @@ jobs:
       uses: actions/[email protected]
       with:
         repo-token: ${{ secrets.DAPR_BOT_TOKEN }}
-        days-before-pr-stale: 60
+        days-before-pr-stale: 90
         days-before-pr-close: 7
-        days-before-issue-stale: 60
+        days-before-issue-stale: 90
         days-before-issue-close: 7
         stale-issue-message: >
           This issue has been automatically marked as stale because it has not had activity in the
-          last 60 days. It will be closed in the next 7 days unless it is tagged (pinned, good first issue, help wanted or triaged/resolved) or other activity
+          last 90 days. It will be closed in the next 7 days unless it is tagged (pinned, good first issue, help wanted or triaged/resolved) or other activity
           occurs. Thank you for your contributions.
         close-issue-message: >
           This issue has been automatically closed because it has not had activity in the
-          last 67 days. If this issue is still valid, please ping a maintainer and ask them to label it as pinned, good first issue, help wanted or triaged/resolved.
+          last 97 days. If this issue is still valid, please ping a maintainer and ask them to label it as pinned, good first issue, help wanted or triaged/resolved.
           Thank you for your contributions.
         stale-pr-message: >
           This pull request has been automatically marked as stale because it has not had
-          activity in the last 60 days. It will be closed in 7 days if no further activity occurs. Please
+          activity in the last 90 days. It will be closed in 7 days if no further activity occurs. Please
           feel free to give a status update now, ping for review, or re-open when it's ready.
           Thank you for your contributions!
         close-pr-message: >
           This pull request has been automatically closed because it has not had
-          activity in the last 67 days. Please feel free to give a status update now, ping for review, or re-open when it's ready.
+          activity in the last 97 days. Please feel free to give a status update now, ping for review, or re-open when it's ready.
           Thank you for your contributions!
         stale-issue-label: 'stale'
         exempt-issue-labels: 'pinned,good first issue,help wanted,triaged/resolved'

go.mod

@@ -11,10 +11,10 @@ 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.1
+	github.com/diagridio/go-etcd-cron v0.6.2
 	github.com/evanphx/json-patch/v5 v5.9.0
 	github.com/go-chi/chi/v5 v5.0.11
 	github.com/go-chi/cors v1.2.1
@@ -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

@@ -1,4 +1,4 @@
-cel.dev/expr v0.19.2 h1:V354PbqIXr9IQdwy4SYA4xa0HXaWq1BUPAGzugBY5V4=
+  cel.dev/expr v0.19.2 h1:V354PbqIXr9IQdwy4SYA4xa0HXaWq1BUPAGzugBY5V4=
 cel.dev/expr v0.19.2/go.mod h1:MrpN08Q+lEBs+bGYdLxxHkZoUSsCp0nSKTs0nTymJgw=
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
@@ -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=
@@ -530,8 +530,8 @@ github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cu
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
 github.com/dgryski/trifles v0.0.0-20200323201526-dd97f9abfb48 h1:fRzb/w+pyskVMQ+UbP35JkH8yB7MYb4q/qhBarqZE6g=
 github.com/dgryski/trifles v0.0.0-20200323201526-dd97f9abfb48/go.mod h1:if7Fbed8SFyPtHLHbg49SI7NAdJiC5WIA09pe59rfAA=
-github.com/diagridio/go-etcd-cron v0.6.1 h1:Ot+ZPYTQZDiSvkSISQp9r6UoWHHtyAI/hYPx76SI/d0=
-github.com/diagridio/go-etcd-cron v0.6.1/go.mod h1:o/UKoR4LS6G8KCa++0HlVXpKn00aUuSbLRxDkXdo4cM=
+github.com/diagridio/go-etcd-cron v0.6.2 h1:slbrZeH6IASSZQIIpVdw7IogUuRSOe3m8ztg1x30WlE=
+github.com/diagridio/go-etcd-cron v0.6.2/go.mod h1:o/UKoR4LS6G8KCa++0HlVXpKn00aUuSbLRxDkXdo4cM=
 github.com/didip/tollbooth/v7 v7.0.1 h1:TkT4sBKoQoHQFPf7blQ54iHrZiTDnr8TceU+MulVAog=
 github.com/didip/tollbooth/v7 v7.0.1/go.mod h1:VZhDSGl5bDSPj4wPsih3PFa4Uh9Ghv8hgacaTm5PRT4=
 github.com/dimfeld/httptreemux v5.0.1+incompatible h1:Qj3gVcDNoOthBAqftuD596rm4wg/adLLz5xh5CmpiCA=
@@ -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
 }