hs-opentelemetry-instrumentation-hw-kafka-client 0.1.0.0 → 1.0.0.0
raw patch · 5 files changed
+347/−45 lines, 5 filesdep +hs-opentelemetry-instrumentation-hw-kafka-clientdep +hspecdep +unordered-containersdep −case-insensitivedep −http-typesdep ~hs-opentelemetry-apidep ~hs-opentelemetry-semantic-conventions
Dependencies added: hs-opentelemetry-instrumentation-hw-kafka-client, hspec, unordered-containers
Dependencies removed: case-insensitive, http-types
Dependency ranges changed: hs-opentelemetry-api, hs-opentelemetry-semantic-conventions
Files
- CHANGELOG.md +11/−0
- LICENSE +1/−1
- hs-opentelemetry-instrumentation-hw-kafka-client.cabal +33/−5
- src/OpenTelemetry/Instrumentation/Kafka.hs +126/−39
- test/Spec.hs +176/−0
CHANGELOG.md view
@@ -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.
LICENSE view
@@ -1,4 +1,4 @@-Copyright (c) 2024, Alberto Fanton+Copyright (c) 2024-2026, Alberto Fanton All rights reserved.
hs-opentelemetry-instrumentation-hw-kafka-client.cabal view
@@ -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
src/OpenTelemetry/Instrumentation/Kafka.hs view
@@ -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
+ test/Spec.hs view
@@ -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"))