diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,35 @@ This package follows the [Haskell Package Versioning Policy](https://pvp.haskell.org/). +## 0.3.0.0 — 2026-05-31 + +### Breaking Changes + +- Upgrade OpenTelemetry support to the `hs-opentelemetry` 1.0 package + family. The library now requires `hs-opentelemetry-api ^>=1.0`, + `hs-opentelemetry-sdk ^>=1.0`, 1.0 exporters, and + `hs-opentelemetry-semantic-conventions >=1.40 && <2`. Downstream + build plans pinned to the 0.x package family must upgrade. + +### New Features + +- Add `producerRecordAttributesWith` and `consumerRecordAttributesWith` + attribute builders that honor the semantic-convention stability mode. +- Add `kafkaHeadersToTextMap` and `textMapToKafkaHeaders` propagation + helpers built on the OpenTelemetry 1.0 `TextMap` carrier. + +### Other Changes + +- Align Kafka messaging attributes with the v1.40 semantic-convention + behavior used by `hs-opentelemetry-instrumentation-hw-kafka-client` + 1.0. Legacy messaging keys remain the default; set + `OTEL_SEMCONV_STABILITY_OPT_IN=messaging` for stable names or + `OTEL_SEMCONV_STABILITY_OPT_IN=messaging/dup` to emit both during + migration. +- Propagation now uses the OpenTelemetry 1.0 `TextMap` carrier + internally while preserving the existing request-header bridge + helpers for callers that imported them directly. + ## 0.2.0.0 — 2026-05-06 Additive release. No breaking changes to existing modules. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -13,6 +13,35 @@ - Resource-safe interpreters that acquire and release Kafka handles via `bracket` - Errors surfaced through `Effectful.Error.Static` (`Error KafkaError`) +## Local Kafka and Jaeger + +This repo includes the same Redpanda + Jaeger `process-compose` +setup used by `shibuya-kafka-adapter`. It uses `rpk container`, so +Docker or a compatible Docker socket provider such as Colima must be +running. From a dev shell, start the services in one terminal: + +```bash +just process-up +``` + +Then create the example topics and run the OpenTelemetry tracing +demo from another terminal: + +```bash +just create-topics +just otel-example +``` + +`just otel-example` produces one record to Redpanda, consumes it back, +and prints matching producer and consumer trace IDs. Jaeger receives +OTLP on `localhost:4318`, and its UI is available at +`http://localhost:16686` while services are running. Stop everything +with: + +```bash +just process-down +``` + ## Usage ### Producer @@ -238,11 +267,18 @@ traced interpreters open a `Producer`-kind span around every record-sending operation and a `Consumer`-kind span around every successful `pollMessage` / per-record success of `pollMessageBatch`, -populated with the OpenTelemetry messaging semantic conventions -(`messaging.system`, `messaging.destination.name`, -`messaging.operation`, `messaging.kafka.destination.partition`, +populated with the OpenTelemetry messaging semantic conventions. +By default the attributes use the legacy-compatible names emitted by +`hs-opentelemetry`'s Kafka instrumentation (`messaging.system`, +`messaging.destination.name`, `messaging.operation`, +`messaging.kafka.destination.partition`, `messaging.kafka.message.offset`, `messaging.kafka.message.key`, -`messaging.kafka.consumer.group`). The current OTel context is +`messaging.kafka.consumer.group`). Set +`OTEL_SEMCONV_STABILITY_OPT_IN=messaging` to emit the stable v1.40 +operation and consumer-group names (`messaging.operation.name`, +`messaging.operation.type`, `messaging.consumer.group.name`) instead, +or `OTEL_SEMCONV_STABILITY_OPT_IN=messaging/dup` to emit both old and +stable names while migrating dashboards. The current OTel context is injected into the outgoing record's headers as W3C `traceparent` / `tracestate` so downstream consumers can extract it and continue the trace. diff --git a/examples/OtelTracing.hs b/examples/OtelTracing.hs --- a/examples/OtelTracing.hs +++ b/examples/OtelTracing.hs @@ -182,7 +182,7 @@ case producerResult of Left (_cs, err) -> do hPutStrLn stderr $ "producer error: " <> show err - shutdownTracerProvider tp + _ <- shutdownTracerProvider tp Nothing exitFailure Right () -> pure () @@ -196,13 +196,13 @@ case consumerResult of Left (_cs, err) -> do hPutStrLn stderr $ "consumer error: " <> show err - shutdownTracerProvider tp + _ <- shutdownTracerProvider tp Nothing exitFailure Right () -> pure () pid <- readIORef producerTraceRef cid <- readIORef consumerTraceRef - shutdownTracerProvider tp + _ <- shutdownTracerProvider tp Nothing case (pid, cid) of (Just p, Just c) -> do putStrLn $ "[otel-tracing] producer trace id: " <> Text.unpack p diff --git a/kafka-effectful.cabal b/kafka-effectful.cabal --- a/kafka-effectful.cabal +++ b/kafka-effectful.cabal @@ -1,6 +1,6 @@ cabal-version: 3.4 name: kafka-effectful -version: 0.2.0.0 +version: 0.3.0.0 synopsis: Effectful effects for hw-kafka-client description: Effectful effects and interpreters for hw-kafka-client, a Haskell @@ -56,8 +56,8 @@ , case-insensitive >=1.2 && <1.3 , containers >=0.6 && <0.8 , effectful-core >=2.5 && <2.7 - , hs-opentelemetry-api ^>=0.3 - , hs-opentelemetry-semantic-conventions ^>=0.1 + , hs-opentelemetry-api ^>=1.0 + , hs-opentelemetry-semantic-conventions >=1.40 && <2 , http-types >=0.12 && <0.13 , hw-kafka-client >=5.3 && <6 , text >=2.0 && <2.2 @@ -148,9 +148,9 @@ , base >=4.21 && <5 , bytestring >=0.11 && <0.13 , effectful-core >=2.5 && <2.7 - , hs-opentelemetry-api ^>=0.3 - , hs-opentelemetry-exporter-otlp ^>=0.1 - , hs-opentelemetry-sdk ^>=0.1 + , hs-opentelemetry-api ^>=1.0 + , hs-opentelemetry-exporter-otlp ^>=1.0 + , hs-opentelemetry-sdk ^>=1.0 , hw-kafka-client >=5.3 && <6 , kafka-effectful , text >=2.0 && <2.2 @@ -184,19 +184,19 @@ Kafka.Effectful.OpenTelemetry.ShibuyaCompatibilityTest build-depends: - , base >=4.21 && <5 - , bytestring >=0.11 && <0.13 - , case-insensitive >=1.2 && <1.3 - , containers >=0.6 && <0.8 - , effectful-core >=2.5 && <2.7 - , hs-opentelemetry-api ^>=0.3 - , hs-opentelemetry-exporter-in-memory ^>=0.0.1 - , hs-opentelemetry-sdk ^>=0.1 - , hs-opentelemetry-semantic-conventions ^>=0.1 - , http-types >=0.12 && <0.13 - , hw-kafka-client >=5.3 && <6 + , base >=4.21 && <5 + , bytestring >=0.11 && <0.13 + , case-insensitive >=1.2 && <1.3 + , containers >=0.6 && <0.8 + , effectful-core >=2.5 && <2.7 + , hs-opentelemetry-api ^>=1.0 + , hs-opentelemetry-exporter-in-memory ^>=1.0 + , hs-opentelemetry-sdk ^>=1.0 + , hs-opentelemetry-semantic-conventions >=1.40 && <2 + , http-types >=0.12 && <0.13 + , hw-kafka-client >=5.3 && <6 , kafka-effectful , tasty ^>=1.5 , tasty-hunit ^>=0.10 - , text >=2.0 && <2.2 - , unordered-containers >=0.2 && <0.3 + , text >=2.0 && <2.2 + , unordered-containers >=0.2 && <0.3 diff --git a/src/Kafka/Effectful/OpenTelemetry.hs b/src/Kafka/Effectful/OpenTelemetry.hs --- a/src/Kafka/Effectful/OpenTelemetry.hs +++ b/src/Kafka/Effectful/OpenTelemetry.hs @@ -28,13 +28,17 @@ -- * Attribute helpers producerRecordAttributes, + producerRecordAttributesWith, consumerRecordAttributes, + consumerRecordAttributesWith, producerSpanName, consumerSpanName, -- * Trace-context propagation extractTraceContextFromRecord, injectTraceContextIntoRecord, + kafkaHeadersToTextMap, + textMapToKafkaHeaders, kafkaHeadersToRequestHeaders, requestHeadersToKafkaHeaders, ) @@ -46,11 +50,15 @@ extractTraceContextFromRecord, injectTraceContextIntoRecord, kafkaHeadersToRequestHeaders, + kafkaHeadersToTextMap, requestHeadersToKafkaHeaders, + textMapToKafkaHeaders, ) import Kafka.Effectful.OpenTelemetry.Semantic ( consumerRecordAttributes, + consumerRecordAttributesWith, consumerSpanName, producerRecordAttributes, + producerRecordAttributesWith, producerSpanName, ) diff --git a/src/Kafka/Effectful/OpenTelemetry/Consumer/Interpreter.hs b/src/Kafka/Effectful/OpenTelemetry/Consumer/Interpreter.hs --- a/src/Kafka/Effectful/OpenTelemetry/Consumer/Interpreter.hs +++ b/src/Kafka/Effectful/OpenTelemetry/Consumer/Interpreter.hs @@ -32,7 +32,6 @@ import Control.Monad (void) import Data.ByteString (ByteString) import Data.Foldable (for_) -import Data.Map.Strict qualified as Map import Effectful (Eff, IOE, (:>)) import Effectful qualified import Effectful.Dispatch.Dynamic (EffectHandler, interpret) @@ -41,7 +40,7 @@ import Effectful.Exception qualified as Exception import Kafka.Consumer (RdKafkaRespErrT (..)) import Kafka.Consumer qualified as K -import Kafka.Consumer.ConsumerProperties (ConsumerProperties (cpProps)) +import Kafka.Consumer.ConsumerProperties (ConsumerProperties) import Kafka.Consumer.Subscription (Subscription) import Kafka.Consumer.Types (ConsumerRecord (crTopic)) import Kafka.Effectful.Consumer.Effect (KafkaConsumer (..)) @@ -49,14 +48,12 @@ extractTraceContextFromRecord, ) import Kafka.Effectful.OpenTelemetry.Semantic ( - consumerRecordAttributes, + consumerRecordAttributesWith, consumerSpanName, ) import Kafka.Types (KafkaError (..)) -import OpenTelemetry.Attributes.Attribute (toAttribute) -import OpenTelemetry.Attributes.Map (AttributeMap, insertAttributeByKey) import OpenTelemetry.Context.ThreadLocal (attachContext, getContext) -import OpenTelemetry.SemanticConventions (messaging_kafka_consumer_group) +import OpenTelemetry.SemanticsConfig (getSemanticsOptions, lookupStability) import OpenTelemetry.Trace.Core ( SpanArguments (kind), SpanKind (Consumer), @@ -189,23 +186,14 @@ Eff es a -> Eff es a withConsumerSpan tracer props cr action = do + semOpts <- Effectful.liftIO $ lookupStability "messaging" <$> getSemanticsOptions inboundCtx <- Effectful.liftIO $ do currentCtx <- getContext extractTraceContextFromRecord cr currentCtx void $ attachContext inboundCtx - inSpan'' tracer (consumerSpanName (crTopic cr)) spanArgs $ \_span -> action + inSpan'' tracer (consumerSpanName (crTopic cr)) (spanArgs semOpts) $ \_span -> action where - spanArgs = + spanArgs semOpts = addAttributesToSpanArguments - (withConsumerGroup (consumerRecordAttributes cr)) + (consumerRecordAttributesWith semOpts props cr) defaultSpanArguments{kind = Consumer} - - withConsumerGroup :: AttributeMap -> AttributeMap - withConsumerGroup attrs = - case Map.lookup "group.id" (cpProps props) of - Just groupId -> - insertAttributeByKey - messaging_kafka_consumer_group - (toAttribute groupId) - attrs - Nothing -> attrs diff --git a/src/Kafka/Effectful/OpenTelemetry/Producer/Interpreter.hs b/src/Kafka/Effectful/OpenTelemetry/Producer/Interpreter.hs --- a/src/Kafka/Effectful/OpenTelemetry/Producer/Interpreter.hs +++ b/src/Kafka/Effectful/OpenTelemetry/Producer/Interpreter.hs @@ -32,6 +32,7 @@ import Control.Concurrent.MVar qualified as Concurrent import Data.Foldable (for_) +import Data.Text qualified as Text import Effectful (Eff, IOE, (:>)) import Effectful qualified import Effectful.Dispatch.Dynamic (EffectHandler, interpret) @@ -41,7 +42,7 @@ injectTraceContextIntoRecord, ) import Kafka.Effectful.OpenTelemetry.Semantic ( - producerRecordAttributes, + producerRecordAttributesWith, producerSpanName, ) import Kafka.Effectful.Producer.Effect (KafkaProducer (..)) @@ -50,15 +51,22 @@ import Kafka.Producer.ProducerProperties (ProducerProperties) import Kafka.Transaction qualified as K import Kafka.Types (KafkaError) +import OpenTelemetry.Attributes.Key (unkey) import OpenTelemetry.Context qualified as Context import OpenTelemetry.Context.ThreadLocal (getContext) +import OpenTelemetry.SemanticConventions (error_type) +import OpenTelemetry.SemanticsConfig (getSemanticsOptions, lookupStability) import OpenTelemetry.Trace.Core ( + Span, SpanArguments (kind), SpanKind (Producer), + SpanStatus (Error), Tracer, + addAttribute, addAttributesToSpanArguments, defaultSpanArguments, inSpan'', + setStatus, ) {- | Run the 'KafkaProducer' effect with OpenTelemetry tracing. @@ -102,32 +110,42 @@ EffectHandler KafkaProducer es handleTracedProducer tracer producer _env = \case ProduceMessage record -> - withProducerSpan tracer record $ \instrumentedRecord -> do + withProducerSpan tracer record $ \span_ instrumentedRecord -> do mbErr <- Effectful.liftIO $ K.produceMessage producer instrumentedRecord - for_ mbErr throwError + for_ mbErr $ \err -> do + recordKafkaError span_ err + throwError err ProduceMessage' record cb -> - withProducerSpan tracer record $ \instrumentedRecord -> do + withProducerSpan tracer record $ \span_ instrumentedRecord -> do res <- Effectful.liftIO $ K.produceMessage' producer instrumentedRecord cb case res of - Left (K.ImmediateError err) -> throwError err + Left (K.ImmediateError err) -> do + recordKafkaError span_ err + throwError err Right () -> pure () ProduceMessageSync record -> - withProducerSpan tracer record $ \instrumentedRecord -> do + withProducerSpan tracer record $ \span_ instrumentedRecord -> do var <- Effectful.liftIO Concurrent.newEmptyMVar res <- Effectful.liftIO $ K.produceMessage' producer instrumentedRecord (Concurrent.putMVar var) case res of - Left (K.ImmediateError err) -> throwError err + Left (K.ImmediateError err) -> do + recordKafkaError span_ err + throwError err Right () -> do Effectful.liftIO $ K.flushProducer producer report <- Effectful.liftIO $ Concurrent.takeMVar var case report of K.DeliverySuccess _ offset -> pure offset - K.DeliveryFailure _ err -> throwError err - K.NoMessageError err -> throwError err + K.DeliveryFailure _ err -> do + recordKafkaError span_ err + throwError err + K.NoMessageError err -> do + recordKafkaError span_ err + throwError err ProduceMessageBatch records -> do -- One span per record so the Producer-kind attributes -- (partition, key) are per-record, matching the upstream @@ -136,10 +154,11 @@ results <- traverse ( \r -> - withProducerSpan tracer r $ \instrumentedRecord -> do + withProducerSpan tracer r $ \span_ instrumentedRecord -> do mbErr <- Effectful.liftIO $ K.produceMessage producer instrumentedRecord + for_ mbErr (recordKafkaError span_) pure (r, mbErr) ) records @@ -175,17 +194,29 @@ (IOE :> es) => Tracer -> ProducerRecord -> - (ProducerRecord -> Eff es a) -> + (Span -> ProducerRecord -> Eff es a) -> Eff es a -withProducerSpan tracer record action = - inSpan'' tracer (producerSpanName (prTopic record)) spanArgs $ \newSpan -> do +withProducerSpan tracer record action = do + semOpts <- Effectful.liftIO $ lookupStability "messaging" <$> getSemanticsOptions + inSpan'' tracer (producerSpanName (prTopic record)) (spanArgs semOpts) $ \newSpan -> do ctx <- getContext instrumentedRecord <- Effectful.liftIO $ injectTraceContextIntoRecord (Context.insertSpan newSpan ctx) record - action instrumentedRecord + action newSpan instrumentedRecord where - spanArgs = + spanArgs semOpts = addAttributesToSpanArguments - (producerRecordAttributes record) + (producerRecordAttributesWith semOpts record) defaultSpanArguments{kind = Producer} + +recordKafkaError :: + (IOE :> es) => + Span -> + KafkaError -> + Eff es () +recordKafkaError span_ err = do + let errText = Text.pack (show err) + Effectful.liftIO $ do + addAttribute span_ (unkey error_type) errText + setStatus span_ (Error errText) diff --git a/src/Kafka/Effectful/OpenTelemetry/Propagation.hs b/src/Kafka/Effectful/OpenTelemetry/Propagation.hs --- a/src/Kafka/Effectful/OpenTelemetry/Propagation.hs +++ b/src/Kafka/Effectful/OpenTelemetry/Propagation.hs @@ -1,17 +1,18 @@ -{- | Bridges between @hw-kafka-client@\'s 'Kafka.Types.Headers' (a list -of @(ByteString, ByteString)@) and @http-types@\'s 'RequestHeaders' -(a list of @(CI ByteString, ByteString)@), plus convenience helpers -that fetch the global propagator and inject\/extract a W3C trace -context against a record\'s headers in one call. +{- | Bridges between @hw-kafka-client@\'s 'Kafka.Types.Headers' and +OpenTelemetry propagation carriers, plus convenience helpers that +fetch the global propagator and inject\/extract trace context against +a record\'s headers in one call. -The header-bridge functions are pure and round-trip the underlying -bytes verbatim; only the case-sensitivity envelope changes (Kafka -headers are case-sensitive, HTTP headers are not). +The traced interpreters use the @hs-opentelemetry-api@ 1.0 +'TextMap' carrier. The older @http-types@ 'RequestHeaders' helpers +remain available for users who imported them directly. @since 0.2.0.0 -} module Kafka.Effectful.OpenTelemetry.Propagation ( -- * Header bridges + kafkaHeadersToTextMap, + textMapToKafkaHeaders, kafkaHeadersToRequestHeaders, requestHeadersToKafkaHeaders, @@ -23,17 +24,49 @@ import Data.Bifunctor (first) import Data.CaseInsensitive qualified as CI +import Data.Text.Encoding qualified as Text import Kafka.Consumer.Types (ConsumerRecord (crHeaders)) import Kafka.Producer.Types (ProducerRecord (prHeaders)) import Kafka.Types (Headers, headersFromList, headersToList) import Network.HTTP.Types (RequestHeaders) import OpenTelemetry.Context (Context) -import OpenTelemetry.Propagator (extract, inject) -import OpenTelemetry.Trace.Core ( - getGlobalTracerProvider, - getTracerProviderPropagators, +import OpenTelemetry.Propagator ( + TextMap, + emptyTextMap, + extract, + getGlobalTextMapPropagator, + inject, + textMapFromList, + textMapToList, ) +{- | Convert @hw-kafka-client@ 'Headers' to the OpenTelemetry 1.0 +'TextMap' propagation carrier. + +Header names and values are decoded as UTF-8, matching upstream +@hs-opentelemetry-instrumentation-hw-kafka-client@ 1.0. +-} +kafkaHeadersToTextMap :: Headers -> TextMap +kafkaHeadersToTextMap = + textMapFromList + . map + ( \(k, v) -> + (Text.decodeUtf8 k, Text.decodeUtf8 v) + ) + . headersToList + +{- | Convert the OpenTelemetry 1.0 'TextMap' propagation carrier back +to @hw-kafka-client@ 'Headers'. +-} +textMapToKafkaHeaders :: TextMap -> Headers +textMapToKafkaHeaders = + headersFromList + . map + ( \(k, v) -> + (Text.encodeUtf8 k, Text.encodeUtf8 v) + ) + . textMapToList + {- | Convert @hw-kafka-client@ 'Headers' (case-sensitive) to @http-types@ 'RequestHeaders' (case-insensitive). Each @(bsKey, bsVal)@ becomes @(CI.mk bsKey, bsVal)@. @@ -52,10 +85,10 @@ {- | Extract a W3C trace context from a 'ConsumerRecord'\'s headers and merge it into the supplied 'Context'. -This fetches the global tracer provider\'s propagator, hands it the -record\'s headers (after the case-insensitivity bridge), and returns -the resulting 'Context'. If the record carries no @traceparent@ -header the propagator returns the input 'Context' unchanged. +This fetches the global text-map propagator, hands it the record\'s +headers, and returns the resulting 'Context'. If the record carries +no @traceparent@ header the propagator returns the input 'Context' +unchanged. This is the building block that the traced consumer interpreter uses to root a per-message Consumer-kind span at the inbound trace @@ -66,8 +99,8 @@ Context -> IO Context extractTraceContextFromRecord record ctx = do - propagator <- getTracerProviderPropagators <$> getGlobalTracerProvider - extract propagator (kafkaHeadersToRequestHeaders (crHeaders record)) ctx + propagator <- getGlobalTextMapPropagator + extract propagator (kafkaHeadersToTextMap (crHeaders record)) ctx {- | Inject the supplied 'Context'\'s W3C trace context into a 'ProducerRecord'\'s headers, returning the augmented record. @@ -86,7 +119,7 @@ ProducerRecord -> IO ProducerRecord injectTraceContextIntoRecord ctx record = do - propagator <- getTracerProviderPropagators <$> getGlobalTracerProvider - extraHeaders <- inject propagator ctx [] - let merged = prHeaders record <> requestHeadersToKafkaHeaders extraHeaders + propagator <- getGlobalTextMapPropagator + extraHeaders <- inject propagator ctx emptyTextMap + let merged = prHeaders record <> textMapToKafkaHeaders extraHeaders pure record{prHeaders = merged} diff --git a/src/Kafka/Effectful/OpenTelemetry/Semantic.hs b/src/Kafka/Effectful/OpenTelemetry/Semantic.hs --- a/src/Kafka/Effectful/OpenTelemetry/Semantic.hs +++ b/src/Kafka/Effectful/OpenTelemetry/Semantic.hs @@ -11,12 +11,11 @@ without having to redefine the keys themselves. The attribute keys and value types follow the OpenTelemetry messaging -spec v1.24 as exposed by -@hs-opentelemetry-semantic-conventions@, and they intentionally -agree with what -@shibuya-kafka-adapter@\'s @Shibuya.Adapter.Kafka.Convert@ produces -(@kafkaSpanAttributes@), so the two libraries can be layered without -attribute key drift. +semantic conventions v1.40 as exposed by +@hs-opentelemetry-semantic-conventions@. Operation and consumer-group +keys follow @hs-opentelemetry-api@ 1.0\'s stability opt-in policy: +legacy by default, stable with @OTEL_SEMCONV_STABILITY_OPT_IN=messaging@, +and both with @OTEL_SEMCONV_STABILITY_OPT_IN=messaging\/dup@. The design parallels the upstream @hs-opentelemetry-instrumentation-hw-kafka-client@\'s @@ -36,33 +35,45 @@ -- * Attribute builders producerRecordAttributes, + producerRecordAttributesWith, consumerRecordAttributes, + consumerRecordAttributesWith, ) where import Data.ByteString (ByteString) +import Data.ByteString qualified as BS import Data.Int (Int64) +import Data.Map.Strict qualified as Map import Data.Text (Text) import Data.Text.Encoding (decodeUtf8') +import Kafka.Consumer.ConsumerProperties (ConsumerProperties (cpProps)) import Kafka.Consumer.Types ( - ConsumerRecord (crKey, crOffset, crPartition, crTopic), + ConsumerRecord (crKey, crOffset, crPartition, crTopic, crValue), Offset (unOffset), ) import Kafka.Producer.Types ( ProducePartition (SpecifiedPartition, UnassignedPartition), - ProducerRecord (prKey, prPartition, prTopic), + ProducerRecord (prKey, prPartition, prTopic, prValue), ) import Kafka.Types (PartitionId (unPartitionId), TopicName (..)) import OpenTelemetry.Attributes.Attribute (toAttribute) import OpenTelemetry.Attributes.Map (AttributeMap, insertAttributeByKey) import OpenTelemetry.SemanticConventions ( + messaging_client_id, + messaging_consumer_group_name, messaging_destination_name, + messaging_kafka_consumer_group, messaging_kafka_destination_partition, messaging_kafka_message_key, messaging_kafka_message_offset, + messaging_message_body_size, messaging_operation, + messaging_operation_name, + messaging_operation_type, messaging_system, ) +import OpenTelemetry.SemanticsConfig (StabilityOpt (..)) {- | The constant @"kafka"@ used as the value of the @messaging.system@ attribute. @@ -111,20 +122,38 @@ is present and decodes as UTF-8. -} producerRecordAttributes :: ProducerRecord -> AttributeMap -producerRecordAttributes record = - addOperation +producerRecordAttributes = producerRecordAttributesWith Old + +{- | Build the OpenTelemetry attribute map for a Producer-kind span +using the supplied messaging semantic-convention stability mode. +-} +producerRecordAttributesWith :: + StabilityOpt -> + ProducerRecord -> + AttributeMap +producerRecordAttributesWith semOpts record = + addOperation semOpts . addDestination . addPartition . addKey + . addBodySize . addSystem $ mempty where addSystem = insertAttributeByKey messaging_system $ toAttribute kafkaMessagingSystem - addOperation = - insertAttributeByKey messaging_operation $ - toAttribute producerOperationName + addOperation = \case + Old -> + insertAttributeByKey messaging_operation $ + toAttribute producerOperationName + Stable -> + insertAttributeByKey messaging_operation_name (toAttribute producerOperationName) + . insertAttributeByKey messaging_operation_type (toAttribute producerOperationName) + StableAndOld -> + insertAttributeByKey messaging_operation (toAttribute producerOperationName) + . insertAttributeByKey messaging_operation_name (toAttribute producerOperationName) + . insertAttributeByKey messaging_operation_type (toAttribute producerOperationName) addDestination = insertAttributeByKey messaging_destination_name $ toAttribute (unTopicName (prTopic record)) @@ -138,6 +167,11 @@ insertAttributeByKey messaging_kafka_message_key $ toAttribute k Nothing -> id + addBodySize = case prValue record of + Just v -> + insertAttributeByKey messaging_message_body_size $ + toAttribute (fromIntegral (BS.length v) :: Int64) + Nothing -> id {- | Build the OpenTelemetry attribute map for a Consumer-kind span describing a single 'ConsumerRecord'. @@ -150,30 +184,92 @@ @messaging.kafka.message.key@ (Text) when the record\'s @crKey@ is present and decodes as UTF-8. -Note: @messaging.kafka.consumer.group@ is not added here because -the consumer-group identifier lives on the -@Kafka.Consumer.ConsumerProperties@ value, not on the record. -Callers that need it (the traced interpreter does) add it on top -of what this helper returns. +Note: consumer-group and client-id attributes need +@Kafka.Consumer.ConsumerProperties@, not just the record. Use +'consumerRecordAttributesWith' when those properties are available. -} consumerRecordAttributes :: ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> AttributeMap -consumerRecordAttributes record = - addOperation +consumerRecordAttributes = consumerRecordAttributesLegacy + +consumerRecordAttributesLegacy :: + ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> + AttributeMap +consumerRecordAttributesLegacy record = + addConsumerCommon Old record + +{- | Build the OpenTelemetry attribute map for a Consumer-kind span +using the supplied messaging semantic-convention stability mode and +consumer properties. +-} +consumerRecordAttributesWith :: + StabilityOpt -> + ConsumerProperties -> + ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> + AttributeMap +consumerRecordAttributesWith semOpts props record = + addConsumerGroup semOpts + . addClientId + $ addConsumerCommon semOpts record + where + addConsumerGroup = \case + Old -> addOldConsumerGroup + Stable -> addStableConsumerGroup + StableAndOld -> addOldConsumerGroup . addStableConsumerGroup + addOldConsumerGroup attrs = + case Map.lookup "group.id" (cpProps props) of + Just groupId -> + insertAttributeByKey + messaging_kafka_consumer_group + (toAttribute groupId) + attrs + Nothing -> attrs + addStableConsumerGroup attrs = + case Map.lookup "group.id" (cpProps props) of + Just groupId -> + insertAttributeByKey + messaging_consumer_group_name + (toAttribute groupId) + attrs + Nothing -> attrs + addClientId attrs = + case Map.lookup "client.id" (cpProps props) of + Just clientId -> + insertAttributeByKey + messaging_client_id + (toAttribute clientId) + attrs + Nothing -> attrs + +addConsumerCommon :: + StabilityOpt -> + ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> + AttributeMap +addConsumerCommon semOpts record = + addOperation semOpts . addDestination . addPartition . addOffset . addKey + . addBodySize . addSystem $ mempty where addSystem = insertAttributeByKey messaging_system $ toAttribute kafkaMessagingSystem - addOperation = - insertAttributeByKey messaging_operation $ - toAttribute consumerOperationName + addOperation = \case + Old -> + insertAttributeByKey messaging_operation $ + toAttribute consumerOperationName + Stable -> + insertAttributeByKey messaging_operation_name (toAttribute consumerOperationName) + . insertAttributeByKey messaging_operation_type (toAttribute consumerOperationName) + StableAndOld -> + insertAttributeByKey messaging_operation (toAttribute consumerOperationName) + . insertAttributeByKey messaging_operation_name (toAttribute consumerOperationName) + . insertAttributeByKey messaging_operation_type (toAttribute consumerOperationName) addDestination = insertAttributeByKey messaging_destination_name $ toAttribute (unTopicName (crTopic record)) @@ -187,6 +283,11 @@ Just k -> insertAttributeByKey messaging_kafka_message_key $ toAttribute k + Nothing -> id + addBodySize = case crValue record of + Just v -> + insertAttributeByKey messaging_message_body_size $ + toAttribute (fromIntegral (BS.length v) :: Int64) Nothing -> id {- | Decode message-key bytes as UTF-8, returning 'Nothing' if the diff --git a/test/Kafka/Effectful/OpenTelemetry/PropagationTest.hs b/test/Kafka/Effectful/OpenTelemetry/PropagationTest.hs --- a/test/Kafka/Effectful/OpenTelemetry/PropagationTest.hs +++ b/test/Kafka/Effectful/OpenTelemetry/PropagationTest.hs @@ -5,6 +5,7 @@ import Data.CaseInsensitive qualified as CI import Data.Text (Text) import Data.Text qualified as Text +import Data.Text.Encoding qualified as Text.Encoding import Kafka.Consumer.Types ( ConsumerRecord (..), Offset (..), @@ -14,7 +15,9 @@ extractTraceContextFromRecord, injectTraceContextIntoRecord, kafkaHeadersToRequestHeaders, + kafkaHeadersToTextMap, requestHeadersToKafkaHeaders, + textMapToKafkaHeaders, ) import Kafka.Producer.Types ( ProducePartition (UnassignedPartition), @@ -27,6 +30,7 @@ headersToList, ) import OpenTelemetry.Context qualified as Context +import OpenTelemetry.Propagator (textMapLookup) import OpenTelemetry.Trace (initializeGlobalTracerProvider) import OpenTelemetry.Trace.Core ( SpanContext (..), @@ -71,7 +75,25 @@ withResource initializeGlobalTracerProvider (\_ -> pure ()) $ \_ -> testGroup "Propagation" - [ testCase "round-trip kafka headers <-> request headers (case-fold)" $ do + [ testCase "round-trip kafka headers <-> TextMap" $ do + let original = + headersFromList + [ ("traceparent", sampleTraceparent) + , ("Custom-Header", "value") + ] + roundTripped = + textMapToKafkaHeaders + (kafkaHeadersToTextMap original) + lookup "traceparent" (headersToList roundTripped) + `shouldBeJust` sampleTraceparent + lookup "Custom-Header" (headersToList roundTripped) + `shouldBeJust` "value" + , testCase "kafkaHeadersToTextMap provides case-insensitive lookup" $ do + let h = headersFromList [("TraceParent", sampleTraceparent)] + tm = kafkaHeadersToTextMap h + textMapLookup "traceparent" tm + `shouldBeJust` Text.Encoding.decodeUtf8 sampleTraceparent + , testCase "request header compatibility helpers still case-fold" $ do let original = headersFromList [ ("traceparent", sampleTraceparent) diff --git a/test/Kafka/Effectful/OpenTelemetry/SemanticTest.hs b/test/Kafka/Effectful/OpenTelemetry/SemanticTest.hs --- a/test/Kafka/Effectful/OpenTelemetry/SemanticTest.hs +++ b/test/Kafka/Effectful/OpenTelemetry/SemanticTest.hs @@ -2,23 +2,28 @@ import Data.ByteString (ByteString) import Data.ByteString qualified as BS -import Data.HashMap.Strict qualified as HashMap +import Data.HashMap.Lazy qualified as HashMap import Data.Int (Int64) import Data.Text (Text) +import Kafka.Consumer.ConsumerProperties qualified as ConsumerProperties import Kafka.Consumer.Types ( + ConsumerGroupId (..), ConsumerRecord (..), Offset (..), Timestamp (NoTimestamp), ) import Kafka.Effectful.OpenTelemetry.Semantic ( consumerRecordAttributes, + consumerRecordAttributesWith, producerRecordAttributes, + producerRecordAttributesWith, ) import Kafka.Producer.Types ( ProducePartition (SpecifiedPartition, UnassignedPartition), ProducerRecord (..), ) import Kafka.Types ( + ClientId (..), PartitionId (..), TopicName (..), headersFromList, @@ -27,8 +32,9 @@ Attribute (AttributeValue), PrimitiveAttribute (IntAttribute, TextAttribute), ) +import OpenTelemetry.SemanticsConfig (StabilityOpt (..)) import Test.Tasty (TestTree, testGroup) -import Test.Tasty.HUnit (assertEqual, assertFailure, testCase, (@?=)) +import Test.Tasty.HUnit (assertFailure, testCase, (@?=)) tests :: TestTree tests = @@ -38,108 +44,111 @@ , consumerTests ] --- Producer-side attribute coverage - producerTests :: TestTree producerTests = testGroup "producerRecordAttributes" - [ testCase "adds messaging.system=kafka" $ do - let attrs = producerRecordAttributes (sampleProducerRecord (Just "k1")) - HashMap.lookup "messaging.system" attrs - @?= Just (textAttr "kafka") - , testCase "adds messaging.destination.name from topic" $ do - let attrs = producerRecordAttributes (sampleProducerRecord (Just "k1")) - HashMap.lookup "messaging.destination.name" attrs - @?= Just (textAttr "orders") - , testCase "adds messaging.operation=send" $ do - let attrs = producerRecordAttributes (sampleProducerRecord (Just "k1")) - HashMap.lookup "messaging.operation" attrs - @?= Just (textAttr "send") - , testCase "adds messaging.kafka.destination.partition for SpecifiedPartition" $ do + [ testCase "legacy helper keeps old operation key" $ do let attrs = producerRecordAttributes (sampleProducerRecord (Just "k1")) + HashMap.lookup "messaging.system" attrs @?= Just (textAttr "kafka") + HashMap.lookup "messaging.destination.name" attrs @?= Just (textAttr "orders") + HashMap.lookup "messaging.operation" attrs @?= Just (textAttr "send") + assertAbsent "messaging.operation.name" attrs + assertAbsent "messaging.operation.type" attrs + , testCase "stable mode uses stable operation keys" $ do + let attrs = producerRecordAttributesWith Stable (sampleProducerRecord (Just "k1")) + HashMap.lookup "messaging.operation.name" attrs @?= Just (textAttr "send") + HashMap.lookup "messaging.operation.type" attrs @?= Just (textAttr "send") + assertAbsent "messaging.operation" attrs + , testCase "duplicate mode emits old and stable operation keys" $ do + let attrs = producerRecordAttributesWith StableAndOld (sampleProducerRecord (Just "k1")) + HashMap.lookup "messaging.operation" attrs @?= Just (textAttr "send") + HashMap.lookup "messaging.operation.name" attrs @?= Just (textAttr "send") + HashMap.lookup "messaging.operation.type" attrs @?= Just (textAttr "send") + , testCase "adds partition, key, and body size when available" $ do + let attrs = producerRecordAttributesWith Old (sampleProducerRecord (Just "hello")) HashMap.lookup "messaging.kafka.destination.partition" attrs @?= Just (intAttr 42) - , testCase "omits messaging.kafka.destination.partition for UnassignedPartition" $ do + HashMap.lookup "messaging.kafka.message.key" attrs + @?= Just (textAttr "hello") + HashMap.lookup "messaging.message.body.size" attrs + @?= Just (intAttr 5) + , testCase "omits partition for UnassignedPartition" $ do let attrs = - producerRecordAttributes + producerRecordAttributesWith + Old (sampleProducerRecord (Just "k1")) { prPartition = UnassignedPartition } - assertEqual - "partition attribute should not be present" - Nothing - (HashMap.lookup "messaging.kafka.destination.partition" attrs) - , testCase "adds messaging.kafka.message.key when valid UTF-8" $ do - let attrs = producerRecordAttributes (sampleProducerRecord (Just "hello")) - HashMap.lookup "messaging.kafka.message.key" attrs - @?= Just (textAttr "hello") - , testCase "omits messaging.kafka.message.key when invalid UTF-8" $ do - let attrs = - producerRecordAttributes + assertAbsent "messaging.kafka.destination.partition" attrs + , testCase "omits invalid or absent keys" $ do + let invalidAttrs = + producerRecordAttributesWith + Old (sampleProducerRecord (Just (BS.pack [0xC3, 0x28]))) - case HashMap.lookup "messaging.kafka.message.key" attrs of - Nothing -> pure () - Just _ -> - assertFailure - "expected messaging.kafka.message.key to be omitted for invalid UTF-8" - , testCase "omits messaging.kafka.message.key when key is Nothing" $ do - let attrs = producerRecordAttributes (sampleProducerRecord Nothing) - assertEqual - "key attribute should not be present" - Nothing - (HashMap.lookup "messaging.kafka.message.key" attrs) + missingAttrs = + producerRecordAttributesWith Old (sampleProducerRecord Nothing) + assertAbsent "messaging.kafka.message.key" invalidAttrs + assertAbsent "messaging.kafka.message.key" missingAttrs ] --- Consumer-side attribute coverage - consumerTests :: TestTree consumerTests = testGroup "consumerRecordAttributes" - [ testCase "adds messaging.system=kafka" $ do - let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1")) - HashMap.lookup "messaging.system" attrs - @?= Just (textAttr "kafka") - , testCase "adds messaging.destination.name from topic" $ do - let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1")) - HashMap.lookup "messaging.destination.name" attrs - @?= Just (textAttr "orders") - , testCase "adds messaging.operation=process" $ do - let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1")) - HashMap.lookup "messaging.operation" attrs - @?= Just (textAttr "process") - , testCase "adds messaging.kafka.destination.partition" $ do + [ testCase "legacy helper keeps old operation key" $ do let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1")) + HashMap.lookup "messaging.system" attrs @?= Just (textAttr "kafka") + HashMap.lookup "messaging.destination.name" attrs @?= Just (textAttr "orders") + HashMap.lookup "messaging.operation" attrs @?= Just (textAttr "process") + assertAbsent "messaging.operation.name" attrs + assertAbsent "messaging.operation.type" attrs + , testCase "old mode uses legacy consumer group key" $ do + let attrs = consumerRecordAttributesWith Old sampleConsumerProps (sampleConsumerRecord (Just "k1")) + HashMap.lookup "messaging.kafka.consumer.group" attrs + @?= Just (textAttr "orders-group") + assertAbsent "messaging.consumer.group.name" attrs + , testCase "stable mode uses stable operation and consumer group keys" $ do + let attrs = consumerRecordAttributesWith Stable sampleConsumerProps (sampleConsumerRecord (Just "k1")) + HashMap.lookup "messaging.operation.name" attrs @?= Just (textAttr "process") + HashMap.lookup "messaging.operation.type" attrs @?= Just (textAttr "process") + HashMap.lookup "messaging.consumer.group.name" attrs + @?= Just (textAttr "orders-group") + assertAbsent "messaging.operation" attrs + assertAbsent "messaging.kafka.consumer.group" attrs + , testCase "duplicate mode emits old and stable operation and group keys" $ do + let attrs = consumerRecordAttributesWith StableAndOld sampleConsumerProps (sampleConsumerRecord (Just "k1")) + HashMap.lookup "messaging.operation" attrs @?= Just (textAttr "process") + HashMap.lookup "messaging.operation.name" attrs @?= Just (textAttr "process") + HashMap.lookup "messaging.operation.type" attrs @?= Just (textAttr "process") + HashMap.lookup "messaging.kafka.consumer.group" attrs + @?= Just (textAttr "orders-group") + HashMap.lookup "messaging.consumer.group.name" attrs + @?= Just (textAttr "orders-group") + , testCase "adds client id, partition, offset, key, and body size" $ do + let attrs = consumerRecordAttributesWith Old sampleConsumerProps (sampleConsumerRecord (Just "hello")) + HashMap.lookup "messaging.client.id" attrs + @?= Just (textAttr "orders-client") HashMap.lookup "messaging.kafka.destination.partition" attrs @?= Just (intAttr 7) - , testCase "adds messaging.kafka.message.offset" $ do - let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "k1")) HashMap.lookup "messaging.kafka.message.offset" attrs @?= Just (intAttr 42) - , testCase "adds messaging.kafka.message.key when valid UTF-8" $ do - let attrs = consumerRecordAttributes (sampleConsumerRecord (Just "hello")) HashMap.lookup "messaging.kafka.message.key" attrs @?= Just (textAttr "hello") - , testCase "omits messaging.kafka.message.key when invalid UTF-8" $ do - let attrs = - consumerRecordAttributes + HashMap.lookup "messaging.message.body.size" attrs + @?= Just (intAttr 5) + , testCase "omits invalid or absent keys" $ do + let invalidAttrs = + consumerRecordAttributesWith + Old + sampleConsumerProps (sampleConsumerRecord (Just (BS.pack [0xC3, 0x28]))) - case HashMap.lookup "messaging.kafka.message.key" attrs of - Nothing -> pure () - Just _ -> - assertFailure - "expected messaging.kafka.message.key to be omitted for invalid UTF-8" - , testCase "omits messaging.kafka.message.key when key is Nothing" $ do - let attrs = consumerRecordAttributes (sampleConsumerRecord Nothing) - assertEqual - "key attribute should not be present" - Nothing - (HashMap.lookup "messaging.kafka.message.key" attrs) + missingAttrs = + consumerRecordAttributesWith Old sampleConsumerProps (sampleConsumerRecord Nothing) + assertAbsent "messaging.kafka.message.key" invalidAttrs + assertAbsent "messaging.kafka.message.key" missingAttrs ] --- Helpers - sampleProducerRecord :: Maybe ByteString -> ProducerRecord sampleProducerRecord key = ProducerRecord @@ -164,8 +173,24 @@ , crValue = Just "value" } +sampleConsumerProps :: ConsumerProperties.ConsumerProperties +sampleConsumerProps = + ConsumerProperties.groupId (ConsumerGroupId "orders-group") + <> ConsumerProperties.clientId (ClientId "orders-client") + textAttr :: Text -> Attribute textAttr = AttributeValue . TextAttribute intAttr :: Int64 -> Attribute intAttr = AttributeValue . IntAttribute + +assertAbsent :: Text -> HashMap.HashMap Text Attribute -> IO () +assertAbsent key attrs = + case HashMap.lookup key attrs of + Nothing -> pure () + Just attr -> + assertFailure $ + "expected " + <> show key + <> " to be absent, got " + <> show attr diff --git a/test/Kafka/Effectful/OpenTelemetry/ShibuyaCompatibilityTest.hs b/test/Kafka/Effectful/OpenTelemetry/ShibuyaCompatibilityTest.hs --- a/test/Kafka/Effectful/OpenTelemetry/ShibuyaCompatibilityTest.hs +++ b/test/Kafka/Effectful/OpenTelemetry/ShibuyaCompatibilityTest.hs @@ -1,6 +1,6 @@ module Kafka.Effectful.OpenTelemetry.ShibuyaCompatibilityTest (tests) where -import Data.HashMap.Strict qualified as HashMap +import Data.HashMap.Lazy qualified as HashMap import Data.Int (Int64) import Data.Text (Text) import Kafka.Consumer.Types ( @@ -22,15 +22,16 @@ import Test.Tasty (TestTree, testGroup) import Test.Tasty.HUnit (assertEqual, testCase) -{- | Tests pinning @Kafka.Effectful.OpenTelemetry.Semantic@\'s -'consumerRecordAttributes' against the attribute set that +{- | Tests pinning @Kafka.Effectful.OpenTelemetry.Semantic@\'s legacy +'consumerRecordAttributes' helper against the attribute set that @shibuya-kafka-adapter@\'s @Shibuya.Adapter.Kafka.Convert.kafkaSpanAttributes@ produces. Both libraries depend on @hs-opentelemetry-semantic-conventions@, so if upstream renames a key in a future release this test fails -together with shibuya — the desired failure mode for a -compatibility pin. +together with shibuya for the legacy-mode compatibility pin. Stable +OpenTelemetry messaging keys are covered separately in +@SemanticTest@. -} tests :: TestTree tests =