Skip to content

feat(grpc): Support span streaming #6204

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
alexander-alderman-webb wants to merge 12 commits into
base: master
Choose a base branch
from
Open

feat(grpc): Support span streaming #6204

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
6acab90
feat(grpc): Support span streaming
alexander-alderman-webb May 5, 2026
8f48c88
remove files
alexander-alderman-webb May 5, 2026
3c26279
add fixture
alexander-alderman-webb May 5, 2026
3deb797
avoid redefinition
alexander-alderman-webb May 5, 2026
42e65e7
change attributes
alexander-alderman-webb May 5, 2026
049f1de
add spandata
alexander-alderman-webb May 5, 2026
71c3619
use const instead of string
alexander-alderman-webb May 5, 2026
26a2672
merge master
alexander-alderman-webb May 6, 2026
e10d7ee
narrow capture_items
alexander-alderman-webb May 6, 2026
0cd6f4c
.
alexander-alderman-webb May 6, 2026
752049d
set attribute in start_span
alexander-alderman-webb May 6, 2026
a789a2e
.
alexander-alderman-webb May 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions sentry_sdk/consts.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -899,6 +899,12 @@ class SPANDATA:
Example: "com.example.ExampleService/exampleMethod"
"""

RPC_RESPONSE_STATUS_CODE = "rpc.response.status_code"
"""
Status code of the RPC returned by the RPC server or generated by the client.
Example: "DEADLINE_EXCEEDED"
"""

SERVER_ADDRESS = "server.address"
"""
Name of the database host.
Expand Down
121 changes: 86 additions & 35 deletions sentry_sdk/integrations/grpc/aio/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
from typing import Callable, Union, AsyncIterable, Any

import sentry_sdk
from sentry_sdk.consts import OP
from sentry_sdk.consts import OP, SPANDATA
from sentry_sdk.integrations import DidNotEnable
from sentry_sdk.integrations.grpc.consts import SPAN_ORIGIN
from sentry_sdk.tracing_utils import has_span_streaming_enabled

