feat: include called service URI in StatusRuntimeException descriptions

interop-tests/src/test/java/org/apache/pekko/grpc/interop/PekkoGrpcJavaClientTester.java

@@ -14,6 +14,7 @@
 package org.apache.pekko.grpc.interop;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
 import com.google.protobuf.ByteString;
@@ -460,7 +461,11 @@ public void statusCodeAndMessage() throws Exception {
 
               final StatusRuntimeException e = (StatusRuntimeException) ex;
               assertEquals(Status.UNKNOWN.getCode(), e.getStatus().getCode());
-              assertEquals(errorMessage, e.getStatus().getDescription());
+              assertTrue(
+                  e.getStatus().getDescription() != null
+                      && e.getStatus().getDescription().startsWith(errorMessage),
+                  () -> "Expected description to start with '" + errorMessage + "' but was '"
+                      + e.getStatus().getDescription() + "'");
 
               return null;
             })
@@ -482,7 +487,11 @@ public void statusCodeAndMessage() throws Exception {
 
               final StatusRuntimeException e = (StatusRuntimeException) ex;
               assertEquals(Status.UNKNOWN.getCode(), e.getStatus().getCode());
-              assertEquals(errorMessage, e.getStatus().getDescription());
+              assertTrue(
+                  e.getStatus().getDescription() != null
+                      && e.getStatus().getDescription().startsWith(errorMessage),
+                  () -> "Expected description to start with '" + errorMessage + "' but was '"
+                      + e.getStatus().getDescription() + "'");
 
               return null;
             })

interop-tests/src/test/scala/org/apache/pekko/grpc/interop/PekkoGrpcScalaClientTester.scala

@@ -287,6 +287,8 @@ class PekkoGrpcScalaClientTester(val settings: Settings, backend: String, testWi
     throwable shouldBe a[StatusRuntimeException]
     val e = throwable.asInstanceOf[StatusRuntimeException]
     assertEquals(expectedStatus.getCode, e.getStatus.getCode)
-    assertEquals(expectedMessage, e.getStatus.getDescription)
+    assert(
+      e.getStatus.getDescription != null && e.getStatus.getDescription.startsWith(expectedMessage),
+      s"Expected description to start with '$expectedMessage' but was '${e.getStatus.getDescription}'")
   }
 }

runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala

@@ -185,7 +185,7 @@ object PekkoHttpClientUtils {
             descriptor.getFullMethodName),
           GrpcEntityHelpers.metadataHeaders(headers.entries),
           source)
-        responseToSource(singleRequest(httpRequest), deserializer)
+        responseToSource(httpRequest.uri, singleRequest(httpRequest), deserializer)
       }
     }
   }
