diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Revision history for hs-opentelemetry-instrumentation-hw-kafka-client +## Unreleased + +## 1.0.0.0 - 2026-05-29 + +* Add required `messaging.system` attribute (`"kafka"`). +* Add `messaging.operation.name` and `messaging.operation.type` (stable convention names). +* Add `messaging.consumer.group.name` (generic, alongside kafka-specific key). +* Add `messaging.client.id` from consumer properties. +* Add `messaging.message.body.size` from message value length. +* Set `error.type` and span error status on produce failures. + ## 0.1.0.0 -- 2025-01-26 * First version. Released on an unsuspecting world. diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2024, Alberto Fanton +Copyright (c) 2024-2026, Alberto Fanton All rights reserved. diff --git a/hs-opentelemetry-instrumentation-hw-kafka-client.cabal b/hs-opentelemetry-instrumentation-hw-kafka-client.cabal --- a/hs-opentelemetry-instrumentation-hw-kafka-client.cabal +++ b/hs-opentelemetry-instrumentation-hw-kafka-client.cabal @@ -1,16 +1,28 @@ cabal-version: 3.0 name: hs-opentelemetry-instrumentation-hw-kafka-client -version: 0.1.0.0 +version: 1.0.0.0 synopsis: OpenTelemetry instrumentation for hw-kafka-client +description: + OpenTelemetry tracing instrumentation for the @hw-kafka-client@ Kafka library. + Propagates trace context through Kafka record headers so that producer and + consumer spans are correlated across services. license: BSD-3-Clause license-file: LICENSE author: Alberto Fanton maintainer: alberto.fanton@protonmail.com +copyright: 2024 Ian Duncan, Mercury Technologies +homepage: https://github.com/iand675/hs-opentelemetry#readme +bug-reports: https://github.com/iand675/hs-opentelemetry/issues +category: OpenTelemetry, Telemetry, Kafka build-type: Simple extra-doc-files: CHANGELOG.md -- extra-source-files: +source-repository head + type: git + location: https://github.com/iand675/hs-opentelemetry + common warnings ghc-options: -Wall @@ -25,12 +37,28 @@ , containers , text , bytestring - , hs-opentelemetry-api - , hs-opentelemetry-semantic-conventions + , hs-opentelemetry-api ^>= 1.0 + , hs-opentelemetry-semantic-conventions >= 1.40 && < 2 , hw-kafka-client , unliftio-core - , case-insensitive - , http-types hs-source-dirs: src default-language: Haskell2010 + +test-suite hs-opentelemetry-instrumentation-hw-kafka-client-test + import: warnings + type: exitcode-stdio-1.0 + main-is: Spec.hs + hs-source-dirs: test + default-language: Haskell2010 + ghc-options: -threaded -rtsopts -with-rtsopts=-N + build-depends: + , base >=4.7 && <5 + , hs-opentelemetry-instrumentation-hw-kafka-client ^>= 1.0 + , hs-opentelemetry-api ^>= 1.0 + , hspec + , text + , bytestring + , hw-kafka-client + , containers + , unordered-containers diff --git a/src/OpenTelemetry/Instrumentation/Kafka.hs b/src/OpenTelemetry/Instrumentation/Kafka.hs --- a/src/OpenTelemetry/Instrumentation/Kafka.hs +++ b/src/OpenTelemetry/Instrumentation/Kafka.hs @@ -17,21 +17,26 @@ -- * Consumer pollMessage, + + -- * Attribute builders (exported for testing) + producerAttributes, + consumerAttributes, ) where -import Control.Monad (void) -import Control.Monad.IO.Class (MonadIO) +import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.IO.Unlift (MonadUnliftIO) -import Data.Bifunctor (first) import Data.ByteString (ByteString) -import qualified Data.CaseInsensitive as CI +import qualified Data.ByteString as BS +import Data.Int (Int64) import qualified Data.Map as M import Data.String (IsString) +import qualified Data.Text as T import Data.Text.Encoding (decodeUtf8') +import qualified Data.Text.Encoding as TE import GHC.Stack.Types (HasCallStack) import Kafka.Consumer ( ConsumerProperties (cpProps), - ConsumerRecord (crHeaders, crKey, crOffset, crPartition, crTopic), + ConsumerRecord (crHeaders, crKey, crOffset, crPartition, crTopic, crValue), KafkaConsumer, Offset (unOffset), ) @@ -40,7 +45,7 @@ KafkaError, KafkaProducer, ProducePartition (SpecifiedPartition, UnassignedPartition), - ProducerRecord (prHeaders, prKey, prPartition, prTopic), + ProducerRecord (prHeaders, prKey, prPartition, prTopic, prValue), ) import qualified Kafka.Producer as KP import Kafka.Types ( @@ -51,31 +56,41 @@ headersFromList, headersToList, ) -import Network.HTTP.Types (RequestHeaders) +import OpenTelemetry.Attributes.Key (unkey) import OpenTelemetry.Attributes.Map (AttributeMap, insertAttributeByKey) import qualified OpenTelemetry.Context as Context import OpenTelemetry.Context.ThreadLocal (attachContext, getContext) -import OpenTelemetry.Propagator (extract, inject) +import OpenTelemetry.Propagator (TextMap, emptyTextMap, extract, getGlobalTextMapPropagator, inject, textMapFromList, textMapToList) import OpenTelemetry.SemanticConventions ( + error_type, + 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 (..), getSemanticsOptions, lookupStability) import OpenTelemetry.Trace.Core ( SpanArguments (kind), SpanKind (Consumer, Producer), + SpanStatus (Error), Tracer, + addAttribute, addAttributesToSpanArguments, callerAttributes, defaultSpanArguments, detectInstrumentationLibrary, getGlobalTracerProvider, - getTracerProviderPropagators, inSpan'', makeTracer, + setStatus, toAttribute, tracerOptions, ) @@ -100,12 +115,30 @@ rightToMaybe _ = Nothing --- | Span attributes with caller information and kafka specifics -producerAttributes :: HasCallStack => ProducerRecord -> AttributeMap -producerAttributes record = +{- | Build producer span attributes. + +Attribute names for operation and consumer-group fields are gated on +@OTEL_SEMCONV_STABILITY_OPT_IN@: + +* @Old@ (default): legacy @messaging.operation@ key. +* @Stable@ (@messaging@): stable @messaging.operation.name@ + @messaging.operation.type@. +* @StableAndOld@ (@messaging\/dup@): both sets. +-} +producerAttributes :: HasCallStack => StabilityOpt -> ProducerRecord -> AttributeMap +producerAttributes semOpts record = let - addOperationName = - insertAttributeByKey messaging_operation producerOperationName + addSystem = + insertAttributeByKey messaging_system (toAttribute ("kafka" :: T.Text)) + addOperationAttrs = case semOpts of + Old -> + insertAttributeByKey messaging_operation (toAttribute (producerOperationName :: T.Text)) + Stable -> + insertAttributeByKey messaging_operation_name (toAttribute (producerOperationName :: T.Text)) + . insertAttributeByKey messaging_operation_type (toAttribute (producerOperationName :: T.Text)) + StableAndOld -> + insertAttributeByKey messaging_operation (toAttribute (producerOperationName :: T.Text)) + . insertAttributeByKey messaging_operation_name (toAttribute (producerOperationName :: T.Text)) + . insertAttributeByKey messaging_operation_type (toAttribute (producerOperationName :: T.Text)) addDestination = insertAttributeByKey messaging_destination_name $ toAttribute . unTopicName . prTopic $ record addPartition = @@ -120,8 +153,12 @@ insertAttributeByKey messaging_kafka_message_key $ toAttribute key Nothing -> id + addBodySize = + case prValue record of + Just v -> insertAttributeByKey messaging_message_body_size (toAttribute (fromIntegral (BS.length v) :: Int64)) + Nothing -> id in - (addOperationName . addDestination . addPartition . addKey) + (addSystem . addOperationAttrs . addDestination . addPartition . addKey . addBodySize) callerAttributes @@ -130,23 +167,53 @@ consumerSpanArgs = defaultSpanArguments {kind = Consumer} --- | Span attributes for consumer with caller information and kafka specifics +{- | Build consumer span attributes. + +Attribute names for operation and consumer-group fields are gated on +@OTEL_SEMCONV_STABILITY_OPT_IN@: + +* @Old@ (default): legacy @messaging.operation@ + @messaging.kafka.consumer.group@ keys. +* @Stable@ (@messaging@): stable @messaging.operation.name@, @messaging.operation.type@, + and @messaging.consumer.group.name@. +* @StableAndOld@ (@messaging\/dup@): both sets. +-} consumerAttributes :: HasCallStack - => ConsumerProperties + => StabilityOpt + -> ConsumerProperties -> ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> AttributeMap -consumerAttributes consumerProperties record = +consumerAttributes semOpts consumerProperties record = let - addOperationName = - insertAttributeByKey messaging_operation consumerOperationName + addSystem = + insertAttributeByKey messaging_system (toAttribute ("kafka" :: T.Text)) + addOperationAttrs = case semOpts of + Old -> + insertAttributeByKey messaging_operation (toAttribute (consumerOperationName :: T.Text)) + Stable -> + insertAttributeByKey messaging_operation_name (toAttribute (consumerOperationName :: T.Text)) + . insertAttributeByKey messaging_operation_type (toAttribute (consumerOperationName :: T.Text)) + StableAndOld -> + insertAttributeByKey messaging_operation (toAttribute (consumerOperationName :: T.Text)) + . insertAttributeByKey messaging_operation_name (toAttribute (consumerOperationName :: T.Text)) + . insertAttributeByKey messaging_operation_type (toAttribute (consumerOperationName :: T.Text)) addDestination = insertAttributeByKey messaging_destination_name $ toAttribute . unTopicName . crTopic $ record addConsumerGroup = - -- NOTE: unfortunately, hw-kafka-client does not expose an API to get the consumer, this a flaky workaround case M.lookup "group.id" $ cpProps consumerProperties of - Just groupId -> insertAttributeByKey messaging_kafka_consumer_group $ toAttribute groupId + Just groupId -> case semOpts of + Old -> + insertAttributeByKey messaging_kafka_consumer_group (toAttribute groupId) + Stable -> + insertAttributeByKey messaging_consumer_group_name (toAttribute groupId) + StableAndOld -> + insertAttributeByKey messaging_kafka_consumer_group (toAttribute groupId) + . insertAttributeByKey messaging_consumer_group_name (toAttribute groupId) Nothing -> id + addClientId = + case M.lookup "client.id" $ cpProps consumerProperties of + Just cid -> insertAttributeByKey messaging_client_id (toAttribute cid) + Nothing -> id addPartition = insertAttributeByKey messaging_kafka_destination_partition $ toAttribute . unPartitionId . crPartition $ record addOffset = @@ -157,8 +224,21 @@ insertAttributeByKey messaging_kafka_message_key $ toAttribute key Nothing -> id + addBodySize = + case crValue record of + Just v -> insertAttributeByKey messaging_message_body_size (toAttribute (fromIntegral (BS.length v) :: Int64)) + Nothing -> id in - (addOperationName . addDestination . addConsumerGroup . addPartition . addOffset . addKey) + ( addSystem + . addOperationAttrs + . addDestination + . addConsumerGroup + . addClientId + . addPartition + . addOffset + . addKey + . addBodySize + ) callerAttributes @@ -169,14 +249,12 @@ return $ makeTracer provider $detectInstrumentationLibrary tracerOptions --- | Convert Kafka headers to HTTP headers format -kafkaHeadersToHttpHeaders :: Headers -> RequestHeaders -kafkaHeadersToHttpHeaders = map (first CI.mk) . headersToList +kafkaHeadersToTextMap :: Headers -> TextMap +kafkaHeadersToTextMap = textMapFromList . map (\(k, v) -> (TE.decodeUtf8 k, TE.decodeUtf8 v)) . headersToList --- | Convert HTTP headers to Kafka headers format -httpHeadersToKafkaHeaders :: RequestHeaders -> Headers -httpHeadersToKafkaHeaders = headersFromList . map (first CI.foldedCase) +textMapToKafkaHeaders :: TextMap -> Headers +textMapToKafkaHeaders = headersFromList . map (\(k, v) -> (TE.encodeUtf8 k, TE.encodeUtf8 v)) . textMapToList {- | Produce a message to Kafka with OpenTelemetry instrumentation. @@ -195,18 +273,26 @@ headers = prHeaders record topicName = prTopic record spanName = producerOperationName <> " " <> unTopicName topicName - attributes = producerAttributes record - spanArguments = addAttributesToSpanArguments attributes producerSpanArgs in do + semOpts <- liftIO $ lookupStability "messaging" <$> getSemanticsOptions + let attributes = producerAttributes semOpts record + spanArguments = addAttributesToSpanArguments attributes producerSpanArgs tracer <- rdkafkaTracer ctxt <- getContext inSpan'' tracer spanName spanArguments $ \newSpan -> do - propagator <- getTracerProviderPropagators <$> getGlobalTracerProvider - extraHeaders <- inject propagator (Context.insertSpan newSpan ctxt) [] - let newKafkaHeaders = headers <> httpHeadersToKafkaHeaders extraHeaders + propagator <- liftIO getGlobalTextMapPropagator + extraTm <- inject propagator (Context.insertSpan newSpan ctxt) emptyTextMap + let newKafkaHeaders = headers <> textMapToKafkaHeaders extraTm let newKafkaRecord = record {prHeaders = newKafkaHeaders} - KP.produceMessage producer newKafkaRecord + result <- KP.produceMessage producer newKafkaRecord + case result of + Just err -> do + let errText = T.pack $ show err + addAttribute newSpan (unkey error_type) errText + setStatus newSpan (Error errText) + Nothing -> pure () + pure result {- | Poll for a single message from Kafka with OpenTelemetry instrumentation. @@ -228,15 +314,16 @@ Left err -> pure $ Left err Right cr -> let - attributes = consumerAttributes consumerProperties cr topicName = crTopic cr spanName = consumerOperationName <> " " <> unTopicName topicName in do + semOpts <- liftIO $ lookupStability "messaging" <$> getSemanticsOptions + let attributes = consumerAttributes semOpts consumerProperties cr tracer <- rdkafkaTracer ctxt <- getContext - propagator <- getTracerProviderPropagators <$> getGlobalTracerProvider - ctx <- extract propagator (kafkaHeadersToHttpHeaders $ crHeaders cr) ctxt - void $ attachContext ctx + propagator <- liftIO getGlobalTextMapPropagator + ctx <- extract propagator (kafkaHeadersToTextMap $ crHeaders cr) ctxt + _ <- attachContext ctx inSpan'' tracer spanName (addAttributesToSpanArguments attributes consumerSpanArgs) $ \_span -> do return $ Right cr diff --git a/test/Spec.hs b/test/Spec.hs new file mode 100644 --- /dev/null +++ b/test/Spec.hs @@ -0,0 +1,176 @@ +{-# LANGUAGE OverloadedStrings #-} + +module Main where + +import qualified Data.ByteString as BS +import qualified Data.HashMap.Strict as H +import Kafka.Consumer (ConsumerRecord (..), Offset (..), Timestamp (..)) +import Kafka.Consumer.ConsumerProperties (ConsumerProperties, clientId, groupId) +import Kafka.Consumer.Types (ConsumerGroupId (..)) +import Kafka.Producer (ProducePartition (..), ProducerRecord (..)) +import Kafka.Types (ClientId (..), PartitionId (..), TopicName (..), headersFromList) +import OpenTelemetry.Attributes.Attribute (Attribute (..), PrimitiveAttribute (..)) +import OpenTelemetry.Instrumentation.Kafka (consumerAttributes, producerAttributes) +import OpenTelemetry.SemanticsConfig (StabilityOpt (..)) +import Test.Hspec + + +main :: IO () +main = hspec spec + + +testProducerRecord :: ProducerRecord +testProducerRecord = + ProducerRecord + { prTopic = TopicName "orders" + , prPartition = SpecifiedPartition 3 + , prKey = Just "order-123" + , prValue = Just "payload-data" + , prHeaders = headersFromList [] + } + + +testConsumerRecord :: ConsumerRecord (Maybe BS.ByteString) (Maybe BS.ByteString) +testConsumerRecord = + ConsumerRecord + { crTopic = TopicName "events" + , crPartition = PartitionId 1 + , crOffset = Offset 42 + , crTimestamp = NoTimestamp + , crHeaders = headersFromList [] + , crKey = Just "event-key" + , crValue = Just "event-body-data" + } + + +testConsumerProps :: ConsumerProperties +testConsumerProps = + groupId (ConsumerGroupId "my-consumer-group") + <> clientId (ClientId "my-client") + + +spec :: Spec +spec = do + describe "producerAttributes (Old — default)" $ do + let attrs = producerAttributes Old testProducerRecord + + it "sets messaging.system to kafka" $ + H.lookup "messaging.system" attrs `shouldBe` Just (AttributeValue (TextAttribute "kafka")) + + it "sets messaging.operation (legacy)" $ + H.lookup "messaging.operation" attrs `shouldBe` Just (AttributeValue (TextAttribute "send")) + + it "does not set messaging.operation.name in Old mode" $ + H.lookup "messaging.operation.name" attrs `shouldBe` Nothing + + it "does not set messaging.operation.type in Old mode" $ + H.lookup "messaging.operation.type" attrs `shouldBe` Nothing + + it "sets messaging.destination.name to topic" $ + H.lookup "messaging.destination.name" attrs `shouldBe` Just (AttributeValue (TextAttribute "orders")) + + it "sets messaging.kafka.destination.partition" $ + H.lookup "messaging.kafka.destination.partition" attrs `shouldSatisfy` (/= Nothing) + + it "sets messaging.kafka.message.key" $ + H.lookup "messaging.kafka.message.key" attrs `shouldBe` Just (AttributeValue (TextAttribute "order-123")) + + it "sets messaging.message.body.size for non-empty value" $ + H.lookup "messaging.message.body.size" attrs `shouldBe` Just (AttributeValue (IntAttribute 12)) + + it "does not set body size when value is Nothing" $ do + let noValueAttrs = producerAttributes Old testProducerRecord {prValue = Nothing} + H.lookup "messaging.message.body.size" noValueAttrs `shouldBe` Nothing + + describe "producerAttributes (Stable)" $ do + let attrs = producerAttributes Stable testProducerRecord + + it "sets messaging.operation.name" $ + H.lookup "messaging.operation.name" attrs `shouldBe` Just (AttributeValue (TextAttribute "send")) + + it "sets messaging.operation.type" $ + H.lookup "messaging.operation.type" attrs `shouldBe` Just (AttributeValue (TextAttribute "send")) + + it "does not set messaging.operation in Stable mode" $ + H.lookup "messaging.operation" attrs `shouldBe` Nothing + + describe "producerAttributes (StableAndOld)" $ do + let attrs = producerAttributes StableAndOld testProducerRecord + + it "sets messaging.operation (legacy)" $ + H.lookup "messaging.operation" attrs `shouldBe` Just (AttributeValue (TextAttribute "send")) + + it "sets messaging.operation.name (stable)" $ + H.lookup "messaging.operation.name" attrs `shouldBe` Just (AttributeValue (TextAttribute "send")) + + it "sets messaging.operation.type (stable)" $ + H.lookup "messaging.operation.type" attrs `shouldBe` Just (AttributeValue (TextAttribute "send")) + + describe "consumerAttributes (Old — default)" $ do + let attrs = consumerAttributes Old testConsumerProps testConsumerRecord + + it "sets messaging.system to kafka" $ + H.lookup "messaging.system" attrs `shouldBe` Just (AttributeValue (TextAttribute "kafka")) + + it "sets messaging.operation (legacy)" $ + H.lookup "messaging.operation" attrs `shouldBe` Just (AttributeValue (TextAttribute "process")) + + it "does not set messaging.operation.name in Old mode" $ + H.lookup "messaging.operation.name" attrs `shouldBe` Nothing + + it "does not set messaging.operation.type in Old mode" $ + H.lookup "messaging.operation.type" attrs `shouldBe` Nothing + + it "sets messaging.destination.name to topic" $ + H.lookup "messaging.destination.name" attrs `shouldBe` Just (AttributeValue (TextAttribute "events")) + + it "sets messaging.kafka.consumer.group (legacy group key)" $ + H.lookup "messaging.kafka.consumer.group" attrs `shouldBe` Just (AttributeValue (TextAttribute "my-consumer-group")) + + it "does not set messaging.consumer.group.name in Old mode" $ + H.lookup "messaging.consumer.group.name" attrs `shouldBe` Nothing + + it "sets messaging.client.id" $ + H.lookup "messaging.client.id" attrs `shouldBe` Just (AttributeValue (TextAttribute "my-client")) + + it "sets messaging.kafka.message.offset" $ + H.lookup "messaging.kafka.message.offset" attrs `shouldSatisfy` (/= Nothing) + + it "sets messaging.message.body.size for non-empty value" $ + H.lookup "messaging.message.body.size" attrs `shouldBe` Just (AttributeValue (IntAttribute 15)) + + it "sets messaging.kafka.message.key" $ + H.lookup "messaging.kafka.message.key" attrs `shouldBe` Just (AttributeValue (TextAttribute "event-key")) + + describe "consumerAttributes (Stable)" $ do + let attrs = consumerAttributes Stable testConsumerProps testConsumerRecord + + it "sets messaging.operation.name" $ + H.lookup "messaging.operation.name" attrs `shouldBe` Just (AttributeValue (TextAttribute "process")) + + it "sets messaging.operation.type" $ + H.lookup "messaging.operation.type" attrs `shouldBe` Just (AttributeValue (TextAttribute "process")) + + it "sets messaging.consumer.group.name (stable group key)" $ + H.lookup "messaging.consumer.group.name" attrs `shouldBe` Just (AttributeValue (TextAttribute "my-consumer-group")) + + it "does not set messaging.operation in Stable mode" $ + H.lookup "messaging.operation" attrs `shouldBe` Nothing + + it "does not set messaging.kafka.consumer.group in Stable mode" $ + H.lookup "messaging.kafka.consumer.group" attrs `shouldBe` Nothing + + describe "consumerAttributes (StableAndOld)" $ do + let attrs = consumerAttributes StableAndOld testConsumerProps testConsumerRecord + + it "sets messaging.operation (legacy)" $ + H.lookup "messaging.operation" attrs `shouldBe` Just (AttributeValue (TextAttribute "process")) + + it "sets messaging.operation.name (stable)" $ + H.lookup "messaging.operation.name" attrs `shouldBe` Just (AttributeValue (TextAttribute "process")) + + it "sets messaging.kafka.consumer.group (legacy)" $ + H.lookup "messaging.kafka.consumer.group" attrs `shouldBe` Just (AttributeValue (TextAttribute "my-consumer-group")) + + it "sets messaging.consumer.group.name (stable)" $ + H.lookup "messaging.consumer.group.name" attrs `shouldBe` Just (AttributeValue (TextAttribute "my-consumer-group"))