Context Propagation for Finagle (Netty) isn't working correctly

[dispalt]

I am not entirely sure how to fix this, so I thought I'd get some help.

I have a play service that calls Finagle MySQL instance, after that particular call most of the tracing information is lost. How do I know? I added log messages to see and I can see the trace information is lost.

Here's a simplified set of log messages:

test 1 is before the finagle call.
test 2 is right after
I generate a stack trace to show you the callstack at that point
test 3 is after converting it to an IO
the last message is logged in play after the request is done (which has the retained context).

2021-04-24 09:28:17 INFO  [staging-gc-akka.actor.default-dispatcher-65] [a470203480f3f4f4] c.g.c.MainController - Test 1
2021-04-24 09:28:17 INFO  [finagle/netty4-4-2] [undefined] c.g.c.MainController - Test 2
java.lang.Exception: Stack trace
	at java.lang.Thread.dumpStack(Thread.java:1336)
	at com.goodcover.controllers.MainController.$anonfun$ping$5(MainController.scala:251)
	at com.twitter.util.Return.map(Try.scala:308)
	at com.twitter.util.Future.$anonfun$map$1(Future.scala:1902)
	at com.twitter.util.Promise$TryTransformer.liftedTree2$1(Promise.scala:257)
	at com.twitter.util.Promise$TryTransformer.k(Promise.scala:257)
	at com.twitter.util.Promise$Transformer.apply(Promise.scala:215)
	at com.twitter.util.Promise$WaitQueue.com$twitter$util$Promise$WaitQueue$$run(Promise.scala:91)
	at com.twitter.util.Promise$WaitQueue$$anon$1.run(Promise.scala:86)
	at com.twitter.concurrent.LocalScheduler$Activation.run(Scheduler.scala:167)
	at com.twitter.concurrent.LocalScheduler$Activation.submit(Scheduler.scala:126)
	at com.twitter.concurrent.LocalScheduler.submit(Scheduler.scala:243)
	at com.twitter.concurrent.Scheduler$.submit(Scheduler.scala:78)
	at com.twitter.util.Promise$WaitQueue.runInScheduler(Promise.scala:86)
	at com.twitter.util.Promise.updateIfEmpty(Promise.scala:783)
	at com.twitter.util.Promise.update(Promise.scala:755)
	at com.twitter.util.Promise.setValue(Promise.scala:731)
	at com.twitter.concurrent.AsyncQueue.offer(AsyncQueue.scala:123)
	at com.twitter.finagle.netty4.transport.ChannelTransport$$anon$2.channelRead(ChannelTransport.scala:169)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at com.twitter.finagle.netty4.decoder.DecoderHandler.channelRead(DecoderHandler.scala:18)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at com.twitter.finagle.netty4.codec.BufCodec$.channelRead(BufCodec.scala:68)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
	at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
	at io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:795)
	at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:480)
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at com.twitter.finagle.util.BlockingTimeTrackingThreadFactory$$anon$1.run(BlockingTimeTrackingThreadFactory.scala:23)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.lang.Thread.run(Thread.java:748)
2021-04-24 09:28:17 INFO  [finagle/netty4-4-2] [undefined] c.g.c.MainController - Test 3
ResultSet(17, 36)
2021-04-24 09:28:17 INFO  [staging-gc-akka.actor.default-dispatcher-65] [a470203480f3f4f4] b.OperationNameFilter - [200] GET - 0:0:0:0:0:0:0:1/ping [3.50ms]

What I've tried to do to fix this:

• Add

kanela.modules {
  executor-service {
    within += "io.netty..*"

• Enable the executor-service-capture-on-submit and disable executor-service
• Enable both.
• Write my own advice which did not work.
• Made a special version of twitter future -> IO which kinda works

      val ctx = Kamon.currentContext()
      IO(Kamon.storeContext(ctx)).flatMap { scope =>
        IO(tf)
          .flatMap { f =>
            f.poll match {
              case Some(Return(value)) => IO.pure(value)
              case Some(Throw(err))    => IO.raiseError[A](err)
              case None                =>
                IO.async[A] { cb =>
                  f.respond {
                    case Return(value) => cb(Right(value))
                    case Throw(value)  => cb(Left(value))
                  }
                }
            }
          }
          .guarantee(IO(scope.close()))
      }

Would love some help on figuring this out!

[dispalt]

@ivantopo I made a small reproduction repo that shows it. It doesn't spin up a mysql instance, so you'll need one handy, but if you run the app and supply a database, user, pass and dest (localhost:3306) it will print the context keys which aren't carried from one pool to the next.

https://github.com/dispalt/finagle-mysql-kamon/blob/master/src/main/scala/App.scala

[dispalt]

It prints something like this

0 [main] [Context{Entries{span=Span.Empty},Tags{the key=the string}}]
1 [finagle/netty4-1-1] [Context{Entries{span=Span.Empty},Tags{}}]
2 [main] [Context{Entries{span=Span.Empty},Tags{the key=the string}}]

  private var i = 0

  def printCtx() = {
    println(s"$i [${Thread.currentThread().getName}] [${Kamon.currentContext()}]")
    i += 1
  }
...
    def prg = for {
      _ <-  Future(Kamon.storeContext(Context.of("the key", "the string")))
      _ <- Future(printCtx())
      _ <- client.select("select * from ids limit 10")(identity)
      _ <- Future(printCtx())

    } yield ()
    Await.result(prg)
    printCtx()