diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,40 @@ # Changelog +## 0.8.0.0 — 2026-11-14 + +### Breaking Changes + +- **Default trace shape changed: each processed event is now its own trace, + linked to the span that appended it.** `KirokuAdapterConfig` and + `KirokuConsumerGroupConfig` gain a strict `telemetry :: AdapterTelemetry` + field, defaulting to `kirokuDefaultTelemetry` + (`defaultAdapterTelemetry {traceCorrelation = LinkOnly}`). A subscription + replays and catches up on events written long ago, and several subscriptions + read the same event, so making processing a child of the appending span + stretched the writer's trace indefinitely. Restore the previous single-trace + shape, now with an added link, with `telemetry = defaultAdapterTelemetry` + (`ParentAndLink`). Every consumer-group member uses the group's value. +- Require `shibuya-core >=0.13 && <0.14` (also in `kiroku-store`). Core 0.13 + renames processing spans from ` process` to `process ` + and adds `messaging.operation.name = "process"`; update trace queries that + match the old name. Full `KirokuAdapterConfig` and `KirokuConsumerGroupConfig` + literals must set `telemetry`. + +### New Features + +- Re-export `AdapterTelemetry (traceCorrelation)`, `defaultAdapterTelemetry`, + `TraceCorrelation (..)` and the new `kirokuDefaultTelemetry` from + `Shibuya.Adapter.Kiroku`. + +### Other Changes + +- Add PostgreSQL-backed correlation tests: both policies, two subscriptions to + one event, consumer-group members, and events without `traceparent`. +- Correct the 0.5.0.0 note below: `kiroku-otel`'s native subscription spans are + independent roots, not part of the writer's or Shibuya's trace. They and + Shibuya's processing spans are separate traces correlated by matching + `kiroku.*` attributes. + ## 0.7.0.1 — 2026-10-28 ### Other Changes @@ -261,9 +296,10 @@ kiroku identity so it appears on Shibuya's per-message span: `kiroku.subscription.name`, `kiroku.event.type`, `kiroku.event.global_position`, and — for a consumer-group member — `kiroku.consumer_group.member`. Combined - with `kiroku-otel`'s native subscription spans, a single trace is followable - from a kiroku subscription into Shibuya's processing; the keys match on both - sides. + with `kiroku-otel`'s native subscription spans, a kiroku subscription can be + followed into Shibuya's processing: the keys match on both sides. (Corrected + in 0.8.0.0: these are separate traces correlated by those attributes, not a + single trace.) * `Shibuya.Adapter.Kiroku.Convert` exposes `KirokuEnvelopeAttrs`; `toEnvelope` and `toIngestedAck` take it. `kirokuAdapter` derives the subscription name and member from its config, so the single-adapter and consumer-group paths are both diff --git a/shibuya-kiroku-adapter.cabal b/shibuya-kiroku-adapter.cabal --- a/shibuya-kiroku-adapter.cabal +++ b/shibuya-kiroku-adapter.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: shibuya-kiroku-adapter -version: 0.7.0.1 +version: 0.8.0.0 synopsis: Kiroku event store adapter for the Shibuya queue processing framework @@ -66,7 +66,7 @@ , hs-opentelemetry-api ^>=1.0 , hs-opentelemetry-semantic-conventions ^>=1.40 , kiroku-store ^>=0.13.0.0 - , shibuya-core >=0.12 && <0.13 + , shibuya-core >=0.13 && <0.14 , shibuya-kiroku-adapter-internal , stm >=2.5 && <2.6 , streamly-core >=0.3 && <0.4 @@ -82,32 +82,34 @@ type: exitcode-stdio-1.0 main-is: Main.hs hs-source-dirs: test + other-modules: CorrelationSpec ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends: - , aeson >=2.1 && <2.3 - , base >=4.18 && <5 - , containers >=0.6 && <0.8 + , aeson >=2.1 && <2.3 + , base >=4.18 && <5 + , containers >=0.6 && <0.8 , directory - , effectful >=2.6.1 && <2.8 - , effectful-core >=2.6.1 && <2.7 || >=2.7.1.1 && <2.8 - , ephemeral-pg >=0.3.1 && <0.4 - , generic-lens >=2.2 && <2.4 - , hasql >=1.10 && <1.11 - , hasql-pool >=1.2 && <1.5 - , hs-opentelemetry-api ^>=1.0 - , hspec >=2.10 && <2.12 - , kiroku-store ^>=0.13.0.0 - , kiroku-test-support ^>=0.1.0.0 - , lens >=5.2 && <5.4 - , shibuya-core >=0.12 && <0.13 + , effectful >=2.6.1 && <2.8 + , effectful-core >=2.6.1 && <2.7 || >=2.7.1.1 && <2.8 + , ephemeral-pg >=0.3.1 && <0.4 + , generic-lens >=2.2 && <2.4 + , hasql >=1.10 && <1.11 + , hasql-pool >=1.2 && <1.5 + , hs-opentelemetry-api ^>=1.0 + , hs-opentelemetry-exporter-in-memory ^>=1.0 + , hspec >=2.10 && <2.12 + , kiroku-store ^>=0.13.0.0 + , kiroku-test-support ^>=0.1.0.0 + , lens >=5.2 && <5.4 + , shibuya-core >=0.13 && <0.14 , shibuya-kiroku-adapter , shibuya-kiroku-adapter-internal - , stm >=2.5 && <2.6 - , streamly-core >=0.3 && <0.4 - , text >=2.0 && <2.2 - , time >=1.12 && <1.15 - , unordered-containers >=0.2 && <0.3 - , uuid >=1.3 && <1.4 + , stm >=2.5 && <2.6 + , streamly-core >=0.3 && <0.4 + , text >=2.0 && <2.2 + , time >=1.12 && <1.15 + , unordered-containers >=0.2 && <0.3 + , uuid >=1.3 && <1.4 -- EP-45 live-store performance, restart, backlog, and retained-memory fixture. -- The executable owns an ephemeral PostgreSQL instance and emits the common @@ -132,7 +134,7 @@ , ephemeral-pg >=0.3.1 && <0.4 , kiroku-store ^>=0.13.0.0 , kiroku-test-support ^>=0.1.0.0 - , shibuya-core >=0.12 && <0.13 + , shibuya-core >=0.13 && <0.14 , shibuya-kiroku-adapter , stm >=2.5 && <2.6 , text >=2.0 && <2.2 @@ -159,7 +161,7 @@ , hasql-pool >=1.2 && <1.5 , kiroku-store ^>=0.13.0.0 , kiroku-test-support - , shibuya-core >=0.12 && <0.13 + , shibuya-core >=0.13 && <0.14 , shibuya-kiroku-adapter , text >=2.0 && <2.2 , vector >=0.13 && <0.14 diff --git a/src/Shibuya/Adapter/Kiroku.hs b/src/Shibuya/Adapter/Kiroku.hs --- a/src/Shibuya/Adapter/Kiroku.hs +++ b/src/Shibuya/Adapter/Kiroku.hs @@ -117,6 +117,21 @@ -- @handlerStallWarnAfter@ to receive advisory store handler-stall events; warnings -- never finalize, retry or checkpoint the item. @retryPolicy@ controls total -- deliveries independently of the delay chosen by 'AckRetry'. +-- +-- == Trace Correlation +-- +-- When an event's metadata carries a W3C @traceparent@ (written, for example, +-- by @kiroku-otel@'s @injectTraceContext@ on append), Shibuya opens a +-- Consumer span named @process \@ around the handler and always +-- adds a link to the appending span. By default ('kirokuDefaultTelemetry', +-- 'LinkOnly') that span is the root of a new trace: subscriptions replay and +-- catch up on events written long ago and several subscriptions read the +-- same event, so making each handler a child of the appending span would +-- stretch the writer's trace indefinitely. Set @telemetry@ to +-- 'defaultAdapterTelemetry' ('ParentAndLink') to make processing the +-- appending span's child instead; a consumer group applies its one value to +-- every member. @kiroku-otel@'s own subscription spans are separate roots, +-- correlated with these spans by the shared @kiroku.*@ attributes. module Shibuya.Adapter.Kiroku ( -- * Adapter kirokuAdapter, @@ -128,6 +143,12 @@ KirokuAdapterConfig (..), defaultKirokuAdapterConfig, + -- * Telemetry + AdapterTelemetry (traceCorrelation), + defaultAdapterTelemetry, + kirokuDefaultTelemetry, + TraceCorrelation (..), + -- * Consumer-group helpers KirokuConsumerGroupConfig (..), defaultConsumerGroupConfig, @@ -178,7 +199,7 @@ import Kiroku.Store.Subscription.Types qualified as Sub import Kiroku.Store.Types (RecordedEvent) import Numeric.Natural (Natural) -import Shibuya.Adapter (Adapter (..)) +import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry) import Shibuya.Adapter.Kiroku.Convert (kirokuEnvelopeAttrs, toIngestedAck) import Shibuya.Adapter.Kiroku.Internal (acquireAllAndTransfer) import Shibuya.App (ProcessorId (..), QueueProcessor (..), mkProcessor) @@ -257,7 +278,18 @@ -- dead-lettered and the checkpoint still advances past it. See -- 'Kiroku.Store.Subscription.Types.selector' for when to prefer it over the -- introspectable 'eventTypeFilter'. - selector :: !(Maybe (RecordedEvent -> Bool)) + selector :: !(Maybe (RecordedEvent -> Bool)), + -- | Telemetry options copied unchanged into the 'Shibuya.Adapter.Adapter'. + -- + -- The default is @'defaultAdapterTelemetry' {traceCorrelation = 'LinkOnly'}@: + -- each processed event's span starts its own trace and links to the span + -- that appended the event. This differs from queue adapters, which default + -- to 'ParentAndLink', because a subscription replays and catches up on + -- events written long ago and several subscriptions read the same event, + -- so parenting would stretch the writer's trace indefinitely and attach + -- every subscriber's work to it. Set @telemetry = 'defaultAdapterTelemetry'@ + -- ('ParentAndLink') to make processing a child of the appending span again. + telemetry :: !AdapterTelemetry } deriving stock (Generic) @@ -265,8 +297,8 @@ -- @bufferSize = defaultStreamBufferSize@, @queueCapacity = 16@, @consumerGroup = 'Nothing'@ -- (ordinary single-consumer subscription), @missingCheckpointPolicy = -- 'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@ --- (deliver every type), and @selector = 'Nothing'@ (no extra predicate --- filtering). Supply the subscription name and target; override individual fields +-- (deliver every type), @selector = 'Nothing'@ (no extra predicate +-- filtering), and @telemetry = 'kirokuDefaultTelemetry'@ ('LinkOnly'). Supply the subscription name and target; override individual fields -- with record-update syntax. -- -- Prefer this over a full record literal so that any field added to @@ -292,9 +324,17 @@ consumerGroup = Nothing, missingCheckpointPolicy = FromBeginning, eventTypeFilter = AllEventTypes, - selector = Nothing + selector = Nothing, + telemetry = kirokuDefaultTelemetry } +-- | The adapter's default telemetry: 'defaultAdapterTelemetry' with +-- @traceCorrelation = 'LinkOnly'@, so each processed event is its own trace +-- linked to the appending span. Shared by 'defaultKirokuAdapterConfig' and +-- 'defaultConsumerGroupConfig'. +kirokuDefaultTelemetry :: AdapterTelemetry +kirokuDefaultTelemetry = defaultAdapterTelemetry {traceCorrelation = LinkOnly} + -- | Convert any synchronous exception thrown by a Shibuya handler into an -- 'AckDecision'. -- @@ -341,7 +381,7 @@ KirokuStore -> KirokuAdapterConfig -> Eff es (Adapter es RecordedEvent) -kirokuAdapter store KirokuAdapterConfig {subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, consumerGroup = cg, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} = do +kirokuAdapter store KirokuAdapterConfig {subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, consumerGroup = cg, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel, telemetry = tel} = do -- Build from 'defaultSubscriptionConfig' and override only the non-default -- fields. Using the smart constructor (rather than a full record literal) -- means any future field added to 'SubscriptionConfigM' is inherited at its @@ -378,7 +418,8 @@ Adapter { adapterName = "kiroku", source = ingestedStream, - shutdown = liftIO cancelAction + shutdown = liftIO cancelAction, + telemetry = tel } -- | Configuration for a whole kiroku consumer group presented as a single @@ -439,7 +480,10 @@ -- selector-filtered single subscription (worker-side, per member, checkpoint -- still advances past rejected events). See -- 'Kiroku.Store.Subscription.Types.selector'. - selector :: !(Maybe (RecordedEvent -> Bool)) + selector :: !(Maybe (RecordedEvent -> Bool)), + -- | Telemetry options given unchanged to /every/ member adapter. Defaults to + -- 'kirokuDefaultTelemetry' ('LinkOnly'); see 'KirokuAdapterConfig'.telemetry. + telemetry :: !AdapterTelemetry } deriving stock (Generic) @@ -447,7 +491,8 @@ -- 'Serial'@ (the only legal per-member concurrency), @batchSize = Sub.defaultBatchSize@, -- @bufferSize = defaultStreamBufferSize@, @queueCapacity = 16@, @missingCheckpointPolicy = -- 'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@ --- (deliver every type), @selector = 'Nothing'@ (no extra predicate filtering). +-- (deliver every type), @selector = 'Nothing'@ (no extra predicate filtering), +-- @telemetry = 'kirokuDefaultTelemetry'@ ('LinkOnly'). -- Supply the subscription name, target, and group size. defaultConsumerGroupConfig :: SubscriptionName -> SubscriptionTarget -> ConsumerGroupSize -> KirokuConsumerGroupConfig @@ -464,7 +509,8 @@ memberConcurrency = Serial, missingCheckpointPolicy = FromBeginning, eventTypeFilter = AllEventTypes, - selector = Nothing + selector = Nothing, + telemetry = kirokuDefaultTelemetry } -- | Map a requested per-member concurrency onto the group's validated Shibuya @@ -506,7 +552,7 @@ KirokuConsumerGroupConfig -> Handler es RecordedEvent -> Eff es (Either PolicyError [(ProcessorId, QueueProcessor es)]) -kirokuConsumerGroupProcessors store cfg@KirokuConsumerGroupConfig {subscriptionName = subName, subscriptionTarget = subTarget, groupSize = n, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} handler = +kirokuConsumerGroupProcessors store cfg@KirokuConsumerGroupConfig {subscriptionName = subName, subscriptionTarget = subTarget, groupSize = n, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel, telemetry = tel} handler = kirokuConsumerGroupProcessorsWith mkMemberAdapter cfg handler where mkMemberAdapter m = @@ -523,7 +569,8 @@ consumerGroup = Just (either (error . show) Prelude.id (mkConsumerGroup m n)), missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, - selector = sel + selector = sel, + telemetry = tel } -- | Factory-parameterized core for 'kirokuConsumerGroupProcessors'. diff --git a/test/CorrelationSpec.hs b/test/CorrelationSpec.hs new file mode 100644 --- /dev/null +++ b/test/CorrelationSpec.hs @@ -0,0 +1,216 @@ +-- | Trace correlation of Shibuya processing spans over a Kiroku subscription. +-- +-- Each case appends an event whose metadata carries a known writer +-- @traceparent@, subscribes with a traced Shibuya app whose tracer exports into +-- memory, and inspects the exported Consumer spans: their trace, their parent +-- and their links. +module CorrelationSpec (spec) where + +import Control.Concurrent.STM (atomically, modifyTVar', newTVarIO, readTVar, registerDelay) +import Control.Concurrent.STM qualified as STM +import Control.Exception qualified as E +import Control.Lens ((&), (.~), (^.)) +import Control.Monad (forM_) +import Control.Monad.IO.Class (liftIO) +import Data.Aeson (Value) +import Data.Aeson qualified as Aeson +import Data.Foldable (toList) +import Data.Generics.Labels () +import Data.IORef (readIORef) +import Data.Maybe (isNothing) +import Data.Text (Text) +import Effectful (runEff) +import Kiroku.Store +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 Shibuya.Adapter (Adapter (..)) +import Shibuya.Adapter.Kiroku + ( AdapterTelemetry, + ConsumerGroupSize, + KirokuAdapterConfig, + TraceCorrelation (..), + defaultAdapterTelemetry, + defaultConsumerGroupConfig, + defaultKirokuAdapterConfig, + kirokuAdapter, + kirokuConsumerGroupProcessors, + mkConsumerGroupSize, + ) +import Shibuya.App (ProcessorId (..), QueueProcessor (..), defaultAppConfig, mkProcessor, runApp, stopApp) +import Shibuya.Core.Ack (AckDecision (..)) +import Shibuya.Telemetry.Effect (runTracing, runTracingNoop) +import Test.Hspec + +spec :: SpecWith KirokuStore +spec = describe "trace correlation" $ do + it "defaults to LinkOnly for single adapters and consumer groups" $ \_ -> do + defaultKirokuAdapterConfig (SubscriptionName "c") AllStreams ^. #telemetry . #traceCorrelation `shouldBe` LinkOnly + defaultConsumerGroupConfig (SubscriptionName "c") AllStreams (sizeOf 2) ^. #telemetry . #traceCorrelation `shouldBe` LinkOnly + + it "an adapter built with a ParentAndLink override exposes it" $ \store -> do + policy <- runEff $ runTracingNoop $ do + Adapter {shutdown, telemetry} <- kirokuAdapter store (defaultKirokuAdapterConfig (SubscriptionName "corr-override") AllStreams & #telemetry .~ defaultAdapterTelemetry) + shutdown + pure (telemetry ^. #traceCorrelation) + policy `shouldBe` ParentAndLink + + it "every consumer-group member reports the configured policy" $ \store -> do + forM_ [LinkOnly, ParentAndLink] $ \policy -> do + let cfg = defaultConsumerGroupConfig (SubscriptionName "corr-group") AllStreams (sizeOf 3) & #telemetry .~ withPolicy policy + policies <- runEff $ runTracingNoop $ do + result <- kirokuConsumerGroupProcessors store cfg (\_ -> pure AckOk) + case result of + Left err -> liftIO (expectationFailure (show err)) >> pure [] + Right processors -> + traverse + ( \(_, processor) -> case processor of + QueueProcessor {adapter = Adapter {shutdown, telemetry}} -> shutdown >> pure (telemetry ^. #traceCorrelation) + BatchingProcessor {adapter = Adapter {shutdown, telemetry}} -> shutdown >> pure (telemetry ^. #traceCorrelation) + ) + processors + policies `shouldBe` replicate 3 policy + + it "LinkOnly (default): the processing span is a linked root" $ \store -> do + appendTraced store "corrlo-1" (Just writerParent) + [sp] <- processAll store [("corr-linkonly", defaultKirokuAdapterConfig (SubscriptionName "corr-linkonly") (category "corrlo"))] + hexTrace (spanContext sp) `shouldNotBe` writerTraceHex + isNothing (spanParent sp) `shouldBe` True + linkIds sp `shouldBe` [(writerTraceHex, writerSpanHex)] + + it "ParentAndLink: the processing span is a child of the writer with one link" $ \store -> do + appendTraced store "corrpal-1" (Just writerParent) + [sp] <- processAll store [("corr-pal", defaultKirokuAdapterConfig (SubscriptionName "corr-pal") (category "corrpal") & #telemetry .~ defaultAdapterTelemetry)] + hexTrace (spanContext sp) `shouldBe` writerTraceHex + fmap hexSpan (spanParent sp) `shouldBe` Just writerSpanHex + linkIds sp `shouldBe` [(writerTraceHex, writerSpanHex)] + + it "LinkOnly: two subscriptions to one event are two traces linking the same writer span" $ \store -> do + appendTraced store "corrfan-1" (Just writerParent) + spans <- + processAll + store + [ ("corr-fan-a", defaultKirokuAdapterConfig (SubscriptionName "corr-fan-a") (category "corrfan")), + ("corr-fan-b", defaultKirokuAdapterConfig (SubscriptionName "corr-fan-b") (category "corrfan")) + ] + case spans of + [a, b] -> do + hexTrace (spanContext a) `shouldNotBe` hexTrace (spanContext b) + mapM_ (\sp -> isNothing (spanParent sp) `shouldBe` True) [a, b] + mapM_ (\sp -> linkIds sp `shouldBe` [(writerTraceHex, writerSpanHex)]) [a, b] + other -> expectationFailure ("expected two processing spans, got " <> show (length other)) + + forM_ [LinkOnly, ParentAndLink] $ \policy -> + it (show policy <> ": an event without traceparent yields no link") $ \store -> do + let cat = if policy == LinkOnly then "corrnolo" else "corrnopal" + appendTraced store (cat <> "-1") Nothing + [sp] <- processAll store [("corr-none-" <> cat, defaultKirokuAdapterConfig (SubscriptionName ("corr-none-" <> cat)) (category cat) & #telemetry .~ withPolicy policy)] + spanLinks sp `shouldBe` [] + isNothing (spanParent sp) `shouldBe` True + +-------------------------------------------------------------------------------- +-- Fixtures +-------------------------------------------------------------------------------- + +writerParent :: Text +writerParent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" + +writerTraceHex :: Text +writerTraceHex = "0af7651916cd43dd8448eb211c80319c" + +writerSpanHex :: Text +writerSpanHex = "b7ad6b7169203331" + +sizeOf :: Int -> ConsumerGroupSize +sizeOf = either (error . show) Prelude.id . mkConsumerGroupSize . fromIntegral + +category :: Text -> SubscriptionTarget +category = Category . CategoryName + +withPolicy :: TraceCorrelation -> AdapterTelemetry +withPolicy policy = defaultAdapterTelemetry & #traceCorrelation .~ policy + +-- | Append one event to the given stream, with the writer's traceparent in its +-- metadata when supplied. +appendTraced :: KirokuStore -> Text -> Maybe Text -> IO () +appendTraced store stream traceparent = do + let metadata = fmap (\tp -> Aeson.object ["traceparent" Aeson..= tp]) traceparent + event = + EventData + { eventId = Nothing, + eventType = EventType "Correlated", + payload = Aeson.object [] :: Value, + metadata = metadata, + causationId = Nothing, + correlationId = Nothing + } + result <- runStoreIO store $ appendToStream (StreamName stream) NoStream [event] + either (\err -> expectationFailure ("append failed: " <> show err)) (const (pure ())) result + +-- | Run one processor per configuration until each has handled one event, and +-- return the exported Consumer spans. +processAll :: KirokuStore -> [(Text, KirokuAdapterConfig)] -> IO [ConsumerSpan] +processAll store configs = do + (spanProcessor, spansRef) <- inMemoryListExporter + handled <- newTVarIO (0 :: Int) + E.bracket + (OTel.createTracerProvider [spanProcessor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator}) + (\provider -> OTel.shutdownTracerProvider provider (Just 5_000_000)) + $ \provider -> do + let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "kiroku-correlation" "" "" emptyAttributes) OTel.tracerOptions + runEff $ runTracing tracer $ do + processors <- + traverse + ( \(name, cfg) -> do + adapter <- kirokuAdapter store cfg + pure (ProcessorId name, mkProcessor adapter (\_ -> liftIO (atomically (modifyTVar' handled (+ 1))) >> pure AckOk)) + ) + configs + result <- runApp defaultAppConfig processors + case result of + Left err -> liftIO (expectationFailure (show err)) + Right appHandle -> do + liftIO $ waitFor handled (length configs) 10_000_000 + stopApp appHandle + consumerSpans =<< readIORef spansRef + +waitFor :: STM.TVar Int -> Int -> Int -> IO () +waitFor var target timeoutMicros = do + deadline <- registerDelay timeoutMicros + reached <- + atomically $ + ((readTVar var >>= STM.check . (>= target)) >> pure True) + `STM.orElse` (readTVar deadline >>= STM.check >> pure False) + if reached then pure () else expectationFailure ("timed out waiting for " <> show target <> " handled events") + +data ConsumerSpan = ConsumerSpan + { spanContext :: OTel.SpanContext, + spanParent :: Maybe OTel.SpanContext, + spanLinks :: [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 + { spanContext = OTel.spanContext sp, + spanParent = parent, + spanLinks = map OTel.frozenLinkContext (toList (appendOnlyBoundedCollectionValues (OTel.hotLinks hot))) + } + +linkIds :: ConsumerSpan -> [(Text, Text)] +linkIds sp = [(hexTrace l, hexSpan l) | l <- spanLinks sp] + +hexTrace :: OTel.SpanContext -> Text +hexTrace ctx = OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) + +hexSpan :: OTel.SpanContext -> Text +hexSpan ctx = OTel.Id.spanIdBaseEncodedText Base16 (OTel.spanId ctx) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -8,6 +8,7 @@ import Control.Lens ((&), (.~), (^.)) import Control.Monad (forM_, unless) import Control.Monad.IO.Class (liftIO) +import CorrelationSpec qualified import Data.Aeson (Value) import Data.Aeson qualified as Aeson import Data.Foldable (toList) @@ -36,7 +37,7 @@ import Kiroku.Store.Subscription.Worker (withFetchBatchHookForTest, withSaveCheckpointHookForTest) import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres) import OpenTelemetry.Attributes (toAttribute) -import Shibuya.Adapter (Adapter (..)) +import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Adapter.Kiroku ( consumerGroupPolicy, defaultConsumerGroupConfig, @@ -195,6 +196,7 @@ `shouldBe` Left (InvalidPolicyCombo "StrictInOrder requires Serial concurrency") acknowledgementSpec + around withTestStore CorrelationSpec.spec around withTestStore $ do describe "kirokuAdapter" $ do it "delivers only matching event types when an eventTypeFilter is set (EP-43)" $ \store -> do @@ -1119,7 +1121,8 @@ modifyIORef' shutdowns (<> [m]) if m == 1 then E.throwIO (userError "member 1 cleanup failed") - else pure () + else pure (), + telemetry = defaultAdapterTelemetry } handler ingested = do let _ = envelopePayload ingested @@ -1287,7 +1290,8 @@ Adapter { adapterName = "stub-kiroku", source = Stream.nil, - shutdown = pure () + shutdown = pure (), + telemetry = defaultAdapterTelemetry } -- | Read the dead letters recorded for a non-group subscription (member 0).