Add tracing to client HTTP transport and improve tracing tags

4 files changed, 69 insertions, 38 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)(
diff --git a/src/main/scala/xyz/driver/core/rest/headers/Traceparent.scala b/src/main/scala/xyz/driver/core/rest/headers/Traceparent.scala
index 9d470ad..866476d 100644
--- a/src/main/scala/xyz/driver/core/rest/headers/Traceparent.scala
+++ b/src/main/scala/xyz/driver/core/rest/headers/Traceparent.scala
@@ -3,20 +3,21 @@ package rest
 package headers
 
 import akka.http.scaladsl.model.headers.{ModeledCustomHeader, ModeledCustomHeaderCompanion}
+import xyz.driver.core.reporting.SpanContext
 
 import scala.util.Try
 
-/** Encapsulates trace context in an HTTP header for propagation across services.
+/** Encapsulates a trace context in an HTTP header for propagation across services.
   *
   * This implementation corresponds to the W3C editor's draft specification (as of 2018-08-28)
   * https://w3c.github.io/distributed-tracing/report-trace-context.html. The 'flags' field is
   * ignored.
   */
-final case class Traceparent(traceId: String, spanId: String) extends ModeledCustomHeader[Traceparent] {
+final case class Traceparent(spanContext: SpanContext) extends ModeledCustomHeader[Traceparent] {
   override def renderInRequests            = true
   override def renderInResponses           = true
   override val companion: Traceparent.type = Traceparent
-  override def value: String               = f"01-$traceId-$spanId-00"
+  override def value: String               = f"01-${spanContext.traceId}-${spanContext.spanId}-00"
 }
 object Traceparent extends ModeledCustomHeaderCompanion[Traceparent] {
   override val name = "traceparent"
@@ -26,8 +27,7 @@ object Traceparent extends ModeledCustomHeaderCompanion[Traceparent] {
       version == "01",
       s"Found unsupported version '$version' in traceparent header. Only version '01' is supported.")
     new Traceparent(
-      traceId,
-      spanId
+      new SpanContext(traceId, spanId)
     )
   }
 }
diff --git a/src/main/scala/xyz/driver/core/rest/package.scala b/src/main/scala/xyz/driver/core/rest/package.scala
index 104261a..3be8f02 100644
--- a/src/main/scala/xyz/driver/core/rest/package.scala
+++ b/src/main/scala/xyz/driver/core/rest/package.scala
@@ -15,6 +15,7 @@ import scalaz.Scalaz.{intInstance, stringInstance}
 import scalaz.syntax.equal._
 import scalaz.{Functor, OptionT}
 import xyz.driver.core.rest.auth.AuthProvider
+import xyz.driver.core.rest.headers.Traceparent
 import xyz.driver.tracing.TracingDirectives
 
 import scala.concurrent.Future
@@ -199,7 +200,8 @@ object `package` {
         h.name === ContextHeaders.AuthenticationTokenHeader || h.name === ContextHeaders.TrackingIdHeader ||
         h.name === ContextHeaders.PermissionsTokenHeader || h.name === ContextHeaders.StacktraceHeader ||
         h.name === ContextHeaders.TraceHeaderName || h.name === ContextHeaders.SpanHeaderName ||
-        h.name === ContextHeaders.OriginatingIpHeader || h.name === ContextHeaders.ClientFingerprintHeader
+        h.name === ContextHeaders.OriginatingIpHeader || h.name === ContextHeaders.ClientFingerprintHeader ||
+        h.name === Traceparent.name
       }
       .map { header =>
         if (header.name === ContextHeaders.AuthenticationTokenHeader) {
diff --git a/src/main/scala/xyz/driver/core/rest/serviceRequestContext.scala b/src/main/scala/xyz/driver/core/rest/serviceRequestContext.scala
index 76f5a0d..d2e4bc3 100644
--- a/src/main/scala/xyz/driver/core/rest/serviceRequestContext.scala
+++ b/src/main/scala/xyz/driver/core/rest/serviceRequestContext.scala
@@ -6,7 +6,11 @@ import xyz.driver.core.auth.{AuthToken, PermissionsToken, User}
 import xyz.driver.core.generators
 import scalaz.Scalaz.{mapEqual, stringInstance}
 import scalaz.syntax.equal._
+import xyz.driver.core.reporting.SpanContext
 import xyz.driver.core.rest.auth.AuthProvider
+import xyz.driver.core.rest.headers.Traceparent
+
+import scala.util.Try
 
 class ServiceRequestContext(
     val trackingId: String = generators.nextUuid().toString,
@@ -45,6 +49,15 @@ class ServiceRequestContext(
     case _ => false
   }
 
+  def spanContext: SpanContext = {
+    val validHeader = Try {
+      contextHeaders(Traceparent.name)
+    }.flatMap { value =>
+      Traceparent.parse(value)
+    }
+    validHeader.map(_.spanContext).getOrElse(SpanContext.fresh())
+  }
+
   override def toString: String = s"ServiceRequestContext($trackingId, $contextHeaders)"
 }