reworked the telemetry package into helpers for SERVER spans and INTERNAL spans

jrhea · jrhea · commit bde2766477a9 · 2026-01-07T16:05:28.000-06:00
diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go
@@ -18,10 +18,12 @@ package telemetry
 
 import (
 	"context"
+	"fmt"
 
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
+	semconv "go.opentelemetry.io/otel/semconv/v1.38.0"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -38,37 +40,96 @@ func Int64Attribute(key string, val int64) Attribute {
 	return attribute.Int64(key, val)
 }
 
-// StartSpan starts a tracing span on the default tracer and returns a function
-// to end the span. The function will record errors and set span status based
-// on the error value.
-func StartSpan(ctx context.Context, spanName string, attributes ...Attribute) (context.Context, func(*error)) {
-	return StartSpanWithTracer(ctx, otel.Tracer(""), spanName, attributes...)
+// StartServerSpan creates a SpanKind=SERVER span at the JSON-RPC boundary.
+// The span name is formatted as $rpcSystem.$rpcService/$rpcMethod
+// (e.g. "jsonrpc.engine/newPayloadV4").
+func StartServerSpan(
+	ctx context.Context,
+	tracer trace.Tracer,
+	rpcSystem string,
+	rpcService string,
+	rpcMethod string,
+	requestID string,
+	additionalAttributes ...Attribute,
+) (context.Context, func(error)) {
+	spanName := fmt.Sprintf("%s.%s/%s", rpcSystem, rpcService, rpcMethod)
+	ctx, span := tracer.Start(
+		ctx,
+		spanName,
+		trace.WithSpanKind(trace.SpanKindServer),
+	)
+
+	// Fast path: noop provider or span not sampled
+	if !span.IsRecording() {
+		return ctx, func(error) { span.End() }
+	}
+
+	// Define required attributes
+	attrs := []Attribute{
+		semconv.RPCSystemKey.String(rpcSystem),
+		semconv.RPCServiceKey.String(rpcService),
+		semconv.RPCMethodKey.String(rpcMethod),
+		semconv.RPCJSONRPCRequestID(requestID),
+	}
+
+	// Add any additional attributes provided
+	if len(additionalAttributes) > 0 {
+		attrs = append(attrs, additionalAttributes...)
+	}
+	span.SetAttributes(attrs...)
+	return ctx, endSpan(span)
 }
 
-// StartSpanWithTracer starts a tracing span on the supplied tracer and returns
-// a function to end the span. The function will record errors and set span
-// status based on the error value.
-func StartSpanWithTracer(ctx context.Context, tracer trace.Tracer, spanName string, attributes ...Attribute) (context.Context, func(*error)) {
-	ctx, span := tracer.Start(ctx, spanName)
+// StartInternalSpan creates a SpanKind=INTERNAL span.
+func StartInternalSpan(
+	ctx context.Context,
+	spanName string,
+	attributes ...Attribute,
+) (context.Context, func(error)) {
+	return StartInternalSpanWithTracer(ctx, otel.Tracer(""), spanName, attributes...)
+}
 
-	// Fast path: noop provider or span not sampled
+// StartInternalSpanWithTracer requires a tracer to be passed in and creates a SpanKind=INTERNAL span.
+func StartInternalSpanWithTracer(
+	ctx context.Context,
+	tracer trace.Tracer,
+	spanName string,
+	attributes ...Attribute,
+) (context.Context, func(error)) {
+	return startInternalSpan(ctx, tracer, spanName, attributes...)
+}
+
+// startInternalSpan creates a SpanKind=INTERNAL span.
+func startInternalSpan(
+	ctx context.Context,
+	tracer trace.Tracer,
+	spanName string,
+	attributes ...Attribute,
+) (context.Context, func(error)) {
+	ctx, span := tracer.Start(
+		ctx,
+		spanName,
+		trace.WithSpanKind(trace.SpanKindInternal),
+	)
+
+	// Fast path
 	if !span.IsRecording() {
-		return ctx, func(*error) { span.End() }
+		return ctx, func(error) { span.End() }
 	}
 
-	// Set span attributes.
 	if len(attributes) > 0 {
 		span.SetAttributes(attributes...)
 	}
+	return ctx, endSpan(span)
+}
 
-	// Define the function to end the span and handle error recording
-	spanEnd := func(err *error) {
-		if *err != nil {
-			// Error occurred, record it and set status on span
-			span.RecordError(*err)
-			span.SetStatus(codes.Error, (*err).Error())
+// endSpan ends the span and handles error recording.
+func endSpan(span trace.Span) func(error) {
+	return func(err error) {
+		if err != nil {
+			span.RecordError(err)
+			span.SetStatus(codes.Error, err.Error())
 		}
 		span.End()
 	}
-	return ctx, spanEnd
 }
diff --git a/rpc/handler.go b/rpc/handler.go
@@ -510,45 +510,38 @@ func (h *handler) handleCall(cp *callProc, msg *jsonrpcMessage) *jsonrpcMessage
 		return h.runMethod(cp.ctx, msg, h.unsubscribeCb, args)
 	}
 
-	// Start root span for the request.
-	var err error
-	attributes := []telemetry.Attribute{
-		telemetry.StringAttribute("rpc.method", msg.Method),
-		telemetry.StringAttribute("rpc.id", string(msg.ID)),
-	}
-	ctx, spanEnd := telemetry.StartSpanWithTracer(cp.ctx, h.tracer(), "rpc.handleCall", attributes...)
-	defer spanEnd(&err)
-
 	// Check method name length
 	if len(msg.Method) > maxMethodNameLength {
-		errMessage := fmt.Sprintf("method name too long: %d > %d", len(msg.Method), maxMethodNameLength)
-		invalidRequestError := &invalidRequestError{errMessage}
-		err = errors.New(invalidRequestError.Error())
-		return msg.errorResponse(invalidRequestError)
+		return msg.errorResponse(&invalidRequestError{fmt.Sprintf("method name too long: %d > %d", len(msg.Method), maxMethodNameLength)})
 	}
-	callb := h.reg.callback(msg.Method)
+	callb, rpcService, rpcMethod := h.reg.callback(msg.Method)
+
+	// If the method is not found, return an error.
 	if callb == nil {
-		methodNotFoundError := &methodNotFoundError{method: msg.Method}
-		err = errors.New(methodNotFoundError.Error())
-		return msg.errorResponse(methodNotFoundError)
+		return msg.errorResponse(&methodNotFoundError{method: msg.Method})
 	}
 
+	// Start root span for the request.
+	var err error
+	ctx, spanEnd := telemetry.StartServerSpan(cp.ctx, h.tracer(), "jsonrpc", rpcService, rpcMethod, string(msg.ID))
+	defer spanEnd(err)
+
 	// Start tracing span before parsing arguments.
-	_, pSpanEnd := telemetry.StartSpanWithTracer(ctx, h.tracer(), "rpc.parsePositionalArguments", attributes...)
+	_, pSpanEnd := telemetry.StartInternalSpanWithTracer(ctx, h.tracer(), "rpc.parsePositionalArguments")
 	args, err := parsePositionalArguments(msg.Params, callb.argTypes)
-	pSpanEnd(&err)
+	pSpanEnd(err)
 	if err != nil {
 		return msg.errorResponse(&invalidParamsError{err.Error()})
 	}
 	start := time.Now()
 
 	// Start tracing span before running the method.
-	rctx, rSpanEnd := telemetry.StartSpanWithTracer(ctx, h.tracer(), "rpc.runMethod", attributes...)
+	rctx, rSpanEnd := telemetry.StartInternalSpanWithTracer(ctx, h.tracer(), "rpc.runMethod")
 	answer := h.runMethod(rctx, msg, callb, args)
 	if answer.Error != nil {
 		err = errors.New(answer.Error.Message)
 	}
-	rSpanEnd(&err)
+	rSpanEnd(err)
 
 	// Collect the statistics for RPC calls if metrics is enabled.
 	rpcRequestGauge.Inc(1)
@@ -616,12 +609,12 @@ func (h *handler) runMethod(ctx context.Context, msg *jsonrpcMessage, callb *cal
 	if err != nil {
 		return msg.errorResponse(err)
 	}
-	_, spanEnd := telemetry.StartSpanWithTracer(ctx, h.tracer(), "rpc.msg.response", attributes...)
+	_, spanEnd := telemetry.StartInternalSpanWithTracer(ctx, h.tracer(), "rpc.encodeJSONResponse", attributes...)
 	response := msg.response(result)
 	if response.Error != nil {
 		err = errors.New(response.Error.Message)
 	}
-	spanEnd(&err)
+	spanEnd(err)
 	return response
 }
 
diff --git a/rpc/service.go b/rpc/service.go
@@ -92,14 +92,14 @@ func (r *serviceRegistry) registerName(name string, rcvr interface{}) error {
 }
 
 // callback returns the callback corresponding to the given RPC method name.
-func (r *serviceRegistry) callback(method string) *callback {
+func (r *serviceRegistry) callback(method string) (cb *callback, service, methodName string) {
 	before, after, found := strings.Cut(method, serviceMethodSeparator)
 	if !found {
-		return nil
+		return nil, "", ""
 	}
 	r.mu.Lock()
 	defer r.mu.Unlock()
-	return r.services[before].callbacks[after]
+	return r.services[before].callbacks[after], before, after
 }
 
 // subscription returns a subscription callback in the given service.
diff --git a/rpc/tracing_test.go b/rpc/tracing_test.go
@@ -86,67 +86,26 @@ func TestTracingHTTP(t *testing.T) {
 	}
 	var rpcSpan *tracetest.SpanStub
 	for i := range spans {
-		if spans[i].Name == "rpc.handleCall" {
+		if spans[i].Name == "jsonrpc.test/echo" {
 			rpcSpan = &spans[i]
 			break
 		}
 	}
 	if rpcSpan == nil {
-		t.Fatalf("rpc.handleCall span not found.")
+		t.Fatalf("jsonrpc.test/echo span not found.")
 	}
 	attrs := attributeMap(rpcSpan.Attributes)
-	if attrs["rpc.method"] != "test_echo" {
-		t.Errorf("expected rpc.method=test_echo, got %v", attrs["rpc.method"])
+	if attrs["rpc.system"] != "jsonrpc" {
+		t.Errorf("expected rpc.system=jsonrpc, got %v", attrs["rpc.system"])
 	}
-	if _, ok := attrs["rpc.id"]; !ok {
-		t.Errorf("expected rpc.id attribute to be set")
+	if attrs["rpc.service"] != "test" {
+		t.Errorf("expected rpc.service=test, got %v", attrs["rpc.service"])
 	}
-}
-
-// TestTracingHTTPMethodNotFound verifies that a span is emitted when rpc method does not exist.
-func TestTracingHTTPMethodNotFound(t *testing.T) {
-	t.Parallel()
-	server, tracer, exporter := newTracingServer(t)
-	httpsrv := httptest.NewServer(server)
-	t.Cleanup(httpsrv.Close)
-	client, err := DialHTTP(httpsrv.URL)
-	if err != nil {
-		t.Fatalf("failed to dial: %v", err)
-	}
-	t.Cleanup(client.Close)
-
-	// Make a RPC call that should fail.
-	var result echoResult
-	if err := client.Call(&result, "testnonexistent", "hello", 42, &echoArgs{S: "world"}); err == nil {
-		t.Fatalf("RPC call should have failed")
-	}
-
-	// Flush spans.
-	if err := tracer.ForceFlush(context.Background()); err != nil {
-		t.Fatalf("failed to flush: %v", err)
-	}
-
-	// Check spans.
-	spans := exporter.GetSpans()
-	if len(spans) == 0 {
-		t.Fatal("no spans were emitted")
-	}
-	var rpcSpan *tracetest.SpanStub
-	for i := range spans {
-		if spans[i].Name == "rpc.handleCall" {
-			rpcSpan = &spans[i]
-			break
-		}
-	}
-	if rpcSpan == nil {
-		t.Fatalf("rpc.handleCall span not found.")
-	}
-	attrs := attributeMap(rpcSpan.Attributes)
-	if attrs["rpc.method"] != "testnonexistent" {
-		t.Errorf("expected rpc.method=testnonexistent, got %v", attrs["rpc.method"])
+	if attrs["rpc.method"] != "echo" {
+		t.Errorf("expected rpc.method=echo, got %v", attrs["rpc.method"])
 	}
-	if _, ok := attrs["rpc.id"]; !ok {
-		t.Errorf("expected rpc.id attribute to be set")
+	if _, ok := attrs["rpc.jsonrpc.request_id"]; !ok {
+		t.Errorf("expected rpc.jsonrpc.request_id attribute to be set")
 	}
 }
 

Original file line number	Diff line number	Diff line change
`@@ -92,14 +92,14 @@ func (r *serviceRegistry) registerName(name string, rcvr interface{}) error {`
`92`	`92`	`}`
`93`	`93`
`94`	`94`	`// callback returns the callback corresponding to the given RPC method name.`
`95`		`-func (r serviceRegistry) callback(method string) callback {`
	`95`	`+func (r serviceRegistry) callback(method string) (cb callback, service, methodName string) {`
`96`	`96`	`before, after, found := strings.Cut(method, serviceMethodSeparator)`
`97`	`97`	`if !found {`
`98`		`- return nil`
	`98`	`+ return nil, "", ""`
`99`	`99`	`}`
`100`	`100`	`r.mu.Lock()`
`101`	`101`	`defer r.mu.Unlock()`
`102`		`- return r.services[before].callbacks[after]`
	`102`	`+ return r.services[before].callbacks[after], before, after`
`103`	`103`	`}`
`104`	`104`
`105`	`105`	`// subscription returns a subscription callback in the given service.`