diff --git a/src/instana/instrumentation/grpcio.py b/src/instana/instrumentation/grpcio.py index c497239f..e9e91668 100644 --- a/src/instana/instrumentation/grpcio.py +++ b/src/instana/instrumentation/grpcio.py @@ -68,7 +68,6 @@ def create_span( argv: Tuple[object, ...], kwargs: Dict[str, Any], call_type: str, - record_exception: bool = True, ) -> object: parent_span = get_current_span() tracer = get_tracer() @@ -80,7 +79,7 @@ def create_span( parent_context = get_current() with tracer.start_as_current_span( - "rpc-client", context=parent_context, record_exception=record_exception + "rpc-client", context=parent_context, record_exception=False ) as span: try: if "metadata" not in kwargs: @@ -94,12 +93,14 @@ def create_span( ) collect_attributes(span, instance, argv, kwargs) span.set_attribute("rpc.call_type", call_type) + except Exception: + logger.debug("grpc.create_span non-fatal error", exc_info=True) - rv = wrapped(*argv, **kwargs) + try: + return wrapped(*argv, **kwargs) except Exception as exc: span.record_exception(exc) - else: - return rv + raise @wrapt.patch_function_wrapper("grpc._channel", "_UnaryUnaryMultiCallable.with_call") def unary_unary_with_call_with_instana( @@ -126,9 +127,7 @@ def unary_unary_call_with_instana( argv: Tuple[object, ...], kwargs: Dict[str, Any], ) -> object: - return create_span( - wrapped, instance, argv, kwargs, call_type="unary", record_exception=False - ) + return create_span(wrapped, instance, argv, kwargs, call_type="unary") @wrapt.patch_function_wrapper("grpc._channel", "_StreamUnaryMultiCallable.__call__") def stream_unary_call_with_instana( @@ -187,24 +186,30 @@ def call_behavior_with_instana( kwargs: Dict[str, Any], ) -> object: tracer = get_tracer() - # Prep any incoming context headers - metadata = argv[0].invocation_metadata - metadata_dict = {} - for c in metadata: - metadata_dict[c.key] = c.value - - ctx = tracer.extract( - Format.BINARY, metadata_dict, disable_w3c_trace_context=True - ) + ctx = None + try: + # Prep any incoming context headers + metadata = getattr(argv[0], "invocation_metadata", None) or () + metadata_dict = {c.key: c.value for c in metadata} + ctx = tracer.extract( + Format.BINARY, metadata_dict, disable_w3c_trace_context=True + ) + except Exception: + logger.debug("grpc.call_behavior extract context non-fatal error", exc_info=True) - with tracer.start_as_current_span("rpc-server", context=ctx) as span: + with tracer.start_as_current_span( + "rpc-server", context=ctx, record_exception=False + ) as span: try: collect_attributes(span, instance, argv, kwargs) - rv = wrapped(*argv, **kwargs) + except Exception: + logger.debug("grpc.call_behavior non-fatal error", exc_info=True) + + try: + return wrapped(*argv, **kwargs) except Exception as exc: span.record_exception(exc) - else: - return rv + raise logger.debug("Instrumenting grpcio") except ImportError: diff --git a/tests/frameworks/test_grpcio.py b/tests/frameworks/test_grpcio.py index 45b2a803..e2b3c4b8 100644 --- a/tests/frameworks/test_grpcio.py +++ b/tests/frameworks/test_grpcio.py @@ -2,10 +2,9 @@ # (c) Copyright Instana Inc. 2020 -import contextlib import random import time -from typing import Generator +from collections.abc import Generator import grpc import pytest @@ -582,15 +581,16 @@ def process_response(future): assert test_span.data["sdk"]["name"] == "test" def test_server_error(self) -> None: - response = None - with self.tracer.start_as_current_span("test"): # noqa: SIM117 - with contextlib.suppress(Exception): - response = self.server_stub.OneQuestionOneErrorResponse( - stan_pb2.QuestionRequest(question="Do u error?") - ) + with ( + self.tracer.start_as_current_span("test"), + pytest.raises(grpc.RpcError) as exc_info, + ): + self.server_stub.OneQuestionOneErrorResponse( + stan_pb2.QuestionRequest(question="Do u error?") + ) assert not get_current_span().is_recording() - assert not response + assert exc_info.value.code() == grpc.StatusCode.UNKNOWN spans = self.recorder.queued_spans() assert len(spans) == 4 @@ -655,6 +655,85 @@ def test_server_error(self) -> None: assert test_span.n == "sdk" assert test_span.data["sdk"]["name"] == "test" + def test_server_error_with_call(self) -> None: + with ( + self.tracer.start_as_current_span("test"), + pytest.raises(grpc.RpcError) as exc_info, + ): + self.server_stub.OneQuestionOneErrorResponse.with_call( + stan_pb2.QuestionRequest(question="Do u error?") + ) + + assert exc_info.value.code() == grpc.StatusCode.UNKNOWN + + spans = self.recorder.queued_spans() + assert len(spans) == 4 + + log_span = get_first_span_by_name(spans, "log") + server_span = get_first_span_by_name(spans, "rpc-server") + client_span = get_first_span_by_name(spans, "rpc-client") + test_span = get_first_span_by_name(spans, "sdk") + + assert log_span + assert server_span + assert client_span + assert test_span + + # Same traceId + assert server_span.t == client_span.t + assert server_span.t == test_span.t + + # Parent relationships + assert server_span.p == client_span.s + assert client_span.p == test_span.s + + # Error logging + assert not test_span.ec + assert client_span.ec == 1 + assert not server_span.ec + + # rpc-client + assert client_span.n == "rpc-client" + assert client_span.k is SpanKind.CLIENT + assert client_span.stack + assert client_span.data["rpc"]["flavor"] == "grpc" + assert ( + client_span.data["rpc"]["call"] == "/stan.Stan/OneQuestionOneErrorResponse" + ) + assert client_span.data["rpc"]["host"] == testenv["grpc_host"] + assert client_span.data["rpc"]["port"] == str(testenv["grpc_port"]) + assert client_span.data["rpc"]["call_type"] == "unary" + assert client_span.data["rpc"]["error"] + + def test_server_error_with_interceptor(self) -> None: + class DummyInterceptor(grpc.UnaryUnaryClientInterceptor): + def intercept_unary_unary(self, continuation, client_call_details, request): + return continuation(client_call_details, request) + + intercepted_channel = grpc.intercept_channel( + self.channel, DummyInterceptor() + ) + intercepted_stub = stan_pb2_grpc.StanStub(intercepted_channel) + + with ( + self.tracer.start_as_current_span("test"), + pytest.raises(grpc.RpcError) as exc_info, + ): + intercepted_stub.OneQuestionOneErrorResponse( + stan_pb2.QuestionRequest(question="Do u error?") + ) + + assert exc_info.value.code() == grpc.StatusCode.UNKNOWN + + spans = self.recorder.queued_spans() + assert len(spans) == 4 + + client_span = get_first_span_by_name(spans, "rpc-client") + assert client_span + assert client_span.ec == 1 + assert client_span.data["rpc"]["call_type"] == "unary" + assert client_span.data["rpc"]["error"] + def test_root_exit_span(self) -> None: agent.options.allow_exit_as_root = True