diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go index 7a3b86e6d5..6bd16da66c 100644 --- a/internal/telemetry/telemetry.go +++ b/internal/telemetry/telemetry.go @@ -40,6 +40,11 @@ func Int64Attribute(key string, val int64) Attribute { return attribute.Int64(key, val) } +// BoolAttribute creates an attribute with a bool value. +func BoolAttribute(key string, val bool) Attribute { + return attribute.Bool(key, val) +} + // StartSpan creates a SpanKind=INTERNAL span. func StartSpan(ctx context.Context, spanName string, attributes ...Attribute) (context.Context, trace.Span, func(error)) { return StartSpanWithTracer(ctx, otel.Tracer(""), spanName, attributes...) diff --git a/rpc/handler.go b/rpc/handler.go index 038323c9ee..de11f1ebbe 100644 --- a/rpc/handler.go +++ b/rpc/handler.go @@ -77,6 +77,7 @@ type handler struct { type callProc struct { ctx context.Context notifiers []*Notifier + isBatch bool } func newHandler(connCtx context.Context, conn jsonWriter, idgen func() ID, reg *serviceRegistry, batchRequestLimit, batchResponseMaxSize int, tracerProvider trace.TracerProvider) *handler { @@ -202,6 +203,7 @@ func (h *handler) handleBatch(msgs []*jsonrpcMessage) { // Process calls on a goroutine because they may block indefinitely: h.startCallProc(func(cp *callProc) { + cp.isBatch = true var ( timer *time.Timer cancel context.CancelFunc @@ -529,7 +531,10 @@ func (h *handler) handleCall(cp *callProc, msg *jsonrpcMessage) *jsonrpcMessage Method: method, RequestID: string(msg.ID), } - ctx, spanEnd := telemetry.StartServerSpan(cp.ctx, h.tracer(), rpcInfo) + attrib := []telemetry.Attribute{ + telemetry.BoolAttribute("rpc.batch", cp.isBatch), + } + ctx, spanEnd := telemetry.StartServerSpan(cp.ctx, h.tracer(), rpcInfo, attrib...) defer spanEnd(err) // Start tracing span before parsing arguments. diff --git a/rpc/tracing_test.go b/rpc/tracing_test.go index f675b831f8..3769cadee5 100644 --- a/rpc/tracing_test.go +++ b/rpc/tracing_test.go @@ -74,12 +74,10 @@ func TestTracingHTTP(t *testing.T) { t.Fatalf("RPC call failed: %v", err) } - // Flush spans. + // Flush and verify that we emitted the expected span. 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") @@ -109,6 +107,60 @@ func TestTracingHTTP(t *testing.T) { } } +// TestTracingBatchHTTP verifies that RPC spans are emitted for batched JSON-RPC calls over HTTP. +func TestTracingBatchHTTP(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 successful batch RPC call. + batch := []BatchElem{ + { + Method: "test_echo", + Args: []any{"hello", 42, &echoArgs{S: "world"}}, + Result: new(echoResult), + }, + { + Method: "test_echo", + Args: []any{"your", 7, &echoArgs{S: "mom"}}, + Result: new(echoResult), + }, + } + if err := client.BatchCall(batch); err != nil { + t.Fatalf("batch RPC call failed: %v", err) + } + + // Flush and verify we emitted spans for each batch element. + 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 found int + for i := range spans { + if spans[i].Name == "jsonrpc.test/echo" { + attrs := attributeMap(spans[i].Attributes) + if attrs["rpc.system"] == "jsonrpc" && + attrs["rpc.service"] == "test" && + attrs["rpc.method"] == "echo" && + attrs["rpc.batch"] == "true" { + found++ + } + } + } + if found != len(batch) { + t.Fatalf("expected %d matching batch spans, got %d", len(batch), found) + } +} + // TestTracingSubscribeUnsubscribe verifies that subscribe and unsubscribe calls // do not emit any spans. // Note: This works because client.newClientConn() does not set a tracer provider.