try:
from grpc.aio import (
Expand Down Expand Up @@ -49,23 +50,49 @@ async def intercept_unary_unary(
) -> "Union[UnaryUnaryCall, Message]":
method = client_call_details.method

with sentry_sdk.start_span(
op=OP.GRPC_CLIENT,
name="unary unary call to %s" % method.decode(),
origin=SPAN_ORIGIN,
) as span:
span.set_data("type", "unary unary")
span.set_data("method", method)

client_call_details = self._update_client_call_details_metadata_from_scope(
client_call_details
)

response = await continuation(client_call_details, request)
status_code = await response.code()
span.set_data("code", status_code.name)

return response
client = sentry_sdk.get_client()
span_streaming = has_span_streaming_enabled(client.options)

if span_streaming:
with sentry_sdk.traces.start_span(
name="unary unary call to %s" % method.decode(),
attributes={
"sentry.op": OP.GRPC_CLIENT,
"sentry.origin": SPAN_ORIGIN,
SPANDATA.RPC_METHOD: method.decode(),
},
) as span:
client_call_details = (
self._update_client_call_details_metadata_from_scope(
client_call_details
)
)

response = await continuation(client_call_details, request)
status_code = await response.code()
span.set_attribute(SPANDATA.RPC_RESPONSE_STATUS_CODE, status_code.name)

return response
else:
with sentry_sdk.start_span(
op=OP.GRPC_CLIENT,
name="unary unary call to %s" % method.decode(),
origin=SPAN_ORIGIN,
) as span:
span.set_data("type", "unary unary")
span.set_data("method", method)

client_call_details = (
self._update_client_call_details_metadata_from_scope(
client_call_details
)
)

response = await continuation(client_call_details, request)
status_code = await response.code()
span.set_data("code", status_code.name)

return response


class SentryUnaryStreamClientInterceptor(
Expand All @@ -80,20 +107,44 @@ async def intercept_unary_stream(
) -> "Union[AsyncIterable[Any], UnaryStreamCall]":
method = client_call_details.method

with sentry_sdk.start_span(
op=OP.GRPC_CLIENT,
name="unary stream call to %s" % method.decode(),
origin=SPAN_ORIGIN,
) as span:
span.set_data("type", "unary stream")
span.set_data("method", method)

client_call_details = self._update_client_call_details_metadata_from_scope(
client_call_details
)

response = await continuation(client_call_details, request)
# status_code = await response.code()
# span.set_data("code", status_code)

return response
client = sentry_sdk.get_client()
span_streaming = has_span_streaming_enabled(client.options)

if span_streaming:
with sentry_sdk.traces.start_span(
name="unary stream call to %s" % method.decode(),
attributes={
"sentry.op": OP.GRPC_CLIENT,
"sentry.origin": SPAN_ORIGIN,
SPANDATA.RPC_METHOD: method.decode(),
},
) as span:
client_call_details = (
self._update_client_call_details_metadata_from_scope(
client_call_details
)
)

response = await continuation(client_call_details, request)

return response
else:
with sentry_sdk.start_span(
op=OP.GRPC_CLIENT,
name="unary stream call to %s" % method.decode(),
origin=SPAN_ORIGIN,
) as span:
span.set_data("type", "unary stream")
span.set_data("method", method)

client_call_details = (
self._update_client_call_details_metadata_from_scope(
client_call_details
)
)

response = await continuation(client_call_details, request)
# status_code = await response.code()
# span.set_data("code", status_code)

return response
71 changes: 50 additions & 21 deletions sentry_sdk/integrations/grpc/aio/server.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from sentry_sdk.integrations.grpc.consts import SPAN_ORIGIN
from sentry_sdk.tracing import TransactionSource
from sentry_sdk.utils import event_from_exception
from sentry_sdk.tracing_utils import has_span_streaming_enabled

from typing import TYPE_CHECKING

Expand Down Expand Up @@ -52,27 +53,55 @@ async def wrapped(request: "Any", context: "ServicerContext") -> "Any":
if not name:
return await handler(request, context)

# What if the headers are empty?
transaction = sentry_sdk.continue_trace(
dict(context.invocation_metadata()),
op=OP.GRPC_SERVER,
name=name,
source=TransactionSource.CUSTOM,
origin=SPAN_ORIGIN,
)

with sentry_sdk.start_transaction(transaction=transaction):
try:
return await handler.unary_unary(request, context)
except AbortError:
raise
except Exception as exc:
event, hint = event_from_exception(
exc,
mechanism={"type": "grpc", "handled": False},
)
sentry_sdk.capture_event(event, hint=hint)
raise
client = sentry_sdk.get_client()
span_streaming = has_span_streaming_enabled(client.options)
if span_streaming:
# What if the headers are empty?
sentry_sdk.traces.continue_trace(
dict(context.invocation_metadata())
)

with sentry_sdk.traces.start_span(
name=name,
attributes={
"sentry.op": OP.GRPC_SERVER,
"sentry.span.source": TransactionSource.CUSTOM,
"sentry.origin": SPAN_ORIGIN,
},
):
Comment thread
alexander-alderman-webb marked this conversation as resolved.
try:
return await handler.unary_unary(request, context)
except AbortError:
raise
except Exception as exc:
event, hint = event_from_exception(
exc,
mechanism={"type": "grpc", "handled": False},
)
sentry_sdk.capture_event(event, hint=hint)
raise
else:
# What if the headers are empty?
transaction = sentry_sdk.continue_trace(
dict(context.invocation_metadata()),
op=OP.GRPC_SERVER,
name=name,
source=TransactionSource.CUSTOM,
origin=SPAN_ORIGIN,
)

with sentry_sdk.start_transaction(transaction=transaction):
try:
return await handler.unary_unary(request, context)
except AbortError:
raise
except Exception as exc:
event, hint = event_from_exception(
exc,
mechanism={"type": "grpc", "handled": False},
)
sentry_sdk.capture_event(event, hint=hint)
raise

elif not handler.request_streaming and handler.response_streaming:
handler_factory = grpc.unary_stream_rpc_method_handler
Expand Down
121 changes: 87 additions & 34 deletions sentry_sdk/integrations/grpc/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import sentry_sdk
from sentry_sdk.consts import OP
from sentry_sdk.consts import OP, SPANDATA
from sentry_sdk.integrations import DidNotEnable
from sentry_sdk.integrations.grpc.consts import SPAN_ORIGIN
from sentry_sdk.tracing_utils import has_span_streaming_enabled

from typing import TYPE_CHECKING

Expand Down Expand Up @@ -30,22 +31,48 @@ def intercept_unary_unary(
) -> "_UnaryOutcome":
method = client_call_details.method

with sentry_sdk.start_span(
op=OP.GRPC_CLIENT,
name="unary unary call to %s" % method,
origin=SPAN_ORIGIN,
) as span:
span.set_data("type", "unary unary")
span.set_data("method", method)

client_call_details = self._update_client_call_details_metadata_from_scope(
client_call_details
)

response = continuation(client_call_details, request)
span.set_data("code", response.code().name)

return response
client = sentry_sdk.get_client()
span_streaming = has_span_streaming_enabled(client.options)
if span_streaming:
with sentry_sdk.traces.start_span(
name="unary unary call to %s" % method,
attributes={
"sentry.op": OP.GRPC_CLIENT,
"sentry.origin": SPAN_ORIGIN,
SPANDATA.RPC_METHOD: method,
},
) as span:
client_call_details = (
self._update_client_call_details_metadata_from_scope(
client_call_details
)
)

response = continuation(client_call_details, request)
span.set_attribute(
SPANDATA.RPC_RESPONSE_STATUS_CODE, response.code().name
)

return response
else:
with sentry_sdk.start_span(
op=OP.GRPC_CLIENT,
name="unary unary call to %s" % method,
origin=SPAN_ORIGIN,
) as span:
span.set_data("type", "unary unary")
span.set_data("method", method)

client_call_details = (
self._update_client_call_details_metadata_from_scope(
client_call_details
)
)

response = continuation(client_call_details, request)
span.set_data("code", response.code().name)

return response

def intercept_unary_stream(
self: "ClientInterceptor",
Expand All @@ -55,23 +82,49 @@ def intercept_unary_stream(
) -> "Union[Iterator[Message], Call]":
method = client_call_details.method

with sentry_sdk.start_span(
op=OP.GRPC_CLIENT,
name="unary stream call to %s" % method,
origin=SPAN_ORIGIN,
) as span:
span.set_data("type", "unary stream")
span.set_data("method", method)

client_call_details = self._update_client_call_details_metadata_from_scope(
client_call_details
)

response: "UnaryStreamCall" = continuation(client_call_details, request)
# Setting code on unary-stream leads to execution getting stuck
# span.set_data("code", response.code().name)

return response
client = sentry_sdk.get_client()
span_streaming = has_span_streaming_enabled(client.options)
response: "UnaryStreamCall"
if span_streaming:
with sentry_sdk.traces.start_span(
name="unary stream call to %s" % method,
attributes={
"sentry.op": OP.GRPC_CLIENT,
"sentry.origin": SPAN_ORIGIN,
SPANDATA.RPC_METHOD: method,
},
) as span:
client_call_details = (
self._update_client_call_details_metadata_from_scope(
client_call_details
)
)

response = continuation(client_call_details, request)
# Setting code on unary-stream leads to execution getting stuck
# span.set_data("code", response.code().name)

return response
else:
with sentry_sdk.start_span(
op=OP.GRPC_CLIENT,
name="unary stream call to %s" % method,
origin=SPAN_ORIGIN,
) as span:
span.set_data("type", "unary stream")
span.set_data("method", method)

client_call_details = (
self._update_client_call_details_metadata_from_scope(
client_call_details
)
)

response = continuation(client_call_details, request)
# Setting code on unary-stream leads to execution getting stuck
# span.set_data("code", response.code().name)

return response

@staticmethod
def _update_client_call_details_metadata_from_scope(
Expand Down
Loading
Loading