Add an HTTP4S client integration (#52)

* Add an HTTP4s client integration

* Add newline

* Update docs

Author: tomverran

build.sbt

@@ -71,12 +71,14 @@ lazy val natchezSlf4j = project
 
 lazy val natchezHttp4s = project
   .in(file("natchez-http4s"))
+  .dependsOn(natchezTestkit)
   .enablePlugins(GitVersioning)
   .settings(common :+ (name := "natchez-http4s"))
   .settings(
     libraryDependencies ++= Seq(
-      "org.tpolecat" %% "natchez-core" % natchezVersion,
-      "org.http4s"   %% "http4s-dsl"   % http4sVersion
+      "org.tpolecat" %% "natchez-core"  % natchezVersion,
+      "org.http4s"   %% "http4s-client" % http4sVersion,
+      "org.http4s"   %% "http4s-dsl"    % http4sVersion
     )
   )
 

docs/docs/docs/natchez-http4s.md

@@ -31,7 +31,8 @@ import cats.effect.{ExitCode, IO, IOApp, Resource, Sync, Timer}
 import cats.syntax.flatMap._
 import cats.syntax.functor._
 import com.ovoenergy.effect.natchez.Datadog
-import com.ovoenergy.effect.natchez.http4s.server.{Configuration, TraceMiddleware}
+import com.ovoenergy.effect.natchez.http4s.Configuration
+import com.ovoenergy.effect.natchez.http4s.server.TraceMiddleware
 import natchez.{EntryPoint, Span, Trace}
 import org.http4s.{HttpApp, HttpRoutes}
 import org.http4s.client.blaze.BlazeClientBuilder
@@ -116,8 +117,8 @@ it is set up to create tags suitable for Datadog but you can use the helper func
 
 ```scala mdoc
 import cats.effect.IO
-import com.ovoenergy.effect.natchez.http4s.server.Configuration
-import com.ovoenergy.effect.natchez.http4s.server.Configuration.TagReader._
+import com.ovoenergy.effect.natchez.http4s.Configuration
+import com.ovoenergy.effect.natchez.http4s.Configuration.TagReader._
 import natchez.TraceValue.BooleanValue
 import cats.syntax.semigroup._
 

natchez-http4s/src/main/scala/com/ovoenergy/effect/natchez/http4s/server/Configuration.scala renamed to natchez-http4s/src/main/scala/com/ovoenergy/effect/natchez/http4s/Configuration.scala

@@ -1,4 +1,4 @@
-package com.ovoenergy.effect.natchez.http4s.server
+package com.ovoenergy.effect.natchez.http4s
 
 import cats.Applicative
 import cats.data.Kleisli
@@ -9,7 +9,7 @@ import cats.kernel.{Monoid, Semigroup}
 import cats.syntax.foldable._
 import cats.syntax.functor._
 import cats.syntax.monoid._
-import com.ovoenergy.effect.natchez.http4s.server.Configuration.TagReader._
+import com.ovoenergy.effect.natchez.http4s.Configuration.TagReader.{MessageReader, RequestReader, ResponseReader}
 import natchez.TraceValue
 import natchez.TraceValue.StringValue
 import org.http4s.util.{CaseInsensitiveString, StringWriter}

natchez-http4s/src/main/scala/com/ovoenergy/effect/natchez/http4s/client/TracedClient.scala

@@ -0,0 +1,44 @@
+package com.ovoenergy.effect.natchez.http4s.client
+
+import cats.data.Kleisli
+import cats.effect.{Resource, Sync}
+import cats.syntax.functor._
+import cats.~>
+import com.ovoenergy.effect.natchez.http4s.Configuration
+import com.ovoenergy.effect.natchez.http4s.server.TraceMiddleware.removeNumericPathSegments
+import natchez.{Span, Trace}
+import org.http4s._
+import org.http4s.client.Client
+
+trait TracedClient[F[_]] {
+  def named(s: String): Client[F]
+}
+
+object TracedClient {
+
+  type Traced[F[_], A] = Kleisli[F, Span[F], A]
+
+  private def dropTracing[F[_]](span: Span[F]): Traced[F, *] ~> F =
+    Kleisli.applyK[F, Span[F]](span)
+
+  private def trace[F[_]]: F ~> Traced[F, *] =
+    Kleisli.liftK
+
+  def apply[F[_]: Sync](client: Client[F], config: Configuration[F]): TracedClient[Traced[F, *]] =
+    name => Client[Traced[F, *]] { req =>
+      Resource(
+        Trace[Traced[F, *]].span(s"$name:http.request:${removeNumericPathSegments(req.uri)}") {
+          for {
+            span        <- Kleisli.ask[F, Span[F]]
+            headers     <- trace(span.kernel.map(_.toHeaders.map { case (k, v) => Header(k, v) }))
+            withHeader  = req.putHeaders(headers.toSeq: _*).mapK(dropTracing(span))
+            reqTags     <- trace(config.request.value.run(req.mapK(dropTracing(span))))
+            _           <- trace(span.put(reqTags.toSeq:_*))
+            (resp, rel) <- client.run(withHeader).mapK(trace[F]).map(_.mapK(trace)).allocated
+            respTags    <- trace(config.response.value.run(resp.mapK(dropTracing(span))))
+            _           <- trace(span.put(respTags.toSeq:_*))
+          } yield resp -> rel
+        }
+      )
+    }
+}

natchez-http4s/src/main/scala/com/ovoenergy/effect/natchez/http4s/server/TraceMiddleware.scala

@@ -7,6 +7,7 @@ import natchez._
 import org.http4s._
 import cats.syntax.flatMap._
 import cats.syntax.functor._
+import com.ovoenergy.effect.natchez.http4s.Configuration
 
 object TraceMiddleware {
 

natchez-http4s/src/test/scala/com/ovoenergy/effect/natchez/http4s/client/TestClient.scala

@@ -0,0 +1,26 @@
+package com.ovoenergy.effect.natchez.http4s.client
+
+import cats.data.Kleisli
+import cats.effect.Sync
+import cats.effect.concurrent.Ref
+import org.http4s.{Request, Response}
+import org.http4s.client.Client
+import cats.syntax.functor._
+
+trait TestClient[F[_]] {
+  def requests: F[List[Request[F]]]
+  def client: Client[F]
+}
+
+object TestClient {
+
+  def apply[F[_]: Sync]: F[TestClient[F]] =
+    Ref.of[F, List[Request[F]]](List.empty).map { ref =>
+      new TestClient[F] {
+        def client: Client[F] =
+          Client.fromHttpApp[F](Kleisli(r => ref.update(_ :+ r).as(Response[F]())))
+        def requests: F[List[Request[F]]] =
+          ref.get
+      }
+    }
+}

natchez-http4s/src/test/scala/com/ovoenergy/effect/natchez/http4s/client/TracedClientTest.scala

@@ -0,0 +1,56 @@
+package com.ovoenergy.effect.natchez.http4s.client
+
+import cats.data.Kleisli
+import cats.effect.{IO, Timer}
+import com.ovoenergy.effect.natchez.TestEntryPoint
+import com.ovoenergy.effect.natchez.TestEntryPoint.TestSpan
+import com.ovoenergy.effect.natchez.http4s.Configuration
+import natchez.{Kernel, Span}
+import org.http4s.Request
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import scala.concurrent.ExecutionContext.global
+
+class TracedClientTest extends AnyWordSpec with Matchers {
+
+  implicit val timer: Timer[IO] = IO.timer(global)
+  val unit: Kleisli[IO, Span[IO], Unit] = Kleisli.pure(())
+  val config: Configuration[IO] = Configuration.default[IO]()
+  type TraceIO[A] = Kleisli[IO, Span[IO], A]
+
+  "TracedClient" should {
+
+    "Add the kernel to requests" in {
+
+      val requests: List[Request[IO]] = (
+        for {
+          client <- TestClient[IO]
+          ep    <- TestEntryPoint[IO]
+          http   = TracedClient(client.client, config)
+          kernel = Kernel(Map("X-Trace-Token" -> "token"))
+          _      <- ep.continue("bar", kernel).use(http.named("foo").status(Request[TraceIO]()).run)
+          reqs   <- client.requests
+        } yield reqs
+      ).unsafeRunSync
+
+      requests.forall(_.headers.exists(_.name.value === "X-Trace-Token")) shouldBe true
+    }
+
+    "Create a new span for HTTP requests" in {
+
+      val spans: List[TestSpan] = (
+        for {
+          client <- TestClient[IO]
+          ep    <- TestEntryPoint[IO]
+          http   = TracedClient(client.client, config)
+          _      <- ep.root("root").use(http.named("foo").status(Request[TraceIO]()).run)
+          reqs   <- ep.spans
+        } yield reqs
+      ).unsafeRunSync
+
+      spans.length shouldBe 2
+      spans.head.name shouldBe "foo:http.request:/"
+    }
+  }
+}

natchez-http4s/src/test/scala/com/ovoenergy/effect/natchez/http4s/server/TraceMiddlewareTest.scala

@@ -4,6 +4,7 @@ import cats.data.Kleisli
 import cats.effect.concurrent.Ref
 import cats.effect.{IO, Resource}
 import cats.{Applicative, Monad}
+import com.ovoenergy.effect.natchez.http4s.Configuration
 import fs2._
 import natchez.TraceValue.{NumberValue, StringValue}
 import natchez.{EntryPoint, Kernel, Span, TraceValue}

natchez-testkit/src/main/scala/com/ovoenergy/effect/natchez/TestEntryPoint.scala

@@ -24,33 +24,35 @@ object TestEntryPoint {
     exitCase: ExitCase[Throwable],
     parent: Option[String],
     completed: Instant,
+    kernel: Kernel,
     name: String
   )
 
   def apply[F[_]: Clock](implicit F: Sync[F]): F[TestEntryPoint[F]] =
     Ref.of[F, List[TestSpan]](List.empty).map { submitted =>
-      def span(myName: String): Span[F] = new Span[F] {
-        def span(name: String): Resource[F, Span[F]] = makeSpan(name, Some(myName))
+
+      def span(myName: String, k: Kernel): Span[F] = new Span[F] {
+        def span(name: String): Resource[F, Span[F]] = makeSpan(name, Some(myName), k)
         def put(fields: (String, TraceValue)*): F[Unit] = F.unit
-        def kernel: F[Kernel] = F.pure(Kernel(Map.empty))
+        def kernel: F[Kernel] = F.pure(k)
       }
 
-      def makeSpan(name: String, parent: Option[String]): Resource[F, Span[F]] =
-        Resource.makeCase(F.delay(span(name))) { (_, ec) =>
+      def makeSpan(name: String, parent: Option[String], kernel: Kernel): Resource[F, Span[F]] =
+        Resource.makeCase(F.delay(span(name, kernel))) { (_, ec) =>
           Clock[F]
             .realTime(TimeUnit.MILLISECONDS)
             .map(Instant.ofEpochMilli)
             .flatMap { time =>
-              val span = TestSpan(ec, parent, time, name)
+              val span = TestSpan(ec, parent, time, kernel, name)
               submitted.update(_ :+ span)
             }
         }
 
       new TestEntryPoint[F] {
         def spans: F[List[TestSpan]] = submitted.get
-        def root(name: String): Resource[F, Span[F]] = makeSpan(name, None)
-        def continue(name: String, k: Kernel): Resource[F, Span[F]] = makeSpan(name, None)
-        def continueOrElseRoot(name: String, k: Kernel): Resource[F, Span[F]] = makeSpan(name, None)
+        def root(name: String): Resource[F, Span[F]] = makeSpan(name, None, Kernel(Map.empty))
+        def continue(name: String, k: Kernel): Resource[F, Span[F]] = makeSpan(name, None, k)
+        def continueOrElseRoot(name: String, k: Kernel): Resource[F, Span[F]] = makeSpan(name, None, k)
       }
     }
 }