otel4s: introduce Telemetry

build.sbt

@@ -231,7 +231,8 @@ lazy val tests = crossProject(JVMPlatform, JSPlatform, NativePlatform)
   )
   .jvmSettings(
     Test / fork := true,
-    javaOptions += "-Dotel.service.name=SkunkTests"
+    javaOptions += "-Dotel.service.name=SkunkTests",
+    libraryDependencies += "org.typelevel" %% "otel4s-sdk-testkit" % otel4sSdkVersion % Test,
   )
   .jsSettings(
     scalaJSLinkerConfig ~= { _.withESFeatures(_.withESVersion(org.scalajs.linker.interface.ESVersion.ES2018)) },

modules/core/shared/src/main/scala/Command.scala

@@ -5,6 +5,7 @@
 package skunk
 
 import cats.Contravariant
+import org.typelevel.otel4s.{Attribute, Attributes}
 import org.typelevel.twiddles.Iso
 import skunk.util.Origin
 import skunk.util.Twiddler
@@ -34,16 +35,32 @@ import skunk.util.Twiddler
 final case class Command[A](
   override val sql:     String,
   override val origin:  Origin,
-  override val encoder: Encoder[A]
+  override val encoder: Encoder[A],
+  override val telemetry: Statement.Telemetry = Statement.Telemetry.empty,
 ) extends Statement[A] {
 
+  /** Attaches a low-cardinality summary used as `db.query.summary` and as the span name. Statement
+    * summaries take precedence over summaries returned by a configured query analyzer. The
+    * method is a shortcut for `addAttributes(DbAttributes.DbQuerySummary(summary))`.
+    */
+  def withQuerySummary(summary: String): Command[A] =
+    copy(telemetry = telemetry.withQuerySummary(summary))
+
+  /** Replaces the additional attributes exported on the logical database span. */
+  def withAttributes(attributes: Attributes): Command[A] =
+    copy(telemetry = telemetry.withAttributes(attributes))
+
+  /** Adds or replaces additional logical database span attributes by key. */
+  def addAttributes(attributes: Attribute[_]*): Command[A] =
+    copy(telemetry = telemetry.addAttributes(attributes: _*))
+
   /**
    * Command is a [[https://typelevel.org/cats/typeclasses/contravariant.html contravariant
    * functor]].
    * @group Transformations
    */
   def contramap[B](f: B => A): Command[B] =
-    Command(sql, origin, encoder.contramap(f))
+    Command(sql, origin, encoder.contramap(f), telemetry)
 
   @deprecated("Use .to[CaseClass] instead of .gcontramap[CaseClass]", "0.6")
   def gcontramap[B](implicit ev: Twiddler.Aux[B, A]): Command[B] =

modules/core/shared/src/main/scala/Query.scala

@@ -5,6 +5,7 @@
 package skunk
 
 import cats.arrow.Profunctor
+import org.typelevel.otel4s.{Attribute, Attributes}
 import org.typelevel.twiddles.Iso
 import skunk.util.Origin
 import skunk.util.Twiddler
@@ -40,15 +41,31 @@ final case class Query[A, B](
   override val origin:  Origin,
   override val encoder: Encoder[A],
   decoder: Decoder[B],
-  isDynamic: Boolean = false
+  isDynamic: Boolean = false,
+  override val telemetry: Statement.Telemetry = Statement.Telemetry.empty,
 ) extends Statement[A] {
 
+  /** Attaches a low-cardinality summary used as `db.query.summary` and as the span name. Statement
+    * summaries take precedence over summaries returned by a configured query analyzer. The
+    * method is a shortcut for `addAttributes(DbAttributes.DbQuerySummary(summary))`.
+    */
+  def withQuerySummary(summary: String): Query[A, B] =
+    copy(telemetry = telemetry.withQuerySummary(summary))
+
+  /** Replaces the additional attributes exported on the logical database span. */
+  def withAttributes(attributes: Attributes): Query[A, B] =
+    copy(telemetry = telemetry.withAttributes(attributes))
+
+  /** Adds or replaces additional logical database span attributes by key. */
+  def addAttributes(attributes: Attribute[_]*): Query[A, B] =
+    copy(telemetry = telemetry.addAttributes(attributes: _*))
+
   /**
    * Query is a profunctor.
    * @group Transformations
    */
   def dimap[C, D](f: C => A)(g: B => D): Query[C, D] =
-    Query(sql, origin, encoder.contramap(f), decoder.map(g), isDynamic)
+    Query(sql, origin, encoder.contramap(f), decoder.map(g), isDynamic, telemetry)
 
   /**
    * Query is a contravariant functor in `A`.

modules/core/shared/src/main/scala/Session.scala

@@ -8,15 +8,12 @@ import cats._
 import cats.effect._
 import cats.effect.std.Console
 import cats.syntax.all._
-import com.comcast.ip4s.*
+import com.comcast.ip4s._
 import fs2.concurrent.Signal
-import fs2.io.net.{ Network, Socket, SocketOption }
+import fs2.io.net.{Network, Socket, SocketOption}
 import fs2.Pipe
 import fs2.Stream
-import org.typelevel.otel4s.metrics.Meter
 import org.typelevel.otel4s.metrics.MeterProvider
-import org.typelevel.otel4s.metrics.Histogram
-import org.typelevel.otel4s.trace.Tracer
 import org.typelevel.otel4s.trace.TracerProvider
 import skunk.codec.all.bool
 import skunk.data._
@@ -27,6 +24,7 @@ import skunk.net.SSLNegotiation
 import skunk.net.protocol.Describe
 import scala.concurrent.duration.Duration
 import skunk.net.protocol.Parse
+import skunk.telemetry.{ConnectionInfo, Telemetry, TelemetryConfig}
 
 /**
  * Represents a live connection to a Postgres database. Operations provided here are safe to use
@@ -491,6 +489,7 @@ object Session {
     val debug: Boolean,
     val typingStrategy: TypingStrategy,
     val redactionStrategy: RedactionStrategy,
+    val telemetryConfig: TelemetryConfig,
     val ssl: SSL,
     val connectionParameters: Map[String, String],
     val socketOptions: List[SocketOption],
@@ -511,6 +510,7 @@ object Session {
       debug: Boolean = self.debug,
       typingStrategy: TypingStrategy = self.typingStrategy,
       redactionStrategy: RedactionStrategy = self.redactionStrategy,
+      telemetryConfig: TelemetryConfig = self.telemetryConfig,
       ssl: SSL = self.ssl,
       connectionParameters: Map[String, String] = self.connectionParameters,
       socketOptions: List[SocketOption] = self.socketOptions,
@@ -519,7 +519,7 @@ object Session {
       queryCacheSize: Int = self.queryCacheSize,
       parseCacheSize: Int = self.parseCacheSize,
     ): Builder[F] =
-      new Builder(connectionType, host, port, unixSocketAddress, unixSocketDirectory, credentials, database, debug, typingStrategy, redactionStrategy, ssl, connectionParameters, socketOptions, readTimeout, commandCacheSize, queryCacheSize, parseCacheSize)
+      new Builder(connectionType, host, port, unixSocketAddress, unixSocketDirectory, credentials, database, debug, typingStrategy, redactionStrategy, telemetryConfig, ssl, connectionParameters, socketOptions, readTimeout, commandCacheSize, queryCacheSize, parseCacheSize)
 
     /** Configures the connection type. */
     def withConnectionType(newConnectionType: ConnectionType): Builder[F] =
@@ -588,6 +588,10 @@ object Session {
     def withRedactionStrategy(newRedactionStrategy: RedactionStrategy): Builder[F] =
       copy(redactionStrategy = newRedactionStrategy)
 
+    /** Configures query capture, query analysis, protocol spans, and pool spans. */
+    def withTelemetryConfig(newTelemetryConfig: TelemetryConfig): Builder[F] =
+      copy(telemetryConfig = newTelemetryConfig)
+
     def withSSL(newSSL: SSL): Builder[F] =
       copy(ssl = newSSL)
 
@@ -633,26 +637,23 @@ object Session {
      */
     def pooled(max: Int): Resource[F, Resource[F, Session[F]]] =
       for {
-        tracer <- Resource.eval(TracerProvider[F].tracer("org.typelevel.skunk").withVersion(BuildInfo.version).get)
-        meter  <- Resource.eval(MeterProvider[F].meter("org.typelevel.skunk").withVersion(BuildInfo.version).get)
-        pool   <- pooledWithTracer(max)(meter)
-      } yield pool(tracer)
+        telemetry <- Resource.eval(Telemetry.create(telemetryConfig, connectionInfo(database.getOrElse(""))))
+        pool      <- pooledWithTelemetry(max)
+      } yield pool(telemetry)
 
-    private def pooledWithTracer(max: Int)(implicit M: Meter[F]): Resource[F, Tracer[F] => Resource[F, Session[F]]] = {
+    private def pooledWithTelemetry(max: Int): Resource[F, Telemetry[F] => Resource[F, Session[F]]] = {
       val logger: String => F[Unit] = s => Console[F].println(s"TLS: $s")
       for {
         dc      <- Resource.eval(Describe.Cache.empty[F](commandCacheSize, queryCacheSize))
         sslOp   <- ssl.toSSLNegotiationOptions(if (debug) logger.some else none)
-        opDuration <- Resource.eval(Otel.OpDurationHistogram[F])
-        pool    <- Pool.ofF({implicit T: Tracer[F] => sessions(sslOp, dc, opDuration)}, max)(Recyclers.full)
+        pool    <- Pool.ofF({implicit T: Telemetry[F] => sessions(sslOp, dc)}, max)(Recyclers.full)
       } yield pool
     }
 
     private def sessions(
       sslOptions:    Option[SSLNegotiation.Options[F]],
       describeCache: Describe.Cache[F],
-      opDuration:    Histogram[F, Double]
-    )(implicit T: Tracer[F]): Resource[F, Session[F]] = {
+    )(implicit T: Telemetry[F]): Resource[F, Session[F]] = {
       val sockets = connectionType match {
         case ConnectionType.TCP =>
           val address = SocketAddress(host, port)
@@ -663,23 +664,45 @@ object Session {
           val filteredSocketOptions = socketOptions.filter(o => o.key != SocketOption.NoDelay)
           Network[F].connect(address, filteredSocketOptions)
       }
-      fromSockets(sockets, sslOptions, describeCache, opDuration)
+
+      for {
+        resolvedCredentials <- Resource.eval(credentials)
+        telemetry = T.withConnection(connectionInfo(database.getOrElse(resolvedCredentials.user)))
+        session <- fromSockets(sockets, sslOptions, describeCache, resolvedCredentials)(telemetry)
+      } yield session
     }
 
     private def fromSockets(
       sockets:           Resource[F, Socket[F]],
       sslOptions:        Option[SSLNegotiation.Options[F]],
       describeCache:     Describe.Cache[F],
-      opDuration:        Histogram[F, Double]
-    )(implicit T: Tracer[F]): Resource[F, Session[F]] =
+      creds:             Credentials,
+    )(implicit T: Telemetry[F]): Resource[F, Session[F]] =
        for {
         namer <- Resource.eval(Namer[F])
         pc    <- Resource.eval(Parse.Cache.empty[F](parseCacheSize))
-        proto <- Protocol[F](debug, namer, sockets, sslOptions, describeCache, pc, readTimeout, redactionStrategy, opDuration)
-        creds <- Resource.eval(credentials)
+        proto <- Protocol[F](debug, namer, sockets, sslOptions, describeCache, pc, readTimeout, redactionStrategy)
         _     <- Resource.eval(proto.startup(creds.user, database.getOrElse(creds.user), creds.password, connectionParameters))
         sess  <- Resource.make(fromProtocol(proto, namer, typingStrategy, redactionStrategy))(_ => proto.cleanup)
       } yield sess
+
+
+    private def connectionInfo(databaseName: String): ConnectionInfo =
+      connectionType match {
+        case ConnectionType.TCP =>
+          ConnectionInfo(
+            databaseName,
+            host.toString,
+            Option.when(port.value != 5432)(port.value.toLong),
+          )
+        case ConnectionType.Unix =>
+          ConnectionInfo(
+            databaseName,
+            unixSocketAddress.fold(s"$unixSocketDirectory/.s.PGSQL.$port")(_.path),
+            None,
+          )
+      }
+
   }
 
   /**
@@ -714,6 +737,7 @@ object Session {
         debug = false,
         typingStrategy = TypingStrategy.BuiltinsOnly,
         redactionStrategy = RedactionStrategy.OptIn,
+        telemetryConfig = TelemetryConfig.default,
         ssl = SSL.None,
         connectionParameters = DefaultConnectionParameters,
         socketOptions = DefaultSocketOptions,

modules/core/shared/src/main/scala/Statement.scala

@@ -4,6 +4,8 @@
 
 package skunk
 
+import org.typelevel.otel4s.{Attribute, Attributes}
+import org.typelevel.otel4s.semconv.attributes.DbAttributes
 import skunk.util.Origin
 import skunk.data.Type
 
@@ -12,15 +14,54 @@ trait Statement[A] {
   def origin:   Origin
   def encoder:  Encoder[A]
   def cacheKey: Statement.CacheKey
+  def telemetry: Statement.Telemetry
 }
 
 object Statement {
 
+  /** Explicit, typed telemetry metadata attached to a statement. */
+  sealed trait Telemetry {
+    /** The low-cardinality query summary, if defined. */
+    def querySummary: Option[String]
+
+    /** Additional attributes exported on the logical database span. */
+    def attributes: Attributes
+
+    /** Shortcut for `addAttributes(DbAttributes.DbQuerySummary(summary))`. */
+    def withQuerySummary(summary: String): Telemetry
+
+    /** Replaces the additional logical database span attributes. */
+    def withAttributes(attributes: Attributes): Telemetry
+
+    /** Adds or replaces additional logical database span attributes by key. */
+    def addAttributes(attributes: Attribute[_]*): Telemetry
+  }
+
+  object Telemetry {
+    val empty: Telemetry = Impl(Attributes.empty)
+
+    private final case class Impl(attributes: Attributes) extends Telemetry {
+      def querySummary: Option[String] =
+        attributes.get(DbAttributes.DbQuerySummary).map(_.value)
+
+      def withQuerySummary(summary: String): Telemetry =
+        copy(attributes = attributes.added(DbAttributes.DbQuerySummary(summary)))
+
+      def withAttributes(attributes: Attributes): Telemetry =
+        copy(attributes = attributes)
+
+      def addAttributes(values: Attribute[_]*): Telemetry =
+        withAttributes(attributes ++ values)
+
+      override def toString: String = s"Telemetry($attributes)"
+    }
+  }
+
   /**
    * A digest of a `Statement`, consisting only of the SQL statement and asserted input/output
    * types. This data type has lawful universal equality/hashing so we can use it as a hash key,
    * which we do internally. There is probably little use for this in end-user code.
    */
   final case class CacheKey(sql: String, encodedTypes: List[Type], decodedTypes: List[Type])
 
-}
+}