This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch feat/service-info-in-status-exception in repository https://gitbox.apache.org/repos/asf/pekko-grpc.git
commit 32ef6af25c6ec12475499efd8b81559ee288a04c Author: 虎鸣 <[email protected]> AuthorDate: Fri Jul 3 04:04:22 2026 +0800 feat: include called service URI in StatusRuntimeException descriptions Motivation: When a gRPC call fails, the StatusRuntimeException description does not indicate which service/method was being called, making debugging harder in environments with multiple services. Modification: Add requestUri parameter to responseToSource, parseResponseStatus, and mapToStatusException. Use augmentDescription to append the request URI to all error descriptions. Update existing tests and add a dedicated test for URI inclusion. Result: StatusRuntimeException messages now include the full request URI (e.g. "When calling rpc service: https://host/Service/Method";), improving debuggability of gRPC failures. Tests: sbt "runtime / Test / testOnly org.apache.pekko.grpc.internal.PekkoHttpClientUtilsSpec" — 4/4 passed References: Refs akka/akka-grpc#1748 --- .../pekko/grpc/internal/PekkoHttpClientUtils.scala | 32 +++++++++++++++------- .../grpc/internal/PekkoHttpClientUtilsSpec.scala | 24 ++++++++++++++-- 2 files changed, 43 insertions(+), 13 deletions(-) diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala b/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala index 508ba6e3..a7ba5977 100644 --- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala +++ b/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,7 +194,7 @@ 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]](() => { @@ -202,7 +202,7 @@ object PekkoHttpClientUtils { { 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) } } diff --git a/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala b/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala index b67a59ed..6e37e76b 100644 --- a/runtime/src/test/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtilsSpec.scala +++ b/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,28 @@ 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) } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
