fixup tracing

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

event.go

@@ -1,12 +1,14 @@
 package kgo
 
 import (
+	"context"
 	"sync"
 
 	"go.unistack.org/micro/v3/broker"
 )
 
 type event struct {
+	ctx   context.Context
 	topic string
 	err   error
 	sync.RWMutex
@@ -14,6 +16,10 @@ type event struct {
 	ack bool
 }
 
+func (p *event) Context() context.Context {
+	return p.ctx
+}
+
 func (p *event) Topic() string {
 	return p.topic
 }

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/twmb/franz-go/pkg/kadm v1.11.0
 	github.com/twmb/franz-go/pkg/kmsg v1.7.0
 	go.opentelemetry.io/otel v1.25.0
-	go.unistack.org/micro/v3 v3.10.58
+	go.unistack.org/micro/v3 v3.10.59
 )
 
 require (

go.sum

@@ -20,8 +20,8 @@ github.com/twmb/franz-go/pkg/kmsg v1.7.0 h1:a457IbvezYfA5UkiBvyV3zj0Is3y1i8EJgqj
 github.com/twmb/franz-go/pkg/kmsg v1.7.0/go.mod h1:se9Mjdt0Nwzc9lnjJ0HyDtLyBnaBDAd7pCje47OhSyw=
 go.opentelemetry.io/otel v1.25.0 h1:gldB5FfhRl7OJQbUHt/8s0a7cE8fbsPAtdpRaApKy4k=
 go.opentelemetry.io/otel v1.25.0/go.mod h1:Wa2ds5NOXEMkCmUou1WA7ZBfLTHWIsp034OVD7AO+Vg=
-go.unistack.org/micro/v3 v3.10.58 h1:2lIQUfb3XdVstcKz7LWm7vVs/HGTzDfG9Q902FcU3xM=
-go.unistack.org/micro/v3 v3.10.58/go.mod h1:erMgt3Bl7vQQ0e9UpQyR5NlLiZ9pKeEJ9+1tfYFaqUg=
+go.unistack.org/micro/v3 v3.10.59 h1:eneYXJLgyu5MZpSvyI0K17CeXvgOoUCN5dWZaPV5lI4=
+go.unistack.org/micro/v3 v3.10.59/go.mod h1:erMgt3Bl7vQQ0e9UpQyR5NlLiZ9pKeEJ9+1tfYFaqUg=
 golang.org/x/crypto v0.22.0 h1:g1v0xeRhjcugydODzvb3mEM9SQ0HGp9s/nh3COQ/C30=
 golang.org/x/crypto v0.22.0/go.mod h1:vr6Su+7cTlO45qkww3VDJlzDn0ctJvRgYbC2NvXHt+M=
 golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4=

kgo.go

@@ -76,9 +76,8 @@ func (k *Broker) Name() string {
 func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, error) {
 	var c *kgo.Client
 	var err error
-	var span tracer.Span
-	ctx, span = k.opts.Tracer.Start(ctx, "Connect")
-	defer span.Finish()
+
+	sp, _ := tracer.SpanFromContext(ctx)
 
 	clientID := "kgo"
 	group := ""
@@ -99,7 +98,9 @@ func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, err
 	select {
 	case <-ctx.Done():
 		if ctx.Err() != nil {
-			span.SetStatus(tracer.SpanStatusError, ctx.Err().Error())
+			if sp != nil {
+				sp.SetStatus(tracer.SpanStatusError, ctx.Err().Error())
+			}
 		}
 		return nil, ctx.Err()
 	default:
@@ -108,7 +109,9 @@ func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, err
 			err = c.Ping(ctx) // check connectivity to cluster
 		}
 		if err != nil {
-			span.SetStatus(tracer.SpanStatusError, err.Error())
+			if sp != nil {
+				sp.SetStatus(tracer.SpanStatusError, err.Error())
+			}
 			return nil, err
 		}
 	}
@@ -231,10 +234,6 @@ func (k *Broker) Publish(ctx context.Context, topic string, msg *broker.Message,
 }
 
 func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
-	var span tracer.Span
-	ctx, span = k.opts.Tracer.Start(ctx, "Publish")
-	defer span.Finish()
-
 	k.Lock()
 	if !k.connected {
 		c, err := k.connect(ctx, k.kopts...)

subscriber.go

@@ -12,6 +12,7 @@ import (
 	"go.unistack.org/micro/v3/logger"
 	"go.unistack.org/micro/v3/metadata"
 	"go.unistack.org/micro/v3/semconv"
+	"go.unistack.org/micro/v3/tracer"
 )
 
 type tp struct {
@@ -91,7 +92,7 @@ func (s *subscriber) poll(ctx context.Context) {
 				ticker.Stop()
 				return
 			case <-ticker.C:
-				dgls, err := ac.Lag(ctx, s.topic)
+				dgls, err := ac.Lag(ctx, s.opts.Group)
 				if err != nil || !dgls.Ok() {
 					continue
 				}
@@ -105,11 +106,11 @@ func (s *subscriber) poll(ctx context.Context) {
 					continue
 				}
 
-				for tp := range s.consumers {
-					if v, ok := lmap[tp.p]; ok {
-						s.kopts.Meter.Counter(semconv.BrokerGroupLag, "topic", s.topic, "group", s.opts.Group, "partition", strconv.Itoa(int(tp.p)), "lag", strconv.Itoa(int(v.Lag)))
-					}
+				s.Lock()
+				for p, l := range lmap {
+					s.kopts.Meter.Counter(semconv.BrokerGroupLag, "topic", s.topic, "group", s.opts.Group, "partition", strconv.Itoa(int(p)), "lag", strconv.Itoa(int(l.Lag)))
 				}
+				s.Unlock()
 
 			}
 		}
@@ -186,7 +187,9 @@ func (s *subscriber) assigned(_ context.Context, c *kgo.Client, assigned map[str
 				kopts:   s.kopts,
 				opts:    s.opts,
 			}
+			s.Lock()
 			s.consumers[tp{topic, partition}] = pc
+			s.Unlock()
 			go pc.consume()
 		}
 	}
@@ -217,6 +220,9 @@ func (pc *consumer) consume() {
 				p.err = nil
 				p.ack = false
 				p.msg.Header = metadata.New(len(record.Headers))
+				p.ctx = record.Context
+				sp, _ := tracer.SpanFromContext(p.ctx)
+				defer sp.Finish()
 				for _, hdr := range record.Headers {
 					p.msg.Header.Set(hdr.Key, string(hdr.Value))
 				}
@@ -225,7 +231,11 @@ func (pc *consumer) consume() {
 				} else if pc.opts.BodyOnly {
 					p.msg.Body = record.Value
 				} else {
-					if err := pc.kopts.Codec.Unmarshal(record.Value, p.msg); err != nil {
+					sp.AddEvent("codec unmarshal start")
+					err := pc.kopts.Codec.Unmarshal(record.Value, p.msg)
+					sp.AddEvent("codec unmarshal stop")
+					if err != nil {
+						sp.SetStatus(tracer.SpanStatusError, err.Error())
 						pc.kopts.Meter.Counter(semconv.SubscribeMessageTotal, "endpoint", record.Topic, "topic", record.Topic, "status", "failure").Inc()
 						p.err = err
 						p.msg.Body = record.Value
@@ -258,10 +268,13 @@ func (pc *consumer) consume() {
 						return
 					}
 				}
+				sp.AddEvent("handler start")
 				err := pc.handler(p)
+				sp.AddEvent("handler stop")
 				if err == nil {
 					pc.kopts.Meter.Counter(semconv.SubscribeMessageTotal, "endpoint", record.Topic, "topic", record.Topic, "status", "success").Inc()
 				} else {
+					sp.SetStatus(tracer.SpanStatusError, err.Error())
 					pc.kopts.Meter.Counter(semconv.SubscribeMessageTotal, "endpoint", record.Topic, "topic", record.Topic, "status", "failure").Inc()
 				}
 				pc.kopts.Meter.Counter(semconv.SubscribeMessageInflight, "endpoint", record.Topic, "topic", record.Topic).Dec()