diff --git a/README.md b/README.md index bf9b2b94..60e70d99 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,12 @@ And this one if you would like to have Json support: libraryDependencies += "dev.profunktor" %% "fs2-rabbit-circe" % Version ``` +OpenTelemetry tracing with otel4s is available as a separate module: + +```scala +libraryDependencies += "dev.profunktor" %% "fs2-rabbit-otel4s" % Version +``` + ## Usage Guide Check the [official guide](https://fs2-rabbit.profunktor.dev/guide.html) for updated compiling examples. diff --git a/build.sbt b/build.sbt index 2eaaec29..fb0750a4 100644 --- a/build.sbt +++ b/build.sbt @@ -1,5 +1,7 @@ import Dependencies.* import microsites.ExtraMdFileConfig +import sbtbuildinfo.BuildInfoPlugin +import sbtbuildinfo.BuildInfoPlugin.autoImport.* import scala.collection.immutable ThisBuild / name := "fs2-rabbit" @@ -84,6 +86,14 @@ def ExamplesDependencies(scalaVersionStr: String): List[ModuleID] = def TestKitDependencies(scalaVersionStr: String): List[ModuleID] = List(Libraries.scalaCheck) +def Otel4sDependencies(scalaVersionStr: String): List[ModuleID] = + List( + Libraries.otel4sCoreTrace, + Libraries.otel4sSemconv, + Libraries.otel4sSemconvExp % Test, + Libraries.otel4sTestkit % Test + ) + def TestsDependencies(scalaVersionStr: String): List[ModuleID] = List( Libraries.disciplineScalaCheck % Test, @@ -102,7 +112,7 @@ lazy val noPublish = List( lazy val `fs2-rabbit-root`: Project = project .in(file(".")) .disablePlugins(MimaPlugin) - .aggregate(`fs2-rabbit`, `fs2-rabbit-circe`, tests, examples, microsite, `fs2-rabbit-testkit`) + .aggregate(`fs2-rabbit`, `fs2-rabbit-circe`, `fs2-rabbit-otel4s`, tests, examples, microsite, `fs2-rabbit-testkit`) .settings(noPublish) lazy val `fs2-rabbit`: Project = project @@ -120,6 +130,19 @@ lazy val `fs2-rabbit-circe`: Project = project .enablePlugins(AutomateHeaderPlugin) .dependsOn(`fs2-rabbit`) +lazy val `fs2-rabbit-otel4s`: Project = project + .in(file("otel4s")) + .settings(commonSettings: _*) + .settings(libraryDependencies ++= Otel4sDependencies(scalaVersion.value)) + .settings( + buildInfoPackage := "dev.profunktor.fs2rabbit.otel4s", + buildInfoOptions += BuildInfoOption.PackagePrivate, + buildInfoKeys := Seq[BuildInfoKey]("version" -> version.value), + Test / parallelExecution := false + ) + .enablePlugins(AutomateHeaderPlugin, BuildInfoPlugin) + .dependsOn(`fs2-rabbit`) + lazy val tests: Project = project .in(file("tests")) .settings(commonSettings: _*) @@ -185,7 +208,7 @@ lazy val microsite: Project = project "-Xlint:-missing-interpolator,_" ) ) - .dependsOn(`fs2-rabbit`, `fs2-rabbit-circe`, `examples`) + .dependsOn(`fs2-rabbit`, `fs2-rabbit-circe`, `fs2-rabbit-otel4s`, `examples`) // CI build addCommandAlias("buildFs2Rabbit", ";clean;+test;mdoc") diff --git a/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/ProcessSpanContext.scala b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/ProcessSpanContext.scala new file mode 100644 index 00000000..95771d88 --- /dev/null +++ b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/ProcessSpanContext.scala @@ -0,0 +1,30 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import dev.profunktor.fs2rabbit.model.{DeliveryTag, ExchangeName, QueueName, RoutingKey} + +final case class ProcessSpanContext( + queueName: QueueName, + exchangeName: ExchangeName, + routingKey: RoutingKey, + destinationName: String, + deliveryTag: DeliveryTag, + messageId: Option[String], + conversationId: Option[String], + redelivered: Boolean +) diff --git a/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/PublishSpanContext.scala b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/PublishSpanContext.scala new file mode 100644 index 00000000..25c57327 --- /dev/null +++ b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/PublishSpanContext.scala @@ -0,0 +1,27 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import dev.profunktor.fs2rabbit.model.{ExchangeName, RoutingKey} + +final case class PublishSpanContext( + exchangeName: ExchangeName, + routingKey: RoutingKey, + destinationName: String, + messageId: Option[String], + conversationId: Option[String] +) diff --git a/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/RabbitTracer.scala b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/RabbitTracer.scala new file mode 100644 index 00000000..5e233048 --- /dev/null +++ b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/RabbitTracer.scala @@ -0,0 +1,159 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import cats.effect.{Concurrent, Resource} +import cats.syntax.functor.* +import cats.syntax.semigroup.* +import dev.profunktor.fs2rabbit.interpreter.RabbitClient +import org.typelevel.otel4s.semconv.attributes.{ErrorAttributes, ServerAttributes} +import org.typelevel.otel4s.trace.{SpanFinalizer, StatusCode, Tracer, TracerProvider} +import org.typelevel.otel4s.{Attribute, Attributes} + +trait RabbitTracer[F[_]] { + def client(client: RabbitClient[F]): TracedRabbitClient[F] +} + +object RabbitTracer { + + sealed trait Config { + private[otel4s] def tracerName: String + private[otel4s] def constAttributes: Attributes + private[otel4s] def clientId: Option[String] + private[otel4s] def publishSpanSetup: PublishSpanContext => Config.SpanSetup + private[otel4s] def processSpanSetup: ProcessSpanContext => Config.SpanSetup + + def withConstAttributes(attributes: Attributes): Config + def addConstAttributes(head: Attribute[?], tail: Attribute[?]*): Config + def withClientId(clientId: String): Config + def withPublishSpanSetup(f: PublishSpanContext => Config.SpanSetup): Config + def withProcessSpanSetup(f: ProcessSpanContext => Config.SpanSetup): Config + def withServerAddress(serverAddress: String, serverPort: Option[Int]): Config + } + + object Config { + + object Defaults { + val tracerName: String = "fs2.rabbit" + + val publishSpanSetup: PublishSpanContext => SpanSetup = + context => SpanSetup(s"publish ${context.destinationName}") + + val processSpanSetup: ProcessSpanContext => SpanSetup = + context => SpanSetup(s"process ${context.destinationName}") + + val spanFinalizationStrategy: SpanFinalizer.Strategy = { + case Resource.ExitCase.Errored(error) => + val errorType = Option(error.getClass.getCanonicalName).getOrElse(error.getClass.getName) + val setStatus = Option(error.getMessage) + .map(message => SpanFinalizer.setStatus(StatusCode.Error, message)) + .getOrElse(SpanFinalizer.setStatus(StatusCode.Error)) + + SpanFinalizer.recordException(error) |+| + SpanFinalizer.addAttribute(ErrorAttributes.ErrorType(errorType)) |+| + setStatus + + case Resource.ExitCase.Canceled => + SpanFinalizer.addAttribute(ErrorAttributes.ErrorType("canceled")) |+| + SpanFinalizer.setStatus(StatusCode.Error, "canceled") + } + } + + sealed trait SpanSetup { + def spanName: String + def attributes: Attributes + def finalizationStrategy: SpanFinalizer.Strategy + } + + object SpanSetup { + def apply( + spanName: String, + attributes: Attributes, + finalizationStrategy: SpanFinalizer.Strategy + ): SpanSetup = + SpanSetupImpl(spanName, attributes, finalizationStrategy) + + private[RabbitTracer] def apply(spanName: String): SpanSetup = + SpanSetup(spanName, Attributes.empty, Defaults.spanFinalizationStrategy) + + final private case class SpanSetupImpl( + spanName: String, + attributes: Attributes, + finalizationStrategy: SpanFinalizer.Strategy + ) extends SpanSetup + } + + val default: Config = + ConfigImpl( + tracerName = Defaults.tracerName, + constAttributes = Attributes.empty, + clientId = None, + publishSpanSetup = Defaults.publishSpanSetup, + processSpanSetup = Defaults.processSpanSetup + ) + + final private case class ConfigImpl( + tracerName: String, + constAttributes: Attributes, + clientId: Option[String], + publishSpanSetup: PublishSpanContext => SpanSetup, + processSpanSetup: ProcessSpanContext => SpanSetup + ) extends Config { + override def withConstAttributes(attributes: Attributes): Config = copy(constAttributes = attributes) + + override def addConstAttributes(head: Attribute[?], tail: Attribute[?]*): Config = + copy(constAttributes = constAttributes + head ++ tail) + + override def withClientId(clientId: String): Config = copy(clientId = Some(clientId)) + + override def withPublishSpanSetup(f: PublishSpanContext => SpanSetup): Config = copy(publishSpanSetup = f) + + override def withProcessSpanSetup(f: ProcessSpanContext => SpanSetup): Config = copy(processSpanSetup = f) + + override def withServerAddress(serverAddress: String, serverPort: Option[Int]): Config = + copy( + constAttributes = constAttributes + + ServerAttributes.ServerAddress(serverAddress) ++ + ServerAttributes.ServerPort.maybe(serverPort.map(_.toLong)) + ) + } + } + + def apply[F[_]](implicit rabbitTracer: RabbitTracer[F]): RabbitTracer[F] = rabbitTracer + + def noop[F[_]: Concurrent]: RabbitTracer[F] = new Noop[F] + + def create[F[_]: Concurrent: TracerProvider](config: Config): F[RabbitTracer[F]] = + TracerProvider[F] + .tracer(config.tracerName) + .withVersion(BuildInfo.version) + .get + .map(implicit tracer => new Impl[F](config)) + + def resource[F[_]: Concurrent: TracerProvider](config: Config): Resource[F, RabbitTracer[F]] = + Resource.eval(create(config)) + + final private class Impl[F[_]: Concurrent: Tracer](config: Config) extends RabbitTracer[F] { + override def client(client: RabbitClient[F]): TracedRabbitClient[F] = + new TracedRabbitClient.Impl[F](client, config) + } + + final private class Noop[F[_]: Concurrent] extends RabbitTracer[F] { + override def client(client: RabbitClient[F]): TracedRabbitClient[F] = + TracedRabbitClient.noop(client) + } +} diff --git a/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/TracedRabbitClient.scala b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/TracedRabbitClient.scala new file mode 100644 index 00000000..3d9187b7 --- /dev/null +++ b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/TracedRabbitClient.scala @@ -0,0 +1,305 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import cats.effect.Concurrent +import cats.syntax.all.* +import dev.profunktor.fs2rabbit.effects.{EnvelopeDecoder, MessageEncoder} +import dev.profunktor.fs2rabbit.interpreter.RabbitClient +import dev.profunktor.fs2rabbit.model.* +import dev.profunktor.fs2rabbit.otel4s.instances.* +import dev.profunktor.fs2rabbit.otel4s.internal.Semconv +import org.typelevel.otel4s.trace.{SpanKind, Tracer} + +trait TracedRabbitClient[F[_]] { + def underlying: RabbitClient[F] + + def createPublisher[A](exchangeName: ExchangeName, routingKey: RoutingKey)(implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[A => F[Unit]] + + def createPublisherWithListener[A]( + exchangeName: ExchangeName, + routingKey: RoutingKey, + flag: PublishingFlag, + listener: PublishReturn => F[Unit] + )(implicit channel: AMQPChannel, encoder: MessageEncoder[F, A]): F[A => F[Unit]] + + def createRoutingPublisher[A](exchangeName: ExchangeName)(implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[RoutingKey => A => F[Unit]] + + def createRoutingPublisherWithListener[A]( + exchangeName: ExchangeName, + flag: PublishingFlag, + listener: PublishReturn => F[Unit] + )(implicit channel: AMQPChannel, encoder: MessageEncoder[F, A]): F[RoutingKey => A => F[Unit]] + + def createBasicPublisher[A](implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[(ExchangeName, RoutingKey, A) => F[Unit]] + + def createBasicPublisherWithListener[A](flag: PublishingFlag, listener: PublishReturn => F[Unit])(implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[(ExchangeName, RoutingKey, A) => F[Unit]] + + def createAckerConsumer[A]( + queueName: QueueName, + basicQos: BasicQos = BasicQos(prefetchSize = 0, prefetchCount = 1), + consumerArgs: Option[ConsumerArgs] = None, + ackMultiple: AckMultiple = AckMultiple(false) + )(implicit + channel: AMQPChannel, + decoder: EnvelopeDecoder[F, A] + ): F[(AckResult => F[Unit], TracedRabbitConsumer[F, A])] + + def createAckerConsumerWithMultipleFlag[A]( + queueName: QueueName, + basicQos: BasicQos = BasicQos(prefetchSize = 0, prefetchCount = 1), + consumerArgs: Option[ConsumerArgs] = None + )(implicit + channel: AMQPChannel, + decoder: EnvelopeDecoder[F, A] + ): F[((AckResult, AckMultiple) => F[Unit], TracedRabbitConsumer[F, A])] + + def createAutoAckConsumer[A]( + queueName: QueueName, + basicQos: BasicQos = BasicQos(prefetchSize = 0, prefetchCount = 1), + consumerArgs: Option[ConsumerArgs] = None + )(implicit channel: AMQPChannel, decoder: EnvelopeDecoder[F, A]): F[TracedRabbitConsumer[F, A]] +} + +object TracedRabbitClient { + + def noop[F[_]: Concurrent](underlying: RabbitClient[F]): TracedRabbitClient[F] = + new Noop[F](underlying) + + final private[otel4s] class Impl[F[_]: Concurrent: Tracer]( + override val underlying: RabbitClient[F], + config: RabbitTracer.Config + ) extends TracedRabbitClient[F] { + + override def createPublisher[A](exchangeName: ExchangeName, routingKey: RoutingKey)(implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[A => F[Unit]] = + Concurrent[F].pure(message => publish(channel, exchangeName, routingKey, None, message)) + + override def createPublisherWithListener[A]( + exchangeName: ExchangeName, + routingKey: RoutingKey, + flag: PublishingFlag, + listener: PublishReturn => F[Unit] + )(implicit channel: AMQPChannel, encoder: MessageEncoder[F, A]): F[A => F[Unit]] = + underlying + .addPublishingListener(listener) + .as(message => publish(channel, exchangeName, routingKey, Some(flag), message)) + + override def createRoutingPublisher[A](exchangeName: ExchangeName)(implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[RoutingKey => A => F[Unit]] = + Concurrent[F].pure(routingKey => message => publish(channel, exchangeName, routingKey, None, message)) + + override def createRoutingPublisherWithListener[A]( + exchangeName: ExchangeName, + flag: PublishingFlag, + listener: PublishReturn => F[Unit] + )(implicit channel: AMQPChannel, encoder: MessageEncoder[F, A]): F[RoutingKey => A => F[Unit]] = + underlying.addPublishingListener(listener).as { routingKey => message => + publish(channel, exchangeName, routingKey, Some(flag), message) + } + + override def createBasicPublisher[A](implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[(ExchangeName, RoutingKey, A) => F[Unit]] = + Concurrent[F].pure((exchangeName, routingKey, message) => + publish(channel, exchangeName, routingKey, None, message) + ) + + override def createBasicPublisherWithListener[A](flag: PublishingFlag, listener: PublishReturn => F[Unit])(implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[(ExchangeName, RoutingKey, A) => F[Unit]] = + underlying.addPublishingListener(listener).as { (exchangeName, routingKey, message) => + publish(channel, exchangeName, routingKey, Some(flag), message) + } + + override def createAckerConsumer[A]( + queueName: QueueName, + basicQos: BasicQos, + consumerArgs: Option[ConsumerArgs], + ackMultiple: AckMultiple + )(implicit + channel: AMQPChannel, + decoder: EnvelopeDecoder[F, A] + ): F[(AckResult => F[Unit], TracedRabbitConsumer[F, A])] = + underlying + .createAckerConsumer(queueName, basicQos, consumerArgs, ackMultiple) + .map { case (acker, stream) => (acker, TracedRabbitConsumer(queueName, stream, config)) } + + override def createAckerConsumerWithMultipleFlag[A]( + queueName: QueueName, + basicQos: BasicQos, + consumerArgs: Option[ConsumerArgs] + )(implicit + channel: AMQPChannel, + decoder: EnvelopeDecoder[F, A] + ): F[((AckResult, AckMultiple) => F[Unit], TracedRabbitConsumer[F, A])] = + underlying + .createAckerConsumerWithMultipleFlag(queueName, basicQos, consumerArgs) + .map { case (acker, stream) => (acker, TracedRabbitConsumer(queueName, stream, config)) } + + override def createAutoAckConsumer[A]( + queueName: QueueName, + basicQos: BasicQos, + consumerArgs: Option[ConsumerArgs] + )(implicit channel: AMQPChannel, decoder: EnvelopeDecoder[F, A]): F[TracedRabbitConsumer[F, A]] = + underlying + .createAutoAckConsumer(queueName, basicQos, consumerArgs) + .map(stream => TracedRabbitConsumer(queueName, stream, config)) + + private def publish[A]( + channel: AMQPChannel, + exchangeName: ExchangeName, + routingKey: RoutingKey, + flag: Option[PublishingFlag], + value: A + )(implicit encoder: MessageEncoder[F, A]): F[Unit] = + encoder.run(value).flatMap { message => + val spanContext = Semconv.publishSpanContext(exchangeName, routingKey, message) + val spanSetup = config.publishSpanSetup(spanContext) + + Tracer[F] + .joinOrRoot(message.properties.headers)(Tracer[F].currentSpanContext) + .flatMap { creationContext => + val spanKind: SpanKind = creationContext.fold[SpanKind](SpanKind.Producer)(_ => SpanKind.Client) + + val builder = Tracer[F] + .spanBuilder(spanSetup.spanName) + .withSpanKind(spanKind) + .withFinalizationStrategy(spanSetup.finalizationStrategy) + .addAttributes( + Semconv.publishAttributes(spanContext, config.clientId) ++ + config.constAttributes ++ + spanSetup.attributes + ) + + creationContext + .fold(builder)(context => builder.addLink(context, Semconv.publishLinkAttributes(spanContext))) + .build + .surround { + val publishedMessage = creationContext.fold( + Tracer[F] + .propagate(message.properties.headers) + .map(headers => message.copy(properties = message.properties.copy(headers = headers))) + )(_ => Concurrent[F].pure(message)) + + publishedMessage.flatMap { tracedMessage => + flag.fold( + underlying.publishingProgram.basicPublish(channel, exchangeName, routingKey, tracedMessage) + )(publishingFlag => + underlying.publishingProgram.basicPublishWithFlag( + channel, + exchangeName, + routingKey, + publishingFlag, + tracedMessage + ) + ) + } + } + } + } + } + + final private class Noop[F[_]: Concurrent](override val underlying: RabbitClient[F]) extends TracedRabbitClient[F] { + override def createPublisher[A](exchangeName: ExchangeName, routingKey: RoutingKey)(implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[A => F[Unit]] = underlying.createPublisher(exchangeName, routingKey) + + override def createPublisherWithListener[A]( + exchangeName: ExchangeName, + routingKey: RoutingKey, + flag: PublishingFlag, + listener: PublishReturn => F[Unit] + )(implicit channel: AMQPChannel, encoder: MessageEncoder[F, A]): F[A => F[Unit]] = + underlying.createPublisherWithListener(exchangeName, routingKey, flag, listener) + + override def createRoutingPublisher[A](exchangeName: ExchangeName)(implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[RoutingKey => A => F[Unit]] = underlying.createRoutingPublisher(exchangeName) + + override def createRoutingPublisherWithListener[A]( + exchangeName: ExchangeName, + flag: PublishingFlag, + listener: PublishReturn => F[Unit] + )(implicit channel: AMQPChannel, encoder: MessageEncoder[F, A]): F[RoutingKey => A => F[Unit]] = + underlying.createRoutingPublisherWithListener(exchangeName, flag, listener) + + override def createBasicPublisher[A](implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[(ExchangeName, RoutingKey, A) => F[Unit]] = underlying.createBasicPublisher + + override def createBasicPublisherWithListener[A](flag: PublishingFlag, listener: PublishReturn => F[Unit])(implicit + channel: AMQPChannel, + encoder: MessageEncoder[F, A] + ): F[(ExchangeName, RoutingKey, A) => F[Unit]] = underlying.createBasicPublisherWithListener(flag, listener) + + override def createAckerConsumer[A]( + queueName: QueueName, + basicQos: BasicQos, + consumerArgs: Option[ConsumerArgs], + ackMultiple: AckMultiple + )(implicit + channel: AMQPChannel, + decoder: EnvelopeDecoder[F, A] + ): F[(AckResult => F[Unit], TracedRabbitConsumer[F, A])] = + underlying + .createAckerConsumer(queueName, basicQos, consumerArgs, ackMultiple) + .map { case (acker, stream) => (acker, TracedRabbitConsumer.noop(queueName, stream)) } + + override def createAckerConsumerWithMultipleFlag[A]( + queueName: QueueName, + basicQos: BasicQos, + consumerArgs: Option[ConsumerArgs] + )(implicit + channel: AMQPChannel, + decoder: EnvelopeDecoder[F, A] + ): F[((AckResult, AckMultiple) => F[Unit], TracedRabbitConsumer[F, A])] = + underlying + .createAckerConsumerWithMultipleFlag(queueName, basicQos, consumerArgs) + .map { case (acker, stream) => (acker, TracedRabbitConsumer.noop(queueName, stream)) } + + override def createAutoAckConsumer[A]( + queueName: QueueName, + basicQos: BasicQos, + consumerArgs: Option[ConsumerArgs] + )(implicit channel: AMQPChannel, decoder: EnvelopeDecoder[F, A]): F[TracedRabbitConsumer[F, A]] = + underlying + .createAutoAckConsumer(queueName, basicQos, consumerArgs) + .map(stream => TracedRabbitConsumer.noop(queueName, stream)) + } +} diff --git a/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/TracedRabbitConsumer.scala b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/TracedRabbitConsumer.scala new file mode 100644 index 00000000..b927492f --- /dev/null +++ b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/TracedRabbitConsumer.scala @@ -0,0 +1,92 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import cats.effect.Concurrent +import cats.syntax.flatMap.* +import dev.profunktor.fs2rabbit.model.{AmqpEnvelope, QueueName} +import dev.profunktor.fs2rabbit.otel4s.instances.* +import dev.profunktor.fs2rabbit.otel4s.internal.Semconv +import fs2.Stream +import org.typelevel.otel4s.trace.{SpanKind, Tracer} + +trait TracedRabbitConsumer[F[_], A] { + def queueName: QueueName + def underlying: Stream[F, AmqpEnvelope[A]] + + final def records: Stream[F, AmqpEnvelope[A]] = underlying + + def process[B](envelope: AmqpEnvelope[A])(fa: F[B]): F[B] + + def recordsWithProcess[B](f: AmqpEnvelope[A] => F[B]): Stream[F, B] = + underlying.evalMap(envelope => process(envelope)(f(envelope))) +} + +object TracedRabbitConsumer { + + private[otel4s] def apply[F[_]: Concurrent: Tracer, A]( + queueName: QueueName, + underlying: Stream[F, AmqpEnvelope[A]], + config: RabbitTracer.Config + ): TracedRabbitConsumer[F, A] = + new Impl[F, A](queueName, underlying, config) + + private[otel4s] def noop[F[_], A]( + queueName: QueueName, + underlying: Stream[F, AmqpEnvelope[A]] + ): TracedRabbitConsumer[F, A] = + new Noop[F, A](queueName, underlying) + + final private class Impl[F[_]: Concurrent: Tracer, A]( + override val queueName: QueueName, + override val underlying: Stream[F, AmqpEnvelope[A]], + config: RabbitTracer.Config + ) extends TracedRabbitConsumer[F, A] { + + override def process[B](envelope: AmqpEnvelope[A])(fa: F[B]): F[B] = { + val spanContext = Semconv.processSpanContext(queueName, envelope) + val spanSetup = config.processSpanSetup(spanContext) + + Tracer[F] + .joinOrRoot(envelope.properties.headers)(Tracer[F].currentSpanContext) + .flatMap { creationContext => + val builder = Tracer[F] + .spanBuilder(spanSetup.spanName) + .root + .withSpanKind(SpanKind.Consumer) + .withFinalizationStrategy(spanSetup.finalizationStrategy) + .addAttributes( + Semconv.processAttributes(spanContext, config.clientId) ++ + config.constAttributes ++ + spanSetup.attributes + ) + + creationContext + .fold(builder)(context => builder.addLink(context, Semconv.processLinkAttributes(spanContext))) + .build + .surround(fa) + } + } + } + + final private class Noop[F[_], A]( + override val queueName: QueueName, + override val underlying: Stream[F, AmqpEnvelope[A]] + ) extends TracedRabbitConsumer[F, A] { + override def process[B](envelope: AmqpEnvelope[A])(fa: F[B]): F[B] = fa + } +} diff --git a/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/instances.scala b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/instances.scala new file mode 100644 index 00000000..b7c9dba8 --- /dev/null +++ b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/instances.scala @@ -0,0 +1,41 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import dev.profunktor.fs2rabbit.model.AmqpFieldValue.StringVal +import dev.profunktor.fs2rabbit.model.Headers +import org.typelevel.otel4s.context.propagation.{TextMapGetter, TextMapUpdater} + +trait Otel4sInstances { + + implicit val headersTextMapGetter: TextMapGetter[Headers] = + new TextMapGetter[Headers] { + override def get(carrier: Headers, key: String): Option[String] = + carrier.getOpt(key).collect { case StringVal(value) => value } + + override def keys(carrier: Headers): Iterable[String] = + carrier.toMap.keys + } + + implicit val headersTextMapUpdater: TextMapUpdater[Headers] = + new TextMapUpdater[Headers] { + override def updated(carrier: Headers, key: String, value: String): Headers = + Headers(carrier.toMap.updated(key, StringVal(value))) + } +} + +object instances extends Otel4sInstances diff --git a/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/internal/Semconv.scala b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/internal/Semconv.scala new file mode 100644 index 00000000..e036fa3e --- /dev/null +++ b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/internal/Semconv.scala @@ -0,0 +1,139 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s +package internal + +import dev.profunktor.fs2rabbit.model.{AmqpEnvelope, AmqpMessage, ExchangeName, QueueName, RoutingKey} +import org.typelevel.otel4s.{Attribute, AttributeKey, Attributes} + +private[otel4s] object Semconv { + + object Const { + val MessagingSystem: Attribute[String] = Attribute("messaging.system", "rabbitmq") + } + + object Keys { + val DestinationAnonymous: AttributeKey[Boolean] = AttributeKey[Boolean]("messaging.destination.anonymous") + val DestinationName: AttributeKey[String] = AttributeKey[String]("messaging.destination.name") + val OperationName: AttributeKey[String] = AttributeKey[String]("messaging.operation.name") + val OperationType: AttributeKey[String] = AttributeKey[String]("messaging.operation.type") + val ClientId: AttributeKey[String] = AttributeKey[String]("messaging.client.id") + val MessageConversationId: AttributeKey[String] = AttributeKey[String]("messaging.message.conversation_id") + val MessageId: AttributeKey[String] = AttributeKey[String]("messaging.message.id") + val RabbitDestinationRoutingKey: AttributeKey[String] = + AttributeKey[String]("messaging.rabbitmq.destination.routing_key") + val RabbitMessageDeliveryTag: AttributeKey[Long] = + AttributeKey[Long]("messaging.rabbitmq.message.delivery_tag") + } + + def publishSpanContext( + exchangeName: ExchangeName, + routingKey: RoutingKey, + message: AmqpMessage[Array[Byte]] + ): PublishSpanContext = + PublishSpanContext( + exchangeName, + routingKey, + producerDestinationName(exchangeName, routingKey), + message.properties.messageId, + message.properties.correlationId + ) + + def processSpanContext[A](queueName: QueueName, envelope: AmqpEnvelope[A]): ProcessSpanContext = + ProcessSpanContext( + queueName, + envelope.exchangeName, + envelope.routingKey, + consumerDestinationName(envelope.exchangeName, envelope.routingKey, queueName), + envelope.deliveryTag, + envelope.properties.messageId, + envelope.properties.correlationId, + envelope.redelivered + ) + + def publishAttributes(context: PublishSpanContext, clientId: Option[String]): Attributes = { + val builder = baseBuilder("publish", "send", context.destinationName, clientId) + builder.addAll(Keys.RabbitDestinationRoutingKey.maybe(nonEmpty(context.routingKey.value))) + builder.addAll(Keys.MessageId.maybe(context.messageId)) + builder.addAll(Keys.MessageConversationId.maybe(context.conversationId)) + builder.result() + } + + def processAttributes(context: ProcessSpanContext, clientId: Option[String]): Attributes = { + val builder = baseBuilder("process", "process", context.destinationName, clientId) + builder.addAll(Keys.RabbitDestinationRoutingKey.maybe(nonEmpty(context.routingKey.value))) + builder.addOne(Keys.RabbitMessageDeliveryTag(context.deliveryTag.value)) + builder.addAll(Keys.MessageId.maybe(context.messageId)) + builder.addAll(Keys.MessageConversationId.maybe(context.conversationId)) + builder.addAll(Keys.DestinationAnonymous.maybe(Option.when(isGeneratedQueueName(context.queueName.value))(true))) + builder.result() + } + + def publishLinkAttributes(context: PublishSpanContext): Attributes = { + val builder = Attributes.newBuilder + builder.addOne(Keys.DestinationName(context.destinationName)) + builder.addAll(Keys.RabbitDestinationRoutingKey.maybe(nonEmpty(context.routingKey.value))) + builder.addAll(Keys.MessageId.maybe(context.messageId)) + builder.result() + } + + def processLinkAttributes(context: ProcessSpanContext): Attributes = { + val builder = Attributes.newBuilder + builder.addOne(Keys.DestinationName(context.destinationName)) + builder.addAll(Keys.RabbitDestinationRoutingKey.maybe(nonEmpty(context.routingKey.value))) + builder.addOne(Keys.RabbitMessageDeliveryTag(context.deliveryTag.value)) + builder.addAll(Keys.MessageId.maybe(context.messageId)) + builder.result() + } + + def producerDestinationName(exchangeName: ExchangeName, routingKey: RoutingKey): String = + joinDestination(exchangeName.value, routingKey.value).getOrElse("amq.default") + + def consumerDestinationName(exchangeName: ExchangeName, routingKey: RoutingKey, queueName: QueueName): String = { + val queue = Option.when(queueName.value != routingKey.value)(queueName.value).getOrElse("") + joinDestination(exchangeName.value, routingKey.value, queue).getOrElse("amq.default") + } + + private def baseBuilder( + operationName: String, + operationType: String, + destinationName: String, + clientId: Option[String] + ): Attributes.Builder = { + val builder = Attributes.newBuilder + builder.addOne(Const.MessagingSystem) + builder.addOne(Keys.OperationName(operationName)) + builder.addOne(Keys.OperationType(operationType)) + builder.addOne(Keys.DestinationName(destinationName)) + builder.addAll(Keys.ClientId.maybe(clientId)) + builder + } + + private def joinDestination(parts: String*): Option[String] = { + val nonEmptyParts = parts.filter(_.nonEmpty) + Option.when(nonEmptyParts.nonEmpty)(nonEmptyParts.mkString(":")) + } + + private def nonEmpty(value: String): Option[String] = + Option.when(value.nonEmpty)(value) + + private def isGeneratedQueueName(queue: String): Boolean = + queue.startsWith("amq.gen-") || queue.startsWith("spring.gen-") || isCanonicalUuid(queue) + + private def isCanonicalUuid(value: String): Boolean = + value.matches("[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}") +} diff --git a/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/syntax.scala b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/syntax.scala new file mode 100644 index 00000000..f5b908b7 --- /dev/null +++ b/otel4s/src/main/scala/dev/profunktor/fs2rabbit/otel4s/syntax.scala @@ -0,0 +1,34 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import dev.profunktor.fs2rabbit.interpreter.RabbitClient +import dev.profunktor.fs2rabbit.model.AmqpEnvelope + +trait RabbitTracingSyntax { + implicit final class RabbitClientTracingOps[F[_]](private val client: RabbitClient[F]) { + def traced(rabbitTracer: RabbitTracer[F]): TracedRabbitClient[F] = + rabbitTracer.client(client) + } + + implicit final class AmqpEnvelopeTracingOps[A](private val envelope: AmqpEnvelope[A]) { + def processTraced[F[_], B](fa: F[B])(implicit consumer: TracedRabbitConsumer[F, A]): F[B] = + consumer.process(envelope)(fa) + } +} + +object syntax extends RabbitTracingSyntax diff --git a/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/HeadersPropagationSpec.scala b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/HeadersPropagationSpec.scala new file mode 100644 index 00000000..dba0e194 --- /dev/null +++ b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/HeadersPropagationSpec.scala @@ -0,0 +1,45 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import dev.profunktor.fs2rabbit.model.AmqpFieldValue.{IntVal, StringVal} +import dev.profunktor.fs2rabbit.model.Headers +import dev.profunktor.fs2rabbit.otel4s.instances.* +import org.scalatest.flatspec.AnyFlatSpecLike +import org.scalatest.matchers.should.Matchers +import org.typelevel.otel4s.context.propagation.{TextMapGetter, TextMapUpdater} + +class HeadersPropagationSpec extends AnyFlatSpecLike with Matchers { + + it should "replace a propagation header and retain unrelated headers" in { + val initial = Headers( + "traceparent" -> StringVal("old"), + "application" -> IntVal(42) + ) + + val updated = TextMapUpdater[Headers].updated(initial, "traceparent", "new") + + TextMapGetter[Headers].get(updated, "traceparent") shouldBe Some("new") + updated.getOpt("application") shouldBe Some(IntVal(42)) + } + + it should "ignore non-string values during propagation extraction" in { + val headers = Headers("traceparent" -> IntVal(42)) + + TextMapGetter[Headers].get(headers, "traceparent") shouldBe None + } +} diff --git a/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitConsumerCoverageSpec.scala b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitConsumerCoverageSpec.scala new file mode 100644 index 00000000..136aba72 --- /dev/null +++ b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitConsumerCoverageSpec.scala @@ -0,0 +1,191 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import cats.effect.{IO, Ref} +import dev.profunktor.fs2rabbit.effects.EnvelopeDecoder +import dev.profunktor.fs2rabbit.model.* +import dev.profunktor.fs2rabbit.otel4s.instances.* +import fs2.Stream +import io.opentelemetry.api.trace.SpanKind as JavaSpanKind +import org.scalatest.flatspec.AnyFlatSpecLike +import org.typelevel.otel4s.semconv.experimental.attributes.MessagingExperimentalAttributes as Messaging +import org.typelevel.otel4s.trace.{SpanFinalizer, Tracer, TracerProvider} +import org.typelevel.otel4s.{Attribute, Attributes} + +class RabbitConsumerCoverageSpec extends AnyFlatSpecLike with RabbitTracingTestSupport { + + private implicit val channel: AMQPChannel = dummyChannel + private implicit val decoder: EnvelopeDecoder[IO, String] = AmqpEnvelope.stringDecoder[IO] + + it should "wrap every consumer constructor and preserve acknowledgement functions without settle spans" in + withTestkit { testkit => + implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider + val first = envelope(queueDeliveryTag = 1L) + + for { + acknowledged <- Ref[IO].of(Vector.empty[(AckResult, AckMultiple)]) + tracer <- RabbitTracer.create[IO](RabbitTracer.Config.default) + client = tracer.client(rabbitClient(new StubConsumingProgram(Stream.emit(first), acknowledged))) + auto <- client.createAutoAckConsumer[String](QueueName("auto")) + manual <- client.createAckerConsumer[String]( + QueueName("manual"), + ackMultiple = AckMultiple(true) + ) + multiple <- client.createAckerConsumerWithMultipleFlag[String](QueueName("multiple")) + autoValues <- auto.records.compile.toList + manualValues <- manual._2.records.compile.toList + multipleValues <- multiple._2.records.compile.toList + _ <- manual._1(AckResult.Ack(DeliveryTag(1L))) + _ <- multiple._1(AckResult.NAck(DeliveryTag(2L)), AckMultiple(false)) + acks <- acknowledged.get + spans <- testkit.finishedSpans + } yield { + autoValues shouldBe List(first) + manualValues shouldBe List(first) + multipleValues shouldBe List(first) + auto.queueName shouldBe QueueName("auto") + manual._2.queueName shouldBe QueueName("manual") + multiple._2.queueName shouldBe QueueName("multiple") + acks shouldBe Vector( + AckResult.Ack(DeliveryTag(1L)) -> AckMultiple(true), + AckResult.NAck(DeliveryTag(2L)) -> AckMultiple(false) + ) + spans shouldBe empty + } + } + + it should "delegate records and create exactly one process span per recordsWithProcess element" in + withTestkit { testkit => + for { + moduleTracer <- testkit.tracerProvider.get("fs2.rabbit") + consumer = { + implicit val tracer: Tracer[IO] = moduleTracer + TracedRabbitConsumer[IO, String]( + QueueName("orders"), + Stream(envelope(queueDeliveryTag = 1L), envelope(queueDeliveryTag = 2L)), + RabbitTracer.Config.default + ) + } + raw <- consumer.records.compile.toList + processed <- consumer.recordsWithProcess(value => IO.pure(value.deliveryTag.value)).compile.toList + explicit <- consumer.process(raw.head)(IO.pure("done")) + spans <- testkit.finishedSpans + } yield { + raw.map(_.deliveryTag.value) shouldBe List(1L, 2L) + processed shouldBe List(1L, 2L) + explicit shouldBe "done" + spans.count(_.getName == "process orders:created:orders") shouldBe 3 + all(spans.map(_.getKind)) shouldBe JavaSpanKind.CONSUMER + all(spans.map(_.getLinks.size)) shouldBe 0 + } + } + + it should "link process spans with semantic link attributes and collapse an equal routing key and queue" in + withTestkit { testkit => + for { + creationTracer <- testkit.tracerProvider.get("creation") + headers <- { + implicit val tracer: Tracer[IO] = creationTracer + creationTracer.rootSpan("create-message").surround(Tracer[IO].propagate(Headers.empty)) + } + moduleTracer <- testkit.tracerProvider.get("fs2.rabbit") + consumer = { + implicit val tracer: Tracer[IO] = moduleTracer + TracedRabbitConsumer[IO, String](QueueName("created"), Stream.empty, RabbitTracer.Config.default) + } + _ <- consumer.process(envelope(headers))(IO.unit) + spans <- testkit.finishedSpans + } yield { + val span = spanNamed(spans, "process orders:created") + assertStringAttribute(span, Messaging.MessagingDestinationName("orders:created")) + span.getLinks.size shouldBe 1 + val attributes = span.getLinks.get(0).getAttributes + stringAttribute(attributes, Messaging.MessagingDestinationName.name) shouldBe Some("orders:created") + stringAttribute(attributes, Messaging.MessagingRabbitmqDestinationRoutingKey.name) shouldBe Some("created") + stringAttribute(attributes, Messaging.MessagingMessageId.name) shouldBe Some("message-1") + longAttribute(attributes, Messaging.MessagingRabbitmqMessageDeliveryTag.name) shouldBe Some(7L) + } + } + + it should "mark generated queue destinations as anonymous" in + withTestkit { testkit => + for { + moduleTracer <- testkit.tracerProvider.get("fs2.rabbit") + consumer = { + implicit val tracer: Tracer[IO] = moduleTracer + TracedRabbitConsumer[IO, String]( + QueueName("amq.gen-random"), + Stream.empty, + RabbitTracer.Config.default + ) + } + _ <- consumer.process(envelope())(IO.unit) + spans <- testkit.finishedSpans + } yield { + val span = spanNamed(spans, "process orders:created:amq.gen-random") + booleanAttribute(span.getAttributes, Messaging.MessagingDestinationAnonymous.name) shouldBe Some(true) + } + } + + it should "apply custom process span setup and finalization" in + withTestkit { testkit => + val finalize: SpanFinalizer.Strategy = { case _ => + SpanFinalizer.addAttribute(Attribute("process.finalized", true)) + } + val config = RabbitTracer.Config.default.withProcessSpanSetup(context => + RabbitTracer.Config.SpanSetup( + s"handle ${context.queueName.value}", + Attributes(Attribute("process.custom", context.redelivered)), + finalize + ) + ) + + for { + moduleTracer <- testkit.tracerProvider.get("fs2.rabbit") + consumer = { + implicit val tracer: Tracer[IO] = moduleTracer + TracedRabbitConsumer[IO, String](QueueName("orders"), Stream.empty, config) + } + redelivered = envelope().copy(redelivered = true) + _ <- consumer.process(redelivered)(IO.unit) + spans <- testkit.finishedSpans + } yield { + val span = spanNamed(spans, "handle orders") + booleanAttribute(span.getAttributes, "process.custom") shouldBe Some(true) + booleanAttribute(span.getAttributes, "process.finalized") shouldBe Some(true) + } + } + + it should "delegate consumer records and processing without spans when using the noop tracer" in + withTestkit { testkit => + val first = envelope() + + for { + acknowledged <- Ref[IO].of(Vector.empty[(AckResult, AckMultiple)]) + consumer <- RabbitTracer + .noop[IO] + .client(rabbitClient(new StubConsumingProgram(Stream.emit(first), acknowledged))) + .createAutoAckConsumer[String](QueueName("orders")) + values <- consumer.recordsWithProcess(value => IO.pure(value.payload.reverse)).compile.toList + spans <- testkit.finishedSpans + } yield { + values shouldBe List("daolyap") + spans shouldBe empty + } + } +} diff --git a/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitPublisherCoverageSpec.scala b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitPublisherCoverageSpec.scala new file mode 100644 index 00000000..d8aea5ff --- /dev/null +++ b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitPublisherCoverageSpec.scala @@ -0,0 +1,269 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import cats.data.Kleisli +import cats.effect.unsafe.implicits.global +import cats.effect.{Deferred, IO, Ref} +import dev.profunktor.fs2rabbit.effects.MessageEncoder +import dev.profunktor.fs2rabbit.model.AmqpFieldValue.{IntVal, StringVal} +import dev.profunktor.fs2rabbit.model.* +import dev.profunktor.fs2rabbit.otel4s.instances.* +import io.opentelemetry.api.trace.{SpanKind as JavaSpanKind, StatusCode as JavaStatusCode} +import io.opentelemetry.api.trace.{Span, SpanContext, TraceFlags, TraceState} +import io.opentelemetry.context.Context +import io.opentelemetry.context.propagation.{TextMapGetter as JavaTextMapGetter, TextMapPropagator, TextMapSetter} +import org.scalatest.flatspec.AnyFlatSpecLike +import org.typelevel.otel4s.context.propagation.TextMapGetter +import org.typelevel.otel4s.oteljava.testkit.OtelJavaTestkit +import org.typelevel.otel4s.semconv.experimental.attributes.MessagingExperimentalAttributes as Messaging +import org.typelevel.otel4s.trace.{SpanFinalizer, TracerProvider} +import org.typelevel.otel4s.{Attribute, Attributes} + +import java.util.Collections + +class RabbitPublisherCoverageSpec extends AnyFlatSpecLike with RabbitTracingTestSupport { + + private implicit val channel: AMQPChannel = dummyChannel + private implicit val encoder: MessageEncoder[IO, AmqpMessage[Array[Byte]]] = Kleisli(IO.pure) + + it should "trace every publisher constructor and forward flags and listeners" in + withTestkit { testkit => + implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider + val flag = PublishingFlag(mandatory = true) + val listener: PublishReturn => IO[Unit] = _ => IO.unit + + for { + published <- Ref[IO].of(Vector.empty[Published]) + listeners <- Ref[IO].of(Vector.empty[PublishReturn => IO[Unit]]) + tracer <- RabbitTracer.create[IO](RabbitTracer.Config.default) + client = tracer.client(rabbitClient(new StubPublishingProgram(published, listeners))) + direct <- client.createPublisher[AmqpMessage[Array[Byte]]](ExchangeName("direct"), RoutingKey("one")) + directFlag <- client.createPublisherWithListener[AmqpMessage[Array[Byte]]]( + ExchangeName("direct"), + RoutingKey("two"), + flag, + listener + ) + routing <- client.createRoutingPublisher[AmqpMessage[Array[Byte]]](ExchangeName("routing")) + routingFlag <- client.createRoutingPublisherWithListener[AmqpMessage[Array[Byte]]]( + ExchangeName("routing"), + flag, + listener + ) + basic <- client.createBasicPublisher[AmqpMessage[Array[Byte]]] + basicFlag <- client.createBasicPublisherWithListener[AmqpMessage[Array[Byte]]](flag, listener) + _ <- direct(message()) + _ <- directFlag(message()) + _ <- routing(RoutingKey("three"))(message()) + _ <- routingFlag(RoutingKey("four"))(message()) + _ <- basic(ExchangeName("basic"), RoutingKey("five"), message()) + _ <- basicFlag(ExchangeName("basic"), RoutingKey("six"), message()) + sent <- published.get + registered <- listeners.get + spans <- testkit.finishedSpans + } yield { + sent.map(value => (value.exchangeName.value, value.routingKey.value, value.flag)) shouldBe Vector( + ("direct", "one", None), + ("direct", "two", Some(flag)), + ("routing", "three", None), + ("routing", "four", Some(flag)), + ("basic", "five", None), + ("basic", "six", Some(flag)) + ) + registered.size shouldBe 3 + spans.count(_.getKind == JavaSpanKind.PRODUCER) shouldBe 6 + } + } + + it should "record publishing failures" in + withTestkit { testkit => + implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider + + for { + published <- Ref[IO].of(Vector.empty[Published]) + listeners <- Ref[IO].of(Vector.empty[PublishReturn => IO[Unit]]) + tracer <- RabbitTracer.create[IO](RabbitTracer.Config.default) + publisher <- tracer + .client( + rabbitClient( + new StubPublishingProgram( + published, + listeners, + _ => IO.raiseError(new IllegalStateException("publish failed")) + ) + ) + ) + .createPublisher[AmqpMessage[Array[Byte]]](ExchangeName("orders"), RoutingKey("failed")) + result <- publisher(message()).attempt + spans <- testkit.finishedSpans + } yield { + result.left.map(_.getMessage) shouldBe Left("publish failed") + val span = spanNamed(spans, "publish orders:failed") + span.getStatus.getStatusCode shouldBe JavaStatusCode.ERROR + stringAttribute(span.getAttributes, "error.type") shouldBe Some(classOf[IllegalStateException].getCanonicalName) + } + } + + it should "finalize a canceled publish span" in + withTestkit { testkit => + implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider + + for { + started <- Deferred[IO, Unit] + published <- Ref[IO].of(Vector.empty[Published]) + listeners <- Ref[IO].of(Vector.empty[PublishReturn => IO[Unit]]) + tracer <- RabbitTracer.create[IO](RabbitTracer.Config.default) + publisher <- tracer + .client( + rabbitClient( + new StubPublishingProgram(published, listeners, _ => started.complete(()).void *> IO.never) + ) + ) + .createPublisher[AmqpMessage[Array[Byte]]](ExchangeName("orders"), RoutingKey("blocked")) + fiber <- publisher(message()).start + _ <- started.get + _ <- fiber.cancel + spans <- testkit.finishedSpans + } yield { + val span = spanNamed(spans, "publish orders:blocked") + span.getStatus.getStatusCode shouldBe JavaStatusCode.ERROR + stringAttribute(span.getAttributes, "error.type") shouldBe Some("canceled") + } + } + + it should "apply custom span setup and attribute precedence without implicit endpoint metadata" in + withTestkit { testkit => + implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider + val finalize: SpanFinalizer.Strategy = { case _ => + SpanFinalizer.addAttribute(Attribute("custom.finalized", true)) + } + val config = RabbitTracer.Config.default + .withConstAttributes( + Attributes( + Messaging.MessagingDestinationName("constant-destination"), + Attribute("attribute.priority", "constant") + ) + ) + .withPublishSpanSetup(_ => + RabbitTracer.Config.SpanSetup( + "custom publish", + Attributes(Attribute("attribute.priority", "span")), + finalize + ) + ) + + for { + published <- Ref[IO].of(Vector.empty[Published]) + listeners <- Ref[IO].of(Vector.empty[PublishReturn => IO[Unit]]) + tracer <- RabbitTracer.create[IO](config) + publisher <- tracer + .client(rabbitClient(new StubPublishingProgram(published, listeners))) + .createPublisher[AmqpMessage[Array[Byte]]](ExchangeName("orders"), RoutingKey("created")) + _ <- publisher(message()) + spans <- testkit.finishedSpans + } yield { + val span = spanNamed(spans, "custom publish") + stringAttribute(span.getAttributes, Messaging.MessagingDestinationName.name) shouldBe Some( + "constant-destination" + ) + stringAttribute(span.getAttributes, "attribute.priority") shouldBe Some("span") + booleanAttribute(span.getAttributes, "custom.finalized") shouldBe Some(true) + stringAttribute(span.getAttributes, "server.address") shouldBe None + longAttribute(span.getAttributes, "server.port") shouldBe None + } + } + + it should "replace malformed propagation input and use the default-exchange destination" in + withTestkit { testkit => + implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider + val malformed = Headers( + "traceparent" -> StringVal("malformed"), + "application" -> IntVal(42) + ) + + for { + published <- Ref[IO].of(Vector.empty[Published]) + listeners <- Ref[IO].of(Vector.empty[PublishReturn => IO[Unit]]) + tracer <- RabbitTracer.create[IO](RabbitTracer.Config.default) + publisher <- tracer + .client(rabbitClient(new StubPublishingProgram(published, listeners))) + .createPublisher[AmqpMessage[Array[Byte]]](ExchangeName(""), RoutingKey("")) + _ <- publisher(message(malformed)) + sent <- published.get.map(_.head) + spans <- testkit.finishedSpans + } yield { + TextMapGetter[Headers].get(sent.message.properties.headers, "traceparent") should not be Some("malformed") + sent.message.properties.headers.getOpt("application") shouldBe Some(IntVal(42)) + val span = spanNamed(spans, "publish amq.default") + span.getKind shouldBe JavaSpanKind.PRODUCER + assertStringAttribute(span, Messaging.MessagingDestinationName("amq.default")) + } + } + + it should "honor a configured non-W3C propagator and expose publish link attributes" in { + val customPropagator = new FixedContextPropagator + + OtelJavaTestkit + .inMemory[IO](_.addTextMapPropagators(customPropagator)) + .use { testkit => + implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider + val headers = Headers("x-rabbit-context" -> StringVal("recognized")) + + for { + published <- Ref[IO].of(Vector.empty[Published]) + listeners <- Ref[IO].of(Vector.empty[PublishReturn => IO[Unit]]) + tracer <- RabbitTracer.create[IO](RabbitTracer.Config.default) + publisher <- tracer + .client(rabbitClient(new StubPublishingProgram(published, listeners))) + .createPublisher[AmqpMessage[Array[Byte]]](ExchangeName("orders"), RoutingKey("created")) + _ <- publisher(message(headers)) + sent <- published.get.map(_.head) + spans <- testkit.finishedSpans + } yield { + sent.message.properties.headers shouldBe headers + val span = spanNamed(spans, "publish orders:created") + span.getKind shouldBe JavaSpanKind.CLIENT + span.getLinks.size shouldBe 1 + val attributes = span.getLinks.get(0).getAttributes + stringAttribute(attributes, Messaging.MessagingDestinationName.name) shouldBe Some("orders:created") + stringAttribute(attributes, Messaging.MessagingRabbitmqDestinationRoutingKey.name) shouldBe Some("created") + stringAttribute(attributes, Messaging.MessagingMessageId.name) shouldBe Some("message-1") + } + } + .unsafeRunSync() + } + + private final class FixedContextPropagator extends TextMapPropagator { + private val remoteContext = SpanContext.createFromRemoteParent( + "11111111111111111111111111111111", + "2222222222222222", + TraceFlags.getSampled, + TraceState.getDefault + ) + + override def fields(): java.util.Collection[String] = Collections.singletonList("x-rabbit-context") + + override def inject[C](context: Context, carrier: C, setter: TextMapSetter[C]): Unit = () + + override def extract[C](context: Context, carrier: C, getter: JavaTextMapGetter[C]): Context = + Option(getter.get(carrier, "x-rabbit-context")) match { + case Some("recognized") => context.`with`(Span.wrap(remoteContext)) + case _ => context + } + } +} diff --git a/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitSyntaxSpec.scala b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitSyntaxSpec.scala new file mode 100644 index 00000000..6516e4c1 --- /dev/null +++ b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitSyntaxSpec.scala @@ -0,0 +1,58 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import cats.effect.{IO, Ref} +import dev.profunktor.fs2rabbit.otel4s.syntax.* +import fs2.Stream +import org.scalatest.flatspec.AnyFlatSpecLike +import org.typelevel.otel4s.trace.Tracer + +class RabbitSyntaxSpec extends AnyFlatSpecLike with RabbitTracingTestSupport { + + it should "bind a RabbitClient to a RabbitTracer" in { + val client = rabbitClient(null.asInstanceOf[dev.profunktor.fs2rabbit.program.PublishingProgram[IO]]) + val traced = client.traced(RabbitTracer.noop[IO]) + + traced.underlying should be theSameInstanceAs client + } + + it should "delegate processTraced to the implicit traced consumer" in + withTestkit { testkit => + val value = envelope() + + for { + processed <- Ref[IO].of(0) + moduleTracer <- testkit.tracerProvider.get("fs2.rabbit") + result <- { + implicit val tracer: Tracer[IO] = moduleTracer + implicit val consumer: TracedRabbitConsumer[IO, String] = + TracedRabbitConsumer[IO, String]( + queueName = dev.profunktor.fs2rabbit.model.QueueName("orders"), + underlying = Stream.empty, + config = RabbitTracer.Config.default + ) + + value.processTraced(processed.updateAndGet(_ + 1)) + } + spans <- testkit.finishedSpans + } yield { + result shouldBe 1 + spans.count(_.getName == "process orders:created:orders") shouldBe 1 + } + } +} diff --git a/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitTracingSpec.scala b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitTracingSpec.scala new file mode 100644 index 00000000..a6f4403d --- /dev/null +++ b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitTracingSpec.scala @@ -0,0 +1,307 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import cats.data.Kleisli +import cats.effect.unsafe.implicits.global +import cats.effect.{IO, Ref} +import dev.profunktor.fs2rabbit.effects.MessageEncoder +import dev.profunktor.fs2rabbit.interpreter.RabbitClient +import dev.profunktor.fs2rabbit.model.* +import dev.profunktor.fs2rabbit.otel4s.instances.* +import dev.profunktor.fs2rabbit.program.PublishingProgram +import fs2.Stream +import io.opentelemetry.api.common.AttributeKey +import io.opentelemetry.api.trace.{SpanKind as JavaSpanKind, StatusCode as JavaStatusCode} +import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator +import io.opentelemetry.sdk.trace.data.SpanData +import org.scalatest.flatspec.AnyFlatSpecLike +import org.scalatest.matchers.should.Matchers +import org.typelevel.otel4s.Attribute +import org.typelevel.otel4s.context.propagation.TextMapGetter +import org.typelevel.otel4s.oteljava.testkit.OtelJavaTestkit +import org.typelevel.otel4s.semconv.experimental.attributes.MessagingExperimentalAttributes as Messaging +import org.typelevel.otel4s.trace.{Tracer, TracerProvider} + +class RabbitTracingSpec extends AnyFlatSpecLike with Matchers { + + private val dummyChannel: AMQPChannel = RabbitChannel(null) + + it should "create a producer publish span and inject its context" in + withTestkit { testkit => + implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider + implicit val channel: AMQPChannel = dummyChannel + implicit val encoder: MessageEncoder[IO, AmqpMessage[Array[Byte]]] = Kleisli(IO.pure) + + for { + captured <- Ref[IO].of(Option.empty[AmqpMessage[Array[Byte]]]) + client = recordingClient(captured) + tracer <- RabbitTracer.create[IO]( + RabbitTracer.Config.default + .withServerAddress("rabbitmq.example.com", Some(5672)) + .withClientId("orders-service") + ) + publisher <- tracer + .client(client) + .createPublisher[AmqpMessage[Array[Byte]]]( + ExchangeName("orders"), + RoutingKey("created") + ) + appTracer <- testkit.tracerProvider.get("application") + _ <- appTracer.rootSpan("ambient").surround(publisher(message())) + published <- captured.get.map(_.getOrElse(fail("message was not published"))) + spans <- testkit.finishedSpans + } yield { + TextMapGetter[Headers].get(published.properties.headers, "traceparent") should not be empty + val span = spanNamed(spans, "publish orders:created") + span.getKind shouldBe JavaSpanKind.PRODUCER + span.getParentSpanContext.isValid shouldBe true + span.getLinks shouldBe empty + assertStringAttribute(span, Messaging.MessagingSystem(Messaging.MessagingSystemValue.Rabbitmq)) + assertStringAttribute(span, Messaging.MessagingOperationName("publish")) + assertStringAttribute(span, Messaging.MessagingOperationType(Messaging.MessagingOperationTypeValue.Send)) + assertStringAttribute(span, Messaging.MessagingDestinationName("orders:created")) + assertStringAttribute(span, Messaging.MessagingRabbitmqDestinationRoutingKey("created")) + assertStringAttribute(span, Messaging.MessagingMessageId("message-1")) + assertStringAttribute(span, Messaging.MessagingMessageConversationId("conversation-1")) + assertStringAttribute(span, Messaging.MessagingClientId("orders-service")) + attribute(span, "server.address") shouldBe "rabbitmq.example.com" + longAttribute(span, "server.port") shouldBe 5672L + } + } + + it should "preserve existing creation context and create a linked client publish span" in + withTestkit { testkit => + implicit val tracerProvider: TracerProvider[IO] = testkit.tracerProvider + implicit val channel: AMQPChannel = dummyChannel + implicit val encoder: MessageEncoder[IO, AmqpMessage[Array[Byte]]] = Kleisli(IO.pure) + + for { + appTracer <- testkit.tracerProvider.get("application") + headers <- { + implicit val tracer: Tracer[IO] = appTracer + appTracer.rootSpan("create-message").surround(Tracer[IO].propagate(Headers.empty)) + } + originalTraceparent = TextMapGetter[Headers] + .get(headers, "traceparent") + .getOrElse(fail("missing source trace context")) + captured <- Ref[IO].of(Option.empty[AmqpMessage[Array[Byte]]]) + tracer <- RabbitTracer.create[IO](RabbitTracer.Config.default) + publisher <- tracer + .client(recordingClient(captured)) + .createPublisher[AmqpMessage[Array[Byte]]]( + ExchangeName("orders"), + RoutingKey("created") + ) + _ <- publisher(message(headers)) + published <- captured.get.map(_.getOrElse(fail("message was not published"))) + spans <- testkit.finishedSpans + } yield { + TextMapGetter[Headers].get(published.properties.headers, "traceparent") shouldBe Some(originalTraceparent) + val span = spanNamed(spans, "publish orders:created") + span.getKind shouldBe JavaSpanKind.CLIENT + span.getLinks.size shouldBe 1 + } + } + + it should "create a root consumer process span linked to message creation context" in + withTestkit { testkit => + for { + appTracer <- testkit.tracerProvider.get("application") + headers <- { + implicit val tracer: Tracer[IO] = appTracer + appTracer.rootSpan("create-message").surround(Tracer[IO].propagate(Headers.empty)) + } + moduleTracer <- testkit.tracerProvider.get("fs2.rabbit") + consumer = { + implicit val tracer: Tracer[IO] = moduleTracer + TracedRabbitConsumer[IO, String]( + QueueName("orders-queue"), + Stream.empty, + RabbitTracer.Config.default + ) + } + envelope = AmqpEnvelope( + DeliveryTag(7L), + "payload", + AmqpProperties.empty.copy(messageId = Some("message-1"), headers = headers), + ExchangeName("orders"), + RoutingKey("created"), + redelivered = false + ) + _ <- appTracer.rootSpan("ambient").surround(consumer.process(envelope)(IO.unit)) + spans <- testkit.finishedSpans + } yield { + val span = spanNamed(spans, "process orders:created:orders-queue") + span.getKind shouldBe JavaSpanKind.CONSUMER + span.getParentSpanContext.isValid shouldBe false + span.getLinks.size shouldBe 1 + assertStringAttribute(span, Messaging.MessagingOperationType(Messaging.MessagingOperationTypeValue.Process)) + assertLongAttribute(span, Messaging.MessagingRabbitmqMessageDeliveryTag(7L)) + assertStringAttribute(span, Messaging.MessagingMessageId("message-1")) + } + } + + it should "record processing failures on the process span" in + withTestkit { testkit => + for { + moduleTracer <- testkit.tracerProvider.get("fs2.rabbit") + consumer = { + implicit val tracer: Tracer[IO] = moduleTracer + TracedRabbitConsumer[IO, String]( + QueueName("orders-queue"), + Stream.empty, + RabbitTracer.Config.default + ) + } + envelope = AmqpEnvelope( + DeliveryTag(8L), + "payload", + AmqpProperties.empty, + ExchangeName("orders"), + RoutingKey("created"), + redelivered = false + ) + _ <- consumer.process(envelope)(IO.raiseError(new IllegalStateException("boom"))).attempt + spans <- testkit.finishedSpans + } yield { + val span = spanNamed(spans, "process orders:created:orders-queue") + span.getStatus.getStatusCode shouldBe JavaStatusCode.ERROR + attribute(span, "error.type") shouldBe classOf[IllegalStateException].getCanonicalName + } + } + + it should "delegate without spans or propagation when using the noop tracer" in + withTestkit { testkit => + implicit val channel: AMQPChannel = dummyChannel + implicit val encoder: MessageEncoder[IO, AmqpMessage[Array[Byte]]] = Kleisli(IO.pure) + + for { + captured <- Ref[IO].of(Option.empty[AmqpMessage[Array[Byte]]]) + publisher <- RabbitTracer + .noop[IO] + .client(recordingClient(captured)) + .createPublisher[AmqpMessage[Array[Byte]]]( + ExchangeName("orders"), + RoutingKey("created") + ) + _ <- publisher(message()) + published <- captured.get.map(_.getOrElse(fail("message was not published"))) + spans <- testkit.finishedSpans + } yield { + TextMapGetter[Headers].get(published.properties.headers, "traceparent") shouldBe None + spans shouldBe empty + } + } + + private def withTestkit[A](run: OtelJavaTestkit[IO] => IO[A]): A = + OtelJavaTestkit + .inMemory[IO](_.addTextMapPropagators(W3CTraceContextPropagator.getInstance())) + .use(run) + .unsafeRunSync() + + private def message(headers: Headers = Headers.empty): AmqpMessage[Array[Byte]] = + AmqpMessage( + "body".getBytes("UTF-8"), + AmqpProperties.empty.copy( + messageId = Some("message-1"), + correlationId = Some("conversation-1"), + headers = headers + ) + ) + + private def recordingClient(captured: Ref[IO, Option[AmqpMessage[Array[Byte]]]]): RabbitClient[IO] = + new RabbitClient[IO]( + null, + null, + null, + null, + null, + new RecordingPublishingProgram(captured) + ) + + private def spanNamed(spans: List[SpanData], name: String): SpanData = + spans.find(_.getName == name).getOrElse(fail(s"missing span [$name], found ${spans.map(_.getName)}")) + + private def attribute(span: SpanData, name: String): String = + span.getAttributes.get(AttributeKey.stringKey(name)) + + private def longAttribute(span: SpanData, name: String): Long = + span.getAttributes.get(AttributeKey.longKey(name)).longValue() + + private def assertStringAttribute(span: SpanData, expected: Attribute[String]): Unit = + attribute(span, expected.key.name) shouldBe expected.value + + private def assertLongAttribute(span: SpanData, expected: Attribute[Long]): Unit = + longAttribute(span, expected.key.name) shouldBe expected.value + + private final class RecordingPublishingProgram(captured: Ref[IO, Option[AmqpMessage[Array[Byte]]]]) + extends PublishingProgram[IO] { + override def basicPublish( + channel: AMQPChannel, + exchangeName: ExchangeName, + routingKey: RoutingKey, + message: AmqpMessage[Array[Byte]] + ): IO[Unit] = captured.set(Some(message)) + + override def basicPublishWithFlag( + channel: AMQPChannel, + exchangeName: ExchangeName, + routingKey: RoutingKey, + flag: PublishingFlag, + message: AmqpMessage[Array[Byte]] + ): IO[Unit] = captured.set(Some(message)) + + override def addPublishingListener(channel: AMQPChannel, listener: PublishReturn => IO[Unit]): IO[Unit] = IO.unit + override def clearPublishingListeners(channel: AMQPChannel): IO[Unit] = IO.unit + + override def createPublisher[A](channel: AMQPChannel, exchangeName: ExchangeName, routingKey: RoutingKey)(implicit + encoder: MessageEncoder[IO, A] + ): IO[A => IO[Unit]] = IO.pure(value => encoder.run(value).flatMap(message => captured.set(Some(message)))) + + override def createPublisherWithListener[A]( + channel: AMQPChannel, + exchangeName: ExchangeName, + routingKey: RoutingKey, + flags: PublishingFlag, + listener: PublishReturn => IO[Unit] + )(implicit encoder: MessageEncoder[IO, A]): IO[A => IO[Unit]] = unsupported + + override def createRoutingPublisher[A](channel: AMQPChannel, exchangeName: ExchangeName)(implicit + encoder: MessageEncoder[IO, A] + ): IO[RoutingKey => A => IO[Unit]] = unsupported + + override def createRoutingPublisherWithListener[A]( + channel: AMQPChannel, + exchangeName: ExchangeName, + flags: PublishingFlag, + listener: PublishReturn => IO[Unit] + )(implicit encoder: MessageEncoder[IO, A]): IO[RoutingKey => A => IO[Unit]] = unsupported + + override def createBasicPublisher[A](channel: AMQPChannel)(implicit + encoder: MessageEncoder[IO, A] + ): IO[(ExchangeName, RoutingKey, A) => IO[Unit]] = unsupported + + override def createBasicPublisherWithListener[A]( + channel: AMQPChannel, + flags: PublishingFlag, + listener: PublishReturn => IO[Unit] + )(implicit encoder: MessageEncoder[IO, A]): IO[(ExchangeName, RoutingKey, A) => IO[Unit]] = unsupported + + private def unsupported[A]: IO[A] = IO.raiseError(new UnsupportedOperationException("unused in test")) + } +} diff --git a/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitTracingTestSupport.scala b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitTracingTestSupport.scala new file mode 100644 index 00000000..5081c8ff --- /dev/null +++ b/otel4s/src/test/scala/dev/profunktor/fs2rabbit/otel4s/RabbitTracingTestSupport.scala @@ -0,0 +1,236 @@ +/* + * Copyright 2017-2026 ProfunKtor + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dev.profunktor.fs2rabbit.otel4s + +import cats.effect.unsafe.implicits.global +import cats.effect.{IO, Ref} +import dev.profunktor.fs2rabbit.arguments.Arguments +import dev.profunktor.fs2rabbit.effects.{EnvelopeDecoder, MessageEncoder} +import dev.profunktor.fs2rabbit.interpreter.RabbitClient +import dev.profunktor.fs2rabbit.model.* +import dev.profunktor.fs2rabbit.program.{AckConsumingProgram, PublishingProgram} +import fs2.Stream +import io.opentelemetry.api.common.{AttributeKey as JavaAttributeKey, Attributes as JavaAttributes} +import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator +import io.opentelemetry.sdk.trace.data.SpanData +import org.scalatest.matchers.should.Matchers +import org.typelevel.otel4s.Attribute +import org.typelevel.otel4s.oteljava.testkit.OtelJavaTestkit + +private[otel4s] final case class Published( + exchangeName: ExchangeName, + routingKey: RoutingKey, + flag: Option[PublishingFlag], + message: AmqpMessage[Array[Byte]] +) + +trait RabbitTracingTestSupport extends Matchers { + + protected val dummyChannel: AMQPChannel = RabbitChannel(null) + + protected def withTestkit[A](run: OtelJavaTestkit[IO] => IO[A]): A = + OtelJavaTestkit + .inMemory[IO](_.addTextMapPropagators(W3CTraceContextPropagator.getInstance())) + .use(run) + .unsafeRunSync() + + protected def message(headers: Headers = Headers.empty): AmqpMessage[Array[Byte]] = + AmqpMessage( + "body".getBytes("UTF-8"), + AmqpProperties.empty.copy( + messageId = Some("message-1"), + correlationId = Some("conversation-1"), + headers = headers + ) + ) + + protected def envelope( + headers: Headers = Headers.empty, + queueDeliveryTag: Long = 7L, + exchangeName: ExchangeName = ExchangeName("orders"), + routingKey: RoutingKey = RoutingKey("created") + ): AmqpEnvelope[String] = + AmqpEnvelope( + DeliveryTag(queueDeliveryTag), + "payload", + AmqpProperties.empty.copy( + messageId = Some("message-1"), + correlationId = Some("conversation-1"), + headers = headers + ), + exchangeName, + routingKey, + redelivered = false + ) + + protected def rabbitClient( + publishingProgram: PublishingProgram[IO], + consumingProgram: AckConsumingProgram[IO] = null + ): RabbitClient[IO] = + new RabbitClient[IO](null, null, null, null, consumingProgram, publishingProgram) + + protected def rabbitClient(consumingProgram: AckConsumingProgram[IO]): RabbitClient[IO] = + new RabbitClient[IO](null, null, null, null, consumingProgram, null) + + protected def spanNamed(spans: List[SpanData], name: String): SpanData = + spans.find(_.getName == name).getOrElse(fail(s"missing span [$name], found ${spans.map(_.getName)}")) + + protected def stringAttribute(attributes: JavaAttributes, name: String): Option[String] = + Option(attributes.get(JavaAttributeKey.stringKey(name))) + + protected def longAttribute(attributes: JavaAttributes, name: String): Option[Long] = + Option(attributes.get(JavaAttributeKey.longKey(name))).map(_.longValue()) + + protected def booleanAttribute(attributes: JavaAttributes, name: String): Option[Boolean] = + Option(attributes.get(JavaAttributeKey.booleanKey(name))).map(_.booleanValue()) + + protected def assertStringAttribute(span: SpanData, expected: Attribute[String]): Unit = + stringAttribute(span.getAttributes, expected.key.name) shouldBe Some(expected.value) + + protected def assertLongAttribute(span: SpanData, expected: Attribute[Long]): Unit = + longAttribute(span.getAttributes, expected.key.name) shouldBe Some(expected.value) + + protected final class StubPublishingProgram( + published: Ref[IO, Vector[Published]], + listeners: Ref[IO, Vector[PublishReturn => IO[Unit]]], + onPublish: Published => IO[Unit] = _ => IO.unit + ) extends PublishingProgram[IO] { + + private def record(value: Published): IO[Unit] = published.update(_ :+ value) *> onPublish(value) + + override def basicPublish( + channel: AMQPChannel, + exchangeName: ExchangeName, + routingKey: RoutingKey, + message: AmqpMessage[Array[Byte]] + ): IO[Unit] = record(Published(exchangeName, routingKey, None, message)) + + override def basicPublishWithFlag( + channel: AMQPChannel, + exchangeName: ExchangeName, + routingKey: RoutingKey, + flag: PublishingFlag, + message: AmqpMessage[Array[Byte]] + ): IO[Unit] = record(Published(exchangeName, routingKey, Some(flag), message)) + + override def addPublishingListener(channel: AMQPChannel, listener: PublishReturn => IO[Unit]): IO[Unit] = + listeners.update(_ :+ listener) + + override def clearPublishingListeners(channel: AMQPChannel): IO[Unit] = listeners.set(Vector.empty) + + override def createPublisher[A](channel: AMQPChannel, exchangeName: ExchangeName, routingKey: RoutingKey)(implicit + encoder: MessageEncoder[IO, A] + ): IO[A => IO[Unit]] = + IO.pure(value => encoder.run(value).flatMap(basicPublish(channel, exchangeName, routingKey, _))) + + override def createPublisherWithListener[A]( + channel: AMQPChannel, + exchangeName: ExchangeName, + routingKey: RoutingKey, + flag: PublishingFlag, + listener: PublishReturn => IO[Unit] + )(implicit encoder: MessageEncoder[IO, A]): IO[A => IO[Unit]] = + addPublishingListener(channel, listener).as(value => + encoder.run(value).flatMap(basicPublishWithFlag(channel, exchangeName, routingKey, flag, _)) + ) + + override def createRoutingPublisher[A](channel: AMQPChannel, exchangeName: ExchangeName)(implicit + encoder: MessageEncoder[IO, A] + ): IO[RoutingKey => A => IO[Unit]] = + IO.pure(routingKey => value => encoder.run(value).flatMap(basicPublish(channel, exchangeName, routingKey, _))) + + override def createRoutingPublisherWithListener[A]( + channel: AMQPChannel, + exchangeName: ExchangeName, + flag: PublishingFlag, + listener: PublishReturn => IO[Unit] + )(implicit encoder: MessageEncoder[IO, A]): IO[RoutingKey => A => IO[Unit]] = + addPublishingListener(channel, listener).as(routingKey => + value => encoder.run(value).flatMap(basicPublishWithFlag(channel, exchangeName, routingKey, flag, _)) + ) + + override def createBasicPublisher[A](channel: AMQPChannel)(implicit + encoder: MessageEncoder[IO, A] + ): IO[(ExchangeName, RoutingKey, A) => IO[Unit]] = + IO.pure((exchangeName, routingKey, value) => + encoder.run(value).flatMap(basicPublish(channel, exchangeName, routingKey, _)) + ) + + override def createBasicPublisherWithListener[A]( + channel: AMQPChannel, + flag: PublishingFlag, + listener: PublishReturn => IO[Unit] + )(implicit encoder: MessageEncoder[IO, A]): IO[(ExchangeName, RoutingKey, A) => IO[Unit]] = + addPublishingListener(channel, listener).as((exchangeName, routingKey, value) => + encoder.run(value).flatMap(basicPublishWithFlag(channel, exchangeName, routingKey, flag, _)) + ) + } + + protected final class StubConsumingProgram( + source: Stream[IO, AmqpEnvelope[String]], + acked: Ref[IO, Vector[(AckResult, AckMultiple)]] + ) extends AckConsumingProgram[IO] { + + private def stream[A]: Stream[IO, AmqpEnvelope[A]] = + source.asInstanceOf[Stream[IO, AmqpEnvelope[A]]] + + override def createAckerConsumer[A]( + channel: AMQPChannel, + queueName: QueueName, + basicQos: BasicQos, + consumerArgs: Option[ConsumerArgs], + ackMultiple: AckMultiple + )(implicit decoder: EnvelopeDecoder[IO, A]): IO[(AckResult => IO[Unit], Stream[IO, AmqpEnvelope[A]])] = + IO.pure((result => acked.update(_ :+ (result -> ackMultiple)), stream[A])) + + override def createAckerConsumerWithMultipleFlag[A]( + channel: AMQPChannel, + queueName: QueueName, + basicQos: BasicQos, + consumerArgs: Option[ConsumerArgs] + )(implicit + decoder: EnvelopeDecoder[IO, A] + ): IO[((AckResult, AckMultiple) => IO[Unit], Stream[IO, AmqpEnvelope[A]])] = + IO.pure(((result, multiple) => acked.update(_ :+ (result -> multiple)), stream[A])) + + override def createAutoAckConsumer[A]( + channel: AMQPChannel, + queueName: QueueName, + basicQos: BasicQos, + consumerArgs: Option[ConsumerArgs] + )(implicit decoder: EnvelopeDecoder[IO, A]): IO[Stream[IO, AmqpEnvelope[A]]] = IO.pure(stream[A]) + + override def createAcker(channel: AMQPChannel, ackMultiple: AckMultiple): IO[AckResult => IO[Unit]] = + IO.pure(result => acked.update(_ :+ (result -> ackMultiple))) + + override def createAckerWithMultipleFlag(channel: AMQPChannel): IO[(AckResult, AckMultiple) => IO[Unit]] = + IO.pure((result, multiple) => acked.update(_ :+ (result -> multiple))) + + override def createConsumer[A]( + queueName: QueueName, + channel: AMQPChannel, + basicQos: BasicQos, + autoAck: Boolean, + noLocal: Boolean, + exclusive: Boolean, + consumerTag: ConsumerTag, + args: Arguments + )(implicit decoder: EnvelopeDecoder[IO, A]): IO[Stream[IO, AmqpEnvelope[A]]] = IO.pure(stream[A]) + + override def basicCancel(channel: AMQPChannel, consumerTag: ConsumerTag): IO[Unit] = IO.unit + } +} diff --git a/project/Dependencies.scala b/project/Dependencies.scala index a1d4dcc0..feed15e5 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -19,15 +19,20 @@ object Dependencies { val scalaCheck = "1.20.0" val scalaTestPlusScalaCheck = "3.2.20.0" val disciplineScalaCheck = "2.2.0" + val otel4s = "1.1.0" } object Libraries { def circe(artifact: String): ModuleID = "io.circe" %% artifact % Version.circe - lazy val amqpClient = "com.rabbitmq" % "amqp-client" % Version.amqpClient - lazy val catsEffect = "org.typelevel" %% "cats-effect" % Version.catsEffect - lazy val fs2Core = "co.fs2" %% "fs2-core" % Version.fs2 - lazy val scodecCats = "org.scodec" %% "scodec-cats" % Version.scodec + lazy val amqpClient = "com.rabbitmq" % "amqp-client" % Version.amqpClient + lazy val catsEffect = "org.typelevel" %% "cats-effect" % Version.catsEffect + lazy val fs2Core = "co.fs2" %% "fs2-core" % Version.fs2 + lazy val otel4sCoreTrace = "org.typelevel" %% "otel4s-core-trace" % Version.otel4s + lazy val otel4sSemconv = "org.typelevel" %% "otel4s-semconv" % Version.otel4s + lazy val otel4sSemconvExp = "org.typelevel" %% "otel4s-semconv-experimental" % Version.otel4s + lazy val otel4sTestkit = "org.typelevel" %% "otel4s-oteljava-testkit" % Version.otel4s + lazy val scodecCats = "org.scodec" %% "scodec-cats" % Version.scodec // Compiler lazy val kindProjector = "org.typelevel" % "kind-projector" % Version.kindProjector cross CrossVersion.full diff --git a/project/plugins.sbt b/project/plugins.sbt index 5cbe6777..095727a7 100644 --- a/project/plugins.sbt +++ b/project/plugins.sbt @@ -13,3 +13,4 @@ addSbtPlugin("org.typelevel" % "sbt-tpolecat" % "0.5.7") addSbtPlugin("com.timushev.sbt" % "sbt-updates" % "0.7.0") addSbtPlugin("com.typesafe" % "sbt-mima-plugin" % "1.1.6") addSbtPlugin("org.scoverage" % "sbt-scoverage" % "2.4.4") +addSbtPlugin("com.eed3si9n" % "sbt-buildinfo" % "0.13.1") diff --git a/site/docs/otel4s.md b/site/docs/otel4s.md new file mode 100644 index 00000000..1acb6c42 --- /dev/null +++ b/site/docs/otel4s.md @@ -0,0 +1,181 @@ +--- +layout: docs +title: "OpenTelemetry tracing" +number: 7 +--- + +# OpenTelemetry tracing + +The `fs2-rabbit-otel4s` module adds otel4s spans and trace-context propagation to +publishers and consumers without requiring the OpenTelemetry Java agent. + +```scala +libraryDependencies += "dev.profunktor" %% "fs2-rabbit-otel4s" % Version +``` + +Create a `RabbitTracer` from the application's otel4s `TracerProvider`, then use +its traced client for publishers and consumers. The tracer should normally be +created once alongside the `RabbitClient`: + +```scala +import cats.effect.Async +import dev.profunktor.fs2rabbit.interpreter.RabbitClient +import dev.profunktor.fs2rabbit.model.* +import dev.profunktor.fs2rabbit.otel4s.* +import org.typelevel.otel4s.trace.TracerProvider + +def program[F[_]: Async: TracerProvider](client: RabbitClient[F])(implicit + channel: AMQPChannel +): F[Unit] = + RabbitTracer + .resource[F]( + RabbitTracer.Config.default + .withServerAddress("rabbitmq.example.com", Some(5672)) + .withClientId("orders-service") + ) + .use { rabbitTracer => + val traced = rabbitTracer.client(client) + + for { + _ <- publish(traced) + _ <- consume(traced) + } yield () + } +``` + +The encoder and decoder instances are the same ones used by the regular +`RabbitClient` API. All publisher variants and the auto-ack and manual-ack consumer +constructors have traced equivalents. + +## Publishing + +Create publishers from the `TracedRabbitClient` in the same way as from a regular +client. Each invocation of the returned function creates a publish span and +injects a creation context into the AMQP headers when the message does not +already carry one: + +```scala +def publish[F[_]: Async](client: TracedRabbitClient[F])(implicit + channel: AMQPChannel +): F[Unit] = + for { + publish <- client.createPublisher[String]( + ExchangeName("orders"), + RoutingKey("created") + ) + _ <- publish("order-123") + } yield () +``` + +The routing publisher, basic publisher, publishing flags, and returned-message +listener variants are also available on `TracedRabbitClient`. + +## Consuming + +`recordsWithProcess` places a process span around the effect returned for each +message: + +```scala +def consume[F[_]: Async](client: TracedRabbitClient[F])(implicit + channel: AMQPChannel +): F[Unit] = + for { + consumer <- client.createAutoAckConsumer[String](QueueName("order-workers")) + _ <- consumer + .recordsWithProcess { envelope => + Async[F].delay(println(envelope.payload)) + } + .compile + .drain + } yield () +``` + +For manual acknowledgement, use `createAckerConsumer` or +`createAckerConsumerWithMultipleFlag`. Acknowledgement and rejection use the +regular fs2-rabbit functions returned by those constructors and do not create +additional spans. + +Use `records` and `consumer.process(envelope)(effect)` when the processing +boundary needs to be placed manually: + +```scala +consumer.records.evalMap { envelope => + consumer.process(envelope)(handle(envelope)) +} +``` + +## Syntax + +Importing `dev.profunktor.fs2rabbit.otel4s.syntax.*` enables two convenience +extensions. They do not change tracing behavior; they are shorter forms of the +regular API: + +```scala +import dev.profunktor.fs2rabbit.otel4s.syntax.* + +val traced = client.traced(rabbitTracer) +// equivalent to rabbitTracer.client(client) +``` + +`processTraced` is useful with a manually chosen processing boundary. It requires +the corresponding `TracedRabbitConsumer` to be in implicit scope: + +```scala +implicit val tracedConsumer: TracedRabbitConsumer[F, String] = consumer + +consumer.records.evalMap { envelope => + envelope.processTraced(handle(envelope)) + // equivalent to consumer.process(envelope)(handle(envelope)) +} +``` + +`recordsWithProcess` is usually simpler when the whole `evalMap` operation should +be traced. + +## Span model + +| Operation | Span name | Kind | Parent and links | +| --- | --- | --- | --- | +| Publish a message without creation context | `publish ` | `PRODUCER` | The ambient span is the parent. A new creation context is injected into the AMQP headers. | +| Publish a message with creation context | `publish ` | `CLIENT` | The ambient span is the parent. The message creation context is preserved and linked. | +| Process a delivered message | `process ` | `CONSUMER` | Always a root span. The message creation context is linked when present. | + +A message published without an existing creation context gets a `PRODUCER` span, +and that span's context is injected into the message headers. If the message +already contains valid trace context, the headers are preserved and publishing +gets a `CLIENT` span linked to that context. + +Processing creates one root `CONSUMER` span per delivered message and links it to +the creation context in the message headers. RabbitMQ pushes messages to consumers, +so the module emits `process` spans rather than `receive` spans. + +The destination is a colon-separated combination of exchange, routing key, and, +for processing, queue name, with empty components removed. The default exchange +and an empty routing key are represented as `amq.default`. + +## Semantic attributes + +Every span has `messaging.system=rabbitmq`, `messaging.operation.name`, +`messaging.operation.type`, and `messaging.destination.name`. When applicable it +also has: + +- `messaging.rabbitmq.destination.routing_key` +- `messaging.message.id` and `messaging.message.conversation_id` +- `messaging.rabbitmq.message.delivery_tag` on process spans +- `messaging.destination.anonymous=true` for generated queue names +- configured `messaging.client.id`, `server.address`, and `server.port` + +Links repeat the destination, routing key, message id, and, for processing, the +delivery tag. Body and envelope size attributes are omitted because the public +fs2-rabbit APIs do not expose a reliable encoded envelope size at every +instrumented boundary. + +## Customization + +The defaults follow the current OpenTelemetry RabbitMQ messaging semantic +conventions. Span names, extra attributes, and finalization behavior can be +customized with `RabbitTracer.Config.withPublishSpanSetup` and +`withProcessSpanSetup`. Use `RabbitTracer.noop` when tracing is disabled. + +Do not enable overlapping RabbitMQ Java-agent instrumentation for the same client; +doing so can create duplicate messaging spans. diff --git a/site/src/main/resources/microsite/data/menu.yml b/site/src/main/resources/microsite/data/menu.yml index f842810f..f4897106 100644 --- a/site/src/main/resources/microsite/data/menu.yml +++ b/site/src/main/resources/microsite/data/menu.yml @@ -24,6 +24,10 @@ options: url: queues.html menu_section: guide + - title: OpenTelemetry tracing + url: otel4s.html + menu_section: guide + - title: Consumers url: consumers/ menu_section: consumer