feat: Add topic to internal telemetry attributes (35336)

open-telemetry · Nov 4, 2024 · 80242c5 · 80242c5
1 parent 908ccf2
commit 80242c5
Showing 1 changed file with 7 additions and 3 deletions.
diff --git a/receiver/kafkareceiver/kafka_receiver.go b/receiver/kafkareceiver/kafka_receiver.go
@@ -29,6 +29,7 @@ const (
 	transport = "kafka"
 	// TODO: update the following attributes to reflect semconv
 	attrInstanceName = "name"
+	attrTopic        = "topic"
 	attrPartition    = "partition"
 )
 
@@ -553,6 +554,7 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 			ctx := c.obsrecv.StartTracesOp(session.Context())
 			attrs := attribute.NewSet(
 				attribute.String(attrInstanceName, c.id.String()),
+				attribute.String(attrTopic, claim.Topic()),
 				attribute.String(attrPartition, strconv.Itoa(int(claim.Partition()))),
 			)
 			c.telemetryBuilder.KafkaReceiverMessages.Add(ctx, 1, metric.WithAttributeSet(attrs))
@@ -562,7 +564,7 @@ func (c *tracesConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSe
 			traces, err := c.unmarshaler.Unmarshal(message.Value)
 			if err != nil {
 				c.logger.Error("failed to unmarshal message", zap.Error(err))
-				c.telemetryBuilder.KafkaReceiverUnmarshalFailedSpans.Add(session.Context(), 1, metric.WithAttributes(attribute.String(attrInstanceName, c.id.String())))
+				c.telemetryBuilder.KafkaReceiverUnmarshalFailedSpans.Add(session.Context(), 1, metric.WithAttributeSet(attrs))
 				if c.messageMarking.After && c.messageMarking.OnError {
 					session.MarkMessage(message, "")
 				}
@@ -630,6 +632,7 @@ func (c *metricsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupS
 			ctx := c.obsrecv.StartMetricsOp(session.Context())
 			attrs := attribute.NewSet(
 				attribute.String(attrInstanceName, c.id.String()),
+				attribute.String(attrTopic, claim.Topic()),
 				attribute.String(attrPartition, strconv.Itoa(int(claim.Partition()))),
 			)
 			c.telemetryBuilder.KafkaReceiverMessages.Add(ctx, 1, metric.WithAttributeSet(attrs))
@@ -639,7 +642,7 @@ func (c *metricsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupS
 			metrics, err := c.unmarshaler.Unmarshal(message.Value)
 			if err != nil {
 				c.logger.Error("failed to unmarshal message", zap.Error(err))
-				c.telemetryBuilder.KafkaReceiverUnmarshalFailedMetricPoints.Add(session.Context(), 1, metric.WithAttributes(attribute.String(attrInstanceName, c.id.String())))
+				c.telemetryBuilder.KafkaReceiverUnmarshalFailedMetricPoints.Add(session.Context(), 1, metric.WithAttributeSet(attrs))
 				if c.messageMarking.After && c.messageMarking.OnError {
 					session.MarkMessage(message, "")
 				}
@@ -707,6 +710,7 @@ func (c *logsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSess
 			ctx := c.obsrecv.StartLogsOp(session.Context())
 			attrs := attribute.NewSet(
 				attribute.String(attrInstanceName, c.id.String()),
+				attribute.String(attrTopic, claim.Topic()),
 				attribute.String(attrPartition, strconv.Itoa(int(claim.Partition()))),
 			)
 			c.telemetryBuilder.KafkaReceiverMessages.Add(ctx, 1, metric.WithAttributeSet(attrs))
@@ -716,7 +720,7 @@ func (c *logsConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSess
 			logs, err := c.unmarshaler.Unmarshal(message.Value)
 			if err != nil {
 				c.logger.Error("failed to unmarshal message", zap.Error(err))
-				c.telemetryBuilder.KafkaReceiverUnmarshalFailedLogRecords.Add(ctx, 1, metric.WithAttributes(attribute.String(attrInstanceName, c.id.String())))
+				c.telemetryBuilder.KafkaReceiverUnmarshalFailedLogRecords.Add(ctx, 1, metric.WithAttributeSet(attrs))
 				if c.messageMarking.After && c.messageMarking.OnError {
 					session.MarkMessage(message, "")
 				}