@@ -194,15 +194,15 @@ object PekkoHttpClientUtils {
    * INTERNAL API
    */
   @InternalApi
-  def responseToSource[O](response: Future[HttpResponse], deserializer: ProtobufSerializer[O])(
+  def responseToSource[O](requestUri: Uri, response: Future[HttpResponse], deserializer: ProtobufSerializer[O])(
       implicit ec: ExecutionContext,
       mat: Materializer): Source[O, Future[GrpcResponseMetadata]] = {
     Source.lazyFutureSource[O, Future[GrpcResponseMetadata]](() => {
       response.map { response =>
         {
           if (response.status != StatusCodes.OK) {
             response.entity.discardBytes()
-            val failure = mapToStatusException(response, immutable.Seq.empty)
+            val failure = mapToStatusException(requestUri, response, immutable.Seq.empty)
             Source.failed(failure).mapMaterializedValue(_ => Future.failed(failure))
           } else {
             Codecs.detect(response) match {
@@ -211,7 +211,7 @@ object PekkoHttpClientUtils {
                 val trailerPromise = Promise[immutable.Seq[HttpHeader]]()
                 // Completed with success or failure based on grpc-status and grpc-message trailing headers
                 val completionFuture: Future[Unit] =
-                  trailerPromise.future.flatMap(trailers => parseResponseStatus(response, trailers))
+                  trailerPromise.future.flatMap(trailers => parseResponseStatus(requestUri, response, trailers))
 
                 val responseData =
                   response.entity match {
@@ -234,7 +234,7 @@ object PekkoHttpClientUtils {
                       Source.single[ByteString](data)
                     case _ =>
                       response.entity.discardBytes()
-                      throw mapToStatusException(response, Seq.empty)
+                      throw mapToStatusException(requestUri, response, Seq.empty)
                   }
                 responseData
                   // This never adds any data to the stream, but makes sure it fails with the correct error code if applicable
@@ -272,25 +272,37 @@ object PekkoHttpClientUtils {
     })
   }.mapMaterializedValue(_.flatten)
 
-  private def parseResponseStatus(response: HttpResponse, trailers: Seq[HttpHeader]): Future[Unit] = {
+  private def parseResponseStatus(requestUri: Uri, response: HttpResponse, trailers: Seq[HttpHeader]): Future[Unit] = {
     val allHeaders = response.headers ++ trailers
     allHeaders.find(_.name == "grpc-status").map(_.value) match {
       case Some("0") =>
         Future.successful(())
       case _ =>
-        Future.failed(mapToStatusException(response, trailers))
+        Future.failed(mapToStatusException(requestUri, response, trailers))
     }
   }
 
-  private def mapToStatusException(response: HttpResponse, trailers: Seq[HttpHeader]): StatusRuntimeException = {
+  private def mapToStatusException(
+      requestUri: Uri,
+      response: HttpResponse,
+      trailers: Seq[HttpHeader]): StatusRuntimeException = {
     val allHeaders = response.headers ++ trailers
     val metadata: io.grpc.Metadata = new MetadataImpl(new HeaderMetadataImpl(allHeaders).asList).toGoogleGrpcMetadata()
     allHeaders.find(_.name == "grpc-status").map(_.value) match {
       case None =>
-        new StatusRuntimeException(mapHttpStatus(response).withDescription("No grpc-status found"), metadata)
+        new StatusRuntimeException(
+          mapHttpStatus(response)
+            .withDescription("No grpc-status found")
+            .augmentDescription(s"When calling rpc service: ${requestUri.toString()}"),
+          metadata)
       case Some(statusCode) =>
         val description = allHeaders.find(_.name == "grpc-message").map(_.value)
-        new StatusRuntimeException(Status.fromCodeValue(statusCode.toInt).withDescription(description.orNull), metadata)
+        new StatusRuntimeException(
+          Status
+            .fromCodeValue(statusCode.toInt)
+            .withDescription(description.orNull)
+            .augmentDescription(s"When calling rpc service: ${requestUri.toString()}"),
+          metadata)
     }
   }
 

runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala

@@ -35,16 +35,19 @@ class PekkoHttpClientUtilsSpec extends TestKit(ActorSystem()) with AnyWordSpecLi
   implicit val patience: PatienceConfig =
     PatienceConfig(5.seconds, Span(100, org.scalatest.time.Millis))
 
+  val testUri = Uri("https://example.com/test.Service/Method")
+
   "The conversion from HttpResponse to Source" should {
     "map a strict 404 response to a failed stream" in {
       val response =
         Future.successful(HttpResponse(NotFound, entity = Strict(GrpcProtocolNative.contentType, ByteString.empty)))
-      val source = PekkoHttpClientUtils.responseToSource(response, null)
+      val source = PekkoHttpClientUtils.responseToSource(testUri, response, null)
 
       val failure = source.run().failed.futureValue
       failure shouldBe a[StatusRuntimeException]
       // https://github.com/grpc/grpc/blob/master/doc/http-grpc-status-mapping.md
       failure.asInstanceOf[StatusRuntimeException].getStatus.getCode should be(Status.Code.UNIMPLEMENTED)
+      failure.asInstanceOf[StatusRuntimeException].getStatus.getDescription should include(testUri.toString())
     }
 
     "map a strict 200 response with non-0 gRPC error code to a failed stream" in {
@@ -54,11 +57,12 @@ class PekkoHttpClientUtilsSpec extends TestKit(ActorSystem()) with AnyWordSpecLi
         Nil
       val response =
         Future.successful(HttpResponse(OK, responseHeaders, Strict(GrpcProtocolNative.contentType, ByteString.empty)))
-      val source = PekkoHttpClientUtils.responseToSource(response, null)
+      val source = PekkoHttpClientUtils.responseToSource(testUri, response, null)
 
       val failure = source.run().failed.futureValue
       failure shouldBe a[StatusRuntimeException]
       failure.asInstanceOf[StatusRuntimeException].getStatus.getCode should be(Status.Code.FAILED_PRECONDITION)
+      failure.asInstanceOf[StatusRuntimeException].getStatus.getDescription should include(testUri.toString())
       failure.asInstanceOf[StatusRuntimeException].getTrailers.get(key) should be("custom-value-in-header")
     }
 
@@ -75,14 +79,29 @@ class PekkoHttpClientUtilsSpec extends TestKit(ActorSystem()) with AnyWordSpecLi
           Map.empty[AttributeKey[?], Any].updated(AttributeKeys.trailer, responseTrailers),
           Strict(GrpcProtocolNative.contentType, ByteString.empty),
           HttpProtocols.`HTTP/1.1`))
-      val source = PekkoHttpClientUtils.responseToSource(response, null)
+      val source = PekkoHttpClientUtils.responseToSource(testUri, response, null)
 
       val failure = source.run().failed.futureValue
       failure.asInstanceOf[StatusRuntimeException].getStatus.getCode should be(Status.Code.FAILED_PRECONDITION)
+      failure.asInstanceOf[StatusRuntimeException].getStatus.getDescription should include(testUri.toString())
       failure.asInstanceOf[StatusRuntimeException].getTrailers.get(key) should be("custom-trailer-value")
       failure.asInstanceOf[StatusRuntimeException].getTrailers.get(keyBin) should be(ByteString("custom-trailer-value"))
     }
 
+    "include the request URI in StatusRuntimeException description for HTTP errors" in {
+      val uri = Uri("https://myhost.example.com/com.example.MyService/DoSomething")
+      val response =
+        Future.successful(HttpResponse(ServiceUnavailable,
+          entity = Strict(GrpcProtocolNative.contentType, ByteString.empty)))
+      val source = PekkoHttpClientUtils.responseToSource(uri, response, null)
+
+      val failure = source.run().failed.futureValue
+      failure shouldBe a[StatusRuntimeException]
+      val sre = failure.asInstanceOf[StatusRuntimeException]
+      sre.getStatus.getDescription should include("myhost.example.com")
+      sre.getStatus.getDescription should include("When calling rpc service")
+    }
+
     lazy val key = Metadata.Key.of("custom-key", Metadata.ASCII_STRING_MARSHALLER)
     lazy val keyBin = Metadata.Key.of("custom-key-bin", Metadata.BINARY_BYTE_MARSHALLER)
   }