fix(grpcio): properly record and re-raise exceptions

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:

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