From bde2766477a90087a79af596664190de0f22a5f6 Mon Sep 17 00:00:00 2001 From: jonny rhea Date: Wed, 7 Jan 2026 15:51:29 -0600 Subject: [PATCH] reworked the telemetry package into helpers for SERVER spans and INTERNAL spans --- internal/telemetry/telemetry.go | 103 +++++++++++++++++++++++++------- rpc/handler.go | 39 +++++------- rpc/service.go | 6 +- rpc/tracing_test.go | 61 ++++--------------- 4 files changed, 111 insertions(+), 98 deletions(-) diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go index 2a6bd45907..cd55a8b745 100644 --- 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...) -} - -// 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) +// 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() } + 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) +} + +// 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...) +} + +// 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() } } - // 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 index 28e92e036e..e525a4f27c 100644 --- 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 index 0f62d7eb7c..8462a5a59a 100644 --- 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 index 62b3999949..067351fb38 100644 --- 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) + if attrs["rpc.method"] != "echo" { + t.Errorf("expected rpc.method=echo, got %v", attrs["rpc.method"]) } - 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 _, 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") } }