diff options
Diffstat (limited to 'src/main/scala/xyz/driver/core/rest/HttpRestServiceTransport.scala')
-rw-r--r-- | src/main/scala/xyz/driver/core/rest/HttpRestServiceTransport.scala | 80 |
1 files changed, 48 insertions, 32 deletions
diff --git a/src/main/scala/xyz/driver/core/rest/HttpRestServiceTransport.scala b/src/main/scala/xyz/driver/core/rest/HttpRestServiceTransport.scala index c3b6bff..e60998f 100644 --- a/src/main/scala/xyz/driver/core/rest/HttpRestServiceTransport.scala +++ b/src/main/scala/xyz/driver/core/rest/HttpRestServiceTransport.scala @@ -6,9 +6,9 @@ import akka.http.scaladsl.model.headers.RawHeader import akka.http.scaladsl.unmarshalling.Unmarshal import akka.stream.Materializer import akka.stream.scaladsl.TcpIdleTimeoutException -import com.typesafe.scalalogging.Logger import org.slf4j.MDC import xyz.driver.core.Name +import xyz.driver.core.reporting.Reporter import xyz.driver.core.rest.errors.{ExternalServiceException, ExternalServiceTimeoutException} import scala.concurrent.{ExecutionContext, Future} @@ -19,7 +19,7 @@ class HttpRestServiceTransport( applicationVersion: String, actorSystem: ActorSystem, executionContext: ExecutionContext, - log: Logger) + reporter: Reporter) extends ServiceTransport { protected implicit val execution: ExecutionContext = executionContext @@ -27,42 +27,58 @@ class HttpRestServiceTransport( protected val httpClient: HttpClient = new SingleRequestHttpClient(applicationName, applicationVersion, actorSystem) def sendRequestGetResponse(context: ServiceRequestContext)(requestStub: HttpRequest): Future[HttpResponse] = { + val tags = Map( + // open tracing semantic tags + "span.kind" -> "client", + "service" -> applicationName.value, + "http.url" -> requestStub.uri.toString, + "http.method" -> requestStub.method.value, + "peer.hostname" -> requestStub.uri.authority.host.toString, + // google's tracing console provides extra search features if we define these tags + "/http/path" -> requestStub.uri.path.toString, + "/http/method" -> requestStub.method.value.toString, + "/http/url" -> requestStub.uri.toString + ) + reporter.traceAsync(s"http_call_rpc", tags) { implicit span => + val requestTime = System.currentTimeMillis() - val requestTime = System.currentTimeMillis() + val request = requestStub + .withHeaders(context.contextHeaders.toSeq.map { + case (ContextHeaders.TrackingIdHeader, _) => + RawHeader(ContextHeaders.TrackingIdHeader, context.trackingId) + case (ContextHeaders.StacktraceHeader, _) => + RawHeader( + ContextHeaders.StacktraceHeader, + Option(MDC.get("stack")) + .orElse(context.contextHeaders.get(ContextHeaders.StacktraceHeader)) + .getOrElse("")) + case (header, headerValue) => RawHeader(header, headerValue) + }: _*) - val request = requestStub - .withHeaders(context.contextHeaders.toSeq.map { - case (ContextHeaders.TrackingIdHeader, _) => - RawHeader(ContextHeaders.TrackingIdHeader, context.trackingId) - case (ContextHeaders.StacktraceHeader, _) => - RawHeader( - ContextHeaders.StacktraceHeader, - Option(MDC.get("stack")) - .orElse(context.contextHeaders.get(ContextHeaders.StacktraceHeader)) - .getOrElse("")) - case (header, headerValue) => RawHeader(header, headerValue) - }: _*) + reporter.debug(s"Sending request to ${request.method} ${request.uri}") - log.debug(s"Sending request to ${request.method} ${request.uri}") + val response = httpClient.makeRequest(request) - val response = httpClient.makeRequest(request) + response.onComplete { + case Success(r) => + val responseLatency = System.currentTimeMillis() - requestTime + reporter.debug( + s"Response from ${request.uri} to request $requestStub is successful in $responseLatency ms: $r") - response.onComplete { - case Success(r) => - val responseLatency = System.currentTimeMillis() - requestTime - log.debug(s"Response from ${request.uri} to request $requestStub is successful in $responseLatency ms: $r") + case Failure(t: Throwable) => + val responseLatency = System.currentTimeMillis() - requestTime + reporter.warn( + s"Failed to receive response from ${request.method.value} ${request.uri} in $responseLatency ms", + t) + }(executionContext) - case Failure(t: Throwable) => - val responseLatency = System.currentTimeMillis() - requestTime - log.warn(s"Failed to receive response from ${request.method} ${request.uri} in $responseLatency ms", t) - }(executionContext) - - response.recoverWith { - case _: TcpIdleTimeoutException => - val serviceCalled = s"${requestStub.method} ${requestStub.uri}" - Future.failed(ExternalServiceTimeoutException(serviceCalled)) - case t: Throwable => Future.failed(t) - } + response.recoverWith { + case _: TcpIdleTimeoutException => + val serviceCalled = s"${requestStub.method.value} ${requestStub.uri}" + Future.failed(ExternalServiceTimeoutException(serviceCalled)) + case t: Throwable => Future.failed(t) + } + }(context.spanContext) } def sendRequest(context: ServiceRequestContext)(requestStub: HttpRequest)( |