fs2 Stream instrumentation

[iRevive]

Currently, it's hard to instrument the fs2 stream properly.

Perhaps we can follow the trace4cats route and encode the stream as WriterT[Stream[F, *], Span[F], A]:

type TracedStream[F[_]] = WriterT[Stream[F, *], Span[F], A]

implicit class StreamOps[F[_], A](private val s: Stream[F, A]) {
  def traced(name: String)(implicit A: Applicative[F], T: Tracer[F]): TracedStream[F, A] =
    Stream.bracket(Tracer[F].span(name).startUnmanaged)(_.end).flatMap { span =>
      s.evalMapChunk(a => Applicative[F].pure((span, a)))
    }
}

implicit class TracedStreamOps[F[_]](private val s: TracedStream[F, A]] {
  def evalMapTraced(name: String, attributes: Attributes): TracedStream[F, A] = ...
}

And a bunch of overloaded syntax methods.

---

But maybe there are better alternatives?

[NthPortal]

I don't think I ended up using it in any final implementation, but I was doing something like the following using SpanOps.Res#resource:

for (res <- Stream.resource(Tracer[F].span("name").resource))
yield someStream.translate(res.trace)

does this not work or keep the resource (span) open for too long or something?

[ValdemarGr]

I think tracing in Resource, Stream and F, can be unified to Resource[F, F ~> F] (which otel4s forms via SpanOps.Res's trace method) plus some combinators.

Natchez does this with typeclasses, the gist in otel4s parlance is akin to

• Tracer[F].span(...).surround{ ...: Stream[F, A] }: Stream[F, A]
• Tracer[F].span(...).surround { ...: Resource[F, A] }: Resource[F, A]

That is, something like

// SpanOps
def surround[A](fa: Stream[F, A]): Stream[F, A] = 
  Stream.resource(resource).flatMap(r => fa.translate(r.trace))
def surround[A](fa: Resource[F, A]): Resource[F, A] = 
  resource.flatMap(r => fa.mapK(r.trace))

The semantics are the expected ones. All effects in the provided resource or stream are traced, but no subsequent effects (evalMap), consistent with semantics over List, Id or otherwise.

As for usecases, websockets and grpc streams, in which library-user must return an un-compiled stream.

To complete the solution, maybe a lower-level joinOrRoot can accompany, with signature def joinOrRootUnmanaged: F ~> F, with semantics

val context = extract(carrier)
[A] =>> (fa: F[A]) => traceScope.withContext(context)(fa)

Such that

def joinOrRoot[A, C: TextMapGetter](carrier: C)(fa: Stream[F, A]): Stream[F, A] = {
  val fk = joinOrRootUnmanaged
  fa.translate(fk)
}

[iRevive]

That sounds reasonable. I think we should further explore this concept

[iRevive]

A follow-up to this topic: #1183.

[ValdemarGr]

It looks like there is no join* which allow F ~> F.
Currently we emulate it like this:

val fa = Resource.make {
   t
    .joinOrRoot(headers) {
      t.span(name).resource.allocated // <- the F ~> F here closes over the joinOrRoot context
    }
  } { case (_, close) => close }
fa.map { case (res, _) => res.trace }

Some streams are not compiled by the user (grpc streaming comes to mind).
I suppose you can do something like this:

val headers: C = ...
val stream = Stream.resource(Tracer...).flatMap(f)
stream.translate(new FunctionK[F, F] {
  def apply[A](fa: [A]): F[A] = Tracer[F].joinOrRoot(headers)(fa)
})

However, this runs the header extraction algorithm on every effect F.

[iRevive]

You can parse headers once and use childScope afterward:

val headers: C = ...
val stream = Stream.resource(Tracer...).flatMap(f)
Stream.eval(Tracer[F].joinOrRoot(headers)(Tracer[F].currentSpanOrNoop)).flatMap { span =>
  stream.translate(new FunctionK[F, F] {
    def apply[A](fa: [A]): F[A] = Tracer[F].childScope(span.context)(fa)
  })
}

But it's still quite cumbersome:(