diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,40 @@ # Changelog +## 0.19.0.0 — 2026-11-14 + +### Breaking Changes + +- Require [`shibuya-core`](mori://shinzui/shibuya/packages/shibuya-core) and + `shibuya-metrics` `^>=0.13.0.0`. Core 0.13 renames processing spans from + ` process` to `process ` and adds + `messaging.operation.name = "process"`; update trace queries, dashboards and + alerts that match the old span name. +- Add the strict field `telemetry :: AdapterTelemetry` to `PgmqAdapterConfig`. + Configurations built from `defaultConfig` with record update need no change; + a full `PgmqAdapterConfig { .. }` literal must set it, normally to + `defaultAdapterTelemetry`. + +### New Features + +- Choose how a processing span relates to the span that sent the message with + `telemetry = defaultAdapterTelemetry {traceCorrelation = LinkOnly}`. The + default, `ParentAndLink`, keeps the processing span in the sender's trace as + its child and now also links to the sender. `LinkOnly` starts each delivery + in its own trace, linked to the sender, which suits delayed sends and long + retry cycles. Without a usable `traceparent` neither adds a link. Re-export + `AdapterTelemetry (traceCorrelation)`, `defaultAdapterTelemetry` and + `TraceCorrelation (..)` from `Shibuya.Adapter.Pgmq`. +- Under `LinkOnly`, a dead-lettered message's `traceparent` names the + consumer's processing span in its own trace; the sender's original header is + still preserved under `x-shibuya-upstream-traceparent`. Dead-letter header + handling itself is unchanged. + +### Other Changes + +- Add PostgreSQL-backed correlation tests covering both policies, redelivery + and the dead-letter headers, and let the example consumer select `LinkOnly` + with `OTEL_TRACE_LINK_ONLY=true`. + ## 0.18.0.0 — 2026-10-27 ### Breaking Changes diff --git a/shibuya-pgmq-adapter.cabal b/shibuya-pgmq-adapter.cabal --- a/shibuya-pgmq-adapter.cabal +++ b/shibuya-pgmq-adapter.cabal @@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-pgmq-adapter -version: 0.18.0.0 +version: 0.19.0.0 synopsis: PGMQ adapter for the Shibuya queue processing framework description: A Shibuya adapter that integrates with pgmq (PostgreSQL Message Queue) @@ -51,7 +51,7 @@ pgmq-core ^>=0.7.0.0, pgmq-effectful ^>=0.7.0.0, pgmq-hasql ^>=0.7.0.0, - shibuya-core ^>=0.12.0.0, + shibuya-core ^>=0.13.0.0, stm ^>=2.5, streamly ^>=0.11, streamly-core ^>=0.3, @@ -101,6 +101,7 @@ Shibuya.Adapter.Pgmq.ConfigSpec Shibuya.Adapter.Pgmq.Convert Shibuya.Adapter.Pgmq.ConvertSpec + Shibuya.Adapter.Pgmq.CorrelationSpec Shibuya.Adapter.Pgmq.IntegrationSpec Shibuya.Adapter.Pgmq.Internal Shibuya.Adapter.Pgmq.InternalSpec @@ -129,7 +130,7 @@ pgmq-migration ^>=0.7.0.0, quickcheck-instances ^>=0.3, random, - shibuya-core ^>=0.12.0.0, + shibuya-core ^>=0.13.0.0, shibuya-pgmq-adapter, stm, streamly ^>=0.11, diff --git a/src/Shibuya/Adapter/Pgmq.hs b/src/Shibuya/Adapter/Pgmq.hs --- a/src/Shibuya/Adapter/Pgmq.hs +++ b/src/Shibuya/Adapter/Pgmq.hs @@ -88,6 +88,11 @@ FifoReadStrategy (..), validateConfig, + -- * Telemetry + AdapterTelemetry (traceCorrelation), + defaultAdapterTelemetry, + TraceCorrelation (..), + -- * Smart Constructors directDeadLetter, topicDeadLetter, @@ -146,7 +151,7 @@ routingKeyToText, topicPatternToText, ) -import Shibuya.Adapter (Adapter (..)) +import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry) import Shibuya.Adapter.Pgmq.Config ( DeadLetterConfig (..), DeadLetterTarget (..), @@ -216,7 +221,8 @@ Adapter { adapterName = "pgmq:" <> queueNameToText validConfig.queueName, source = messageSource, - shutdown = liftIO $ atomically $ writeTVar shutdownVar True + shutdown = liftIO $ atomically $ writeTVar shutdownVar True, + telemetry = validConfig.telemetry } -- | Poll in chunks so a shutdown can release messages that were read from pgmq diff --git a/src/Shibuya/Adapter/Pgmq/Config.hs b/src/Shibuya/Adapter/Pgmq/Config.hs --- a/src/Shibuya/Adapter/Pgmq/Config.hs +++ b/src/Shibuya/Adapter/Pgmq/Config.hs @@ -42,6 +42,7 @@ import Pgmq.Effectful (PgmqRuntimeError) import Pgmq.Types (QueueName, RoutingKey) import Pgmq.Types qualified as Pgmq +import Shibuya.Adapter (AdapterTelemetry, defaultAdapterTelemetry) -- | Runtime resources and callbacks used by the adapter. data PgmqAdapterEnv = PgmqAdapterEnv @@ -99,7 +100,18 @@ -- See 'PrefetchConfig' for the visibility-timeout trade-off and the bounded, -- loss-free shutdown behaviour (buffered messages are redelivered after -- their visibility timeout rather than released immediately). - prefetchConfig :: !(Maybe PrefetchConfig) + prefetchConfig :: !(Maybe PrefetchConfig), + -- | Telemetry options copied unchanged into the 'Shibuya.Adapter.Adapter'. + -- + -- The default is 'defaultAdapterTelemetry', whose @traceCorrelation@ is + -- 'Shibuya.Adapter.ParentAndLink': the processing span is a child of the + -- span that sent the message and also links to it, so a job appears in + -- the sending request's trace. Choose 'Shibuya.Adapter.LinkOnly' with + -- @defaultAdapterTelemetry {traceCorrelation = LinkOnly}@ when messages + -- are sent with long delays or retried over long visibility timeouts: + -- each delivery then starts its own trace and is related to the sender + -- only by the link, instead of extending a trace that ended long ago. + telemetry :: !AdapterTelemetry } deriving stock (Show, Eq, Generic) @@ -295,5 +307,6 @@ haltVisibilityTimeout = Nothing, maxRetries = 3, fifoConfig = Nothing, - prefetchConfig = Nothing + prefetchConfig = Nothing, + telemetry = defaultAdapterTelemetry } diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -3,6 +3,7 @@ import Shibuya.Adapter.Pgmq.ChaosSpec qualified as ChaosSpec import Shibuya.Adapter.Pgmq.ConfigSpec qualified as ConfigSpec import Shibuya.Adapter.Pgmq.ConvertSpec qualified as ConvertSpec +import Shibuya.Adapter.Pgmq.CorrelationSpec qualified as CorrelationSpec import Shibuya.Adapter.Pgmq.DeadLetterSpec qualified as DeadLetterSpec import Shibuya.Adapter.Pgmq.IntegrationSpec qualified as IntegrationSpec import Shibuya.Adapter.Pgmq.InternalSpec qualified as InternalSpec @@ -19,4 +20,5 @@ describe "Shibuya.Adapter.Pgmq Properties" PropertySpec.spec describe "Integration Tests" IntegrationSpec.spec describe "Chaos Tests" ChaosSpec.spec + describe "Trace Correlation Tests" CorrelationSpec.spec DeadLetterSpec.spec diff --git a/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs b/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs --- a/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs +++ b/test/Shibuya/Adapter/Pgmq/ChaosSpec.hs @@ -40,7 +40,7 @@ import Pgmq.Hasql.Sessions qualified as Sessions import Pgmq.Hasql.Statements.Types (BindTopic (..), ReadMessage (..), SendMessage (..), SendMessageWithHeaders (..)) import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), QueueName, parseQueueName, parseRoutingKey, parseTopicPattern) -import Shibuya.Adapter (Adapter (..)) +import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Adapter.Pgmq ( PgmqAcknowledgementException, PgmqAdapterConfig (..), @@ -386,7 +386,7 @@ Vector.length messages `shouldBe` 1 spans <- readIORef spansRef named <- mapM (\sp -> do hot <- readIORef (OTel.spanHot sp); pure (OTel.hotName hot, sp)) spans - case (Vector.uncons messages, [sp | ("dlq-identity process", sp) <- named]) of + case (Vector.uncons messages, [sp | ("process dlq-identity", sp) <- named]) of (Just (msg, _), [consumer]) -> case msg.headers of Just (Object hdrs) -> do let stringHeader k = case KeyMap.lookup k hdrs of @@ -695,7 +695,7 @@ case ingestedResult of Nothing -> liftIO (expectationFailure "message should not auto-DLQ") >> pure "" Just ingested -> do - let adapter = Adapter "pgmq:test-terminal-ack-failure" (Stream.fromList [ingested]) (pure ()) + let adapter = Adapter "pgmq:test-terminal-ack-failure" (Stream.fromList [ingested]) (pure ()) defaultAdapterTelemetry processor = mkProcessor adapter (\_ -> pure $ AckDeadLetter $ PoisonPill "fail loudly") appResult <- runApp defaultAppConfig [(processorId, processor)] case appResult of diff --git a/test/Shibuya/Adapter/Pgmq/ConfigSpec.hs b/test/Shibuya/Adapter/Pgmq/ConfigSpec.hs --- a/test/Shibuya/Adapter/Pgmq/ConfigSpec.hs +++ b/test/Shibuya/Adapter/Pgmq/ConfigSpec.hs @@ -2,7 +2,7 @@ import Data.Either (isLeft, isRight) import Pgmq.Types (parseRoutingKey) -import Shibuya.Adapter.Pgmq (parseQueueName) +import Shibuya.Adapter.Pgmq (AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry, parseQueueName) import Shibuya.Adapter.Pgmq.Config import Test.Hspec @@ -55,6 +55,14 @@ it "sets fifoConfig to Nothing" $ do config.fifoConfig `shouldBe` Nothing + + it "uses the default adapter telemetry, whose correlation is ParentAndLink" $ do + config.telemetry `shouldBe` defaultAdapterTelemetry + config.telemetry.traceCorrelation `shouldBe` ParentAndLink + + it "validates a LinkOnly override and keeps it" $ do + let linkOnly = config {telemetry = defaultAdapterTelemetry {traceCorrelation = LinkOnly}} + fmap (\c -> c.telemetry.traceCorrelation) (validateConfig linkOnly) `shouldBe` Right LinkOnly -- | Tests for defaultPollingConfig defaultPollingConfigSpec :: Spec diff --git a/test/Shibuya/Adapter/Pgmq/CorrelationSpec.hs b/test/Shibuya/Adapter/Pgmq/CorrelationSpec.hs new file mode 100644 --- /dev/null +++ b/test/Shibuya/Adapter/Pgmq/CorrelationSpec.hs @@ -0,0 +1,286 @@ +-- | Trace correlation of PGMQ processing spans against a real PostgreSQL. +-- +-- Each case sends a message carrying a known producer @traceparent@, runs a +-- traced Shibuya app over the adapter with an in-memory span exporter, and +-- inspects the exported Consumer spans: their trace, their parent and their +-- links. The dead-letter cases also read the dead-letter queue message's +-- headers, which 'Shibuya.Adapter.Pgmq.Internal.mergeDlqHeaders' writes from +-- the active processing span. +-- +-- Run with: cabal test shibuya-pgmq-adapter-test --test-option=--match --test-option=Correlation +module Shibuya.Adapter.Pgmq.CorrelationSpec (spec) where + +import Control.Concurrent (threadDelay) +import Control.Exception qualified as Exception +import Control.Monad (void) +import Data.Aeson (Value (..), object, (.=)) +import Data.Aeson.Key qualified as Key +import Data.Aeson.KeyMap qualified as KeyMap +import Data.Foldable (toList) +import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) +import Data.Maybe (isNothing) +import Data.Text (Text) +import Data.Text.Encoding qualified as Text +import Data.Vector qualified as Vector +import Effectful (Eff, IOE, liftIO, runEff, (:>)) +import Effectful.Error.Static (Error, runErrorNoCallStack) +import Hasql.Pool qualified as Pool +import OpenTelemetry.Attributes (emptyAttributes) +import OpenTelemetry.Exporter.InMemory (inMemoryListExporter) +import OpenTelemetry.Trace.Core qualified as OTel +import OpenTelemetry.Trace.Id (Base (..)) +import OpenTelemetry.Trace.Id qualified as OTel.Id +import OpenTelemetry.Trace.Id.Generator (IdGenerator (..)) +import OpenTelemetry.Util (appendOnlyBoundedCollectionValues) +import Pgmq.Effectful (Pgmq, PgmqRuntimeError, runPgmq) +import Pgmq.Hasql.Sessions qualified as Sessions +import Pgmq.Hasql.Statements.Types (ReadMessage (..), SendMessageWithHeaders (..)) +import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..)) +import Shibuya.Adapter (Adapter (..)) +import Shibuya.Adapter.Pgmq + ( AdapterTelemetry (traceCorrelation), + PgmqAdapterConfig (..), + PollingConfig (..), + TraceCorrelation (..), + defaultAdapterTelemetry, + defaultConfig, + directDeadLetter, + mkPgmqAdapterEnv, + pgmqAdapter, + ) +import Shibuya.App (ProcessorId (..), ShutdownConfig (..), defaultAppConfig, defaultShutdownConfig, mkProcessor, runApp, stopAppGracefully) +import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), RetryDelay (..)) +import Shibuya.Core.Ingested (Message (..)) +import Shibuya.Core.Types (Attempt (..), Envelope (..)) +import Shibuya.Handler (Handler) +import Shibuya.Telemetry.Effect (Tracing, runTracing) +import Shibuya.Telemetry.Propagation (extractTraceContext) +import Shibuya.Telemetry.Semantic (processSpanName) +import System.Environment (lookupEnv) +import Test.Hspec +import TmpPostgres (TestFixture (..), runPgmqSession, withPgmqDb, withTestFixture) + +spec :: Spec +spec = around withTempDbFixture $ describe "Correlation" $ do + it "ParentAndLink: process span is a child of the producer with one link" $ \fixture -> do + run <- runCorrelated fixture defaultAdapterTelemetry False (countingHandler AckOk) 1 + run.adapterPolicy `shouldBe` ParentAndLink + [sp] <- pure run.spans + sp.name `shouldBe` processSpanName processorName + hexTrace sp.context `shouldBe` producerTraceHex + fmap hexSpan sp.parent `shouldBe` Just producerSpanHex + linkIds sp `shouldBe` [(producerTraceHex, producerSpanHex)] + + it "LinkOnly: process span is a linked root" $ \fixture -> do + run <- runCorrelated fixture linkOnly False (countingHandler AckOk) 1 + run.adapterPolicy `shouldBe` LinkOnly + [sp] <- pure run.spans + hexTrace sp.context `shouldNotBe` producerTraceHex + isNothing sp.parent `shouldBe` True + linkIds sp `shouldBe` [(producerTraceHex, producerSpanHex)] + + it "ParentAndLink: every redelivery stays in the producer's trace" $ \fixture -> do + run <- runCorrelated fixture defaultAdapterTelemetry False retryOnceHandler 2 + length run.spans `shouldBe` 2 + map (hexTrace . (.context)) run.spans `shouldBe` [producerTraceHex, producerTraceHex] + map (fmap hexSpan . (.parent)) run.spans `shouldBe` [Just producerSpanHex, Just producerSpanHex] + map linkIds run.spans `shouldBe` replicate 2 [(producerTraceHex, producerSpanHex)] + + it "LinkOnly: each redelivery is its own linked trace" $ \fixture -> do + run <- runCorrelated fixture linkOnly False retryOnceHandler 2 + case run.spans of + [first, second] -> do + hexTrace first.context `shouldNotBe` hexTrace second.context + mapM_ (\sp -> hexTrace sp.context `shouldNotBe` producerTraceHex) [first, second] + mapM_ (\sp -> isNothing sp.parent `shouldBe` True) [first, second] + mapM_ (\sp -> linkIds sp `shouldBe` [(producerTraceHex, producerSpanHex)]) [first, second] + other -> expectationFailure ("expected two processing spans, got " <> show (length other)) + + it "LinkOnly: DLQ traceparent is the consumer's, upstream preserved" $ \fixture -> do + run <- runCorrelated fixture linkOnly True (countingHandler (AckDeadLetter (PoisonPill "correlation"))) 1 + [sp] <- pure run.spans + isNothing sp.parent `shouldBe` True + linkIds sp `shouldBe` [(producerTraceHex, producerSpanHex)] + hdrs <- readDlqHeaders fixture + let stringHeader :: Text -> Maybe Text + stringHeader k = case KeyMap.lookup (Key.fromText k) hdrs of + Just (String v) -> Just v + _ -> Nothing + active = [(Text.encodeUtf8 k, Text.encodeUtf8 v) | k <- ["traceparent", "tracestate"], Just v <- [stringHeader k]] + case extractTraceContext active of + Nothing -> expectationFailure ("expected a decodable consumer traceparent in the DLQ, got " <> show hdrs) + Just ctx -> do + hexTrace ctx `shouldBe` hexTrace sp.context + hexSpan ctx `shouldBe` hexSpan sp.context + hexTrace ctx `shouldNotBe` producerTraceHex + hexSpan ctx `shouldNotBe` producerSpanHex + stringHeader "x-shibuya-upstream-traceparent" `shouldBe` Just producerParent + stringHeader "x-shibuya-upstream-tracestate" `shouldBe` Just producerState + +-------------------------------------------------------------------------------- +-- Fixtures +-------------------------------------------------------------------------------- + +producerParent :: Text +producerParent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" + +producerState :: Text +producerState = "vendor=opaque" + +producerTraceHex :: Text +producerTraceHex = "0af7651916cd43dd8448eb211c80319c" + +producerSpanHex :: Text +producerSpanHex = "b7ad6b7169203331" + +processorName :: Text +processorName = "pgmq-correlation" + +linkOnly :: AdapterTelemetry +linkOnly = defaultAdapterTelemetry {traceCorrelation = LinkOnly} + +data CorrelatedRun = CorrelatedRun + { adapterPolicy :: TraceCorrelation, + spans :: [ConsumerSpan] + } + +-- | Send one message with the producer headers, process it with the given +-- telemetry until the handler has been called @deliveries@ times, and return +-- the exported Consumer spans in the order they ended. +runCorrelated :: + TestFixture -> + AdapterTelemetry -> + Bool -> + (IORef Int -> Handler '[Tracing, Pgmq, Error PgmqRuntimeError, IOE] Value) -> + Int -> + IO CorrelatedRun +runCorrelated TestFixture {pool, queueName, dlqName} telemetry withDlq handler deliveries = do + runPgmqSession pool $ + void $ + Sessions.sendMessageWithHeaders $ + SendMessageWithHeaders + { queueName = queueName, + messageBody = MessageBody (String "correlated"), + messageHeaders = MessageHeaders (object ["traceparent" .= producerParent, "tracestate" .= producerState]), + delay = Just 0 + } + let config = + (defaultConfig queueName) + { visibilityTimeout = 5, + batchSize = 1, + polling = StandardPolling {pollInterval = 0.05}, + deadLetterConfig = if withDlq then Just (directDeadLetter dlqName True) else Nothing, + telemetry = telemetry + } + calls <- newIORef (0 :: Int) + (spanProcessor, spansRef) <- inMemoryListExporter + policy <- + Exception.bracket + (OTel.createTracerProvider [spanProcessor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator}) + (\p -> OTel.shutdownTracerProvider p (Just 5_000_000)) + $ \provider -> do + let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "pgmq-correlation" "" "" emptyAttributes) OTel.tracerOptions + runAdapterIO pool $ runTracing tracer $ do + adapter <- requireAdapter pool config + result <- runApp defaultAppConfig [(ProcessorId processorName, mkProcessor adapter (handler calls))] + case result of + Left err -> liftIO (expectationFailure (show err)) + Right appHandle -> do + liftIO $ waitForCalls calls deliveries 5_000_000 + drained <- stopAppGracefully defaultShutdownConfig {drainTimeout = 5} appHandle + liftIO $ drained `shouldBe` True + pure adapter.telemetry.traceCorrelation + readIORef calls `shouldReturn` deliveries + exported <- readIORef spansRef + consumers <- consumerSpans exported + pure CorrelatedRun {adapterPolicy = policy, spans = consumers} + +countingHandler :: (IOE :> es) => AckDecision -> IORef Int -> Handler es Value +countingHandler decision calls _ = do + liftIO $ atomicModifyIORef' calls (\n -> (n + 1, ())) + pure decision + +-- | Ask for an immediate redelivery on the first attempt, succeed afterwards. +retryOnceHandler :: (IOE :> es) => IORef Int -> Handler es Value +retryOnceHandler calls message = do + liftIO $ atomicModifyIORef' calls (\n -> (n + 1, ())) + pure $ case message.envelope.attempt of + Just (Attempt n) | n > 0 -> AckOk + _ -> AckRetry (RetryDelay 0) + +readDlqHeaders :: TestFixture -> IO (KeyMap.KeyMap Value) +readDlqHeaders TestFixture {pool, dlqName} = do + messages <- runPgmqSession pool $ Sessions.readMessage $ ReadMessage dlqName 30 (Just 10) Nothing + case Vector.toList messages of + [msg] -> case msg.headers of + Just (Object hdrs) -> pure hdrs + other -> expectationFailure ("expected DLQ headers, got " <> show other) >> pure KeyMap.empty + other -> expectationFailure ("expected one DLQ message, got " <> show (length other)) >> pure KeyMap.empty + +waitForCalls :: IORef Int -> Int -> Int -> IO () +waitForCalls ref target = go + where + go remaining + | remaining <= 0 = pure () + | otherwise = do + n <- readIORef ref + if n >= target then pure () else threadDelay 20_000 >> go (remaining - 20_000) + +withTempDbFixture :: (TestFixture -> IO ()) -> IO () +withTempDbFixture action = do + skipDb <- lookupEnv "PGMQ_TEST_SKIP_DB" + case skipDb of + Just "1" -> pendingWith "Database tests skipped (PGMQ_TEST_SKIP_DB=1)" + _ -> do + result <- withPgmqDb $ \pool -> withTestFixture pool action + case result of + Left err -> expectationFailure $ "Failed to start temp database: " <> show err + Right () -> pure () + +runAdapterIO :: Pool.Pool -> Eff '[Pgmq, Error PgmqRuntimeError, IOE] a -> IO a +runAdapterIO pool action = do + result <- runEff $ runErrorNoCallStack $ runPgmq pool action + either (\err -> error ("Pgmq error: " <> show err)) pure result + +requireAdapter :: + (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es, Tracing :> es) => + Pool.Pool -> + PgmqAdapterConfig -> + Eff es (Adapter es Value) +requireAdapter pool config = + pgmqAdapter (mkPgmqAdapterEnv pool) config + >>= either (\err -> liftIO (error ("Invalid PGMQ adapter config: " <> show err))) pure + +-------------------------------------------------------------------------------- +-- Span inspection +-------------------------------------------------------------------------------- + +data ConsumerSpan = ConsumerSpan + { name :: Text, + context :: OTel.SpanContext, + parent :: Maybe OTel.SpanContext, + links :: [OTel.SpanContext] + } + +consumerSpans :: [OTel.ImmutableSpan] -> IO [ConsumerSpan] +consumerSpans exported = traverse describeSpan [sp | sp <- exported, OTel.spanKind sp == OTel.Consumer] + where + describeSpan sp = do + hot <- readIORef (OTel.spanHot sp) + parent <- traverse OTel.getSpanContext (OTel.spanParent sp) + pure + ConsumerSpan + { name = OTel.hotName hot, + context = OTel.spanContext sp, + parent, + links = map OTel.frozenLinkContext (toList (appendOnlyBoundedCollectionValues (OTel.hotLinks hot))) + } + +linkIds :: ConsumerSpan -> [(Text, Text)] +linkIds sp = [(hexTrace l, hexSpan l) | l <- sp.links] + +hexTrace :: OTel.SpanContext -> Text +hexTrace ctx = OTel.Id.traceIdBaseEncodedText Base16 ctx.traceId + +hexSpan :: OTel.SpanContext -> Text +hexSpan ctx = OTel.Id.spanIdBaseEncodedText Base16 ctx.spanId diff --git a/test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs b/test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs --- a/test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs +++ b/test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs @@ -34,6 +34,7 @@ VisibilityTimeoutQuery (..), ) import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), MessageId, QueueName) +import Shibuya.Adapter (defaultAdapterTelemetry) import Shibuya.Adapter.Pgmq.Config ( FifoConfig (..), FifoReadStrategy (..), @@ -448,7 +449,8 @@ haltVisibilityTimeout = Nothing, maxRetries = 3, fifoConfig = Nothing, - prefetchConfig = Nothing + prefetchConfig = Nothing, + telemetry = defaultAdapterTelemetry } -- | Create config with custom batch size @@ -465,5 +467,6 @@ haltVisibilityTimeout = Nothing, maxRetries = 3, fifoConfig = Nothing, - prefetchConfig = Nothing + prefetchConfig = Nothing, + telemetry = defaultAdapterTelemetry } diff --git a/test/Shibuya/Adapter/Pgmq/InternalSpec.hs b/test/Shibuya/Adapter/Pgmq/InternalSpec.hs --- a/test/Shibuya/Adapter/Pgmq/InternalSpec.hs +++ b/test/Shibuya/Adapter/Pgmq/InternalSpec.hs @@ -18,6 +18,7 @@ import Pgmq.Effectful.Effect qualified as PgmqEffect import Pgmq.Hasql.Statements.Types (ReadGrouped (..), ReadGroupedWithPoll (..), ReadMessage (..), ReadWithPollMessage (..)) import Pgmq.Types (Message (..), MessageBody (..), MessageId (..), parseQueueName, queueNameToText) +import Shibuya.Adapter (defaultAdapterTelemetry) import Shibuya.Adapter.Pgmq.Config ( FifoConfig (..), FifoReadStrategy (..), @@ -104,7 +105,8 @@ haltVisibilityTimeout = Nothing, maxRetries = 3, fifoConfig = Nothing, - prefetchConfig = Nothing + prefetchConfig = Nothing, + telemetry = defaultAdapterTelemetry } ReadMessage { queueName = queryQueueName, @@ -143,7 +145,8 @@ haltVisibilityTimeout = Nothing, maxRetries = 3, fifoConfig = Nothing, - prefetchConfig = Nothing + prefetchConfig = Nothing, + telemetry = defaultAdapterTelemetry } ReadWithPollMessage { queueName = queryQueueName, @@ -190,7 +193,8 @@ haltVisibilityTimeout = Nothing, maxRetries = 3, fifoConfig = Nothing, - prefetchConfig = Nothing + prefetchConfig = Nothing, + telemetry = defaultAdapterTelemetry } ReadGrouped { queueName = queryQueueName, @@ -503,7 +507,8 @@ haltVisibilityTimeout = Nothing, maxRetries = 3, fifoConfig = Nothing, - prefetchConfig = Nothing + prefetchConfig = Nothing, + telemetry = defaultAdapterTelemetry } runStubPgmq ::