OpenTelemetry Distributed Tracing with ZIO
Tags: distributedtracing, observability, scala, tutorial
Published: 2021-11-01 16:29:00

---
title: OpenTelemetry Distributed Tracing with ZIO
published: true
date: 2021-11-01 00:00:00 UTC
tags: distributedtracing,observability,scala,tutorial
canonical_url: https://tuleism.github.io/blog/2021/opentelemetry-distributed-tracing-zio/
description: A tutorial on implementing distributed tracing for Scala applications using OpenTelemetry and various libraries from the ZIO ecosystem.
---

## Introduction

This post is some quick notes on using [ZIO](https://zio.dev/) and [zio-telemetry](https://github.com/zio/zio-telemetry) to implement [OpenTelemetry](https://opentelemetry.io/) distributed tracing for Scala applications.
The source code is available [here](https://github.com/tuleism/opentelemetry-distributed-tracing-zio).

This is not an introduction to any of these technologies, but here are a few good reads:

- NewRelic's introduction to [Distributed Tracing](https://newrelic.com/resources/ebooks/quick-introduction-distributed-tracing).
- Lightstep's concise summary for [OpenTelemetry](https://opentelemetry.lightstep.com/).

## Initial implementation
For demonstration purpose, we will perform [manual instrumentation](https://opentelemetry.io/docs/concepts/instrumenting/#manual-instrumentation) on a [modified](Hhttps://github.com/tuleism/opentelemetry-distributed-tracing-zio/commit/d9ba391dc9725e3e4ddae9c05cfe2641d8a435cf) version of the [zio-grpc](https://scalapb.github.io/zio-grpc/)'s [helloworld example](https://github.com/scalapb/zio-grpc/tree/master/examples/helloworld), in which we incorporate both [gRPC](https://grpc.io/) and HTTP communications:

- **Original**: `hello-client` sends a `HelloRequest` with `name` *x* and `hello-server` returns a `HelloResponse` with `message` *Hello, x*.
- **Modified**: in addition to the original behavior, client sends an optional integer field `guess` and server performs an HTTP request to [HTTPBin](https://httpbin.org/) based on its value.

![Initial Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/vnqh607w1uvszevnsdz4.png) 

### Add the new flag `guess`:

```protobuf
// The greeting service definition.
service Greeter {
  // Sends a greeting
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

// The request message containing the user's name.
message HelloRequest {
  string name = 1;
  google.protobuf.Int32Value guess = 2;
}

// The response message containing the greetings
message HelloReply {
  string message = 1;
}
```

### Add zio-grpc dependency

```scala
resolvers += Resolver.sonatypeRepo("snapshots")

addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.4.3")

addSbtPlugin("com.thesamet" % "sbt-protoc" % "1.0.4")

val zioGrpcVersion = "0.5.1+12-93cdbe22-SNAPSHOT"

libraryDependencies ++= Seq(
  "com.thesamet.scalapb.zio-grpc" %% "zio-grpc-codegen" % zioGrpcVersion,
  "com.thesamet.scalapb"          %% "compilerplugin"   % "0.11.5"
)
```

### Set up `build.sbt`

- Generate Scala code from `helloworld.proto`.
- Depend on [sttp](https://sttp.softwaremill.com) for HTTP client.

```scala
val grpcVersion = "1.41.0"
val sttpVersion = "3.3.15"

val scalaPBRuntime = "com.thesamet.scalapb" %% "scalapb-runtime-grpc" % scalapb.compiler.Version.scalapbVersion

val grpcRuntimeDeps = Seq(
  "io.grpc"      % "grpc-netty" % grpcVersion,
  scalaPBRuntime,
  scalaPBRuntime % "protobuf"
)

val sttpZioDeps = Seq(
  "com.softwaremill.sttp.client3" %% "async-http-client-backend-zio" % sttpVersion
)

lazy val root = Project("opentelemetry-distributed-tracing-zio", file(".")).aggregate(zio)

lazy val zio = commonProject("zio").settings(
  Compile / PB.targets := Seq(
    scalapb.gen(grpc = true)          -> (Compile / sourceManaged).value,
    scalapb.zio_grpc.ZioCodeGenerator -> (Compile / sourceManaged).value
  ),
  libraryDependencies ++= grpcRuntimeDeps ++ sttpZioDeps
)
```

### Client implementation

- Create a gRPC client pointing to localhost:9000.
- Send a single `HelloRequest`.
- Send 5 `HelloRequest`s in parallel.
- Send a single `HelloRequest` with an invalid guess.
- Print "Done" and exit.

```scala
object ZClient extends zio.App {
  private val clientLayer = GreeterClient.live(
    ZManagedChannel(
      ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext()
    )
  )

  private val singleHello = GreeterClient.sayHello(HelloRequest("World"))

  private val multipleHellos = ZIO.collectAllParN(5)(
    List(
      GreeterClient.sayHello(HelloRequest("1", Some(1))),
      GreeterClient.sayHello(HelloRequest("2", Some(2))),
      GreeterClient.sayHello(HelloRequest("3", Some(3))),
      GreeterClient.sayHello(HelloRequest("4", Some(4))),
      GreeterClient.sayHello(HelloRequest("5", Some(5)))
    )
  )

  private val invalidHello = GreeterClient.sayHello(HelloRequest("Invalid", Some(-1))).ignore

  private def myAppLogic =
    singleHello *> multipleHellos *> invalidHello *> putStrLn("Done")

  def run(args: List[String]): URIO[ZEnv, ExitCode] =
    myAppLogic.provideCustomLayer(clientLayer).exitCode
}
```

### Server implementation

- Fail the request if `guess` is less than 0.
- Based on the value of `guess`, delay for some time and then send a request to HTTPBin.

```scala
type ZGreeterEnv = Clock with Random with SttpClient
```

```scala
object ZGreeterImpl extends RGreeter[ZGreeterEnv] {

  def sayHello(request: HelloRequest): ZIO[ZGreeterEnv, Status, HelloReply] = {
    val guess = request.guess.getOrElse(0)
    for {
      _      <- ZIO.fail(Status.INVALID_ARGUMENT).when(guess < 0)
      code   <- ???
      delayMs = ???
      _      <- httpRequest(code)
                  .delay(delayMs.millis)
                  .mapError(ex => Status.INTERNAL.withCause(ex))
    } yield HelloReply(s"Hello, ${request.name}")
  }

  def httpRequest(code: Int): RIO[SttpClient, Unit] =
    send(basicRequest.get(uri"https://httpbin.org/status/$code")).unit
}
```

### Run it

- To run the server:

```bash
$ sbt "zio/runMain com.github.tuleism.ZServer"

[info] running (fork) com.github.tuleism.ZServer
[info] Server is running. Press Ctrl-C to stop.
```

- To run the client:

```bash
$ sbt "zio/runMain com.github.tuleism.ZClient"

[info] running (fork) com.github.tuleism.ZClient
[info] Done
[success] Total time: 12 s
```

At this point, we only know that it takes roughly 12 seconds for the client to initialize and finish its work.

Let's add distributed tracing to gain more insights into this.

## Common tracing requirements

- For both client and server, we need to acquire a [Tracer](), an object responsible for creating and managing [Spans]().
- Tracing data is sent to [Jaeger](https://www.jaegertracing.io/), which acts as a standalone [collector](https://opentelemetry.lightstep.com/the-collector-and-exporters/).

![Instrumented Diagram](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/oug7wvdrcos2luchiq3e.png)

### Add new dependencies

- [zio-telemetry's OpenTelemetry module](https://zio.github.io/zio-telemetry/docs/overview/overview_opentelemetry).
- We also depend on [zio-config](https://zio.github.io/zio-config/) to read tracing config from file and [zio-magic](https://github.com/kitlangton/zio-magic) to ease [ZLayer](https://zio.dev/1.x/datatypes/contextual/zlayer/) wiring.

```scala
val openTelemetryVersion = "1.6.0"
val zioConfigVersion     = "1.0.10"
val zioMagicVersion      = "0.3.9"
val zioTelemetryVersion  = "0.8.2"

val openTelemetryDeps = Seq(
  "io.opentelemetry" % "opentelemetry-exporter-jaeger"    % openTelemetryVersion,
  "io.opentelemetry" % "opentelemetry-sdk"                % openTelemetryVersion,
  "io.opentelemetry" % "opentelemetry-extension-noop-api" % s"$openTelemetryVersion-alpha"
)

val zioConfigDeps = Seq(
  "dev.zio" %% "zio-config"          % zioConfigVersion,
  "dev.zio" %% "zio-config-magnolia" % zioConfigVersion,
  "dev.zio" %% "zio-config-typesafe" % zioConfigVersion
)

val zioMagicDeps = Seq(
  "io.github.kitlangton" %% "zio-magic" % zioMagicVersion
)

val zioTelemetryDeps = Seq(
  "dev.zio"                       %% "zio-opentelemetry"                   % zioTelemetryVersion,
  "com.softwaremill.sttp.client3" %% "zio-telemetry-opentelemetry-backend" % sttpVersion
)
```

### Add a config layer

```
tracing {
  enable = false
  enable = ${?TRACING_ENABLE}
  endpoint = "http://127.0.0.1:14250"
  endpoint = ${?JAEGER_ENDPOINT}
}
```

```scala
case class AppConfig(tracing: TracingConfig)

case class TracingConfig(enable: Boolean, endpoint: String)

object AppConfig {
  private val configDescriptor = descriptor[AppConfig]

  val live: Layer[ReadError[String], Has[AppConfig]] = TypesafeConfig.fromDefaultLoader(configDescriptor)
}
```

### Add a `Tracer` layer

- Depend on the configuration, we either create a *noop* `Tracer` or one that sends data to Jaeger.
- Once we have it, we can construct a `Tracing` layer, which give us access to many [useful operations in zio-telemetry](https://www.javadoc.io/static/dev.zio/zio-opentelemetry_2.13/0.8.2/zio/telemetry/opentelemetry/Tracing$.html).

```scala
object ZTracer {
  private val InstrumentationName = "com.github.tuleism"

  private def managed(serviceName: String, endpoint: String) = {
    val resource = Resource.builder().put(ResourceAttributes.SERVICE_NAME, serviceName).build()
    for {
      spanExporter   <- ZManaged.fromAutoCloseable(
                          Task(JaegerGrpcSpanExporter.builder().setEndpoint(endpoint).build())
                        )
      spanProcessor  <- ZManaged.fromAutoCloseable(UIO(SimpleSpanProcessor.create(spanExporter)))
      tracerProvider <- UIO(
                          SdkTracerProvider.builder().addSpanProcessor(spanProcessor).setResource(resource).build()
                        ).toManaged_
      openTelemetry  <- UIO(OpenTelemetrySdk.builder().setTracerProvider(tracerProvider).build()).toManaged_
      tracer         <- UIO(openTelemetry.getTracer(InstrumentationName)).toManaged_
    } yield tracer
  }

  def live(serviceName: String): RLayer[Has[TracingConfig], Has[Tracer]] =
    (
      for {
        config <- ZIO.service[TracingConfig].toManaged_
        tracer <- if (!config.enable) {
                    Task(NoopOpenTelemetry.getInstance().getTracer(InstrumentationName)).toManaged_
                  } else {
                    managed(serviceName, config.endpoint)
                  }
      } yield tracer
    ).toLayer
}
```

## New server

### Instrument the HTTP client

- Use out-of-the-box [sttp backend](https://sttp.softwaremill.com/en/latest/backends/wrappers/zio-opentelemetry.html).
- We also add additional HTTP specific attributes according to the OpenTelemetry's [semantic convention](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/http.md#name).

```scala
object SttpTracing {
  private val wrapper = new ZioTelemetryOpenTelemetryTracer {
    def before[T](request: Request[T, Nothing]): RIO[Tracing, Unit] =
      Tracing.setAttribute(SemanticAttributes.HTTP_METHOD.getKey, request.method.method) *>
        Tracing.setAttribute(SemanticAttributes.HTTP_URL.getKey, request.uri.toString()) *>
        ZIO.unit

    def after[T](response: Response[T]): RIO[Tracing, Unit] =
      Tracing.setAttribute(SemanticAttributes.HTTP_STATUS_CODE.getKey, response.code.code) *>
        ZIO.unit
  }

  val live = AsyncHttpClientZioBackend.layer().flatMap { hasBackend =>
    ZIO
      .service[Tracing.Service]
      .map { tracing =>
        ZioTelemetryOpenTelemetryBackend(hasBackend.get, tracing, wrapper)
      }
      .toLayer
  }
}
```

### Instrument the gRPC server

We can add `Tracing` without changing our server implementation with a [ZTransform](https://scalapb.github.io/zio-grpc/docs/decorating/). For each request:

- We use [zio-telemetry's spanFrom](https://www.javadoc.io/static/dev.zio/zio-opentelemetry_2.13/0.8.2/zio/telemetry/opentelemetry/TracingSyntax$$OpenTelemetryZioOps.html#spanFrom[C](propagator:io.opentelemetry.context.propagation.TextMapPropagator,carrier:C,getter:io.opentelemetry.context.propagation.TextMapGetter[C],spanName:String,spanKind:io.opentelemetry.api.trace.SpanKind,toErrorStatus:PartialFunction[E,io.opentelemetry.api.trace.StatusCode]):zio.ZIO[Rwithzio.telemetry.opentelemetry.Tracing,E,A]), which extracts the propagated context (through [gRPC Metadata](https://grpc.io/docs/what-is-grpc/core-concepts/#metadata), using [W3C Trace Context format](https://w3c.github.io/trace-context/)) and starts a new child `Span` right after.
- We have access to a [RequestContext](https://scalapb.github.io/zio-grpc/docs/context) and thus the full method name used for `Span`'s name.
- We also add additional gRPC specific attributes according to the OpenTelemetry's [semantic convention](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/rpc.md#grpc-attributes).

```scala
object GrpcTracing {
  private val propagator: TextMapPropagator = W3CTraceContextPropagator.getInstance()

  private val metadataGetter: TextMapGetter[Metadata] = new TextMapGetter[Metadata] {
    override def keys(carrier: Metadata): java.lang.Iterable[String] =
      carrier.keys()

    override def get(carrier: Metadata, key: String): String =
      carrier.get(
        Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER)
      )
  }

  private def withSemanticAttributes[R, A](effect: ZIO[R, Status, A]): ZIO[Tracing with R, Status, A] =
    Tracing.setAttribute(SemanticAttributes.RPC_SYSTEM.getKey, "grpc") *>
      effect
        .tapBoth(
          status =>
            Tracing.setAttribute(
              SemanticAttributes.RPC_GRPC_STATUS_CODE.getKey,
              status.getCode.value()
            ),
          _ =>
            Tracing.setAttribute(SemanticAttributes.RPC_GRPC_STATUS_CODE.getKey, Status.OK.getCode.value())
        )

  def serverTracingTransform[R]: ZTransform[R, Status, R with Tracing with Has[RequestContext]] =
    new ZTransform[R, Status, R with Tracing with Has[RequestContext]] {

      def effect[A](io: ZIO[R, Status, A]): ZIO[R with Tracing with Has[RequestContext], Status, A] =
        for {
          rc       <- ZIO.service[RequestContext]
          metadata <- rc.metadata.wrap(identity)
          result   <- withSemanticAttributes(io)
                        .spanFrom(
                          propagator,
                          metadata,
                          metadataGetter,
                          rc.methodDescriptor.getFullMethodName,
                          SpanKind.SERVER,
                          { case _ => StatusCode.ERROR }
                        )
        } yield result

      def stream[A](io: ZStream[R, Status, A]): ZStream[R with Tracing with Has[RequestContext], Status, A] =
        ???
    }
}
```

### Update Server Main

- Add required layers for Tracing.
- Transform the original `ZGreeterImpl`.

```scala
import zio.magic._

object ZServer extends ServerMain {
  private val requirements =
    ZLayer
      .wire[ZEnv with ZGreeterEnv](
        ZEnv.live,
        AppConfig.live.narrow(_.tracing),
        ZTracer.live("hello-server"),
        Tracing.live,
        SttpTracing.live
      )
      .orDie

  def services: ServiceList[Any] =
    ServiceList
      .add(ZGreeterImpl.transform[ZGreeterEnv, Has[RequestContext]](GrpcTracing.serverTracingTransform))
      .provideLayer(requirements)
}
```

## New client

### Inject current context into [gRPC Metadata](https://grpc.io/docs/what-is-grpc/core-concepts/#metadata) for [context propagation](https://opentelemetry.lightstep.com/core-concepts/context-propagation/)

```scala
object GrpcTracing {

  ...

  private val metadataSetter: TextMapSetter[Metadata] = (carrier, key, value) =>
    carrier.put(Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER), value)
    
  val contextPropagationClientInterceptor: ZClientInterceptor[Tracing] = ZClientInterceptor.headersUpdater {
    (_, _, metadata) =>
      metadata.wrapM(Tracing.inject(propagator, _, metadataSetter))
  }
  
  ...
  
}
```

```scala
object ZClient extends zio.App {
  private val clientLayer = GreeterClient.live(
    ZManagedChannel(
      ManagedChannelBuilder.forAddress("localhost", 9000).usePlaintext(),
      Seq(GrpcTracing.contextPropagationClientInterceptor)
    )
  )
  
  ...
}
```

### Start a `Span` for each request

- Use `ZTransform` to record the relevant gRPC attributes.

```scala
object GrpcTracing {

  ...

  def clientTracingTransform[R]: ZTransform[R, Status, R with Tracing] =
    new ZTransform[R, Status, R with Tracing] {
      def effect[A](io: ZIO[R, Status, A]): ZIO[R with Tracing, Status, A] = withSemanticAttributes(io)

      def stream[A](io: ZStream[R, Status, A]): ZStream[R with Tracing, Status, A] = ???
    }
}
```

- Unlike the server, we don't have access to a `RequestContext` object, so we have to set the method name manually.
- We also start additional `Span`s.

```scala
object ZClient extends zio.App {
  
  ...
  
  
  private def errorToStatusCode[E]: PartialFunction[E, StatusCode] = { case _ => StatusCode.ERROR }

  private def sayHello(request: HelloRequest) =
    GreeterClient
      .sayHello(request)
      .span(
        GreeterGrpc.METHOD_SAY_HELLO.getFullMethodName,
        SpanKind.CLIENT,
        errorToStatusCode
      )
      
  private val singleHello = sayHello(HelloRequest("World"))
    .span("singleHello", toErrorStatus = errorToStatusCode)

  private val multipleHellos = ZIO
    .collectAllParN(5)(
      List(
        sayHello(HelloRequest("1", Some(1))),
        sayHello(HelloRequest("2", Some(2))),
        sayHello(HelloRequest("3", Some(3))),
        sayHello(HelloRequest("4", Some(4))),
        sayHello(HelloRequest("5", Some(5)))
      )
    )
    .span("multipleHellos", toErrorStatus = errorToStatusCode)

  private val invalidHello = sayHello(HelloRequest("Invalid", Some(-1))).ignore
    .span("invalidHello", toErrorStatus = errorToStatusCode)
}
```

### Add required layers

```scala
object ZClient extends zio.App {

  ...
  
  private val requirements = ZLayer
    .wire[ZEnv with Tracing](
      ZEnv.live,
      AppConfig.live.narrow(_.tracing),
      ZTracer.live("hello-client"),
      Tracing.live
    ) >+> clientLayer

  def run(args: List[String]): URIO[ZEnv, ExitCode] =
    myAppLogic.provideCustomLayer(requirements).exitCode
}
```

## Showtime

### Run Jaeger through Docker

- Tracing data can be sent to port 14250.
- We can view [Jaeger UI](https://github.com/jaegertracing/jaeger-ui) at http://localhost:16686.

```bash
$ docker run --rm --name jaeger \
  -p 16686:16686 \
  -p 14250:14250 \
  jaegertracing/all-in-one:1.25
```

### Start the server

```bash
$ TRACING_ENABLE=true sbt "zio/runMain com.github.tuleism.ZServer"

[info] running (fork) com.github.tuleism.ZServer
[info] Server is running. Press Ctrl-C to stop.
```

### Start the client

```bash
$ TRACING_ENABLE=true sbt "zio/runMain com.github.tuleism.ZClient"

[info] running (fork) com.github.tuleism.ZClient
[info] Done
[success] Total time: 12 s
```

### Distributed Tracing in action!

- Now we can see the details for `multipleHellos`:

![multipleHellos Span](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/j54n9vrfngkjb6idaddh.png) 

- And which `guess` is causing the longest delay.

![bad guess](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ojdo73cv5o525j3inn2b.png) 

## Integration with Logging

- Let's add tracing context into log messages following the [specification](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/logs/overview.md#trace-context-in-legacy-formats).
- We're going to use [izumi logstage](https://izumi.7mind.io/logstage/index.html), our favorite logging library.
- See the [diffs](https://github.com/tuleism/opentelemetry-distributed-tracing-zio/commit/d9ba391dc9725e3e4ddae9c05cfe2641d8a435cf).

### Add logging dependency

```scala
val izumiVersion         = "1.0.8"

val loggingDeps = Seq(
  "io.7mind.izumi" %% "logstage-core"          % izumiVersion,
  "io.7mind.izumi" %% "logstage-adapter-slf4j" % izumiVersion
)
```

### Setup logging

- Add `trace_id`, `span_id` to logging context if current trace context is valid.

```scala
object Logging {
  private def baseLogger = IzLogger()

  val live: ZLayer[Has[Tracing.Service], Nothing, Has[LogZIO.Service]] =
    (
      for {
        tracing <- ZIO.service[Tracing.Service]
      } yield LogZIO.withDynamicContext(baseLogger)(
        Tracing.getCurrentSpanContext
          .map(spanContext =>
            if (spanContext.isValid)
              CustomContext(
                "trace_id"    -> spanContext.getTraceId,
                "span_id"     -> spanContext.getSpanId,
                "trace_flags" -> spanContext.getTraceFlags.asHex()
              )
            else
              CustomContext.empty
          )
          .provide(Has(tracing))
      )
    ).toLayer
}
```

### Add a few log messages

- E.g for `singleHello`.

```scala
object ZClient extends zio.App {

  ...

  private val singleHello = (
    for {
      _ <- log.info("singleHello")
      _ <- sayHello(HelloRequest("World"))
    } yield ()
  ).span("singleHello", toErrorStatus = errorToStatusCode)
  
}
```

### Sample Logs

```bash
[info] running (fork) com.github.tuleism.ZClient
[info] I 2021-11-01T22:59:10.881 (ZClient.scala:37)  …tuleism.ZClient.singleHello [24:zio-default-async-11] trace_id=9c8a7ebb87381293bc8937a5f7673cb9, span_id=cb7c9a440472e1be, trace_flags=01 singleHello
[info] I 2021-11-01T22:59:14.064 (ZClient.scala:44)  …eism.ZClient.multipleHellos [21:zio-default-async-8 ] trace_id=fe405246fbaa5f876c19f14fa649a99f, span_id=bef19494bef4106e, trace_flags=01 multipleHellos
[info] I 2021-11-01T22:59:18.171 (ZClient.scala:60)  …uleism.ZClient.invalidHello [26:zio-default-async-13] trace_id=be5ccd425e0cfb01fd97274abd0c4d72, span_id=ea6499fb9a7c8d28, trace_flags=01 invalidHello
[info] I 2021-11-01T22:59:18.272 (ZClient.scala:66)  ….tuleism.ZClient.myAppLogic [15:zio-default-async-2 ] Done
[success] Total time: 12 s
```

## Extra notes
- If we receive an HTTP 5xx response, we should set the `Span` status to error according to the [semantic convention](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/http.md#status). However, it is currently [not possible](https://github.com/zio/zio-telemetry/issues/444) with `zio-telemetry`.
- We need [a better way](https://github.com/scalapb/zio-grpc/issues/309) to implement tracing for `zio-grpc` client.