diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,70 @@ # Changelog +## 0.4.0.0 — 2026-04-29 + +### Breaking Changes + +- Tracks the `shibuya-core 0.4.0.0` release, which adds an + `attempt :: !(Maybe Attempt)` field to the `Envelope` record + exported from `Shibuya.Core.Types`. The adapter's + `consumerRecordToEnvelope` now sets this field to `Nothing`, + consistent with the upstream guidance that adapters which cannot + observe broker-side redeliveries (e.g. Kafka) report `Nothing`. + Downstream code that pattern-matches on `Envelope` with positional + patterns or with non-punned record patterns that name every field + must be updated. + +### Other Changes + +- Bumps the `shibuya-core` build-depends pin to `^>=0.4` in all three + packages of this repo. +- Drops three orphan `NFData` instances (`MessageId`, `Cursor`, + `Envelope a`) from `shibuya-kafka-adapter-bench/bench/Main.hs`; + these instances have been provided upstream by `shibuya-core` since + `0.2.0.0` and the orphans had become duplicate-instance hazards + whenever the bench resolved against a `shibuya-core` newer than + `0.1`. +- `shibuya-kafka-adapter-bench` and `shibuya-kafka-adapter-jitsurei` + are re-released at `0.4.0.0` to track the shared version of this + repo; neither has user-visible changes of its own. + +## 0.3.0.0 — 2026-04-22 + +Telemetry wire-format change. No Haskell API break — +`Shibuya.Adapter.Kafka.Tracing.traced`'s signature is unchanged — but +operators with dashboards filtering on the old attribute-key strings +or span name must update their queries. + +### Changed + +- Per-message spans now follow the OpenTelemetry messaging + semantic-conventions span-name pattern `" "`, + yielding e.g. `"orders process"` in place of the previous constant + `"shibuya.process.message"`. +- The `messaging.operation` attribute is now set to `"process"` on + every consumer span. +- The Kafka partition is now emitted as the typed Kafka-specific key + `messaging.kafka.destination.partition` (Int64), replacing the + never-defined `messaging.destination.partition.id` (Text). If the + envelope's partition text does not parse as an integer, the + shibuya-namespaced `shibuya.partition` is emitted as a defensive + fallback. +- The Kafka offset is now emitted as `messaging.kafka.message.offset` + (Int64), derived from `Envelope.cursor` when it is a `CursorInt`. + +### Aligned with + +- `Shibuya.Telemetry.Semantic` as of sibling `shibuya` plan 2 + (`docs/plans/2-align-opentelemetry-semantic-conventions.md` in the + shibuya repo). Attribute keys for the generic `messaging.*` + namespace are sourced from that module, which in turn derives them + from typed `AttributeKey` values in + `OpenTelemetry.SemanticConventions`. +- `OpenTelemetry.SemanticConventions` (new direct `build-depends`) + for the typed Kafka-specific keys + `messaging_kafka_destination_partition` and + `messaging_kafka_message_offset`. + ## 0.2.0.0 — 2026-04-18 Additive release. Adds one new exposed module, no changes to existing diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ ## Tracing (opt-in) -`Shibuya.Adapter.Kafka.Tracing.traced` is an opt-in stream transformer that wraps each emitted `Ingested` so that the downstream handler's eventual `finalize` call runs inside a Consumer-kind `shibuya.process.message` OpenTelemetry span. The span inherits the envelope's W3C `traceparent` as parent (from `Envelope.traceContext`) or opens a fresh root span when no parent is present, and is populated with the v1.27 messaging-conventions attributes (`messaging.system=kafka`, `messaging.destination.name`, `messaging.message.id`, and `messaging.destination.partition.id` when the partition is known). A caller that does not import this module pays nothing — no spans are opened and the adapter's public surface is unchanged. +`Shibuya.Adapter.Kafka.Tracing.traced` is an opt-in stream transformer that wraps each emitted `Ingested` so that the downstream handler's eventual `finalize` call runs inside a Consumer-kind OpenTelemetry span named following the messaging convention `" "` — e.g. `"orders process"` for a topic named `orders`. The span inherits the envelope's W3C `traceparent` as parent (from `Envelope.traceContext`) or opens a fresh root span when no parent is present, and is populated with the spec-aligned messaging attributes (`messaging.system=kafka`, `messaging.destination.name`, `messaging.operation=process`, `messaging.message.id`) and the Kafka-specific typed attributes `messaging.kafka.destination.partition` (Int64) and `messaging.kafka.message.offset` (Int64) when available. A caller that does not import this module pays nothing — no spans are opened and the adapter's public surface is unchanged. Typical wiring: diff --git a/shibuya-kafka-adapter.cabal b/shibuya-kafka-adapter.cabal --- a/shibuya-kafka-adapter.cabal +++ b/shibuya-kafka-adapter.cabal @@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-kafka-adapter -version: 0.2.0.0 +version: 0.4.0.0 synopsis: Kafka adapter for the Shibuya queue processing framework description: A Shibuya adapter that integrates with Apache Kafka via kafka-effectful @@ -53,20 +53,21 @@ QuasiQuotes build-depends: - , base ^>=4.21.0.0 - , bytestring ^>=0.12 - , containers ^>=0.7 - , effectful-core ^>=2.6.1.0 - , hs-opentelemetry-api ^>=0.3 - , hw-kafka-client >=5.3 && <6 - , hw-kafka-streamly ^>=0.1 - , kafka-effectful ^>=0.1 - , shibuya-core ^>=0.1.0.0 - , stm ^>=2.5 - , streamly ^>=0.11 - , streamly-core ^>=0.3 - , text ^>=2.1 - , time ^>=1.14 + , base ^>=4.21.0.0 + , bytestring ^>=0.12 + , containers ^>=0.7 + , effectful-core ^>=2.6.1.0 + , hs-opentelemetry-api ^>=0.3 + , hs-opentelemetry-semantic-conventions ^>=0.1 + , hw-kafka-client >=5.3 && <6 + , hw-kafka-streamly ^>=0.1 + , kafka-effectful ^>=0.1 + , shibuya-core ^>=0.4 + , stm ^>=2.5 + , streamly ^>=0.11 + , streamly-core ^>=0.3 + , text ^>=2.1 + , time ^>=1.14 hs-source-dirs: src default-language: GHC2024 @@ -97,12 +98,13 @@ Shibuya.Adapter.Kafka.TracingTest build-depends: - , async ^>=2.2 - , base ^>=4.21.0.0 + , async ^>=2.2 + , base ^>=4.21.0.0 , bytestring , containers , effectful-core - , hs-opentelemetry-api ^>=0.3 + , hs-opentelemetry-api ^>=0.3 + , hs-opentelemetry-semantic-conventions ^>=0.1 , hw-kafka-client , kafka-effectful , process @@ -112,7 +114,7 @@ , stm , streamly , streamly-core - , tasty ^>=1.5 - , tasty-hunit ^>=0.10 + , tasty ^>=1.5 + , tasty-hunit ^>=0.10 , text , time diff --git a/src/Shibuya/Adapter/Kafka/Convert.hs b/src/Shibuya/Adapter/Kafka/Convert.hs --- a/src/Shibuya/Adapter/Kafka/Convert.hs +++ b/src/Shibuya/Adapter/Kafka/Convert.hs @@ -35,6 +35,7 @@ * @partition@: @Just (show partitionId)@ * @enqueuedAt@: converted from Kafka timestamp if available * @traceContext@: extracted from @traceparent@/@tracestate@ headers +* @attempt@: 'Nothing' (Kafka does not expose a redelivery counter) * @payload@: the @crValue@ field (@Maybe ByteString@) -} consumerRecordToEnvelope :: @@ -47,6 +48,7 @@ , partition = Just (Text.pack (show (unPartitionId cr.crPartition))) , enqueuedAt = timestampToUTCTime cr.crTimestamp , traceContext = extractTraceHeaders cr.crHeaders + , attempt = Nothing , payload = cr.crValue } diff --git a/src/Shibuya/Adapter/Kafka/Tracing.hs b/src/Shibuya/Adapter/Kafka/Tracing.hs --- a/src/Shibuya/Adapter/Kafka/Tracing.hs +++ b/src/Shibuya/Adapter/Kafka/Tracing.hs @@ -3,17 +3,25 @@ Exposes a single stream transformer, 'traced', that wraps each 'Ingested' emitted by 'Shibuya.Adapter.Kafka.kafkaAdapter' so that when a downstream handler eventually calls the envelope's 'AckHandle.finalize', the call is -enclosed in an OpenTelemetry Consumer-kind span named -@shibuya.process.message@ (see 'Shibuya.Telemetry.Semantic.processMessageSpanName'). +enclosed in an OpenTelemetry Consumer-kind span following the messaging +semantic-conventions span-name pattern @" "@, +e.g. @"orders process"@ for a topic named @orders@ (see +'Shibuya.Telemetry.Semantic.processSpanName'). The span: * Is a child of the W3C trace context carried on the envelope, when present. The parent is recovered via 'Shibuya.Telemetry.Propagation.extractTraceContext' from @envelope.traceContext@. When absent, the span is a fresh root span. -* Carries the v1.27 messaging-conventions attribute set: - @messaging.system@, @messaging.destination.name@, @messaging.message.id@, - and — when a partition is known — @messaging.destination.partition.id@. +* Carries the spec-aligned messaging attributes @messaging.system=kafka@, + @messaging.destination.name@, @messaging.operation=process@, and + @messaging.message.id@. +* Carries the Kafka-specific typed attributes + @messaging.kafka.destination.partition@ (Int64, from @envelope.partition@ + parsed as an integer) and @messaging.kafka.message.offset@ (Int64, from + @envelope.cursor@ when it is a 'CursorInt'). If the partition text fails + to parse as an integer, the shibuya-namespaced @shibuya.partition@ is + emitted as a defensive fallback. Typical usage: @@ -32,13 +40,17 @@ ) where +import Data.Int (Int64) import Data.Text (Text) +import Data.Text.Read qualified as TR import Effectful (Eff, IOE, (:>)) import Kafka.Types (TopicName (..)) +import OpenTelemetry.Attributes (unkey) +import OpenTelemetry.SemanticConventions qualified as Sem import OpenTelemetry.Trace.Core (Span) import Shibuya.Core.AckHandle (AckHandle (..)) import Shibuya.Core.Ingested (Ingested (..)) -import Shibuya.Core.Types (Envelope (..), MessageId (..)) +import Shibuya.Core.Types (Cursor (..), Envelope (..), MessageId (..)) import Shibuya.Telemetry.Effect ( Tracing, addAttribute, @@ -48,18 +60,22 @@ import Shibuya.Telemetry.Propagation (extractTraceContext) import Shibuya.Telemetry.Semantic ( attrMessagingDestinationName, - attrMessagingDestinationPartitionId, attrMessagingMessageId, + attrMessagingOperation, attrMessagingSystem, + attrShibuyaPartition, consumerSpanArgs, - processMessageSpanName, + processSpanName, ) import Streamly.Data.Stream (Stream) import Streamly.Data.Stream qualified as Stream {- | Rewrite each 'Ingested' so that its 'AckHandle.finalize' opens a Consumer-kind span parented on the envelope's carried trace context -(when any) and populated with the v1.27 messaging attributes. +(when any), named @" process"@ per +'Shibuya.Telemetry.Semantic.processSpanName', and populated with the +spec-aligned messaging attributes plus the Kafka-specific +@messaging.kafka.*@ attributes. The topic name is supplied by the caller rather than parsed from the envelope's 'MessageId' — see the Decision Log of @@ -78,13 +94,15 @@ parentCtx = envelope.traceContext >>= extractTraceContext wrappedFinalize decision = withExtractedContext parentCtx $ - withSpan' processMessageSpanName consumerSpanArgs $ + withSpan' (processSpanName topicName) consumerSpanArgs $ \sp -> do populateAttrs sp topicName envelope finalize decision pure ing{ack = AckHandle wrappedFinalize} --- | Set the messaging-conventions v1.27 attribute set on a span. +{- | Set the spec-aligned messaging attribute set on a span, plus the +Kafka-specific typed attributes when the envelope carries them. +-} populateAttrs :: (Tracing :> es, IOE :> es) => Span -> @@ -92,10 +110,30 @@ Envelope v -> Eff es () populateAttrs sp topicName envelope = do + -- Generic messaging.* attributes (wire-names from shibuya-core's + -- aligned Semantic module, which derives them from typed AttributeKeys). addAttribute sp attrMessagingSystem ("kafka" :: Text) addAttribute sp attrMessagingDestinationName topicName + addAttribute sp attrMessagingOperation ("process" :: Text) let MessageId msgIdText = envelope.messageId addAttribute sp attrMessagingMessageId msgIdText + + -- Kafka-specific typed attributes (wire-names from AttributeKey + -- values in OpenTelemetry.SemanticConventions). case envelope.partition of - Just p -> addAttribute sp attrMessagingDestinationPartitionId p + Just p -> case TR.decimal p of + Right (n :: Int64, "") -> + addAttribute sp (unkey Sem.messaging_kafka_destination_partition) n + _ -> + -- Defensive: the partition text didn't parse as an int. + -- Emit the shibuya-namespaced fallback so the information + -- is not lost. + addAttribute sp attrShibuyaPartition p Nothing -> pure () + case envelope.cursor of + Just (CursorInt off) -> + addAttribute + sp + (unkey Sem.messaging_kafka_message_offset) + (fromIntegral off :: Int64) + _ -> pure () diff --git a/test/Shibuya/Adapter/Kafka/TracingTest.hs b/test/Shibuya/Adapter/Kafka/TracingTest.hs --- a/test/Shibuya/Adapter/Kafka/TracingTest.hs +++ b/test/Shibuya/Adapter/Kafka/TracingTest.hs @@ -21,8 +21,9 @@ import Data.Text.Encoding qualified as TE import Effectful (Eff, IOE, liftIO, runEff, (:>)) import Kafka.Types (TopicName (..)) -import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..), lookupAttribute) +import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..), lookupAttribute, unkey) import OpenTelemetry.Processor.Span (ShutdownResult (..), SpanProcessor (..)) +import OpenTelemetry.SemanticConventions qualified as Sem import OpenTelemetry.Trace.Core ( ImmutableSpan (..), SpanContext (..), @@ -53,6 +54,7 @@ [ testCase "envelope traceparent becomes span parent" testParentedSpan , testCase "missing traceContext yields root span" testRootSpan , testCase "messaging attributes populated from envelope" testAttributes + , testCase "span name follows spec pattern" testSpanName , testCase "AckDecision threads through wrapped finalize" testAckPassthrough ] @@ -99,6 +101,7 @@ , partition = Just "2" , enqueuedAt = Nothing , traceContext = tc + , attempt = Nothing , payload = () } @@ -176,6 +179,7 @@ case spans of [s] -> do let attrs = s.spanAttributes + -- Generic messaging.* attributes. assertEqual "messaging.system" (Just (AttributeValue (TextAttribute "kafka"))) @@ -185,13 +189,36 @@ (Just (AttributeValue (TextAttribute "orders"))) (lookupAttribute attrs "messaging.destination.name") assertEqual + "messaging.operation" + (Just (AttributeValue (TextAttribute "process"))) + (lookupAttribute attrs "messaging.operation") + assertEqual "messaging.message.id" (Just (AttributeValue (TextAttribute "orders-2-42"))) (lookupAttribute attrs "messaging.message.id") + -- Kafka-specific typed attributes (Int64 on the wire). assertEqual - "messaging.destination.partition.id" - (Just (AttributeValue (TextAttribute "2"))) + "messaging.kafka.destination.partition" + (Just (AttributeValue (IntAttribute 2))) + (lookupAttribute attrs (unkey Sem.messaging_kafka_destination_partition)) + assertEqual + "messaging.kafka.message.offset" + (Just (AttributeValue (IntAttribute 42))) + (lookupAttribute attrs (unkey Sem.messaging_kafka_message_offset)) + -- The deleted pre-alignment key must NOT be present. + assertEqual + "messaging.destination.partition.id absent" + Nothing (lookupAttribute attrs "messaging.destination.partition.id") + other -> assertFailure ("expected exactly one span, got " <> show (length other)) + +testSpanName :: Assertion +testSpanName = withRecordingTracer $ \spansRef tracer -> do + ackRef <- newIORef Nothing + runOneThroughTraced tracer (mkEnvelope Nothing) ackRef + spans <- readIORef spansRef + case spans of + [s] -> assertEqual "span name" "orders process" s.spanName other -> assertFailure ("expected exactly one span, got " <> show (length other)) testAckPassthrough :: Assertion