Merge branch 'main' into update-to-21

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
 

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 {

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)),
 		),

src/checkoutservice/main.go

@@ -181,8 +181,7 @@ func main() {
 	}
 
 	var srv = grpc.NewServer(
-		grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor()),
-		grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor()),
+		grpc.StatsHandler(otelgrpc.NewServerHandler()),
 	)
 	pb.RegisterCheckoutServiceServer(srv, svc)
 	healthpb.RegisterHealthServer(srv, svc)
@@ -343,8 +342,7 @@ func (cs *checkoutService) prepareOrderItemsAndShippingQuoteFromCart(ctx context
 func createClient(ctx context.Context, svcAddr string) (*grpc.ClientConn, error) {
 	return grpc.DialContext(ctx, svcAddr,
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()),
-		grpc.WithStreamInterceptor(otelgrpc.StreamClientInterceptor()),
+		grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
 	)
 }
 

src/productcatalogservice/main.go

@@ -141,8 +141,7 @@ func main() {
 	}
 
 	srv := grpc.NewServer(
-		grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor()),
-		grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor()),
+		grpc.StatsHandler(otelgrpc.NewServerHandler()),
 	)
 
 	reflection.Register(srv)
@@ -278,7 +277,6 @@ func (p *productCatalog) checkProductFailure(ctx context.Context, id string) boo
 func createClient(ctx context.Context, svcAddr string) (*grpc.ClientConn, error) {
 	return grpc.DialContext(ctx, svcAddr,
 		grpc.WithTransportCredentials(insecure.NewCredentials()),
-		grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()),
-		grpc.WithStreamInterceptor(otelgrpc.StreamClientInterceptor()),
+		grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
 	)
 }