Add error recording

utr1903 · Feb 4, 2024 · c38a80d · c38a80d
1 parent 829cb8a
commit c38a80d
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 4 deletions.
diff --git a/apps/golang/commons/otel/kafka/kafka_consumer_interceptor.go b/apps/golang/commons/otel/kafka/kafka_consumer_interceptor.go
@@ -58,7 +58,7 @@ func (k *KafkaConsumer) Intercept(
 	consumerGroup string,
 ) (
 	context.Context,
-	func(),
+	func(err error),
 ) {
 	consumeStartTime := time.Now()
 
@@ -82,11 +82,17 @@ func (k *KafkaConsumer) Intercept(
 	)
 
 	// Record consumer latency
-	endConsume := func() {
+	endConsume := func(
+		err error,
+	) {
 		elapsedTime := float64(time.Since(consumeStartTime)) / float64(time.Millisecond)
+		attrs := semconv.WithMessagingKafkaConsumerAttributes(msg, consumerGroup)
+		if err != nil {
+			attrs = append(attrs, semconv.ErrorType.String(err.Error()))
+		}
 		k.latency.Record(ctx, elapsedTime,
 			metric.WithAttributes(
-				semconv.WithMessagingKafkaConsumerAttributes(msg, consumerGroup)...,
+				attrs...,
 			))
 		span.End()
 	}

diff --git a/apps/golang/kafkaconsumer/consumer/consumer.go b/apps/golang/kafkaconsumer/consumer/consumer.go
@@ -203,7 +203,6 @@ func (g *groupHandler) consumeMessage(
 	// Create consumer span (parent)
 	ctx := context.Background()
 	ctx, endConsume := g.Consumer.Intercept(ctx, msg, g.Opts.ConsumerGroupId)
-	defer endConsume()
 
 	// Parse name out of the message
 	name := string(msg.Value)
@@ -214,13 +213,15 @@ func (g *groupHandler) consumeMessage(
 	err := g.storeIntoDb(ctx, name)
 	if err != nil {
 		g.logger.Log(logrus.ErrorLevel, ctx, name, "Consuming message is failed.")
+		endConsume(err)
 		return nil
 	}
 
 	// Acknowledge message
 	session.MarkMessage(msg, "")
 	g.logger.Log(logrus.InfoLevel, ctx, name, "Consuming message is succeeded.")
 
+	endConsume(nil)
 	return nil
 }