tracing, rpc: add support for tracing interceptors in dRPC

This change brings the support for client and tracing interceptors
in dRPC that start spans, propagate metadata via DRPC context
and record error codes for unary and streaming RPCs .

Epic: CRDB-49359
Fixes: #148350 and #144372
Release note: None

Author: Nukitt

pkg/BUILD.bazel

@@ -830,6 +830,7 @@ ALL_TESTS = [
     "//pkg/util/timeutil:timeutil_test",
     "//pkg/util/tochar:tochar_test",
     "//pkg/util/tracing/collector:collector_test",
+    "//pkg/util/tracing/drpcinterceptor:drpcinterceptor_test",
     "//pkg/util/tracing/goexectrace:goexectrace_test",
     "//pkg/util/tracing/grpcinterceptor:grpcinterceptor_test",
     "//pkg/util/tracing/service:service_test",
@@ -2818,6 +2819,8 @@ GO_TARGETS = [
     "//pkg/util/tochar:tochar_test",
     "//pkg/util/tracing/collector:collector",
     "//pkg/util/tracing/collector:collector_test",
+    "//pkg/util/tracing/drpcinterceptor:drpcinterceptor",
+    "//pkg/util/tracing/drpcinterceptor:drpcinterceptor_test",
     "//pkg/util/tracing/goexectrace:goexectrace",
     "//pkg/util/tracing/goexectrace:goexectrace_test",
     "//pkg/util/tracing/grpcinterceptor:grpcinterceptor",

pkg/rpc/BUILD.bazel

@@ -64,6 +64,7 @@ go_library(
         "//pkg/util/sysutil",
         "//pkg/util/timeutil",
         "//pkg/util/tracing",
+        "//pkg/util/tracing/drpcinterceptor",
         "//pkg/util/tracing/grpcinterceptor",
         "//pkg/util/tracing/tracingutil",
         "//pkg/util/uuid",

pkg/rpc/context.go

@@ -38,6 +38,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+	"github.com/cockroachdb/cockroach/pkg/util/tracing/drpcinterceptor"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing/grpcinterceptor"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing/tracingutil"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
@@ -651,6 +652,10 @@ func NewContext(ctx context.Context, opts ContextOptions) *Context {
 			grpcinterceptor.ClientInterceptor(tracer, tagger))
 		rpcCtx.clientStreamInterceptors = append(rpcCtx.clientStreamInterceptors,
 			grpcinterceptor.StreamClientInterceptor(tracer, tagger))
+		rpcCtx.clientUnaryInterceptorsDRPC = append(rpcCtx.clientUnaryInterceptorsDRPC,
+			drpcinterceptor.ClientInterceptor(tracer, tagger))
+		rpcCtx.clientStreamInterceptorsDRPC = append(rpcCtx.clientStreamInterceptorsDRPC,
+			drpcinterceptor.StreamClientInterceptor(tracer, tagger))
 	}
 	// Note that we do not consult rpcCtx.Knobs.StreamClientInterceptor. That knob
 	// can add another interceptor, but it can only do it dynamically, based on

pkg/rpc/drpc.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/rpc/rpcbase"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
+	"github.com/cockroachdb/cockroach/pkg/util/tracing/drpcinterceptor"
 	"github.com/cockroachdb/errors"
 	"storj.io/drpc"
 	"storj.io/drpc/drpcclient"
@@ -232,6 +233,11 @@ func NewDRPCServer(_ context.Context, rpcCtx *Context, opts ...ServerOption) (DR
 		streamInterceptors = append(streamInterceptors, a.AuthDRPCStream())
 	}
 
+	if tracer := rpcCtx.Stopper.Tracer(); tracer != nil {
+		unaryInterceptors = append(unaryInterceptors, drpcinterceptor.ServerInterceptor(tracer))
+		streamInterceptors = append(streamInterceptors, drpcinterceptor.StreamServerInterceptor(tracer))
+	}
+
 	mux := drpcmux.NewWithInterceptors(unaryInterceptors, streamInterceptors)
 
 	d.Server = drpcserver.NewWithOptions(mux, drpcserver.Options{

pkg/util/tracing/drpcinterceptor/BUILD.bazel

@@ -0,0 +1,43 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "drpcinterceptor",
+    srcs = ["drpc_interceptor.go"],
+    importpath = "github.com/cockroachdb/cockroach/pkg/util/tracing/drpcinterceptor",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/util/ctxutil",
+        "//pkg/util/tracing",
+        "//pkg/util/tracing/tracingutil",
+        "@io_opentelemetry_go_otel//attribute",
+        "@io_opentelemetry_go_otel//codes",
+        "@io_storj_drpc//:drpc",
+        "@io_storj_drpc//drpcclient",
+        "@io_storj_drpc//drpcerr",
+        "@io_storj_drpc//drpcmetadata",
+        "@io_storj_drpc//drpcmux",
+    ],
+)
+
+go_test(
+    name = "drpcinterceptor_test",
+    srcs = ["drpc_interceptor_test.go"],
+    deps = [
+        ":drpcinterceptor",
+        "//pkg/testutils",
+        "//pkg/util",
+        "//pkg/util/leaktest",
+        "//pkg/util/protoutil",
+        "//pkg/util/tracing",
+        "//pkg/util/tracing/tracingpb",
+        "//pkg/util/tracing/tracingutil",
+        "@com_github_cockroachdb_errors//:errors",
+        "@com_github_gogo_protobuf//types",
+        "@com_github_stretchr_testify//require",
+        "@io_storj_drpc//:drpc",
+        "@io_storj_drpc//drpcclient",
+        "@io_storj_drpc//drpcconn",
+        "@io_storj_drpc//drpcmux",
+        "@io_storj_drpc//drpcserver",
+    ],
+)