added back rpc.batch attribute + test

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...)

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.

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.