ovotech / natchez-extras Goto Github PK

Hello!

This line should probably be changed to https://github.com/ovotech/natchez-extras/blob/master/natchez-extras-fs2/src/test/scala/com/ovoenergy/natchez/extras/fs2/AllocatedSpanTest.scala.

Great work on the project!

Howdy folks! We're evaluating using this library, but one thing that stood out to me was usage of 1.0.0-M* in http4s integration.

Is there a particular feature 1.0 brings that is required for the integration?

Problem with 1.0.0-M* lineage in http4s is that milestones are not binary compatible (and the date for final release of 1.x is not really set), unlike the stable 0.23.* lineage which is using CE3

Given a comment on the first line could we have our span name be or have selectLatestFooBar in it?

-- Name: selectLatestFooBar
select 
foo,
bar
from baz;

Currently we do this for metrics and would be nice for traces as well.

We parse the sql like so:

  private val re = """--\s*Name:\s*(\w+)""".r

  def extractQueryNameOrSql(sql: String): String = {
    re.findFirstMatchIn(sql).flatMap(m => Option(m.group(1))).getOrElse(sql)
  }

If it happened to not have a comment in it then you could default back to your current way you display the sql (com.ovoenergy.natchez.extras.doobie.TracedTransactor$#formatQuery)

We are currently on 6.25 of natchez-extras-doobie but would likely upgrade to newer version if we need newer version of natchez.

I noticed that the Stream that checks for queued spans runs every 0.5s, until the first span is created (and I guess being enqueued). Maybe I'm using this endpoint wrong, but I reproduced this issue in this mini project:

https://github.com/joshm1/http4s-playground/blob/master/src/main/scala/example3/Server.scala

It never called the flatMap function in submitOnce:

  private def submitOnce[F[_]: Sync](client: Client[F], queue: Queue[F, SubmittableSpan]): F[Unit] =
    queue // breakpoint is hit every 0.5s, until I curl my endpoint to create the first span(s)
      .tryDequeueChunk1(maxSize = 1000)
      .flatMap { items => // this function is never invoked
        items.traverse { traces => ... }
      }
      .as(())

Issue was noticed by putting a breakpoint in submitOnce and seeing that it stopped getting hit after my first curl localhost:8888/foo . Also did some "println debugging" by modifying this line

.repeatEval(Sync[F].delay(println("tick")) >> sem.acquire >> submitOnce(http, queue) >> sem.release)

And I saw "tick" in the console every 0.5s, then it stopped after I ran curl. It never called the release function in the resource.

I've read all of the source code in the natchez-http4s module, including the test, but I'm struggling with how to use this module in my existing application.

Any help is greatly appreciated! Thanks!

Edit:

I removed the original code in this post and created the following example project to better show what I'd like to do and the errors I'm having:

https://github.com/joshm1/http4s-playground/blob/master/src/main/scala/Server.scala

It would be useful to have the ability to send custom events to Datadog.
https://docs.datadoghq.com/developers/events/dogstatsd/?tab=java

I've noticed that a Datadog module was recently added directly to Natchez (here) and I was wondering if it's worth chatting about whether it can be integrated into effect-utils.

Can perhaps the internal natchez-datadog module be entirely replaced and the extra functionality be contributed back to natchez or would there be any reason for maintaining a custom datadog module?

Datadog supports sending one of a few types to tell it what kind of app a particular service is, for example is it a database or is it an API. This doesn't affect very much but changes a few labels in the UI.

I incorrectly thought the span type was a tag but in fact it is a field you need to set in CompletedSpan as per the docs: https://docs.datadoghq.com/api/?lang=python#tracing

The tricky part of this is deciding how to get this type information through Natchez into Datadog given that Natchez only allows you to identify spans by a single string (currently we encode a name/service/resource into this string by using colons to separate them, do we need to add an optional fourth type?)

Once upon a time this was a repository for loosely related utilities based around cats-effect, hence the name effect-utils - recently though we've moved to more well supported third party utilities (like log4cats) leaving this repository as solely concerned with Datadog

I think therefore we should come up with a better name for this repository / set of libraries so we can promote them more effectively both inside & outside OVO

Would be good to get automatic dependency updates.
https://github.com/scala-steward-org/repos/blob/master/repos-github.md

We have some legacy http4s apps that are in concrete IO, is there any easy way to lift IO into TracedIO?

Thanks!

example

case class ApplicationRoutes(services: ApplicationServices) {
    private val authMiddleware = AuthMiddleware(services.bar)
  
  private val clientClientRoutes: HttpRoutes[IO] = ClientClientRoutes(services.baz).routes(authMiddleware)

  private val adminClientRoutes: HttpRoutes[IO] = AdminClientRoutes(services.foo).routes(authMiddleware)

  private val newRoutes: HttpRoutes[IO] = adminClientRoutes <+> clientClientRoutes

  val httpApp: HttpApp[IO] = appRoutes.orNotFound
}

object AppMain extends IOApp {
  override def run(): IO[ExitCode] = {
    Resources.make(config).use {
      val httpApp = ApplicationRoutes(ApplicationServices()).appRoutes

      // Would be great to wrap `httpApp` here with `TraceMiddleware`

      BlazeServerBuilder[IO]
        .bindHttp(httpPort, httpHost)
        .withHttpApp(httpApp)
        .withServiceErrorHandler(errorHandler)
        .resource
        .run
    }
  }
}

I would be great if TraceMiddleware could work with Http instead of HttpApp, so it could be used with HttpRoutes.

Self explanatory! A microsite might be nice

I attempted to PR this but had some issues (type inference on getTypeMap which is strange, something about Class[_] vs Class[_ <: Object]) which I don't feel comfortable in submitting the PR.
Could this be project be upgraded to RC4?
The Transactor now needs a LoggingHandler which is effectful.
Here's the code I came up with:

package com.ovoenergy.natchez.extras.doobie

import cats.data.Kleisli
import cats.effect.Async
import cats.implicits._
import com.ovoenergy.natchez.extras.core.Config
import com.ovoenergy.natchez.extras.core.Config.ServiceAndResource
import doobie.WeakAsync
import doobie.free.KleisliInterpreter
import doobie.util.log
import doobie.util.log.LogHandler
import doobie.util.transactor.Transactor
import natchez.{Span, Trace}

import java.sql.{Connection, PreparedStatement, ResultSet}

object TracedTransactor {
  private val DefaultResourceName = "db.execute"

  type Traced[F[_], A] = Kleisli[F, Span[F], A]
  def apply[F[_]: Async: LogHandler](
    service: String,
    transactor: Transactor[F]
  ): Transactor[Traced[F, *]] = {
    val kleisliTransactor = transactor
      .mapK(Kleisli.liftK[F, Span[F]])(implicitly, Async.asyncForKleisli(implicitly))

    implicit val lh: LogHandler[Traced[F, *]] = new LogHandler[Traced[F, *]] {
      override def run(
        logEvent: log.LogEvent
      ): Traced[F, Unit] = Kleisli.liftF(implicitly[LogHandler[F]].run(logEvent))
    }

    trace(ServiceAndResource(s"$service-db", DefaultResourceName), kleisliTransactor)
  }

  private def formatQuery(q: String): String =
    q.replace("\n", " ").replaceAll("\\s+", " ").trim()

  def trace[F[_]: Trace: Async: LogHandler](
    config: Config,
    transactor: Transactor[F]
  ): Transactor[F] =
    transactor
      .copy(
        interpret0 = createInterpreter(config)(WeakAsync.doobieWeakAsyncForAsync[F]).ConnectionInterpreter
      )

  private def createInterpreter[F[_]: Trace: LogHandler](
    config: Config
  )(W: WeakAsync[F]): KleisliInterpreter[F] = {
    new KleisliInterpreter[F](implicitly[LogHandler[F]])(W) {
      override lazy val PreparedStatementInterpreter: PreparedStatementInterpreter =
        new PreparedStatementInterpreter {

          type TracedOp[A] = Kleisli[F, PreparedStatement, A] //PreparedStatement => F[A]

          def runTraced[A](f: TracedOp[A]): TracedOp[A] =
            Kleisli {
              case TracedStatement(p, sql) =>
                Trace[F].span(config.fullyQualifiedSpanName(formatQuery(sql)))(
                  Trace[F].put("span.type" -> "db") >> f(p)
                )
              case a =>
                f(a)
            }

          override val executeBatch: TracedOp[Array[Int]] =
            runTraced(super.executeBatch)

          override val executeLargeBatch: TracedOp[Array[Long]] =
            runTraced(super.executeLargeBatch)

          override val execute: TracedOp[Boolean] =
            runTraced(super.execute)

          override val executeUpdate: TracedOp[Int] =
            runTraced(super.executeUpdate)

          override val executeQuery: TracedOp[ResultSet] =
            runTraced(super.executeQuery)
        }

      override lazy val ConnectionInterpreter: ConnectionInterpreter =
        new ConnectionInterpreter {
          override def prepareStatement(a: String): Kleisli[F, Connection, PreparedStatement] =
            super.prepareStatement(a).map(TracedStatement(_, a): PreparedStatement)
          override def getTypeMap = primitive(_.getTypeMap)
        }
    }
  }
}

The submitter uses a Fiber to run a background task which means we have to use a Semaphore to ensure it isn't stopped between removing data from the queue & submitting it when the app shuts down.

My theory is we could operate at a slightly higher level and use FS2 for this but it needs some careful consideration to ensure we don't introduce any new race conditions. I have a branch I can produce upon request.

We noticed the services we migrated to this stopped tracing properly with services written in other languages using builtin datadog tracing. The issue is the headers in the Kernel aren't named according to the B3 Propagation standard:
https://github.com/openzipkin/b3-propagation

Looks like datadog supports a bunch of headers for reference:
https://github.com/DataDog/dd-trace-java/blob/614976ed2cbcef3b528b1a0eca3f06b8eb1dab54/dd-java-agent/instrumentation/opentelemetry/src/main/java/datadog/trace/instrumentation/opentelemetry/OtelContextPropagators.java#L67-L78

The 7.0.0-M2 artifacts depend on natchez-extras-core, but it was not published because of publish / skip := true. It would be nice to have it on Maven Central too.

Currently the automated tests are running in @ovotech's internal CircleCI environment, this means it is not suitable to run unit tests automatically on external branches and PRs, which in turn makes it convoluted to accept external contributions.

Would there be interest in a sttp module for natchez?