diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,43 @@ # Changelog +## 0.5.0.0 — 2026-05-05 + +### Breaking Changes + +- `Envelope` gained an `attributes :: !(HashMap Text Attribute)` field + carrying adapter-supplied OpenTelemetry attributes for the per-message + processing span. Direct constructions of `Envelope` must add the + field; pass `Data.HashMap.Strict.empty` when the adapter has nothing + to contribute (the common case). `Envelope`'s `NFData` instance is + now hand-written rather than derived (because `Attribute` from + `hs-opentelemetry-api` does not ship `NFData`); the strictness shape + is unchanged for every other field. + +### New Features + +- `Shibuya.Runner.Supervised`'s `processOne` now applies + `envelope.attributes` to its Consumer-kind span after setting the + framework-default `messaging.*` attributes, so adapter-supplied + keys override framework defaults of the same name. This lets + broker-aware adapters (Kafka in particular) emit typed attributes + (`messaging.kafka.destination.partition`, + `messaging.kafka.message.offset`) and override the + `messaging.system` default — without opening a second span. The + per-record `Shibuya.Adapter.Kafka.Tracing.traced` wrapper that + previously bolted these on can now be removed; see the audit + document `docs/plans/9-otel-audit-findings.md` (Finding F1, P0) + for the full motivation. +- `Shibuya.Telemetry.Propagation.currentTraceHeaders :: + (Tracing :> es, IOE :> es) => Eff es (Maybe TraceHeaders)` looks + up the currently-active OTel span (via thread-local context) and + encodes its trace context as W3C headers, ready for an adapter + to attach to an outgoing message. Returns `Nothing` when tracing + is disabled or there is no active span. The intended call sites + are adapter-side DLQ writes (so the failing-consumer's trace + links to the resulting DLQ message) and ad-hoc producer paths. + See the audit document `docs/plans/9-otel-audit-findings.md` + (Findings F3 and F5). + ## 0.4.0.0 — 2026-04-29 ### Breaking Changes diff --git a/shibuya-core.cabal b/shibuya-core.cabal --- a/shibuya-core.cabal +++ b/shibuya-core.cabal @@ -1,6 +1,6 @@ cabal-version: 3.14 name: shibuya-core -version: 0.4.0.0 +version: 0.5.0.0 synopsis: Supervised queue processing framework for Haskell description: A supervised queue processing framework inspired by Broadway (Elixir). diff --git a/src/Shibuya/Core/Types.hs b/src/Shibuya/Core/Types.hs --- a/src/Shibuya/Core/Types.hs +++ b/src/Shibuya/Core/Types.hs @@ -19,9 +19,11 @@ ) where -import Control.DeepSeq (NFData) +import Control.DeepSeq (NFData (..)) import Data.ByteString (ByteString) +import Data.HashMap.Strict (HashMap) import Data.String (IsString) +import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..)) import Shibuya.Prelude -- | Stable identity for idempotency & observability. @@ -68,8 +70,41 @@ -- 'Just (Attempt 0)' on first delivery; 'Nothing' if the adapter -- does not track redeliveries (e.g., Kafka). attempt :: !(Maybe Attempt), + -- | Adapter-supplied OpenTelemetry attributes for the per-message + -- processing span. + -- + -- The framework's @processOne@ adds these to its Consumer-kind + -- span after setting the spec-aligned @messaging.*@ defaults, so + -- adapter-supplied keys override framework defaults of the same + -- name. Use 'Data.HashMap.Strict.empty' when the adapter has + -- nothing to contribute (the common case). + -- + -- Adapters that emit broker-specific typed attributes + -- (e.g. Kafka's @messaging.kafka.destination.partition@) should + -- populate this field at envelope-construction time. The + -- previous opt-in @Shibuya.Adapter.Kafka.Tracing.traced@ + -- transformer existed only to bolt these on; this field replaces + -- that mechanism without the duplicate-span hazard. + attributes :: !(HashMap Text Attribute), -- | The actual message payload payload :: !msg } deriving stock (Eq, Show, Functor, Generic) - deriving anyclass (NFData) + +-- | Manual 'NFData' so the @attributes@ field's 'Attribute' values do +-- not require an upstream NFData instance (which +-- @hs-opentelemetry-api@ does not currently ship). Forces every other +-- field deeply and reduces 'attributes' to WHNF — every 'Attribute' +-- leaf is a small primitive ('Text', 'Bool', 'Double', 'Int64'), so +-- WHNF is enough to evaluate the contained values when the HashMap +-- is itself in WHNF. +instance (NFData msg) => NFData (Envelope msg) where + rnf e = + rnf e.messageId `seq` + rnf e.cursor `seq` + rnf e.partition `seq` + rnf e.enqueuedAt `seq` + rnf e.traceContext `seq` + rnf e.attempt `seq` + e.attributes `seq` + rnf e.payload diff --git a/src/Shibuya/Runner/Supervised.hs b/src/Shibuya/Runner/Supervised.hs --- a/src/Shibuya/Runner/Supervised.hs +++ b/src/Shibuya/Runner/Supervised.hs @@ -36,11 +36,13 @@ writeTVar, ) import Control.Monad (unless) +import Data.HashMap.Strict qualified as HashMap import Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef) import Data.Text qualified as Text import Effectful (Eff, IOE, liftIO, withEffToIO, (:>)) import Effectful.Dispatch.Static (unsafeEff_) import Effectful.Internal.Unlift (Limit (..), Persistence (..), UnliftStrategy (..)) +import OpenTelemetry.Attributes (toAttribute) import OpenTelemetry.Trace.Core qualified as OTel import Shibuya.Adapter (Adapter (..)) import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..)) @@ -66,6 +68,7 @@ import Shibuya.Telemetry.Effect ( Tracing, addAttribute, + addAttributes, addEvent, recordException, setStatus, @@ -373,17 +376,26 @@ withExtractedContext parentCtx $ withSpan' (processSpanName pidText) consumerSpanArgs $ \traceSpan -> do - -- Add messaging attributes + -- Build the messaging.* attribute set: framework defaults first, + -- then the adapter's HashMap layered on top so adapter keys with + -- the same name override the framework's default. The explicit + -- 'HashMap.union' here is left-biased (left wins), which keeps + -- the precedence rule local and obvious instead of relying on + -- the order of repeated 'addAttribute' / 'addAttributes' calls + -- against the underlying mutable Span. let MessageId msgIdText = ingested.envelope.messageId - addAttribute traceSpan attrMessagingSystem ("shibuya" :: Text) - addAttribute traceSpan attrMessagingDestinationName pidText - addAttribute traceSpan attrMessagingOperation ("process" :: Text) - addAttribute traceSpan attrMessagingMessageId msgIdText - - -- Add partition if present - case ingested.envelope.partition of - Just p -> addAttribute traceSpan attrShibuyaPartition p - Nothing -> pure () + frameworkAttrs = + HashMap.fromList $ + [ (attrMessagingSystem, toAttribute ("shibuya" :: Text)), + (attrMessagingDestinationName, toAttribute pidText), + (attrMessagingOperation, toAttribute ("process" :: Text)), + (attrMessagingMessageId, toAttribute msgIdText) + ] + <> case ingested.envelope.partition of + Just p -> [(attrShibuyaPartition, toAttribute p)] + Nothing -> [] + mergedAttrs = HashMap.union ingested.envelope.attributes frameworkAttrs + addAttributes traceSpan mergedAttrs -- Increment in-flight and add inflight attributes now <- liftIO getCurrentTime diff --git a/src/Shibuya/Telemetry/Propagation.hs b/src/Shibuya/Telemetry/Propagation.hs --- a/src/Shibuya/Telemetry/Propagation.hs +++ b/src/Shibuya/Telemetry/Propagation.hs @@ -6,15 +6,20 @@ -- * Injection injectTraceContext, + currentTraceHeaders, -- * Re-export TraceHeaders, ) where +import Effectful (Eff, IOE, liftIO, (:>)) +import OpenTelemetry.Context qualified as Ctx +import OpenTelemetry.Context.ThreadLocal qualified as Ctx import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C import OpenTelemetry.Trace.Core (Span, SpanContext) import Shibuya.Core.Types (TraceHeaders) +import Shibuya.Telemetry.Effect (Tracing, isTracingEnabled) -- | Extract SpanContext from W3C trace headers. -- Returns Nothing if headers are missing or malformed. @@ -49,3 +54,37 @@ [ ("traceparent", traceparent), ("tracestate", tracestate) ] + +-- | Look up the currently-active OTel span and encode its context as +-- W3C trace headers, ready to attach to an outgoing message. +-- +-- Returns 'Nothing' when tracing is disabled or when there is no +-- active span at the call site (e.g. a producer running outside any +-- 'Shibuya.Telemetry.Effect.withSpan'/@withSpan'@ scope). +-- +-- This is the higher-level helper most call sites want — adapter +-- code that forwards a message (a DLQ write from @AckDeadLetter@, +-- a producer publishing a follow-on event) can do: +-- +-- @ +-- consumerHeaders <- currentTraceHeaders +-- let outgoingHeaders = +-- maybe originalHeaders (\\h -> h <> originalHeaders) consumerHeaders +-- @ +-- +-- and the failing-consumer's trace shows up linked to the resulting +-- message in the downstream trace store. The lower-level +-- 'injectTraceContext' is still exported for callers that already +-- hold a 'Span' handle from inside a 'withSpan''. +currentTraceHeaders :: + (Tracing :> es, IOE :> es) => + Eff es (Maybe TraceHeaders) +currentTraceHeaders = do + enabled <- isTracingEnabled + if not enabled + then pure Nothing + else liftIO $ do + ctx <- Ctx.getContext + case Ctx.lookupSpan ctx of + Nothing -> pure Nothing + Just sp -> Just <$> injectTraceContext sp diff --git a/test/Shibuya/Core/RetrySpec.hs b/test/Shibuya/Core/RetrySpec.hs --- a/test/Shibuya/Core/RetrySpec.hs +++ b/test/Shibuya/Core/RetrySpec.hs @@ -2,6 +2,7 @@ module Shibuya.Core.RetrySpec (spec) where +import Data.HashMap.Strict qualified as HashMap import Data.Time (nominalDiffTimeToSeconds) import Effectful (runEff) import Shibuya.Core.Ack (AckDecision (..), RetryDelay (..)) @@ -22,6 +23,7 @@ enqueuedAt = Nothing, traceContext = Nothing, attempt = a, + attributes = HashMap.empty, payload = () } diff --git a/test/Shibuya/Core/TypesSpec.hs b/test/Shibuya/Core/TypesSpec.hs --- a/test/Shibuya/Core/TypesSpec.hs +++ b/test/Shibuya/Core/TypesSpec.hs @@ -2,6 +2,7 @@ module Shibuya.Core.TypesSpec (spec) where +import Data.HashMap.Strict qualified as HashMap import Data.Time (UTCTime (..), fromGregorian) import Shibuya.Core.Types import Test.Hspec @@ -88,6 +89,7 @@ enqueuedAt = Just testTime, traceContext = Nothing, attempt = Nothing, + attributes = HashMap.empty, payload = msg } diff --git a/test/Shibuya/Runner/SupervisedSpec.hs b/test/Shibuya/Runner/SupervisedSpec.hs --- a/test/Shibuya/Runner/SupervisedSpec.hs +++ b/test/Shibuya/Runner/SupervisedSpec.hs @@ -5,6 +5,7 @@ import Control.Concurrent.NQE.Supervisor (Strategy (..)) import Control.Concurrent.STM (readTVarIO) import Control.Monad (forM, replicateM) +import Data.HashMap.Strict qualified as HashMap import Data.IORef (IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef) import Data.Text qualified as Text import Data.Time (UTCTime (..), fromGregorian) @@ -854,6 +855,7 @@ enqueuedAt = Just testTime, traceContext = Nothing, attempt = Nothing, + attributes = HashMap.empty, payload = "message-" <> show i } ackHandle = AckHandle $ \_ -> pure () @@ -877,6 +879,7 @@ enqueuedAt = Just testTime, traceContext = Nothing, attempt = Nothing, + attributes = HashMap.empty, payload = "message-" <> show i } ackHandle = AckHandle $ \_ -> pure () diff --git a/test/Shibuya/RunnerSpec.hs b/test/Shibuya/RunnerSpec.hs --- a/test/Shibuya/RunnerSpec.hs +++ b/test/Shibuya/RunnerSpec.hs @@ -2,6 +2,7 @@ module Shibuya.RunnerSpec (spec) where +import Data.HashMap.Strict qualified as HashMap import Data.IORef (IORef, modifyIORef', newIORef, readIORef) import Data.Text qualified as Text import Data.Time (UTCTime (..), fromGregorian) @@ -206,6 +207,7 @@ enqueuedAt = Just testTime, traceContext = Nothing, attempt = Nothing, + attributes = HashMap.empty, payload = "message-" <> show i } ackHandle = AckHandle $ \_ -> pure () -- No-op ack @@ -230,6 +232,7 @@ enqueuedAt = Just testTime, traceContext = Nothing, attempt = Nothing, + attributes = HashMap.empty, payload = "message-" <> show i } ackHandle = trackingAckHandle tracking msgId diff --git a/test/Shibuya/Telemetry/PropagationSpec.hs b/test/Shibuya/Telemetry/PropagationSpec.hs --- a/test/Shibuya/Telemetry/PropagationSpec.hs +++ b/test/Shibuya/Telemetry/PropagationSpec.hs @@ -5,9 +5,21 @@ import Data.ByteArray.Encoding (Base (..)) import Data.ByteString (ByteString) +import Effectful (runEff) +import OpenTelemetry.Attributes (emptyAttributes) +import OpenTelemetry.Trace.Core + ( InstrumentationLibrary (..), + createTracerProvider, + defaultSpanArguments, + emptyTracerProviderOptions, + makeTracer, + shutdownTracerProvider, + tracerOptions, + ) import OpenTelemetry.Trace.Core qualified as OTel import OpenTelemetry.Trace.Id qualified as OTel.Id -import Shibuya.Telemetry.Propagation (extractTraceContext) +import Shibuya.Telemetry.Effect (runTracing, runTracingNoop, withSpan) +import Shibuya.Telemetry.Propagation (currentTraceHeaders, extractTraceContext) import Test.Hspec spec :: Spec @@ -119,3 +131,46 @@ Nothing -> expectationFailure "Expected to extract trace context despite malformed tracestate" Just ctx -> do OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "0af7651916cd43dd8448eb211c80319c" + + describe "currentTraceHeaders" $ do + it "returns Nothing when tracing is disabled" $ do + result <- runEff $ runTracingNoop currentTraceHeaders + result `shouldBe` Nothing + + it "returns Nothing outside any active span" $ do + provider <- createTracerProvider [] emptyTracerProviderOptions + let tracer = + makeTracer + provider + ( InstrumentationLibrary + { libraryName = "shibuya-test", + libraryVersion = "", + librarySchemaUrl = "", + libraryAttributes = emptyAttributes + } + ) + tracerOptions + result <- runEff $ runTracing tracer currentTraceHeaders + _ <- shutdownTracerProvider provider + result `shouldBe` Nothing + + it "returns headers carrying the active span's traceparent" $ do + provider <- createTracerProvider [] emptyTracerProviderOptions + let tracer = + makeTracer + provider + ( InstrumentationLibrary + { libraryName = "shibuya-test", + libraryVersion = "", + librarySchemaUrl = "", + libraryAttributes = emptyAttributes + } + ) + tracerOptions + result <- runEff $ runTracing tracer $ withSpan "outer" defaultSpanArguments currentTraceHeaders + _ <- shutdownTracerProvider provider + case result of + Nothing -> expectationFailure "expected currentTraceHeaders to return Just" + Just hs -> case lookup "traceparent" hs of + Nothing -> expectationFailure "expected traceparent header to be present" + Just _ -> pure () diff --git a/test/Shibuya/Telemetry/SemanticSpec.hs b/test/Shibuya/Telemetry/SemanticSpec.hs --- a/test/Shibuya/Telemetry/SemanticSpec.hs +++ b/test/Shibuya/Telemetry/SemanticSpec.hs @@ -14,6 +14,7 @@ import Data.HashMap.Strict (HashMap) import Data.HashMap.Strict qualified as HashMap import Data.IORef (readIORef) +import Data.Int (Int64) import Data.Text (Text) import Effectful (runEff) import OpenTelemetry.Attributes @@ -21,6 +22,7 @@ PrimitiveAttribute (..), emptyAttributes, getAttributeMap, + toAttribute, ) import OpenTelemetry.Exporter.InMemory.Span (inMemoryListExporter) import OpenTelemetry.Trace.Core @@ -49,17 +51,7 @@ it "emits a process span with conventions-aligned attributes and events" $ do (processor, spansRef) <- inMemoryListExporter provider <- createTracerProvider [processor] emptyTracerProviderOptions - let tracer = - makeTracer - provider - ( InstrumentationLibrary - { libraryName = "shibuya-test", - libraryVersion = "", - librarySchemaUrl = "", - libraryAttributes = emptyAttributes - } - ) - tracerOptions + let tracer = mkTestTracer provider runEff $ runTracing tracer $ do let envelope = @@ -70,6 +62,7 @@ enqueuedAt = Nothing, traceContext = Nothing, attempt = Nothing, + attributes = HashMap.empty, payload = ("hello" :: Text) } ingested = @@ -102,7 +95,79 @@ evNames `shouldContain` ["shibuya.handler.completed"] _ -> expectationFailure $ "expected exactly one span, got " <> show (length spans) + + it "applies envelope.attributes onto the framework span (P0 fix, plan 9 F1/F2)" $ do + -- The adapter contributes broker-specific typed attributes via + -- 'Envelope.attributes'. The framework's processOne span must carry + -- them, and adapter-supplied keys must override framework defaults + -- of the same name (here: messaging.system flips from "shibuya" to + -- "kafka" because the adapter set it). + (processor, spansRef) <- inMemoryListExporter + provider <- createTracerProvider [processor] emptyTracerProviderOptions + let tracer = mkTestTracer provider + + runEff $ runTracing tracer $ do + let envelope = + Envelope + { messageId = MessageId "orders-2-42", + cursor = Nothing, + partition = Nothing, + enqueuedAt = Nothing, + traceContext = Nothing, + attempt = Nothing, + attributes = + HashMap.fromList + [ ("messaging.system", toAttribute ("kafka" :: Text)), + ( "messaging.kafka.destination.partition", + toAttribute (2 :: Int64) + ), + ( "messaging.kafka.message.offset", + toAttribute (42 :: Int64) + ) + ], + payload = ("hello" :: Text) + } + ingested = + Ingested + { envelope = envelope, + ack = AckHandle (\_ -> pure ()), + lease = Nothing + } + adapter = listAdapter [ingested] + handler _ = pure AckOk + procId = ProcessorId "orders-consumer" + _ <- runWithMetrics 4 procId adapter handler + pure () + + _ <- shutdownTracerProvider provider + spans <- readIORef spansRef + case spans of + [s] -> do + let attrs = getAttributeMap (spanAttributes s) + -- Adapter override wins. + attrs `shouldHaveTextAttribute` ("messaging.system", "kafka") + -- Adapter-typed attributes appear on the framework span. + attrs `shouldHaveIntAttribute` ("messaging.kafka.destination.partition", 2) + attrs `shouldHaveIntAttribute` ("messaging.kafka.message.offset", 42) + -- Framework defaults still set where the adapter did not override. + attrs `shouldHaveTextAttribute` ("messaging.destination.name", "orders-consumer") + attrs `shouldHaveTextAttribute` ("messaging.operation", "process") + attrs `shouldHaveTextAttribute` ("messaging.message.id", "orders-2-42") + _ -> + expectationFailure $ + "expected exactly one span, got " <> show (length spans) where + mkTestTracer p = + makeTracer + p + ( InstrumentationLibrary + { libraryName = "shibuya-test", + libraryVersion = "", + librarySchemaUrl = "", + libraryAttributes = emptyAttributes + } + ) + tracerOptions shouldHaveTextAttribute :: HashMap Text Attribute -> (Text, Text) -> Expectation shouldHaveTextAttribute attrs (k, expected) = case HashMap.lookup k attrs of