Merge pull request #653 from yianni/john/kotel-tombstone-attr

kotel/tracer: add tombstone attribute
twmb · Dec 28, 2023 · 1ec2d22 · 1ec2d22
2 parents dec75b3 + dcab256
commit 1ec2d22
Showing 1 changed file with 9 additions and 0 deletions.
diff --git a/plugin/kotel/tracer.go b/plugin/kotel/tracer.go
@@ -137,6 +137,9 @@ func (t *Tracer) WithProcessSpan(r *kgo.Record) (context.Context, trace.Span) {
 	if t.consumerGroup != "" {
 		attrs = append(attrs, semconv.MessagingKafkaConsumerGroupKey.String(t.consumerGroup))
 	}
+	if r.Key != nil && r.Value == nil {
+		attrs = append(attrs, semconv.MessagingKafkaMessageTombstoneKey.Bool(true))
+	}
 	opts := []trace.SpanStartOption{
 		trace.WithAttributes(attrs...),
 		trace.WithSpanKind(trace.SpanKindConsumer),
@@ -169,6 +172,9 @@ func (t *Tracer) OnProduceRecordBuffered(r *kgo.Record) {
 	if t.clientID != "" {
 		attrs = append(attrs, semconv.MessagingKafkaClientIDKey.String(t.clientID))
 	}
+	if r.Key != nil && r.Value == nil {
+		attrs = append(attrs, semconv.MessagingKafkaMessageTombstoneKey.Bool(true))
+	}
 	opts := []trace.SpanStartOption{
 		trace.WithAttributes(attrs...),
 		trace.WithSpanKind(trace.SpanKindProducer),
@@ -221,6 +227,9 @@ func (t *Tracer) OnFetchRecordBuffered(r *kgo.Record) {
 	if t.consumerGroup != "" {
 		attrs = append(attrs, semconv.MessagingKafkaConsumerGroupKey.String(t.consumerGroup))
 	}
+	if r.Key != nil && r.Value == nil {
+		attrs = append(attrs, semconv.MessagingKafkaMessageTombstoneKey.Bool(true))
+	}
 	opts := []trace.SpanStartOption{
 		trace.WithAttributes(attrs...),
 		trace.WithSpanKind(trace.SpanKindConsumer),