diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,142 @@ # Revision history for pgmq-effectful +## 0.2.0.0 -- 2026-04-23 + +### Breaking Changes + +* **OpenTelemetry semantic conventions updated to spec v1.24.** The + traced interpreter now emits attribute names compliant with + OpenTelemetry Semantic Conventions + [v1.24](https://github.com/open-telemetry/semantic-conventions/tree/v1.24.0), + sourced as typed `AttributeKey` values from the + [`hs-opentelemetry-semantic-conventions`](https://hackage.haskell.org/package/hs-opentelemetry-semantic-conventions) + library. Breaking consequences: + + * `messaging.operation.type` → `messaging.operation`. Values change + from the post-v1.24 vocabulary (`"send"`, `"receive"`) to the + v1.24 vocabulary (`"publish"`, `"receive"`). Note the verb change: + `send` → `publish`. + * `db.operation.name` → `db.operation`. Values are the pgmq SQL + function name (`"pgmq.send"`, `"pgmq.read"`, `"pgmq.archive"`, …), + not the previous free-form label. + * `messaging.destination.routing_key` (ad-hoc, not in v1.24) is + gone. Topic-send operations now emit the routing key as + `messaging.destination.name`, since for pgmq topics the routing + key is the logical destination. `ValidateRoutingKey`, + `ValidateTopicPattern`, and `TestRouting` no longer emit a + routing-key attribute at all. + * Span names changed from `"pgmq "` to the v1.24 + `" "` form. Examples: + `"pgmq send"` → `"publish my-queue"`, + `"pgmq read"` → `"receive my-queue"`, + `"pgmq archive"` → `"pgmq.archive my-queue"`, + `"pgmq list_queues"` → `"pgmq.list_queues"`. + * Queue management operations (`createQueue`, `dropQueue`, + `createPartitionedQueue`, `createUnloggedQueue`) moved from + `Producer` span kind to `Internal`. v1.24 reserves `Producer` for + message publishes; queue administration is not a publish. + * Span status on failure now carries a short non-PII label + (`"pool.acquisition_timeout"`, `"pool.connection.networking"`, + `"pool.session.statement"`, …) instead of the raw `show` of + `UsageError` (which bakes in SQL text and parameter values). + Full detail remains on the standard `exception` event via + `OpenTelemetry.Trace.Core.recordException`. + + Dashboards, saved queries, and alerts keyed on the old attribute + names or span-name format need to be updated. + +* **Trace context propagation now uses the tracer provider's + configured propagator** (W3C by default, but B3 / Datadog / … work + out of the box when the provider is configured with them). The old + implementation hard-wired `hs-opentelemetry-propagator-w3c`. + + * `Pgmq.Effectful.Telemetry.injectTraceContext` / `extractTraceContext` + take a `TracerProvider` (previously an `OTel.Span` / raw carrier). + * `Pgmq.Effectful.Telemetry.TraceHeaders` is now + `Network.HTTP.Types.RequestHeaders` (case-insensitive header names), + matching the carrier type every propagator uses. The previous + `[(ByteString, ByteString)]` alias is gone. + * `Pgmq.Effectful.Traced.sendMessageTraced` takes a `TracerProvider`. + * `Pgmq.Effectful.Traced.readMessageWithContext` takes a + `TracerProvider` and now returns + `Vector (Message, OpenTelemetry.Context.Context)`. + Callers that specifically need the raw `SpanContext` can recover + it via `Context.lookupSpan >>= getSpanContext`. + * Two new helpers, `traceHeadersToJson` and `jsonToTraceHeaders`, + handle the pgmq-over-jsonb serialization boundary. + + The `hs-opentelemetry-propagator-w3c` dependency was dropped from + the library; the SDK still installs the W3C propagator as the + default. + +* Renamed the interpreter error type from `PgmqError` to + `PgmqRuntimeError` and replaced its opaque `PgmqPoolError UsageError` + constructor with three structured constructors: + + data PgmqRuntimeError + = PgmqAcquisitionTimeout + | PgmqConnectionError Hasql.Errors.ConnectionError + | PgmqSessionError Hasql.Errors.SessionError + + The old `PgmqError`/`PgmqPoolError` names are retained with a + DEPRECATED pragma and will be removed in 0.3.0.0. + +* `runPgmq`'s error constraint changed from `Error PgmqError :> es` to + `Error PgmqRuntimeError :> es`. Update any + `runError @PgmqError` annotation to `runError @PgmqRuntimeError`. + +* `runPgmqTraced` and `runPgmqTracedWith` now require + `Error PgmqRuntimeError :> es`. Previously they had *no* error + constraint and threw a `fail`-derived `IOError` outside the Error + effect channel, which meant any `runError` wrapper around a traced + program was a no-op. Code that relied on that silent swallowing now + receives typed errors; update call sites to wrap with + `runError @PgmqRuntimeError`. + +### New Features + +* `fromUsageError :: Hasql.Pool.UsageError -> PgmqRuntimeError` — + convert raw hasql-pool errors into the pgmq-effectful error type. + Useful when layering `pgmq-effectful` over code that already calls + `Pool.use` directly. + +* `isTransient :: PgmqRuntimeError -> Bool` — classification helper for + retry logic. Returns True for acquisition timeouts, networking + connection errors, unrecognized libpq connection errors, and + session-level connection drops; False for authentication, + compatibility, missing-types, statement, script, and driver errors. + +* New test suite `pgmq-effectful-test` asserts that both interpreters + surface typed `PgmqRuntimeError` values through the Error channel. + +### Migration Guide + +Before: + + import Pgmq.Effectful (PgmqError (..), runPgmq) + + handler = + runEff . runError @PgmqError . runPgmq pool $ action + +After: + + import Pgmq.Effectful (PgmqRuntimeError (..), runPgmq) + + handler = + runEff . runError @PgmqRuntimeError . runPgmq pool $ action + +For retry logic: + + import Pgmq.Effectful (isTransient) + + retryIfTransient action = do + result <- runError @PgmqRuntimeError action + case result of + Right a -> pure (Right a) + Left (_, err) + | isTransient err -> retryIfTransient action + | otherwise -> pure (Left err) + ## 0.1.3.0 -- 2026-03-12 ### Other Changes diff --git a/pgmq-effectful.cabal b/pgmq-effectful.cabal --- a/pgmq-effectful.cabal +++ b/pgmq-effectful.cabal @@ -1,6 +1,6 @@ cabal-version: 3.4 name: pgmq-effectful -version: 0.1.3.0 +version: 0.2.0.0 synopsis: Effectful effects for PGMQ (PostgreSQL Message Queue) description: Effectful effects and interpreters for pgmq-hs, a Haskell client @@ -47,20 +47,65 @@ TypeOperators build-depends: - , aeson >=2.0 && <2.3 - , base >=4.18 && <5 - , bytestring >=0.11 && <0.13 - , effectful-core ^>=2.5 || ^>=2.6 - , hasql ^>=1.10 - , hasql-pool ^>=1.4 - , hs-opentelemetry-api >=0.2 && <0.4 - , hs-opentelemetry-propagator-w3c >=0.0 && <0.2 - , pgmq-core >=0.1.3 && <0.2 - , pgmq-hasql >=0.1.3 && <0.2 - , text >=2.0 && <2.2 - , unliftio >=0.2 && <0.3 - , unordered-containers >=0.2 && <0.3 - , vector ^>=0.13 + , aeson >=2.0 && <2.3 + , base >=4.18 && <5 + , bytestring >=0.11 && <0.13 + , case-insensitive >=1.2 && <1.3 + , effectful-core ^>=2.5 || ^>=2.6 + , hasql ^>=1.10 + , hasql-pool ^>=1.4 + , hs-opentelemetry-api >=0.2 && <0.4 + , hs-opentelemetry-semantic-conventions >=0.1 && <0.2 + , http-types >=0.12 && <0.13 + , pgmq-core >=0.2 && <0.3 + , pgmq-hasql >=0.2 && <0.3 + , text >=2.0 && <2.2 + , unliftio >=0.2 && <0.3 + , unordered-containers >=0.2 && <0.3 + , vector ^>=0.13 hs-source-dirs: src default-language: GHC2024 + +test-suite pgmq-effectful-test + import: warnings + default-language: GHC2024 + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: Main.hs + ghc-options: -threaded -rtsopts -with-rtsopts=-N + other-modules: + ClassificationSpec + EphemeralDb + PlainInterpreterSpec + TracedInterpreterSpec + + default-extensions: + DataKinds + ImportQualifiedPost + LambdaCase + OverloadedRecordDot + OverloadedStrings + TypeOperators + + build-depends: + , aeson >=2.0 && <2.3 + , base >=4.18 && <5 + , effectful-core ^>=2.5 || ^>=2.6 + , ephemeral-pg >=0.2.1 + , hasql ^>=1.10 + , hasql-pool ^>=1.4 + , hs-opentelemetry-api >=0.2 && <0.4 + , hs-opentelemetry-exporter-in-memory >=0.0 && <0.1 + , hs-opentelemetry-propagator-w3c >=0.0 && <0.2 + , hs-opentelemetry-sdk >=0.0 && <0.2 + , hs-opentelemetry-semantic-conventions >=0.1 && <0.2 + , pgmq-core >=0.2 && <0.3 + , pgmq-effectful + , pgmq-migration >=0.2 && <0.3 + , random ^>=1.2 + , tasty ^>=1.5 + , tasty-hunit ^>=0.10 + , text >=2.0 && <2.2 + , unordered-containers >=0.2 && <0.3 + , vector ^>=0.13 diff --git a/src/Pgmq/Effectful.hs b/src/Pgmq/Effectful.hs --- a/src/Pgmq/Effectful.hs +++ b/src/Pgmq/Effectful.hs @@ -4,7 +4,6 @@ -- * Interpreters runPgmq, - PgmqError (..), -- ** Traced Interpreters runPgmqTraced, @@ -12,6 +11,14 @@ TracingConfig (..), defaultTracingConfig, + -- * Errors + PgmqRuntimeError (..), + fromUsageError, + isTransient, + + -- ** Deprecated Error Types + PgmqError (..), + -- ** Traced Operations sendMessageTraced, readMessageWithContext, @@ -185,7 +192,13 @@ validateRoutingKey, validateTopicPattern, ) -import Pgmq.Effectful.Interpreter (PgmqError (..), runPgmq) +import Pgmq.Effectful.Interpreter + ( PgmqError (..), + PgmqRuntimeError (..), + fromUsageError, + isTransient, + runPgmq, + ) import Pgmq.Effectful.Interpreter.Traced ( TracingConfig (..), defaultTracingConfig, diff --git a/src/Pgmq/Effectful/Interpreter.hs b/src/Pgmq/Effectful/Interpreter.hs --- a/src/Pgmq/Effectful/Interpreter.hs +++ b/src/Pgmq/Effectful/Interpreter.hs @@ -3,28 +3,90 @@ runPgmq, -- * Error Types + PgmqRuntimeError (..), + fromUsageError, + isTransient, + + -- * Legacy Error Types (deprecated; will be removed in 0.3.0) PgmqError (..), ) where +import Control.Exception (Exception) import Effectful (Eff, IOE, (:>)) import Effectful qualified import Effectful.Dispatch.Dynamic (interpret) import Effectful.Error.Static (Error, throwError) +import GHC.Generics (Generic) +import Hasql.Errors qualified as HasqlErrors import Hasql.Pool (Pool, UsageError) import Hasql.Pool qualified as Pool import Hasql.Session qualified import Pgmq.Effectful.Effect (Pgmq (..)) import Pgmq.Hasql.Sessions qualified as Sessions --- | Error type for pgmq operations. +-- | Structured runtime error for pgmq-effectful operations. +-- +-- Constructors mirror 'Hasql.Pool.UsageError' so the mapping is direct. +-- The inner 'HasqlErrors.ConnectionError' and 'HasqlErrors.SessionError' +-- types are sum types with detailed constructors; see the Hasql +-- documentation for their full shapes. +data PgmqRuntimeError + = -- | Timed out waiting for a connection from the pool. + PgmqAcquisitionTimeout + | -- | Failed to establish a connection to PostgreSQL. + PgmqConnectionError HasqlErrors.ConnectionError + | -- | Error during session execution (SQL statement, decoding, etc.). + PgmqSessionError HasqlErrors.SessionError + deriving stock (Show, Eq, Generic) + +instance Exception PgmqRuntimeError + +-- | Convert hasql-pool's 'UsageError' into the structured +-- 'PgmqRuntimeError'. +fromUsageError :: UsageError -> PgmqRuntimeError +fromUsageError = \case + Pool.AcquisitionTimeoutUsageError -> PgmqAcquisitionTimeout + Pool.ConnectionUsageError e -> PgmqConnectionError e + Pool.SessionUsageError e -> PgmqSessionError e + +-- | Is this error plausibly transient — i.e., worth retrying? +-- +-- Returns 'True' for acquisition timeouts, networking connection errors, +-- uncategorized libpq connection errors, and session-level connection +-- drops. All other errors (authentication failure, compatibility +-- mismatches, missing types, statement errors, driver bugs) are treated +-- as permanent. +-- +-- Note: 'HasqlErrors.OtherConnectionError' is classed as transient here +-- despite hasql's documentation calling it \"not transient by default\", +-- because it is the catch-all for unrecognized libpq errors and classing +-- the unknown as transient errs toward letting retries happen. +isTransient :: PgmqRuntimeError -> Bool +isTransient = \case + PgmqAcquisitionTimeout -> True + PgmqConnectionError e -> case e of + HasqlErrors.NetworkingConnectionError _ -> True + HasqlErrors.AuthenticationConnectionError _ -> False + HasqlErrors.CompatibilityConnectionError _ -> False + HasqlErrors.OtherConnectionError _ -> True + PgmqSessionError e -> case e of + HasqlErrors.ConnectionSessionError _ -> True + HasqlErrors.StatementSessionError {} -> False + HasqlErrors.ScriptSessionError {} -> False + HasqlErrors.MissingTypesSessionError _ -> False + HasqlErrors.DriverSessionError _ -> False + +-- | Legacy error type. Retained for one release cycle; migrate to +-- 'PgmqRuntimeError'. Will be removed in pgmq-effectful 0.3.0. newtype PgmqError = PgmqPoolError UsageError deriving stock (Show) +{-# DEPRECATED PgmqError, PgmqPoolError "Use PgmqRuntimeError instead (available from Pgmq.Effectful). The legacy types will be removed in pgmq-effectful 0.3.0." #-} -- | Run the Pgmq effect using a connection pool. --- Errors are thrown via the 'Error' effect. +-- Errors are thrown via the 'Error' effect as 'PgmqRuntimeError'. runPgmq :: - (IOE :> es, Error PgmqError :> es) => + (IOE :> es, Error PgmqRuntimeError :> es) => Pool -> Eff (Pgmq : es) a -> Eff es a @@ -92,12 +154,12 @@ -- Internal helper runSession :: - (IOE :> es, Error PgmqError :> es) => + (IOE :> es, Error PgmqRuntimeError :> es) => Pool -> Hasql.Session.Session a -> Eff es a runSession pool session = do result <- Effectful.liftIO $ Pool.use pool session case result of - Left err -> throwError $ PgmqPoolError err + Left err -> throwError $ fromUsageError err Right a -> pure a diff --git a/src/Pgmq/Effectful/Interpreter/Traced.hs b/src/Pgmq/Effectful/Interpreter/Traced.hs --- a/src/Pgmq/Effectful/Interpreter/Traced.hs +++ b/src/Pgmq/Effectful/Interpreter/Traced.hs @@ -3,11 +3,46 @@ -- This module provides a traced version of the Pgmq interpreter that -- creates OpenTelemetry spans for all PGMQ operations. -- +-- == OpenTelemetry Semantic Conventions +-- +-- Spans emitted by this interpreter follow OpenTelemetry +-- [Semantic Conventions v1.24](https://github.com/open-telemetry/semantic-conventions/tree/v1.24.0) +-- for messaging clients and database clients. +-- +-- For every operation the interpreter emits: +-- +-- * @db.system = "postgresql"@ +-- * @db.operation = "pgmq."@ (for example @"pgmq.send"@, @"pgmq.read"@, +-- @"pgmq.archive"@). +-- +-- For /messaging/ operations (the publish and receive families) the +-- interpreter additionally emits: +-- +-- * @messaging.system = "pgmq"@ +-- * @messaging.operation@ — one of @"publish"@ (every @send*@ variant, +-- including topic sends) or @"receive"@ (every @read*@ and @pop@ +-- variant). The @"process"@ operation is intentionally not emitted +-- here — it belongs to the consumer, which should open its own +-- @process@ span around application-level handling of a received +-- message (see 'Pgmq.Effectful.Traced.readMessageWithContext'). +-- * @messaging.destination.name@ — the queue name, or for topic sends +-- the routing key (which is the logical destination for pgmq topics). +-- +-- == Span Names +-- +-- Span names follow the v1.24 @" "@ form: +-- +-- * Messaging spans: @"publish my-queue"@, @"receive my-queue"@. +-- * Lifecycle and observability spans: @"pgmq.archive my-queue"@, +-- @"pgmq.set_vt my-queue"@, @"pgmq.list_queues"@ (no destination). +-- -- == Span Kinds -- --- * Producer spans: Queue creation, message send operations --- * Consumer spans: Message read operations --- * Internal spans: Message lifecycle (delete, archive, visibility timeout) +-- * 'OTel.Producer': every @send*@ / @send_topic*@ variant. +-- * 'OTel.Consumer': every @read*@ variant and @pop@. +-- * 'OTel.Internal': lifecycle (create, drop, archive, delete, set_vt, +-- purge, bind_topic, …) and observability (metrics, list_queues, +-- list_topic_bindings, validate_*). -- -- == Usage -- @@ -22,7 +57,7 @@ -- pool <- ... -- -- runEff --- . runError @PgmqError +-- . runError @PgmqRuntimeError -- . runPgmqTraced pool tracer -- $ do -- createQueue "my-queue" @@ -40,18 +75,24 @@ where import Control.Monad (when) +import Data.Int (Int64) import Data.Text (Text) import Data.Text qualified as T import Effectful (Eff, IOE, (:>)) import Effectful qualified import Effectful.Dispatch.Dynamic (interpret) -import Hasql.Pool (Pool, UsageError) +import Effectful.Error.Static (Error, throwError) +import Hasql.Errors qualified as HasqlErrors +import Hasql.Pool (Pool) import Hasql.Pool qualified as Pool import Hasql.Session qualified -import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..)) -import OpenTelemetry.Attributes.Map qualified as AttrMap +import OpenTelemetry.Attributes.Map (AttributeMap, insertByKey) import OpenTelemetry.Trace.Core qualified as OTel import Pgmq.Effectful.Effect (Pgmq (..)) +import Pgmq.Effectful.Interpreter + ( PgmqRuntimeError (..), + fromUsageError, + ) import Pgmq.Effectful.Telemetry import Pgmq.Hasql.Sessions qualified as Sessions import Pgmq.Hasql.Statements.Types qualified as Types @@ -76,9 +117,44 @@ includeMessageBodies = False } +-- | Describes how a 'Pgmq' operation maps onto OpenTelemetry semantic +-- conventions v1.24. +data OpInfo = OpInfo + { -- | The pgmq SQL function being invoked (@"pgmq.send"@ etc.). + opDbFunction :: !Text, + -- | The v1.24 @messaging.operation@ verb, if any. One of + -- @"publish"@ or @"receive"@; @Nothing@ for lifecycle/observability + -- operations. + opMessagingKind :: !(Maybe Text), + -- | The span kind. + opSpanKind :: !OTel.SpanKind, + -- | The logical destination (queue name, or the routing key for + -- topic sends). @Nothing@ for operations without a per-queue scope + -- (for example @list_queues@, @metrics_all@). + opDestination :: !(Maybe Text), + -- | A pre-known message id (for per-message operations like delete, + -- archive, set_vt). + opMessageId :: !(Maybe MessageId), + -- | Batch size, for batch operations. + opBatchCount :: !(Maybe Int) + } + +-- | An 'OpInfo' with no messaging verb, no destination, no message id, +-- and no batch count. +defaultOpInfo :: Text -> OTel.SpanKind -> OpInfo +defaultOpInfo fn kind = + OpInfo + { opDbFunction = fn, + opMessagingKind = Nothing, + opSpanKind = kind, + opDestination = Nothing, + opMessageId = Nothing, + opBatchCount = Nothing + } + -- | Run the Pgmq effect with OpenTelemetry instrumentation. runPgmqTraced :: - (IOE :> es) => + (IOE :> es, Error PgmqRuntimeError :> es) => Pool -> OTel.Tracer -> Eff (Pgmq : es) a -> @@ -87,323 +163,319 @@ -- | Run the Pgmq effect with custom tracing configuration. runPgmqTracedWith :: - (IOE :> es) => + (IOE :> es, Error PgmqRuntimeError :> es) => Pool -> TracingConfig -> Eff (Pgmq : es) a -> Eff es a runPgmqTracedWith pool config = interpret $ \_ -> \case - -- Queue Management (Producer spans) + -- Queue Management (Internal spans) CreateQueue q -> - withTracedSession config "pgmq create_queue" OTel.Producer q pool $ + withTracedOp config pool (queueOp "pgmq.create" OTel.Internal q) $ Sessions.createQueue q DropQueue q -> - withTracedSession config "pgmq drop_queue" OTel.Producer q pool $ + withTracedOp config pool (queueOp "pgmq.drop_queue" OTel.Internal q) $ Sessions.dropQueue q CreatePartitionedQueue pq@(Types.CreatePartitionedQueue qn _ _) -> - withTracedSession config "pgmq create_partitioned_queue" OTel.Producer qn pool $ + withTracedOp config pool (queueOp "pgmq.create_partitioned" OTel.Internal qn) $ Sessions.createPartitionedQueue pq CreateUnloggedQueue q -> - withTracedSession config "pgmq create_unlogged_queue" OTel.Producer q pool $ + withTracedOp config pool (queueOp "pgmq.create_unlogged" OTel.Internal q) $ Sessions.createUnloggedQueue q DetachArchive _q -> pure () EnableNotifyInsert cfg@(Types.EnableNotifyInsert qn _) -> - withTracedSession config "pgmq enable_notify_insert" OTel.Internal qn pool $ + withTracedOp config pool (queueOp "pgmq.enable_notify_insert" OTel.Internal qn) $ Sessions.enableNotifyInsert cfg DisableNotifyInsert q -> - withTracedSession config "pgmq disable_notify_insert" OTel.Internal q pool $ + withTracedOp config pool (queueOp "pgmq.disable_notify_insert" OTel.Internal q) $ Sessions.disableNotifyInsert q CreateFifoIndex q -> - withTracedSession config "pgmq create_fifo_index" OTel.Internal q pool $ + withTracedOp config pool (queueOp "pgmq.create_fifo_index" OTel.Internal q) $ Sessions.createFifoIndex q CreateFifoIndexesAll -> - withTracedSessionNoQueue config "pgmq create_fifo_indexes_all" OTel.Internal pool $ + withTracedOp config pool (defaultOpInfo "pgmq.create_fifo_indexes_all" OTel.Internal) $ Sessions.createFifoIndexesAll - -- Message Operations - Send (Producer spans) + -- Message Operations - Send (Producer spans, messaging.operation=publish) SendMessage msg@(Types.SendMessage qn _ _) -> - withTracedSession config "pgmq send" OTel.Producer qn pool $ + withTracedOp config pool (publishOp "pgmq.send" qn) $ Sessions.sendMessage msg SendMessageForLater msg@(Types.SendMessageForLater qn _ _) -> - withTracedSession config "pgmq send" OTel.Producer qn pool $ + withTracedOp config pool (publishOp "pgmq.send" qn) $ Sessions.sendMessageForLater msg BatchSendMessage msg@(Types.BatchSendMessage qn bodies _) -> - withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $ + withTracedOp config pool (publishBatchOp "pgmq.send_batch" qn (length bodies)) $ Sessions.batchSendMessage msg BatchSendMessageForLater msg@(Types.BatchSendMessageForLater qn bodies _) -> - withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $ + withTracedOp config pool (publishBatchOp "pgmq.send_batch" qn (length bodies)) $ Sessions.batchSendMessageForLater msg SendMessageWithHeaders msg@(Types.SendMessageWithHeaders qn _ _ _) -> - withTracedSession config "pgmq send" OTel.Producer qn pool $ + withTracedOp config pool (publishOp "pgmq.send" qn) $ Sessions.sendMessageWithHeaders msg SendMessageWithHeadersForLater msg@(Types.SendMessageWithHeadersForLater qn _ _ _) -> - withTracedSession config "pgmq send" OTel.Producer qn pool $ + withTracedOp config pool (publishOp "pgmq.send" qn) $ Sessions.sendMessageWithHeadersForLater msg BatchSendMessageWithHeaders msg@(Types.BatchSendMessageWithHeaders qn bodies _ _) -> - withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $ + withTracedOp config pool (publishBatchOp "pgmq.send_batch" qn (length bodies)) $ Sessions.batchSendMessageWithHeaders msg BatchSendMessageWithHeadersForLater msg@(Types.BatchSendMessageWithHeadersForLater qn bodies _ _) -> - withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $ + withTracedOp config pool (publishBatchOp "pgmq.send_batch" qn (length bodies)) $ Sessions.batchSendMessageWithHeadersForLater msg - -- Message Operations - Read (Consumer spans) + -- Message Operations - Read (Consumer spans, messaging.operation=receive) ReadMessage query@(Types.ReadMessage qn _ _ _) -> - withTracedSession config "pgmq read" OTel.Consumer qn pool $ + withTracedOp config pool (receiveOp "pgmq.read" qn) $ Sessions.readMessage query ReadWithPoll query@(Types.ReadWithPollMessage qn _ _ _ _ _) -> - withTracedSession config "pgmq read_with_poll" OTel.Consumer qn pool $ + withTracedOp config pool (receiveOp "pgmq.read_with_poll" qn) $ Sessions.readWithPoll query Pop query@(Types.PopMessage qn _) -> - withTracedSession config "pgmq pop" OTel.Consumer qn pool $ + withTracedOp config pool (receiveOp "pgmq.pop" qn) $ Sessions.pop query -- FIFO Read (Consumer spans) ReadGrouped query@(Types.ReadGrouped qn _ _) -> - withTracedSession config "pgmq read_grouped" OTel.Consumer qn pool $ + withTracedOp config pool (receiveOp "pgmq.read_grouped" qn) $ Sessions.readGrouped query ReadGroupedWithPoll query@(Types.ReadGroupedWithPoll qn _ _ _ _) -> - withTracedSession config "pgmq read_grouped_with_poll" OTel.Consumer qn pool $ + withTracedOp config pool (receiveOp "pgmq.read_grouped_with_poll" qn) $ Sessions.readGroupedWithPoll query -- Round-robin FIFO Read (Consumer spans) ReadGroupedRoundRobin query@(Types.ReadGrouped qn _ _) -> - withTracedSession config "pgmq read_grouped_rr" OTel.Consumer qn pool $ + withTracedOp config pool (receiveOp "pgmq.read_grouped_rr" qn) $ Sessions.readGroupedRoundRobin query ReadGroupedRoundRobinWithPoll query@(Types.ReadGroupedWithPoll qn _ _ _ _) -> - withTracedSession config "pgmq read_grouped_rr_with_poll" OTel.Consumer qn pool $ + withTracedOp config pool (receiveOp "pgmq.read_grouped_rr_with_poll" qn) $ Sessions.readGroupedRoundRobinWithPoll query -- Message Lifecycle (Internal spans) DeleteMessage query@(Types.MessageQuery qn msgId) -> - withTracedSessionWithMsgId config "pgmq delete" OTel.Internal qn msgId pool $ + withTracedOp config pool ((queueOp "pgmq.delete" OTel.Internal qn) {opMessageId = Just msgId}) $ Sessions.deleteMessage query BatchDeleteMessages query@(Types.BatchMessageQuery qn msgIds) -> - withTracedSessionWithCount config "pgmq delete_batch" OTel.Internal qn (length msgIds) pool $ + withTracedOp config pool ((queueOp "pgmq.delete" OTel.Internal qn) {opBatchCount = Just (length msgIds)}) $ Sessions.batchDeleteMessages query ArchiveMessage query@(Types.MessageQuery qn msgId) -> - withTracedSessionWithMsgId config "pgmq archive" OTel.Internal qn msgId pool $ + withTracedOp config pool ((queueOp "pgmq.archive" OTel.Internal qn) {opMessageId = Just msgId}) $ Sessions.archiveMessage query BatchArchiveMessages query@(Types.BatchMessageQuery qn msgIds) -> - withTracedSessionWithCount config "pgmq archive_batch" OTel.Internal qn (length msgIds) pool $ + withTracedOp config pool ((queueOp "pgmq.archive" OTel.Internal qn) {opBatchCount = Just (length msgIds)}) $ Sessions.batchArchiveMessages query DeleteAllMessagesFromQueue q -> - withTracedSession config "pgmq purge_queue" OTel.Internal q pool $ + withTracedOp config pool (queueOp "pgmq.purge_queue" OTel.Internal q) $ Sessions.deleteAllMessagesFromQueue q ChangeVisibilityTimeout query@(Types.VisibilityTimeoutQuery qn msgId _) -> - withTracedSessionWithMsgId config "pgmq set_vt" OTel.Internal qn msgId pool $ + withTracedOp config pool ((queueOp "pgmq.set_vt" OTel.Internal qn) {opMessageId = Just msgId}) $ Sessions.changeVisibilityTimeout query BatchChangeVisibilityTimeout query@(Types.BatchVisibilityTimeoutQuery qn msgIds _) -> - withTracedSessionWithCount config "pgmq set_vt_batch" OTel.Internal qn (length msgIds) pool $ + withTracedOp config pool ((queueOp "pgmq.set_vt" OTel.Internal qn) {opBatchCount = Just (length msgIds)}) $ Sessions.batchChangeVisibilityTimeout query -- Timestamp-based VT (pgmq 1.10.0+) SetVisibilityTimeoutAt query@(Types.VisibilityTimeoutAtQuery qn msgId _) -> - withTracedSessionWithMsgId config "pgmq set_vt_at" OTel.Internal qn msgId pool $ + withTracedOp config pool ((queueOp "pgmq.set_vt" OTel.Internal qn) {opMessageId = Just msgId}) $ Sessions.setVisibilityTimeoutAt query BatchSetVisibilityTimeoutAt query@(Types.BatchVisibilityTimeoutAtQuery qn msgIds _) -> - withTracedSessionWithCount config "pgmq set_vt_at_batch" OTel.Internal qn (length msgIds) pool $ + withTracedOp config pool ((queueOp "pgmq.set_vt" OTel.Internal qn) {opBatchCount = Just (length msgIds)}) $ Sessions.batchSetVisibilityTimeoutAt query -- Topic Management (Internal spans, pgmq 1.11.0+) BindTopic params@(Types.BindTopic _ qn) -> - withTracedSession config "pgmq bind_topic" OTel.Internal qn pool $ + withTracedOp config pool (queueOp "pgmq.bind_topic" OTel.Internal qn) $ Sessions.bindTopic params UnbindTopic params@(Types.UnbindTopic _ qn) -> - withTracedSession config "pgmq unbind_topic" OTel.Internal qn pool $ + withTracedOp config pool (queueOp "pgmq.unbind_topic" OTel.Internal qn) $ Sessions.unbindTopic params ValidateRoutingKey key -> - withTracedSessionWithRoutingKey config "pgmq validate_routing_key" OTel.Internal key pool $ + withTracedOp config pool (defaultOpInfo "pgmq.validate_routing_key" OTel.Internal) $ Sessions.validateRoutingKey key - ValidateTopicPattern _pat -> - withTracedSessionNoQueue config "pgmq validate_topic_pattern" OTel.Internal pool $ - Sessions.validateTopicPattern _pat + ValidateTopicPattern pat -> + withTracedOp config pool (defaultOpInfo "pgmq.validate_topic_pattern" OTel.Internal) $ + Sessions.validateTopicPattern pat TestRouting key -> - withTracedSessionWithRoutingKey config "pgmq test_routing" OTel.Internal key pool $ + withTracedOp config pool (defaultOpInfo "pgmq.test_routing" OTel.Internal) $ Sessions.testRouting key ListTopicBindings -> - withTracedSessionNoQueue config "pgmq list_topic_bindings" OTel.Internal pool $ + withTracedOp config pool (defaultOpInfo "pgmq.list_topic_bindings" OTel.Internal) $ Sessions.listTopicBindings ListTopicBindingsForQueue q -> - withTracedSession config "pgmq list_topic_bindings" OTel.Internal q pool $ + withTracedOp config pool (queueOp "pgmq.list_topic_bindings" OTel.Internal q) $ Sessions.listTopicBindingsForQueue q -- Topic Sending (Producer spans, pgmq 1.11.0+) SendTopic msg@(Types.SendTopic rk _ _) -> - withTracedSessionWithRoutingKey config "pgmq send_topic" OTel.Producer rk pool $ + withTracedOp config pool (publishTopicOp "pgmq.send_topic" rk) $ Sessions.sendTopic msg SendTopicWithHeaders msg@(Types.SendTopicWithHeaders rk _ _ _) -> - withTracedSessionWithRoutingKey config "pgmq send_topic" OTel.Producer rk pool $ + withTracedOp config pool (publishTopicOp "pgmq.send_topic" rk) $ Sessions.sendTopicWithHeaders msg BatchSendTopic msg@(Types.BatchSendTopic rk bodies _) -> - withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $ + withTracedOp config pool (publishTopicBatchOp "pgmq.send_batch_topic" rk (length bodies)) $ Sessions.batchSendTopic msg BatchSendTopicForLater msg@(Types.BatchSendTopicForLater rk bodies _) -> - withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $ + withTracedOp config pool (publishTopicBatchOp "pgmq.send_batch_topic" rk (length bodies)) $ Sessions.batchSendTopicForLater msg BatchSendTopicWithHeaders msg@(Types.BatchSendTopicWithHeaders rk bodies _ _) -> - withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $ + withTracedOp config pool (publishTopicBatchOp "pgmq.send_batch_topic" rk (length bodies)) $ Sessions.batchSendTopicWithHeaders msg BatchSendTopicWithHeadersForLater msg@(Types.BatchSendTopicWithHeadersForLater rk bodies _ _) -> - withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $ + withTracedOp config pool (publishTopicBatchOp "pgmq.send_batch_topic" rk (length bodies)) $ Sessions.batchSendTopicWithHeadersForLater msg -- Notification Management (Internal spans, pgmq 1.11.0+) ListNotifyInsertThrottles -> - withTracedSessionNoQueue config "pgmq list_notify_insert_throttles" OTel.Internal pool $ + withTracedOp config pool (defaultOpInfo "pgmq.list_notify_insert_throttles" OTel.Internal) $ Sessions.listNotifyInsertThrottles UpdateNotifyInsert params@(Types.UpdateNotifyInsert qn _) -> - withTracedSession config "pgmq update_notify_insert" OTel.Internal qn pool $ + withTracedOp config pool (queueOp "pgmq.update_notify_insert" OTel.Internal qn) $ Sessions.updateNotifyInsert params -- Queue Observability (Internal spans) ListQueues -> - withTracedSessionNoQueue config "pgmq list_queues" OTel.Internal pool $ + withTracedOp config pool (defaultOpInfo "pgmq.list_queues" OTel.Internal) $ Sessions.listQueues QueueMetrics q -> - withTracedSession config "pgmq metrics" OTel.Internal q pool $ + withTracedOp config pool (queueOp "pgmq.metrics" OTel.Internal q) $ Sessions.queueMetrics q AllQueueMetrics -> - withTracedSessionNoQueue config "pgmq metrics_all" OTel.Internal pool $ + withTracedOp config pool (defaultOpInfo "pgmq.metrics_all" OTel.Internal) $ Sessions.allQueueMetrics --- Internal helpers +-- --------------------------------------------------------------------- +-- OpInfo constructors +-- --------------------------------------------------------------------- -withTracedSession :: - (IOE :> es) => - TracingConfig -> - Text -> - OTel.SpanKind -> - QueueName -> - Pool -> - Hasql.Session.Session a -> - Eff es a -withTracedSession config spanName kind queueName pool session = do - let args = OTel.defaultSpanArguments {OTel.kind = kind} - Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do - addQueueAttributes s queueName - runSessionIO config s pool session +-- | 'OpInfo' for a non-messaging operation scoped to a single queue. +queueOp :: Text -> OTel.SpanKind -> QueueName -> OpInfo +queueOp fn kind qn = (defaultOpInfo fn kind) {opDestination = Just (queueNameToText qn)} -withTracedSessionNoQueue :: - (IOE :> es) => - TracingConfig -> - Text -> - OTel.SpanKind -> - Pool -> - Hasql.Session.Session a -> - Eff es a -withTracedSessionNoQueue config spanName kind pool session = do - let args = OTel.defaultSpanArguments {OTel.kind = kind} - Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do - addBaseAttributes s - runSessionIO config s pool session +-- | 'OpInfo' for a @publish@ (send) on a queue. +publishOp :: Text -> QueueName -> OpInfo +publishOp fn qn = + (queueOp fn OTel.Producer qn) {opMessagingKind = Just "publish"} -withTracedSessionWithMsgId :: - (IOE :> es) => +-- | 'OpInfo' for a batch @publish@ on a queue. +publishBatchOp :: Text -> QueueName -> Int -> OpInfo +publishBatchOp fn qn n = (publishOp fn qn) {opBatchCount = Just n} + +-- | 'OpInfo' for a topic send (publish to a routing key). +publishTopicOp :: Text -> RoutingKey -> OpInfo +publishTopicOp fn rk = + (defaultOpInfo fn OTel.Producer) + { opMessagingKind = Just "publish", + opDestination = Just (routingKeyToText rk) + } + +-- | 'OpInfo' for a batch topic send. +publishTopicBatchOp :: Text -> RoutingKey -> Int -> OpInfo +publishTopicBatchOp fn rk n = (publishTopicOp fn rk) {opBatchCount = Just n} + +-- | 'OpInfo' for a @receive@ (read, pop) on a queue. +receiveOp :: Text -> QueueName -> OpInfo +receiveOp fn qn = + (queueOp fn OTel.Consumer qn) {opMessagingKind = Just "receive"} + +-- --------------------------------------------------------------------- +-- Span assembly +-- --------------------------------------------------------------------- + +-- | Span name: @" "@ for messaging +-- spans, else @" "@; if there is no +-- destination, the operation name alone. +spanNameFor :: OpInfo -> Text +spanNameFor info = + let op = case info.opMessagingKind of + Just mk -> mk + Nothing -> info.opDbFunction + in case info.opDestination of + Just d -> op <> " " <> d + Nothing -> op + +operationAttributes :: OpInfo -> AttributeMap +operationAttributes info = + let base = + insertByKey db_system ("postgresql" :: Text) + . insertByKey db_operation info.opDbFunction + withMsgKind = case info.opMessagingKind of + Just mk -> + insertByKey messaging_system ("pgmq" :: Text) + . insertByKey messaging_operation mk + Nothing -> id + withDest = case info.opDestination of + Just d -> insertByKey messaging_destination_name d + Nothing -> id + withMsgId = case info.opMessageId of + Just (MessageId mid) -> + insertByKey messaging_message_id (T.pack (show mid)) + Nothing -> id + withCount = case info.opBatchCount of + Just n -> insertByKey messaging_batch_messageCount (fromIntegral n :: Int64) + Nothing -> id + in (withMsgId . withCount . withDest . withMsgKind . base) mempty + +withTracedOp :: + (IOE :> es, Error PgmqRuntimeError :> es) => TracingConfig -> - Text -> - OTel.SpanKind -> - QueueName -> - MessageId -> Pool -> + OpInfo -> Hasql.Session.Session a -> Eff es a -withTracedSessionWithMsgId config spanName kind queueName msgId pool session = do - let args = OTel.defaultSpanArguments {OTel.kind = kind} - Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do - addQueueAttributes s queueName - addMessageIdAttribute s msgId - runSessionIO config s pool session +withTracedOp config pool info session = do + let args = + OTel.addAttributesToSpanArguments + (operationAttributes info) + OTel.defaultSpanArguments {OTel.kind = info.opSpanKind} + result <- + Effectful.liftIO $ + OTel.inSpan' config.tracer (spanNameFor info) args $ \s -> + runSessionIO config s pool session + throwOnLeft result -withTracedSessionWithCount :: - (IOE :> es) => - TracingConfig -> - Text -> - OTel.SpanKind -> - QueueName -> - Int -> - Pool -> - Hasql.Session.Session a -> +-- --------------------------------------------------------------------- +-- Session runner and error recording +-- --------------------------------------------------------------------- + +-- | Rethrow a runtime error via the 'Error' effect when present. +throwOnLeft :: + (Error PgmqRuntimeError :> es) => + Either PgmqRuntimeError a -> Eff es a -withTracedSessionWithCount config spanName kind queueName count pool session = do - let args = OTel.defaultSpanArguments {OTel.kind = kind} - Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do - addQueueAttributes s queueName - addBatchCountAttribute s count - runSessionIO config s pool session +throwOnLeft = \case + Left err -> throwError err + Right a -> pure a runSessionIO :: TracingConfig -> OTel.Span -> Pool -> Hasql.Session.Session a -> - IO a + IO (Either PgmqRuntimeError a) runSessionIO config s pool session = do result <- Pool.use pool session case result of Left err -> do - when config.recordExceptions $ recordUsageError s err - OTel.setStatus s (OTel.Error $ T.pack $ show err) - fail $ "PgmqPoolError: " <> show err + let runtimeErr = fromUsageError err + when config.recordExceptions $ + OTel.recordException s mempty Nothing err + OTel.setStatus s (OTel.Error (errorStatusDescription runtimeErr)) + pure $ Left runtimeErr Right a -> do OTel.setStatus s OTel.Ok - pure a - -recordUsageError :: OTel.Span -> UsageError -> IO () -recordUsageError s err = do - OTel.addEvent s $ - OTel.NewEvent - { OTel.newEventName = "exception", - OTel.newEventTimestamp = Nothing, - OTel.newEventAttributes = - AttrMap.fromList - [ ("exception.type", AttributeValue $ TextAttribute "UsageError"), - ("exception.message", AttributeValue $ TextAttribute $ T.pack $ show err) - ] - } - -addQueueAttributes :: OTel.Span -> QueueName -> IO () -addQueueAttributes s queueName = do - addBaseAttributes s - OTel.addAttribute s messagingDestinationName (queueNameToText queueName) - -addBaseAttributes :: OTel.Span -> IO () -addBaseAttributes s = do - OTel.addAttribute s messagingSystem ("pgmq" :: Text) - OTel.addAttribute s dbSystem ("postgresql" :: Text) - -addMessageIdAttribute :: OTel.Span -> MessageId -> IO () -addMessageIdAttribute s (MessageId msgId) = - OTel.addAttribute s messagingMessageId (T.pack $ show msgId) - -addBatchCountAttribute :: OTel.Span -> Int -> IO () -addBatchCountAttribute s count = - OTel.addAttribute s messagingBatchMessageCount count - -addRoutingKeyAttribute :: OTel.Span -> RoutingKey -> IO () -addRoutingKeyAttribute s rk = - OTel.addAttribute s messagingRoutingKey (routingKeyToText rk) - -withTracedSessionWithRoutingKey :: - (IOE :> es) => - TracingConfig -> - Text -> - OTel.SpanKind -> - RoutingKey -> - Pool -> - Hasql.Session.Session a -> - Eff es a -withTracedSessionWithRoutingKey config spanName kind routingKey pool session = do - let args = OTel.defaultSpanArguments {OTel.kind = kind} - Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do - addBaseAttributes s - addRoutingKeyAttribute s routingKey - runSessionIO config s pool session + pure $ Right a -withTracedSessionWithRoutingKeyAndCount :: - (IOE :> es) => - TracingConfig -> - Text -> - OTel.SpanKind -> - RoutingKey -> - Int -> - Pool -> - Hasql.Session.Session a -> - Eff es a -withTracedSessionWithRoutingKeyAndCount config spanName kind routingKey count pool session = do - let args = OTel.defaultSpanArguments {OTel.kind = kind} - Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do - addBaseAttributes s - addRoutingKeyAttribute s routingKey - addBatchCountAttribute s count - runSessionIO config s pool session +-- | Short, non-PII span-status description derived from a +-- 'PgmqRuntimeError'. The detail lives inside the 'recordException' +-- event; the span status description is a coarse label that backends +-- can group on without risking query text or credentials leaking into +-- dashboards. +errorStatusDescription :: PgmqRuntimeError -> Text +errorStatusDescription = \case + PgmqAcquisitionTimeout -> "pool.acquisition_timeout" + PgmqConnectionError e -> "pool.connection." <> connectionLabel e + PgmqSessionError e -> "pool.session." <> sessionLabel e + where + connectionLabel :: HasqlErrors.ConnectionError -> Text + connectionLabel = \case + HasqlErrors.NetworkingConnectionError _ -> "networking" + HasqlErrors.AuthenticationConnectionError _ -> "authentication" + HasqlErrors.CompatibilityConnectionError _ -> "compatibility" + HasqlErrors.OtherConnectionError _ -> "other" + sessionLabel :: HasqlErrors.SessionError -> Text + sessionLabel = \case + HasqlErrors.ConnectionSessionError _ -> "connection" + HasqlErrors.StatementSessionError {} -> "statement" + HasqlErrors.ScriptSessionError {} -> "script" + HasqlErrors.MissingTypesSessionError _ -> "missing_types" + HasqlErrors.DriverSessionError _ -> "driver" diff --git a/src/Pgmq/Effectful/Telemetry.hs b/src/Pgmq/Effectful/Telemetry.hs --- a/src/Pgmq/Effectful/Telemetry.hs +++ b/src/Pgmq/Effectful/Telemetry.hs @@ -2,114 +2,148 @@ -- -- This module provides: -- --- * W3C Trace Context propagation via message headers --- * OpenTelemetry semantic conventions for messaging systems +-- * Trace-context propagation through pgmq message headers, using the +-- tracer provider's /configured/ propagator (W3C Trace Context by +-- default, but any propagator installed on the provider will work — +-- B3, Datadog, …). The pgmq header payload is a JSON object whose +-- keys are header names (lower-cased) and whose values are strings. +-- * Re-exports of typed 'AttributeKey' values from +-- "OpenTelemetry.SemanticConventions" for the v1.24 attributes used +-- by the pgmq instrumentation. +-- +-- The semantic-conventions names mirror OpenTelemetry specification +-- v1.24 as generated into @hs-opentelemetry-semantic-conventions@ +-- 0.1.0.0. module Pgmq.Effectful.Telemetry - ( -- * Trace Context Operations + ( -- * Trace Context Propagation injectTraceContext, extractTraceContext, + traceHeadersToJson, + jsonToTraceHeaders, mergeTraceHeaders, TraceHeaders, - -- * Semantic Conventions - messagingSystem, - messagingOperationType, - messagingDestinationName, - messagingMessageId, - messagingBatchMessageCount, - messagingRoutingKey, - dbSystem, - dbOperationName, + -- * Semantic Convention Keys (re-exported from + + -- "OpenTelemetry.SemanticConventions") + messaging_system, + messaging_operation, + messaging_destination_name, + messaging_message_id, + messaging_batch_messageCount, + db_system, + db_operation, ) where +import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Aeson (Value (..)) import Data.Aeson.Key qualified as Key import Data.Aeson.KeyMap qualified as KM -import Data.ByteString (ByteString) -import Data.Text (Text) +import Data.CaseInsensitive qualified as CI +import Data.Text qualified as T import Data.Text.Encoding qualified as TE -import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C -import OpenTelemetry.Trace.Core qualified as OTel - --- | Headers for W3C Trace Context propagation. --- Contains traceparent and optionally tracestate. -type TraceHeaders = [(ByteString, ByteString)] - --- | Inject trace context into message headers (for producers). --- Creates traceparent and tracestate headers from current span context. -injectTraceContext :: OTel.Span -> IO TraceHeaders -injectTraceContext otelSpan = do - (traceparent, tracestate) <- W3C.encodeSpanContext otelSpan - pure - [ ("traceparent", traceparent), - ("tracestate", tracestate) - ] - --- | Extract trace context from message headers (for consumers). --- Returns SpanContext that can be used as parent link. -extractTraceContext :: TraceHeaders -> Maybe OTel.SpanContext -extractTraceContext headers = do - let traceparent = lookup "traceparent" headers - tracestate = lookup "tracestate" headers - W3C.decodeSpanContext traceparent tracestate - --- | Merge trace headers into existing PGMQ message headers. --- Preserves existing headers while adding trace context. -mergeTraceHeaders :: TraceHeaders -> Maybe Value -> Value -mergeTraceHeaders traceHeaders existingHeaders = - let traceObj = - Object $ - KM.fromList - [ (Key.fromText "traceparent", toJsonValue $ lookup "traceparent" traceHeaders), - (Key.fromText "tracestate", toJsonValue $ lookup "tracestate" traceHeaders) - ] - in case existingHeaders of - Just (Object obj) -> Object $ KM.union (toKeyMap traceObj) obj - Just v -> Object $ KM.fromList [(Key.fromText "_original", v), (Key.fromText "_trace", traceObj)] - Nothing -> traceObj - where - toJsonValue :: Maybe ByteString -> Value - toJsonValue = maybe Null (String . TE.decodeUtf8) - - toKeyMap :: Value -> KM.KeyMap Value - toKeyMap (Object o) = o - toKeyMap _ = KM.empty - --- OpenTelemetry Semantic Conventions for Messaging --- See: https://opentelemetry.io/docs/specs/semconv/messaging/ +import Network.HTTP.Types (RequestHeaders) +import OpenTelemetry.Context qualified as Ctxt +import OpenTelemetry.Propagator (Propagator, extract, inject) +import OpenTelemetry.SemanticConventions + ( db_operation, + db_system, + messaging_batch_messageCount, + messaging_destination_name, + messaging_message_id, + messaging_operation, + messaging_system, + ) +import OpenTelemetry.Trace.Core + ( TracerProvider, + getTracerProviderPropagators, + ) --- | The messaging system identifier. --- Value: "pgmq" -messagingSystem :: Text -messagingSystem = "messaging.system" +-- | Carrier for propagated trace context. Same shape as +-- @http-types@' @RequestHeaders@ — @[(HeaderName, ByteString)]@, where +-- @HeaderName@ is case-insensitive — because that is the carrier type +-- used by every propagator in the @hs-opentelemetry@ ecosystem. +type TraceHeaders = RequestHeaders --- | The type of messaging operation. --- Values: "send", "receive", "process" -messagingOperationType :: Text -messagingOperationType = "messaging.operation.type" +-- | Inject trace context into carrier headers, using the tracer +-- provider's configured propagator. +-- +-- Callers typically pass a 'Ctxt.Context' obtained from +-- 'OpenTelemetry.Context.ThreadLocal.getContext' (with any current +-- span inserted) so the propagator can write the current span's +-- context onto the carrier. +injectTraceContext :: + (MonadIO m) => + TracerProvider -> + Ctxt.Context -> + m TraceHeaders +injectTraceContext provider ctxt = + let propagator = getTracerProviderPropagators provider + in injectWith propagator ctxt --- | The destination queue name. -messagingDestinationName :: Text -messagingDestinationName = "messaging.destination.name" +-- | Extract trace context from carrier headers, using the tracer +-- provider's configured propagator. Returns an updated 'Ctxt.Context' +-- carrying the extracted span context (if any) — suitable for starting +-- a child \"process\" span linked to the producer's trace. +extractTraceContext :: + (MonadIO m) => + TracerProvider -> + TraceHeaders -> + Ctxt.Context -> + m Ctxt.Context +extractTraceContext provider headers ctxt = + let propagator = getTracerProviderPropagators provider + in extract propagator headers ctxt --- | The unique message identifier. -messagingMessageId :: Text -messagingMessageId = "messaging.message.id" +-- | Propagator-generic inject wrapper (mainly useful for tests that +-- want to pin the propagator explicitly). +injectWith :: + (MonadIO m) => + Propagator Ctxt.Context RequestHeaders RequestHeaders -> + Ctxt.Context -> + m TraceHeaders +injectWith propagator ctxt = liftIO $ inject propagator ctxt [] --- | Number of messages in a batch operation. -messagingBatchMessageCount :: Text -messagingBatchMessageCount = "messaging.batch.message_count" +-- | Encode carrier headers as a JSON object for storage in pgmq +-- message headers (jsonb). Header names are lower-cased (via +-- case-insensitive folding) so the on-wire representation is stable +-- regardless of the original case the propagator emitted. +traceHeadersToJson :: TraceHeaders -> Value +traceHeadersToJson = Object . traceHeadersKeyMap --- | The routing key for topic-based message routing. -messagingRoutingKey :: Text -messagingRoutingKey = "messaging.destination.routing_key" +-- | Decode a JSON object of trace headers back into the carrier +-- shape. Non-object inputs and non-string values are ignored. +jsonToTraceHeaders :: Value -> TraceHeaders +jsonToTraceHeaders = \case + Object obj -> + [ (CI.mk (TE.encodeUtf8 (Key.toText k)), TE.encodeUtf8 v) + | (k, String v) <- KM.toList obj + ] + _ -> [] --- | The database system. --- Value: "postgresql" -dbSystem :: Text -dbSystem = "db.system" +-- | Merge propagated trace headers into an existing pgmq headers +-- value. Existing keys win — we never overwrite user-supplied headers. +mergeTraceHeaders :: TraceHeaders -> Maybe Value -> Value +mergeTraceHeaders traceHeaders existingHeaders = + let traceObj = traceHeadersKeyMap traceHeaders + in case existingHeaders of + Just (Object obj) -> Object (KM.union obj traceObj) + Just v -> + Object + ( KM.fromList + [ (Key.fromText (T.pack "_original"), v), + (Key.fromText (T.pack "_trace"), Object traceObj) + ] + ) + Nothing -> Object traceObj --- | The database operation name. -dbOperationName :: Text -dbOperationName = "db.operation.name" +traceHeadersKeyMap :: TraceHeaders -> KM.KeyMap Value +traceHeadersKeyMap = + KM.fromList + . fmap + ( \(name, value) -> + ( Key.fromText (TE.decodeUtf8 (CI.foldedCase name)), + String (TE.decodeUtf8 value) + ) + ) diff --git a/src/Pgmq/Effectful/Traced.hs b/src/Pgmq/Effectful/Traced.hs --- a/src/Pgmq/Effectful/Traced.hs +++ b/src/Pgmq/Effectful/Traced.hs @@ -1,20 +1,29 @@ --- | Traced operations for PGMQ with automatic trace context propagation. +-- | Traced operations for PGMQ with automatic trace-context propagation. -- --- This module provides higher-level operations that automatically inject --- W3C Trace Context into message headers (for producers) and extract --- trace context from received messages (for consumers). +-- Higher-level helpers layered on top of the Pgmq effect: -- +-- * 'sendMessageTraced' writes the current trace context onto the +-- outgoing message headers using the tracer provider's /configured/ +-- propagator (W3C Trace Context by default; swap in B3, Datadog, or +-- any other propagator by configuring the provider). +-- * 'readMessageWithContext' extracts the trace context from each +-- received message and returns an 'OTel.Context' suitable for +-- starting a child @process@ span that links back to the producer's +-- trace. +-- -- == Usage -- -- @ --- -- Producer: Send with trace context --- sendMessageTraced tracer queueName body Nothing +-- -- Producer +-- provider <- OTel.getGlobalTracerProvider +-- sendMessageTraced provider queueName body Nothing -- --- -- Consumer: Read with trace context extraction --- messagesWithCtx <- readMessageWithContext readQuery --- forM_ messagesWithCtx $ \(msg, maybeParentCtx) -> --- -- maybeParentCtx contains the SpanContext from the producer --- processMessage msg maybeParentCtx +-- -- Consumer +-- messagesWithCtx <- readMessageWithContext provider readQuery +-- forM_ messagesWithCtx $ \\(msg, parentCtx) -> +-- -- parentCtx carries the propagated parent context +-- OTel.inSpan tracer \"process\" OTel.defaultSpanArguments +-- (processMessage msg) -- @ module Pgmq.Effectful.Traced ( -- * Traced Send Operations @@ -27,10 +36,6 @@ where import Data.Aeson (Value (..)) -import Data.Aeson.Key qualified as Key -import Data.Aeson.KeyMap qualified as KM -import Data.Text (Text) -import Data.Text.Encoding qualified as TE import Data.Vector (Vector) import Data.Vector qualified as V import Effectful (Eff, IOE, (:>)) @@ -40,34 +45,39 @@ import OpenTelemetry.Trace.Core qualified as OTel import Pgmq.Effectful.Effect (Pgmq, readMessage, sendMessageWithHeaders) import Pgmq.Effectful.Telemetry + ( extractTraceContext, + injectTraceContext, + jsonToTraceHeaders, + mergeTraceHeaders, + ) import Pgmq.Hasql.Statements.Types qualified as Types import Pgmq.Types --- | A message paired with its extracted trace context (if present). -type MessageWithContext = (Message, Maybe OTel.SpanContext) +-- | A message paired with the trace context extracted from its +-- headers. Use 'OpenTelemetry.Context.ThreadLocal.attachContext' (or +-- pass it to an @inSpan''@-shaped primitive that lets you set the +-- parent) to start a @process@ span linked to the producer's trace. +type MessageWithContext = (Message, Ctxt.Context) --- | Send a message with trace context automatically injected into headers. --- --- This creates traceparent and tracestate headers from the current span --- context, which can be extracted by consumers to link traces. +-- | Send a message with trace context injected into its headers. -- --- If existing headers are provided, the trace headers are merged in. +-- The current 'Ctxt.Context' is fetched via +-- 'OpenTelemetry.Context.ThreadLocal.getContext' and handed to the +-- tracer provider's configured propagator, which writes the propagated +-- fields (traceparent/tracestate for W3C, x-b3-* for B3, etc.) onto +-- the message headers. Any user-supplied headers win against +-- propagator-written ones (the merge is additive). sendMessageTraced :: (Pgmq :> es, IOE :> es) => - OTel.Tracer -> + OTel.TracerProvider -> QueueName -> MessageBody -> Maybe Value -> Eff es MessageId -sendMessageTraced _tracer queueName body existingHeaders = do - -- Get current span context and inject into headers +sendMessageTraced provider queueName body existingHeaders = do ctx <- Effectful.liftIO CtxtLocal.getContext - traceHeaders <- case Ctxt.lookupSpan ctx of - Just s -> Effectful.liftIO $ injectTraceContext s - Nothing -> pure [] - + traceHeaders <- injectTraceContext provider ctx let mergedHeaders = MessageHeaders $ mergeTraceHeaders traceHeaders existingHeaders - sendMessageWithHeaders $ Types.SendMessageWithHeaders { queueName = queueName, @@ -76,51 +86,23 @@ delay = Nothing } --- | Read messages and extract trace context from headers. --- --- Returns messages paired with their extracted SpanContext (if present). --- The SpanContext can be used to create a child span that links to the --- producer's trace. +-- | Read messages and extract trace context from their headers. -- --- @ --- messagesWithCtx <- readMessageWithContext readQuery --- forM_ messagesWithCtx $ \(msg, maybeParentCtx) -> --- inSpan' tracer "process" (linkToParent maybeParentCtx) $ do --- processMessage msg --- @ +-- The returned 'Ctxt.Context' for each message is the +-- 'Ctxt.empty' context updated with whatever fields the tracer +-- provider's propagator was able to parse from the headers — so a +-- caller can open a child @process@ span linked to the producer's +-- trace without knowing which propagator is in use. readMessageWithContext :: - (Pgmq :> es) => + (Pgmq :> es, IOE :> es) => + OTel.TracerProvider -> Types.ReadMessage -> Eff es (Vector MessageWithContext) -readMessageWithContext readQuery = do +readMessageWithContext provider readQuery = do messages <- readMessage readQuery - pure $ V.map extractContext messages - where - extractContext :: Message -> MessageWithContext - extractContext msg = - let ctx = extractTraceContextFromMessage msg - in (msg, ctx) - - extractTraceContextFromMessage :: Message -> Maybe OTel.SpanContext - extractTraceContextFromMessage msg = do - hdrs <- msg.headers - traceHeaders <- parseTraceHeaders hdrs - extractTraceContext traceHeaders - - parseTraceHeaders :: Value -> Maybe TraceHeaders - parseTraceHeaders (Object obj) = do - traceparent <- KM.lookup (Key.fromText "traceparent") obj >>= asText - let tracestate = KM.lookup (Key.fromText "tracestate") obj >>= asText - pure $ - catMaybes - [ Just ("traceparent", TE.encodeUtf8 traceparent), - ("tracestate",) . TE.encodeUtf8 <$> tracestate - ] - parseTraceHeaders _ = Nothing - - asText :: Value -> Maybe Text - asText (String t) = Just t - asText _ = Nothing - - catMaybes :: [Maybe a] -> [a] - catMaybes = foldr (\x acc -> maybe acc (: acc) x) [] + Effectful.liftIO $ + V.forM messages $ \msg -> do + parentCtx <- case msg.headers of + Just hdrs -> extractTraceContext provider (jsonToTraceHeaders hdrs) Ctxt.empty + Nothing -> pure Ctxt.empty + pure (msg, parentCtx) diff --git a/test/ClassificationSpec.hs b/test/ClassificationSpec.hs new file mode 100644 --- /dev/null +++ b/test/ClassificationSpec.hs @@ -0,0 +1,66 @@ +{-# LANGUAGE OverloadedStrings #-} + +module ClassificationSpec (tests) where + +import Data.HashSet qualified as HashSet +import Hasql.Errors qualified as HasqlErrors +import Pgmq.Effectful + ( PgmqRuntimeError (..), + isTransient, + ) +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (assertBool, testCase) + +tests :: TestTree +tests = + testGroup + "isTransient classification" + [ testCase "acquisition timeout is transient" $ + assertBool "expected transient" (isTransient PgmqAcquisitionTimeout), + testCase "networking connection error is transient" $ + assertBool + "expected transient" + ( isTransient + (PgmqConnectionError (HasqlErrors.NetworkingConnectionError "refused")) + ), + testCase "authentication error is not transient" $ + assertBool + "expected permanent" + ( not $ + isTransient + (PgmqConnectionError (HasqlErrors.AuthenticationConnectionError "bad password")) + ), + testCase "compatibility error is not transient" $ + assertBool + "expected permanent" + ( not $ + isTransient + (PgmqConnectionError (HasqlErrors.CompatibilityConnectionError "version mismatch")) + ), + testCase "other connection error is treated as transient" $ + assertBool + "expected transient" + ( isTransient + (PgmqConnectionError (HasqlErrors.OtherConnectionError "libpq says no")) + ), + testCase "connection-drop session error is transient" $ + assertBool + "expected transient" + ( isTransient + (PgmqSessionError (HasqlErrors.ConnectionSessionError "dropped")) + ), + testCase "driver session error is not transient" $ + assertBool + "expected permanent" + ( not $ + isTransient + (PgmqSessionError (HasqlErrors.DriverSessionError "bug")) + ), + testCase "missing-types session error is not transient" $ + assertBool + "expected permanent" + ( not $ + isTransient + (PgmqSessionError (HasqlErrors.MissingTypesSessionError HashSet.empty)) + ) + ] diff --git a/test/EphemeralDb.hs b/test/EphemeralDb.hs new file mode 100644 --- /dev/null +++ b/test/EphemeralDb.hs @@ -0,0 +1,44 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | Test database infrastructure using ephemeral-pg. +-- +-- This is a copy of pgmq-hasql\'s EphemeralDb helper, adjusted for +-- pgmq-effectful\'s test suite. Keeping a local copy avoids cross-package +-- test-helper sharing (which would require a shared test-helper library +-- or fragile cross-package hs-source-dirs). +module EphemeralDb + ( withPgmqDb, + withPgmqPool, + StartError, + ) +where + +import EphemeralPg + ( StartError, + connectionSettings, + withCached, + ) +import Hasql.Pool qualified as Pool +import Hasql.Pool.Config qualified as PoolConfig +import Pgmq.Migration qualified as Migration + +-- | Run an action with a temporary PostgreSQL database that has the +-- pgmq schema installed. +withPgmqDb :: (Pool.Pool -> IO a) -> IO (Either StartError a) +withPgmqDb action = withCached $ \db -> do + let connSettings = connectionSettings db + poolConfig = + PoolConfig.settings + [ PoolConfig.size 3, + PoolConfig.staticConnectionSettings connSettings + ] + pool <- Pool.acquire poolConfig + installResult <- Pool.use pool Migration.migrate + case installResult of + Left poolErr -> error $ "Failed to install pgmq schema: " <> show poolErr + Right (Left migrationErr) -> error $ "Migration failed: " <> show migrationErr + Right (Right ()) -> action pool + +-- | Alias for 'withPgmqDb' kept for parallelism with pgmq-hasql. +withPgmqPool :: (Pool.Pool -> IO a) -> IO (Either StartError a) +withPgmqPool = withPgmqDb diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,24 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Main (main) where + +import ClassificationSpec qualified +import EphemeralDb (withPgmqPool) +import PlainInterpreterSpec qualified +import Test.Tasty (defaultMain, testGroup) +import TracedInterpreterSpec qualified + +main :: IO () +main = do + result <- withPgmqPool $ \pool -> do + let tree = + testGroup + "pgmq-effectful" + [ ClassificationSpec.tests, + PlainInterpreterSpec.tests pool, + TracedInterpreterSpec.tests pool + ] + defaultMain tree + case result of + Left err -> error $ "Failed to start temp database: " <> show err + Right () -> pure () diff --git a/test/PlainInterpreterSpec.hs b/test/PlainInterpreterSpec.hs new file mode 100644 --- /dev/null +++ b/test/PlainInterpreterSpec.hs @@ -0,0 +1,68 @@ +{-# LANGUAGE OverloadedStrings #-} + +module PlainInterpreterSpec (tests) where + +import Effectful (runEff) +import Effectful.Error.Static (runError) +import Hasql.Pool qualified as Pool +import Hasql.Pool.Config qualified as PoolConfig +import Pgmq.Effectful + ( MessageId (..), + MessageQuery (..), + PgmqRuntimeError (..), + deleteMessage, + listQueues, + parseQueueName, + runPgmq, + ) +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (assertFailure, testCase) + +tests :: Pool.Pool -> TestTree +tests pool = + testGroup + "Plain interpreter error propagation" + [ testCase "statement error surfaces PgmqSessionError" $ do + bogus <- case parseQueueName "queue_that_does_not_exist_xyz" of + Right q -> pure q + Left err -> assertFailure ("could not build test queue name: " <> show err) >> fail "" + result <- + runEff + . runError @PgmqRuntimeError + . runPgmq pool + $ deleteMessage + MessageQuery + { queueName = bogus, + messageId = MessageId 1 + } + case result of + Left (_cs, PgmqSessionError _) -> pure () + Left (_cs, other) -> + assertFailure $ + "Expected PgmqSessionError, got " <> show other + Right _ -> + assertFailure + "Expected an error deleting from missing queue, got success", + testCase "connection error surfaces PgmqConnectionError" $ do + let badCfg = + PoolConfig.settings + [ PoolConfig.size 1, + PoolConfig.staticConnectionSettings + "host=127.0.0.1 port=1 user=nobody dbname=nonexistent connect_timeout=1" + ] + badPool <- Pool.acquire badCfg + result <- + runEff + . runError @PgmqRuntimeError + . runPgmq badPool + $ listQueues + Pool.release badPool + case result of + Left (_cs, PgmqConnectionError _) -> pure () + Left (_cs, other) -> + assertFailure $ + "Expected PgmqConnectionError, got " <> show other + Right _ -> + assertFailure + "Expected a connection error, got success" + ] diff --git a/test/TracedInterpreterSpec.hs b/test/TracedInterpreterSpec.hs new file mode 100644 --- /dev/null +++ b/test/TracedInterpreterSpec.hs @@ -0,0 +1,311 @@ +{-# LANGUAGE OverloadedStrings #-} + +module TracedInterpreterSpec (tests) where + +import Control.Monad (unless) +import Control.Monad.IO.Class (liftIO) +import Data.HashMap.Strict qualified as HM +import Data.IORef (IORef, readIORef) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Vector qualified as V +import Effectful (runEff) +import Effectful.Error.Static (runError) +import Hasql.Pool qualified as Pool +import OpenTelemetry.Attributes qualified as Attrs +import OpenTelemetry.Context qualified as Ctxt +import OpenTelemetry.Context.ThreadLocal qualified as CtxtLocal +import OpenTelemetry.Exporter.InMemory.Span qualified as InMemoryExporter +import OpenTelemetry.Propagator (extract) +import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C +import OpenTelemetry.Trace.Core qualified as OTel +import OpenTelemetry.Trace.Id.Generator.Default (defaultIdGenerator) +import OpenTelemetry.Util (appendOnlyBoundedCollectionValues) +import Pgmq.Effectful + ( MessageBody (..), + MessageId (..), + MessageQuery (..), + PgmqRuntimeError (..), + QueueName, + ReadMessage (..), + SendMessage (..), + createQueue, + deleteMessage, + parseQueueName, + queueNameToText, + readMessage, + runPgmqTraced, + sendMessage, + sendMessageTraced, + ) +import Pgmq.Effectful.Telemetry (jsonToTraceHeaders) +import Pgmq.Types qualified as Pgmq +import System.Random (randomRIO) +import Test.Tasty (TestTree, testGroup) +import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase) + +tests :: Pool.Pool -> TestTree +tests pool = + testGroup + "Traced interpreter" + [ testGroup + "Error propagation" + [ testCase "statement error surfaces PgmqSessionError via Error channel" $ do + (tracer, _provider, _spansRef) <- setupTracer + bogus <- mkQueueName "queue_that_does_not_exist_xyz2" + result <- + runEff + . runError @PgmqRuntimeError + . runPgmqTraced pool tracer + $ deleteMessage + MessageQuery + { queueName = bogus, + messageId = MessageId 1 + } + case result of + Left (_cs, PgmqSessionError _) -> pure () + Left (_cs, other) -> + assertFailure $ + "Expected PgmqSessionError, got " <> show other + Right _ -> + assertFailure + "Expected an error, got success" + ], + testGroup + "v1.24 semantic conventions" + [ testCase "publish emits messaging.* + db.* attributes" $ do + (tracer, _provider, spansRef) <- setupTracer + queue <- mkUniqueQueue "publish_attrs" + result <- + runEff . runError @PgmqRuntimeError . runPgmqTraced pool tracer $ do + createQueue queue + sendMessage + SendMessage + { queueName = queue, + messageBody = MessageBody "hello", + delay = Nothing + } + _ <- assertRight result + spans <- readIORef spansRef + let publishSpans = filter ((==) "publish" . firstWord . OTel.spanName) spans + unless (length publishSpans == 1) $ + assertFailure $ + "Expected exactly one publish span, got " + <> show (map OTel.spanName spans) + s <- singleSpan publishSpans "publish" + assertEqual + "span name carries destination" + ("publish " <> queueNameToText queue) + (OTel.spanName s) + assertSpanKindProducer s + assertAttrText s "messaging.system" "pgmq" + assertAttrText s "messaging.operation" "publish" + assertAttrText s "messaging.destination.name" (queueNameToText queue) + assertAttrText s "db.system" "postgresql" + assertAttrText s "db.operation" "pgmq.send", + testCase "receive emits messaging.* + db.* attributes" $ do + (tracer, _provider, spansRef) <- setupTracer + queue <- mkUniqueQueue "receive_attrs" + result <- + runEff . runError @PgmqRuntimeError . runPgmqTraced pool tracer $ do + createQueue queue + _ <- + sendMessage + SendMessage + { queueName = queue, + messageBody = MessageBody "one", + delay = Nothing + } + readMessage + ReadMessage + { queueName = queue, + delay = 30, + batchSize = Just 1, + conditional = Nothing + } + _ <- assertRight result + spans <- readIORef spansRef + let receiveSpans = filter ((==) "receive" . firstWord . OTel.spanName) spans + unless (length receiveSpans == 1) $ + assertFailure $ + "Expected exactly one receive span, got " + <> show (map OTel.spanName spans) + s <- singleSpan receiveSpans "receive" + assertEqual + "span name carries destination" + ("receive " <> queueNameToText queue) + (OTel.spanName s) + assertSpanKindConsumer s + assertAttrText s "messaging.system" "pgmq" + assertAttrText s "messaging.operation" "receive" + assertAttrText s "messaging.destination.name" (queueNameToText queue) + assertAttrText s "db.system" "postgresql" + assertAttrText s "db.operation" "pgmq.read", + testCase "error path records exception event and Error status" $ do + (tracer, _provider, spansRef) <- setupTracer + bogus <- mkQueueName "queue_that_does_not_exist_xyz3" + _ <- + runEff + . runError @PgmqRuntimeError + . runPgmqTraced pool tracer + $ deleteMessage + MessageQuery {queueName = bogus, messageId = MessageId 1} + spans <- readIORef spansRef + s <- case spans of + [only] -> pure only + _ -> + assertFailure + ( "Expected exactly one span, got " + <> show (map OTel.spanName spans) + ) + >> fail "" + let events = + V.toList + (appendOnlyBoundedCollectionValues (OTel.spanEvents s)) + hasExceptionEvent = + any ((==) "exception" . OTel.eventName) events + assertBool "span has exception event" hasExceptionEvent + case OTel.spanStatus s of + OTel.Error desc -> + assertBool + ( "span status description should be a short non-PII label, got " + <> T.unpack desc + ) + ("pool.session.statement" `T.isPrefixOf` desc) + other -> + assertFailure $ "expected Error status, got " <> show other, + testCase "W3C traceparent round-trips through message headers" $ do + (tracer, provider, _spansRef) <- setupTracer + queue <- mkUniqueQueue "w3c_roundtrip" + parentSpan <- + OTel.createSpan + tracer + Ctxt.empty + "parent" + OTel.defaultSpanArguments + parentTraceId <- OTel.traceId <$> OTel.getSpanContext parentSpan + _ <- CtxtLocal.attachContext (Ctxt.insertSpan parentSpan Ctxt.empty) + result <- + runEff . runError @PgmqRuntimeError . runPgmqTraced pool tracer $ do + createQueue queue + _ <- sendMessageTraced provider queue (MessageBody "payload") Nothing + readMessage + ReadMessage + { queueName = queue, + delay = 30, + batchSize = Just 1, + conditional = Nothing + } + OTel.endSpan parentSpan Nothing + msgs <- assertRight result + msg <- case V.toList msgs of + [m] -> pure m + other -> + assertFailure ("expected 1 message, got " <> show (length other)) + >> fail "" + case Pgmq.headers msg of + Nothing -> + assertFailure + "expected trace headers on received message, got none" + Just hdrs -> do + let propagator = OTel.getTracerProviderPropagators provider + ctx <- + liftIO $ + extract propagator (jsonToTraceHeaders hdrs) Ctxt.empty + case Ctxt.lookupSpan ctx of + Nothing -> + assertFailure $ + "propagator did not extract a span from headers: " <> show hdrs + Just extracted -> do + extractedTraceId <- OTel.traceId <$> OTel.getSpanContext extracted + assertEqual + "trace id round-trips via message headers" + parentTraceId + extractedTraceId + ] + ] + +-- --------------------------------------------------------------------- +-- Helpers +-- --------------------------------------------------------------------- + +-- | Build a tracer, tracer provider (with in-memory list processor and +-- W3C Trace Context propagator), and a handle to the exported spans. +setupTracer :: IO (OTel.Tracer, OTel.TracerProvider, IORef [OTel.ImmutableSpan]) +setupTracer = do + (processor, spansRef) <- InMemoryExporter.inMemoryListExporter + let opts = + OTel.emptyTracerProviderOptions + { OTel.tracerProviderOptionsIdGenerator = defaultIdGenerator, + OTel.tracerProviderOptionsPropagators = W3C.w3cTraceContextPropagator + } + provider <- OTel.createTracerProvider [processor] opts + let tracer = + OTel.makeTracer provider "pgmq-effectful-test" OTel.tracerOptions + pure (tracer, provider, spansRef) + +mkQueueName :: String -> IO QueueName +mkQueueName raw = + case parseQueueName (T.pack raw) of + Right q -> pure q + Left err -> + assertFailure + ("could not build queue name " <> show raw <> ": " <> show err) + >> fail "" + +mkUniqueQueue :: String -> IO QueueName +mkUniqueQueue prefix = do + suffix <- randomRIO (100000 :: Int, 999999) + mkQueueName (prefix <> "_" <> show suffix) + +firstWord :: Text -> Text +firstWord = T.takeWhile (/= ' ') + +assertRight :: (Show a) => Either a b -> IO b +assertRight = \case + Right b -> pure b + Left e -> + assertFailure ("expected Right, got Left " <> show e) >> fail "" + +singleSpan :: [OTel.ImmutableSpan] -> String -> IO OTel.ImmutableSpan +singleSpan spans label = case spans of + [s] -> pure s + _ -> + assertFailure + ( "expected exactly one " + <> label + <> " span, got " + <> show (map OTel.spanName spans) + ) + >> fail "" + +-- | 'OTel.SpanKind' has no 'Eq' instance, so assert via pattern. +assertSpanKindProducer :: OTel.ImmutableSpan -> IO () +assertSpanKindProducer s = case OTel.spanKind s of + OTel.Producer -> pure () + other -> assertFailure ("expected Producer span kind, got " <> show other) + +assertSpanKindConsumer :: OTel.ImmutableSpan -> IO () +assertSpanKindConsumer s = case OTel.spanKind s of + OTel.Consumer -> pure () + other -> assertFailure ("expected Consumer span kind, got " <> show other) + +-- | Assert a span carries the given attribute as a 'TextAttribute' with +-- the given value. Produces a readable failure message with the full +-- attribute map when the assertion fails. +assertAttrText :: OTel.ImmutableSpan -> Text -> Text -> IO () +assertAttrText s key expected = do + let attrMap = Attrs.getAttributeMap (OTel.spanAttributes s) + case HM.lookup key attrMap of + Just (Attrs.AttributeValue (Attrs.TextAttribute actual)) -> + assertEqual ("attribute " <> T.unpack key) expected actual + other -> + assertFailure $ + "expected text attribute " + <> T.unpack key + <> "=" + <> T.unpack expected + <> "; got " + <> show other + <> " in " + <> show attrMap