[accountingservice] add attributes to kafka spans (#1286)

* add attributes to kafka spans Signed-off-by: Pierre Tessier <[email protected]> * add attributes to kafka spans Signed-off-by: Pierre Tessier <[email protected]> --------- Signed-off-by: Pierre Tessier <[email protected]>
open-telemetry · Dec 4, 2023 · 0244786 · 0244786
1 parent 8124353
commit 0244786
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 3 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -19,7 +19,10 @@ release.
   ([#1236](https://github.com/open-telemetry/opentelemetry-demo/pull/1236))
 * [cartservice] Add .NET memory, CPU, and thread metrics
   ([#1265](https://github.com/open-telemetry/opentelemetry-demo/pull/1265))
-* enable browser traffic in loadgenerator using playwright ([#1266](https://github.com/open-telemetry/opentelemetry-demo/pull/1266))
+* enable browser traffic in loadgenerator using playwright
+  ([#1266](https://github.com/open-telemetry/opentelemetry-demo/pull/1266))
+* [accountingservice] Add additional attributes to Kafka spans
+  ([#1286](https://github.com/open-telemetry/opentelemetry-demo/pull/1286))
 
 ## 1.6.0
 

diff --git a/src/accountingservice/kafka/consumer.go b/src/accountingservice/kafka/consumer.go
@@ -24,7 +24,7 @@ func StartConsumerGroup(ctx context.Context, brokers []string, log *logrus.Logge
 	saramaConfig.Version = ProtocolVersion
 	// So we can know the partition and offset of messages.
 	saramaConfig.Producer.Return.Successes = true
-	saramaConfig.Consumer.Interceptors = []sarama.ConsumerInterceptor{NewOTelInterceptor()}
+	saramaConfig.Consumer.Interceptors = []sarama.ConsumerInterceptor{NewOTelInterceptor(GroupID)}
 
 	consumerGroup, err := sarama.NewConsumerGroup(brokers, GroupID, saramaConfig)
 	if err != nil {

diff --git a/src/accountingservice/kafka/trace_interceptor.go b/src/accountingservice/kafka/trace_interceptor.go
@@ -21,12 +21,13 @@ type OTelInterceptor struct {
 
 // NewOTelInterceptor processes span for intercepted messages and add some
 // headers with the span data.
-func NewOTelInterceptor() *OTelInterceptor {
+func NewOTelInterceptor(groupID string) *OTelInterceptor {
 	oi := OTelInterceptor{}
 	oi.tracer = otel.Tracer("github.com/open-telemetry/opentelemetry-demo/accountingservice/sarama")
 
 	oi.fixedAttrs = []attribute.KeyValue{
 		semconv.MessagingSystem("kafka"),
+		semconv.MessagingKafkaConsumerGroup(groupID),
 		semconv.NetTransportTCP,
 	}
 	return &oi
@@ -50,6 +51,8 @@ func (oi *OTelInterceptor) OnConsume(msg *sarama.ConsumerMessage) {
 		trace.WithAttributes(
 			semconv.MessagingDestinationKindTopic,
 			semconv.MessagingDestinationName(msg.Topic),
+			semconv.MessagingKafkaMessageOffset(int(msg.Offset)),
+			semconv.MessagingMessagePayloadSizeBytes(len(msg.Value)),
 			semconv.MessagingOperationReceive,
 			semconv.MessagingKafkaDestinationPartition(int(msg.Partition)),
 		),