From 60df872d5ac368e2e20185e4426a4c4352b09ae6 Mon Sep 17 00:00:00 2001 From: jonny rhea Date: Tue, 6 Jan 2026 17:26:56 -0600 Subject: [PATCH] moved otel logic to internal/telemetry and simplified logic --- internal/telemetry/telemetry.go | 55 +++++++++++++++++++++ rpc/handler.go | 86 ++++++++++----------------------- rpc/tracing_test.go | 46 ++++++++++++++++-- 3 files changed, 123 insertions(+), 64 deletions(-) create mode 100644 internal/telemetry/telemetry.go diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go new file mode 100644 index 0000000000..4652c15543 --- /dev/null +++ b/internal/telemetry/telemetry.go @@ -0,0 +1,55 @@ +package telemetry + +import ( + "context" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" +) + +type Attribute = attribute.KeyValue + +func StringAttribute(key, val string) Attribute { + return attribute.String(key, val) +} + +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) + + // Fast path: noop provider or span not sampled + if !span.IsRecording() { + return ctx, func(*error) { span.End() } + } + + // Set span attributes. + if len(attributes) > 0 { + span.SetAttributes(attributes...) + } + + // 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 and parent + 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 c7aa0b8434..28e92e036e 100644 --- a/rpc/handler.go +++ b/rpc/handler.go @@ -28,11 +28,9 @@ import ( "sync" "time" + "github.com/ethereum/go-ethereum/internal/telemetry" "github.com/ethereum/go-ethereum/log" "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/attribute" - "go.opentelemetry.io/otel/codes" - semconv "go.opentelemetry.io/otel/semconv/v1.24.0" "go.opentelemetry.io/otel/trace" ) @@ -512,22 +510,31 @@ 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 { - return msg.errorResponse(&invalidRequestError{fmt.Sprintf("method name too long: %d > %d", 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) } callb := h.reg.callback(msg.Method) if callb == nil { - return msg.errorResponse(&methodNotFoundError{method: msg.Method}) + methodNotFoundError := &methodNotFoundError{method: msg.Method} + err = errors.New(methodNotFoundError.Error()) + return msg.errorResponse(methodNotFoundError) } - // Start root span for the request. - var err error - ctx, spanEnd := h.startSpan(cp.ctx, msg, "rpc.handleCall") - defer spanEnd(&err) - // Start tracing span before parsing arguments. - _, pSpanEnd := h.startSpan(ctx, msg, "rpc.parsePositionalArguments") + _, pSpanEnd := telemetry.StartSpanWithTracer(ctx, h.tracer(), "rpc.parsePositionalArguments", attributes...) args, err := parsePositionalArguments(msg.Params, callb.argTypes) pSpanEnd(&err) if err != nil { @@ -536,7 +543,7 @@ func (h *handler) handleCall(cp *callProc, msg *jsonrpcMessage) *jsonrpcMessage start := time.Now() // Start tracing span before running the method. - rctx, rSpanEnd := h.startSpan(ctx, msg, "rpc.runMethod") + rctx, rSpanEnd := telemetry.StartSpanWithTracer(ctx, h.tracer(), "rpc.runMethod", attributes...) answer := h.runMethod(rctx, msg, callb, args) if answer.Error != nil { err = errors.New(answer.Error.Message) @@ -552,7 +559,6 @@ func (h *handler) handleCall(cp *callProc, msg *jsonrpcMessage) *jsonrpcMessage } rpcServingTimer.UpdateSince(start) updateServeTimeHistogram(msg.Method, answer.Error == nil, time.Since(start)) - return answer } @@ -593,36 +599,6 @@ func (h *handler) handleSubscribe(cp *callProc, msg *jsonrpcMessage) *jsonrpcMes return h.runMethod(ctx, msg, callb, args) } -// startSpan starts a tracing span for an RPC call and returns a function to -// end the span. The function will record errors and set span status based on -// the error value. -func (h *handler) startSpan(ctx context.Context, msg *jsonrpcMessage, spanName string) (context.Context, func(*error)) { - ctx, span := h.tracer().Start(ctx, spanName) - - // Fast path: noop provider or span not sampled - if !span.IsRecording() { - return ctx, func(*error) { span.End() } - } - - // Set span attributes. - span.SetAttributes(semconv.RPCMethodKey.String(msg.Method)) - id := strings.TrimSpace(string(msg.ID)) - if id != "" && id != "null" { - span.SetAttributes(attribute.String("rpc.id", id)) - } - - // 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 and parent - span.RecordError(*err) - span.SetStatus(codes.Error, (*err).Error()) - } - span.End() - } - return ctx, spanEnd -} - // tracer returns the OpenTelemetry Tracer for RPC call tracing. func (h *handler) tracer() trace.Tracer { if h.tracerProvider == nil { @@ -635,28 +611,18 @@ func (h *handler) tracer() trace.Tracer { } // runMethod runs the Go callback for an RPC method. -func (h *handler) runMethod(ctx context.Context, msg *jsonrpcMessage, callb *callback, args []reflect.Value) *jsonrpcMessage { +func (h *handler) runMethod(ctx context.Context, msg *jsonrpcMessage, callb *callback, args []reflect.Value, attributes ...telemetry.Attribute) *jsonrpcMessage { result, err := callb.call(ctx, msg.Method, args) if err != nil { return msg.errorResponse(err) } - - // If parent span is recording, start a span for the response. - // Note: This prevents msg.response spans from being created when - // the parent span is not recording. This is needed bc we aren't - // currently recording spans for subscriptions, but we do record spans - // for other RPC methods and this is called for both. - parentSpan := trace.SpanFromContext(ctx) - if parentSpan.IsRecording() { - _, spanEnd := h.startSpan(ctx, msg, "rpc.msg.response") - response := msg.response(result) - if response.Error != nil { - err = errors.New(response.Error.Message) - } - spanEnd(&err) - return response + _, spanEnd := telemetry.StartSpanWithTracer(ctx, h.tracer(), "rpc.msg.response", attributes...) + response := msg.response(result) + if response.Error != nil { + err = errors.New(response.Error.Message) } - return msg.response(result) + spanEnd(&err) + return response } // unsubscribe is the callback function for all *_unsubscribe calls. diff --git a/rpc/tracing_test.go b/rpc/tracing_test.go index 56aed88ac7..1772d2ef08 100644 --- a/rpc/tracing_test.go +++ b/rpc/tracing_test.go @@ -72,16 +72,13 @@ func TestTracingHTTP(t *testing.T) { if err := client.Call(&result, "test_echo", "hello", 42, &echoArgs{S: "world"}); err != nil { t.Fatalf("RPC call failed: %v", err) } - if err := tracer.ForceFlush(context.Background()); err != nil { t.Fatalf("failed to flush: %v", err) } - 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" { @@ -92,7 +89,6 @@ func TestTracingHTTP(t *testing.T) { if rpcSpan == nil { t.Fatalf("rpc.handleCall 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"]) @@ -102,8 +98,50 @@ func TestTracingHTTP(t *testing.T) { } } +func TestTracingHTTPShouldFail(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) + + var result echoResult + if err := client.Call(&result, "testnonexistent", "hello", 42, &echoArgs{S: "world"}); err == nil { + t.Fatalf("RPC call should have failed") + } + if err := tracer.ForceFlush(context.Background()); err != nil { + t.Fatalf("failed to flush: %v", err) + } + 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") + } +} + // TestTracingSubscribeUnsubscribe verifies that subscribe and unsubscribe calls // do not emit any spans. +// Note: This works because client.newClientConn() does not set a tracer provider. func TestTracingSubscribeUnsubscribe(t *testing.T) { t.Parallel() server, tracer, exporter := newTracingServer(t)