diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,28 @@ # Changelog +## 0.2.0.0 — 2026-04-22 + +### Breaking Changes + +- `Shibuya.Telemetry.Semantic`: rename `processMessageSpanName :: Text` to + `processSpanName :: Text -> Text`. The span name is now built from the + destination (processor id), yielding e.g. `"shibuya-consumer process"`, in + line with the OpenTelemetry messaging-spans recommendation. +- `Shibuya.Telemetry.Semantic`: remove `attrMessagingDestinationPartitionId` + (replaced by the Shibuya-specific `attrShibuyaPartition`). +- `Shibuya.Telemetry.Semantic`: remove `eventHandlerException`. + +### New Features + +- `Shibuya.Telemetry.Semantic`: add `attrMessagingOperation` and + `attrShibuyaPartition` attribute keys. Messaging attribute keys are now + sourced from the typed `AttributeKey` values exported by + `OpenTelemetry.SemanticConventions`, so upstream renames surface as + compile errors rather than silent wire-format drift. +- `Shibuya.Core.Types`: add `NFData` instances for `MessageId`, `Cursor`, + and `Envelope a` (when `a` itself has an `NFData` instance). Benchmark + authors no longer need to declare these as orphans. + ## 0.1.0.0 — 2026-02-24 Initial release. 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.1.0.0 +version: 0.2.0.0 synopsis: Supervised queue processing framework for Haskell description: A supervised queue processing framework inspired by Broadway (Elixir). @@ -65,11 +65,13 @@ base ^>=4.21.0.0, bytestring ^>=0.12.2.0, containers ^>=0.7, + deepseq ^>=1.5, effectful ^>=2.6.1.0, effectful-core ^>=2.6.1.0, generic-lens ^>=2.3.0.0, hs-opentelemetry-api ^>=0.3, hs-opentelemetry-propagator-w3c ^>=0.1, + hs-opentelemetry-semantic-conventions ^>=0.1, lens ^>=5.3.5, nqe ^>=0.6, stm ^>=2.5, @@ -115,6 +117,7 @@ Shibuya.RunnerSpec Shibuya.Telemetry.EffectSpec Shibuya.Telemetry.PropagationSpec + Shibuya.Telemetry.SemanticSpec build-depends: QuickCheck ^>=2.15, @@ -122,6 +125,7 @@ bytestring, effectful, hs-opentelemetry-api, + hs-opentelemetry-exporter-in-memory ^>=0.0, hspec ^>=2.11, memory, nqe, @@ -132,3 +136,5 @@ text, time, unliftio, + unordered-containers, + vector, 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 @@ -16,6 +16,7 @@ ) where +import Control.DeepSeq (NFData) import Data.ByteString (ByteString) import Data.String (IsString) import Shibuya.Prelude @@ -25,6 +26,7 @@ newtype MessageId = MessageId {unMessageId :: Text} deriving stock (Eq, Ord, Show, Generic) deriving newtype (IsString) + deriving anyclass (NFData) -- | Optional cursor / offset / global position. -- Used to track position in ordered streams. @@ -32,6 +34,7 @@ = CursorInt !Int | CursorText !Text deriving stock (Eq, Ord, Show, Generic) + deriving anyclass (NFData) -- | W3C Trace Context headers for distributed tracing. -- Contains traceparent and optionally tracestate headers. @@ -54,3 +57,4 @@ payload :: !msg } deriving stock (Eq, Show, Functor, Generic) + deriving anyclass (NFData) 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 @@ -74,19 +74,20 @@ ) import Shibuya.Telemetry.Propagation (extractTraceContext) import Shibuya.Telemetry.Semantic - ( attrMessagingDestinationPartitionId, + ( attrMessagingDestinationName, attrMessagingMessageId, + attrMessagingOperation, attrMessagingSystem, attrShibuyaAckDecision, attrShibuyaInflightCount, attrShibuyaInflightMax, + attrShibuyaPartition, consumerSpanArgs, eventAckDecision, eventHandlerCompleted, - eventHandlerException, eventHandlerStarted, mkEvent, - processMessageSpanName, + processSpanName, ) import Streamly.Data.Fold qualified as Fold import Streamly.Data.Stream qualified as Stream @@ -160,7 +161,7 @@ runInIO $ -- Catch ProcessorHalt to prevent propagation via link -- (Halt is intentional, not a failure - other processors should continue) - ( runIngesterAndProcessor metricsVar doneVar inboxSize concurrency adapter handler + ( runIngesterAndProcessor metricsVar procId doneVar inboxSize concurrency adapter handler `catch` \(ProcessorHalt _) -> pure () -- Convert halt to graceful exit ) `finally` unregisterProcessor master procId @@ -205,7 +206,7 @@ runIngesterWithMetrics metricsVar adapter.source inbox -- Drain remaining messages from inbox - drainInboxWithMetrics metricsVar handler inbox + drainInboxWithMetrics metricsVar procId handler inbox -- Mark done liftIO $ atomically $ writeTVar doneVar True @@ -224,13 +225,14 @@ runIngesterAndProcessor :: (IOE :> es, Tracing :> es) => TVar ProcessorMetrics -> + ProcessorId -> TVar Bool -> Natural -> Concurrency -> Adapter es msg -> Handler es msg -> Eff es () -runIngesterAndProcessor metricsVar doneVar inboxSize concurrency adapter handler = do +runIngesterAndProcessor metricsVar procId doneVar inboxSize concurrency adapter handler = do -- Create bounded inbox (this is where inboxSize is used for backpressure) inbox <- liftIO $ newBoundedInbox inboxSize @@ -247,7 +249,7 @@ UIO.withAsync ingesterWithSignal $ \_ -> -- Processor: process messages, exit when stream done and inbox empty - runInIO $ processUntilDrained metricsVar concurrency handler inbox streamDoneVar + runInIO $ processUntilDrained metricsVar procId concurrency handler inbox streamDoneVar -- Mark done when processor exits liftIO $ atomically $ writeTVar doneVar True @@ -294,12 +296,13 @@ processUntilDrained :: (IOE :> es, Tracing :> es) => TVar ProcessorMetrics -> + ProcessorId -> Concurrency -> Handler es msg -> Inbox (Ingested es msg) -> TVar Bool -> Eff es () -processUntilDrained metricsVar concurrency handler inbox streamDoneVar = do +processUntilDrained metricsVar procId concurrency handler inbox streamDoneVar = do haltRef <- liftIO $ newIORef Nothing let maxConc = case concurrency of @@ -309,7 +312,7 @@ withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do let inboxStream = inboxToStream inbox streamDoneVar haltRef - processAction = runInIO . processOne metricsVar maxConc haltRef handler + processAction = runInIO . processOne metricsVar procId maxConc haltRef handler case concurrency of Serial -> @@ -333,10 +336,11 @@ drainInboxWithMetrics :: (IOE :> es, Tracing :> es) => TVar ProcessorMetrics -> + ProcessorId -> Handler es msg -> Inbox (Ingested es msg) -> Eff es () -drainInboxWithMetrics metricsVar handler inbox = do +drainInboxWithMetrics metricsVar procId handler inbox = do haltRef <- liftIO $ newIORef Nothing go haltRef where @@ -344,7 +348,7 @@ empty <- liftIO $ mailboxEmpty inbox unless empty $ do ingested <- liftIO $ receive inbox - processOne metricsVar 1 haltRef handler ingested + processOne metricsVar procId 1 haltRef handler ingested -- Check if halted halted <- liftIO $ readIORef haltRef case halted of @@ -356,25 +360,29 @@ processOne :: (IOE :> es, Tracing :> es) => TVar ProcessorMetrics -> + ProcessorId -> Int -> IORef (Maybe HaltReason) -> Handler es msg -> Ingested es msg -> Eff es () -processOne metricsVar maxConc haltRef handler ingested = do +processOne metricsVar procId maxConc haltRef handler ingested = do -- Extract parent context from message headers for distributed tracing let parentCtx = ingested.envelope.traceContext >>= extractTraceContext + ProcessorId pidText = procId withExtractedContext parentCtx $ - withSpan' processMessageSpanName consumerSpanArgs $ \traceSpan -> do + withSpan' (processSpanName pidText) consumerSpanArgs $ \traceSpan -> do -- Add messaging attributes 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 attrMessagingDestinationPartitionId p + Just p -> addAttribute traceSpan attrShibuyaPartition p Nothing -> pure () -- Increment in-flight and add inflight attributes @@ -406,7 +414,6 @@ ) ( \ex -> do recordException traceSpan ex - addEvent traceSpan (mkEvent eventHandlerException []) pure $ Left $ HandlerException $ Text.pack $ show ex ) diff --git a/src/Shibuya/Telemetry/Semantic.hs b/src/Shibuya/Telemetry/Semantic.hs --- a/src/Shibuya/Telemetry/Semantic.hs +++ b/src/Shibuya/Telemetry/Semantic.hs @@ -1,26 +1,31 @@ -- | Semantic conventions for Shibuya OpenTelemetry instrumentation. -- Provides span names, attribute keys, and helpers following OTel conventions. +-- +-- Attribute keys for the OTel @messaging.*@ namespace are derived from the +-- typed @AttributeKey@ values exported by @OpenTelemetry.SemanticConventions@, +-- so an upstream rename surfaces as a Haskell compile error rather than a +-- silent wire-format drift. module Shibuya.Telemetry.Semantic ( -- * Span Names - processMessageSpanName, + processSpanName, ingestSpanName, -- * Attribute Keys (OTel Messaging Conventions) attrMessagingSystem, attrMessagingMessageId, attrMessagingDestinationName, - attrMessagingDestinationPartitionId, + attrMessagingOperation, -- * Attribute Keys (Shibuya-specific) attrShibuyaProcessorId, attrShibuyaInflightCount, attrShibuyaInflightMax, attrShibuyaAckDecision, + attrShibuyaPartition, -- * Event Names eventHandlerStarted, eventHandlerCompleted, - eventHandlerException, eventAckDecision, -- * SpanArguments Helpers @@ -34,7 +39,8 @@ import Data.HashMap.Strict qualified as HashMap import Data.Text (Text) -import OpenTelemetry.Attributes (Attribute) +import OpenTelemetry.Attributes (Attribute, AttributeKey (..)) +import OpenTelemetry.SemanticConventions qualified as Sem import OpenTelemetry.Trace.Core ( NewEvent (..), SpanArguments (..), @@ -46,8 +52,18 @@ -------------------------------------------------------------------------------- -- | Span name for processing a single message. -processMessageSpanName :: Text -processMessageSpanName = "shibuya.process.message" +-- +-- Follows the OpenTelemetry messaging-spans recommendation that consumer +-- spans be named @\" \"@ — for Shibuya the +-- destination is the processor id and the operation is @\"process\"@, +-- yielding e.g. @\"shibuya-consumer process\"@. +-- +-- This intentionally departs from +-- @hs-opentelemetry-instrumentation-hw-kafka-client@'s +-- @\"process \"@ order in favor of the spec's order; see +-- @docs/plans/2-align-opentelemetry-semantic-conventions.md@. +processSpanName :: Text -> Text +processSpanName destination = destination <> " process" -- | Span name for the ingester lifecycle. ingestSpanName :: Text @@ -56,63 +72,94 @@ -------------------------------------------------------------------------------- -- Attribute Keys (OTel Messaging Semantic Conventions) -- See: https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/ +-- +-- Each constant is the wire string of a typed 'AttributeKey' exported by +-- 'OpenTelemetry.SemanticConventions'. Using @unkey@ (the AttributeKey +-- newtype selector) keeps these strings in lock-step with the upstream +-- code-generated keys. -------------------------------------------------------------------------------- --- | The messaging system identifier. +-- | The messaging system identifier (@messaging.system@). attrMessagingSystem :: Text -attrMessagingSystem = "messaging.system" +attrMessagingSystem = unkey Sem.messaging_system --- | The message identifier. +-- | The message identifier (@messaging.message.id@). attrMessagingMessageId :: Text -attrMessagingMessageId = "messaging.message.id" +attrMessagingMessageId = unkey Sem.messaging_message_id --- | The destination (queue/topic) name. +-- | The destination (queue/topic) name (@messaging.destination.name@). attrMessagingDestinationName :: Text -attrMessagingDestinationName = "messaging.destination.name" +attrMessagingDestinationName = unkey Sem.messaging_destination_name --- | The partition identifier. -attrMessagingDestinationPartitionId :: Text -attrMessagingDestinationPartitionId = "messaging.destination.partition.id" +-- | The messaging operation (@messaging.operation@). +-- +-- One of the spec-defined enum values: @publish@, @receive@, @process@, +-- @settle@, @create@. Shibuya's per-message span uses @"process"@. +attrMessagingOperation :: Text +attrMessagingOperation = unkey Sem.messaging_operation -------------------------------------------------------------------------------- -- Attribute Keys (Shibuya-specific) +-- +-- These keys are intentionally outside the upstream OTel spec. The +-- semantic-conventions guide is explicit that non-spec attributes should +-- carry a vendor prefix; @shibuya.*@ satisfies that and cannot collide +-- with any current or future spec key. -------------------------------------------------------------------------------- --- | The processor identifier. +-- | The processor identifier (@shibuya.processor.id@). +-- Shibuya-specific: identifies a deployment-level processor instance. attrShibuyaProcessorId :: Text attrShibuyaProcessorId = "shibuya.processor.id" --- | Current number of in-flight messages. +-- | Current number of in-flight messages (@shibuya.inflight.count@). +-- Shibuya-specific: a property of Shibuya's supervision/concurrency model. attrShibuyaInflightCount :: Text attrShibuyaInflightCount = "shibuya.inflight.count" --- | Maximum concurrency limit. +-- | Maximum concurrency limit (@shibuya.inflight.max@). +-- Shibuya-specific: a property of Shibuya's supervision/concurrency model. attrShibuyaInflightMax :: Text attrShibuyaInflightMax = "shibuya.inflight.max" --- | The ack decision made by the handler. +-- | The ack decision made by the handler (@shibuya.ack.decision@). +-- Shibuya-specific: there is no upstream key for an explicit ack decision. attrShibuyaAckDecision :: Text attrShibuyaAckDecision = "shibuya.ack.decision" +-- | A generic partition identifier (@shibuya.partition@). +-- +-- Shibuya-specific: in semantic-conventions v1.24 there is no portable +-- @messaging.destination.partition.id@ key — only system-specific ones +-- (@messaging.kafka.destination.partition@, +-- @messaging.eventhubs.destination.partition.id@). System-specific +-- adapters that know which broker is in use may emit those in addition, +-- but a generic Shibuya envelope only carries an opaque partition string, +-- which we surface under this vendor-prefixed key. +attrShibuyaPartition :: Text +attrShibuyaPartition = "shibuya.partition" + -------------------------------------------------------------------------------- -- Event Names +-- +-- Shibuya-namespaced; the upstream spec does not define handler.* events. +-- The standard @exception@ event (with @exception.type@/@exception.message@/ +-- @exception.stacktrace@) is emitted by 'OpenTelemetry.Trace.Core.recordException' +-- and so is not duplicated here. -------------------------------------------------------------------------------- --- | Event recorded when handler execution starts. +-- | Event recorded when handler execution starts (@shibuya.handler.started@). eventHandlerStarted :: Text -eventHandlerStarted = "handler.started" +eventHandlerStarted = "shibuya.handler.started" --- | Event recorded when handler execution completes successfully. +-- | Event recorded when handler execution completes successfully +-- (@shibuya.handler.completed@). eventHandlerCompleted :: Text -eventHandlerCompleted = "handler.completed" - --- | Event recorded when handler throws an exception. -eventHandlerException :: Text -eventHandlerException = "handler.exception" +eventHandlerCompleted = "shibuya.handler.completed" --- | Event recorded when ack decision is made. +-- | Event recorded when ack decision is made (@shibuya.ack.decision@). eventAckDecision :: Text -eventAckDecision = "ack.decision" +eventAckDecision = "shibuya.ack.decision" -------------------------------------------------------------------------------- -- SpanArguments Helpers diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -9,6 +9,7 @@ import Shibuya.RunnerSpec qualified import Shibuya.Telemetry.EffectSpec qualified import Shibuya.Telemetry.PropagationSpec qualified +import Shibuya.Telemetry.SemanticSpec qualified import Test.Hspec main :: IO () @@ -20,3 +21,4 @@ Shibuya.Runner.SupervisedSpec.spec Shibuya.Telemetry.EffectSpec.spec Shibuya.Telemetry.PropagationSpec.spec + Shibuya.Telemetry.SemanticSpec.spec diff --git a/test/Shibuya/Telemetry/SemanticSpec.hs b/test/Shibuya/Telemetry/SemanticSpec.hs new file mode 100644 --- /dev/null +++ b/test/Shibuya/Telemetry/SemanticSpec.hs @@ -0,0 +1,124 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | Asserts that spans emitted by 'processOne' carry the wire names +-- defined by the OpenTelemetry messaging semantic conventions and the +-- shibuya-namespaced fallbacks. This is the guard that catches drift +-- between Shibuya's emitted attributes and the upstream spec — a future +-- rename in @hs-opentelemetry-semantic-conventions@ will break both +-- compilation (because @Shibuya.Telemetry.Semantic@ derives the strings +-- from typed @AttributeKey@s) and these assertions (because the wire +-- string changed). +module Shibuya.Telemetry.SemanticSpec (spec) where + +import Data.Foldable (toList) +import Data.HashMap.Strict (HashMap) +import Data.HashMap.Strict qualified as HashMap +import Data.IORef (readIORef) +import Data.Text (Text) +import Effectful (runEff) +import OpenTelemetry.Attributes + ( Attribute (..), + PrimitiveAttribute (..), + emptyAttributes, + getAttributeMap, + ) +import OpenTelemetry.Exporter.InMemory.Span (inMemoryListExporter) +import OpenTelemetry.Trace.Core + ( Event (..), + ImmutableSpan (..), + InstrumentationLibrary (..), + createTracerProvider, + emptyTracerProviderOptions, + makeTracer, + shutdownTracerProvider, + tracerOptions, + ) +import OpenTelemetry.Util (appendOnlyBoundedCollectionValues) +import Shibuya.Adapter.Mock (listAdapter) +import Shibuya.Core.Ack (AckDecision (..)) +import Shibuya.Core.AckHandle (AckHandle (..)) +import Shibuya.Core.Ingested (Ingested (..)) +import Shibuya.Core.Types (Envelope (..), MessageId (..)) +import Shibuya.Runner.Metrics (ProcessorId (..)) +import Shibuya.Runner.Supervised (runWithMetrics) +import Shibuya.Telemetry.Effect (runTracing) +import Test.Hspec + +spec :: Spec +spec = describe "Shibuya.Telemetry.Semantic (wire-format)" $ do + 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 + + runEff $ runTracing tracer $ do + let envelope = + Envelope + { messageId = MessageId "m-1", + cursor = Nothing, + partition = Nothing, + enqueuedAt = Nothing, + traceContext = Nothing, + payload = ("hello" :: Text) + } + ingested = + Ingested + { envelope = envelope, + ack = AckHandle (\_ -> pure ()), + lease = Nothing + } + adapter = listAdapter [ingested] + handler _ = pure AckOk + procId = ProcessorId "test-proc" + _ <- runWithMetrics 4 procId adapter handler + pure () + + _ <- shutdownTracerProvider provider + spans <- readIORef spansRef + case spans of + [s] -> do + spanName s `shouldBe` "test-proc process" + let attrs = getAttributeMap (spanAttributes s) + attrs `shouldHaveTextAttribute` ("messaging.system", "shibuya") + attrs `shouldHaveTextAttribute` ("messaging.message.id", "m-1") + attrs `shouldHaveTextAttribute` ("messaging.destination.name", "test-proc") + attrs `shouldHaveTextAttribute` ("messaging.operation", "process") + attrs `shouldHaveTextAttribute` ("shibuya.ack.decision", "ack_ok") + attrs `shouldHaveIntAttribute` ("shibuya.inflight.count", 1) + attrs `shouldHaveIntAttribute` ("shibuya.inflight.max", 1) + let evNames = map eventName (toList (appendOnlyBoundedCollectionValues (spanEvents s))) + evNames `shouldContain` ["shibuya.handler.started"] + evNames `shouldContain` ["shibuya.handler.completed"] + _ -> + expectationFailure $ "expected exactly one span, got " <> show (length spans) + where + shouldHaveTextAttribute :: HashMap Text Attribute -> (Text, Text) -> Expectation + shouldHaveTextAttribute attrs (k, expected) = + case HashMap.lookup k attrs of + Just (AttributeValue (TextAttribute v)) -> v `shouldBe` expected + Just other -> + expectationFailure $ + "attribute " <> show k <> " was not a Text: " <> show other + Nothing -> + expectationFailure $ + "attribute " <> show k <> " missing; have keys " <> show (HashMap.keys attrs) + shouldHaveIntAttribute :: HashMap Text Attribute -> (Text, Int) -> Expectation + shouldHaveIntAttribute attrs (k, expected) = + case HashMap.lookup k attrs of + Just (AttributeValue (IntAttribute v)) -> v `shouldBe` fromIntegral expected + Just other -> + expectationFailure $ + "attribute " <> show k <> " was not an Int: " <> show other + Nothing -> + expectationFailure $ + "attribute " <> show k <> " missing; have keys " <> show (HashMap.keys attrs)