unistack-org/micro-broker-kgo
4
0
Fork 2
Code Issues 3 Pull Requests 2 Actions 8 Packages Projects Releases 44 Wiki Activity

fixup for attrs
Some checks failed
build / test (push) Failing after 1m31s
Details
codeql / analyze (go) (push) Failing after 1m53s
Details
build / lint (push) Successful in 9m15s
Details

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2024-03-15 09:38:31 +03:00
parent 56288f46b1
commit 9c4d88bb69
1 changed files with 4 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
tracer.go
View File

@ -7,7 +7,6 @@ import (
"unicode/utf8" "unicode/utf8"
"github.com/twmb/franz-go/pkg/kgo" "github.com/twmb/franz-go/pkg/kgo"
"go.opentelemetry.io/otel/attribute"
semconv "go.opentelemetry.io/otel/semconv/v1.18.0" semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
"go.unistack.org/micro/v3/tracer" "go.unistack.org/micro/v3/tracer"
) )
@ -61,7 +60,7 @@ func (m *hookTracer) OnFetchBatchRead(meta kgo.BrokerMetadata, topic string, _ i
// hook. // hook.
func (m *hookTracer) OnProduceRecordBuffered(r *kgo.Record) { func (m *hookTracer) OnProduceRecordBuffered(r *kgo.Record) {
// Set up span options. // Set up span options.
attrs := []attribute.KeyValue{ attrs := []interface{}{
semconv.MessagingSystemKey.String("kafka"), semconv.MessagingSystemKey.String("kafka"),
semconv.MessagingDestinationKindTopic, semconv.MessagingDestinationKindTopic,
semconv.MessagingDestinationName(r.Topic), semconv.MessagingDestinationName(r.Topic),
@ -111,7 +110,7 @@ func (m *hookTracer) OnProduceRecordUnbuffered(r *kgo.Record, err error) {
// processing. // processing.
func (m *hookTracer) OnFetchRecordBuffered(r *kgo.Record) { func (m *hookTracer) OnFetchRecordBuffered(r *kgo.Record) {
// Set up the span options. // Set up the span options.
attrs := []attribute.KeyValue{ attrs := []interface{}{
semconv.MessagingSystemKey.String("kafka"), semconv.MessagingSystemKey.String("kafka"),
semconv.MessagingSourceKindTopic, semconv.MessagingSourceKindTopic,
semconv.MessagingSourceName(r.Topic), semconv.MessagingSourceName(r.Topic),
@ -160,7 +159,7 @@ func (m *hookTracer) OnFetchRecordUnbuffered(r *kgo.Record, _ bool) {
// iteration of your processing for the record. // iteration of your processing for the record.
func (m *hookTracer) WithProcessSpan(r *kgo.Record) (context.Context, tracer.Span) { func (m *hookTracer) WithProcessSpan(r *kgo.Record) (context.Context, tracer.Span) {
// Set up the span options. // Set up the span options.
attrs := []attribute.KeyValue{ attrs := []interface{}{
semconv.MessagingSystemKey.String("kafka"), semconv.MessagingSystemKey.String("kafka"),
semconv.MessagingSourceKindTopic, semconv.MessagingSourceKindTopic,
semconv.MessagingSourceName(r.Topic), semconv.MessagingSourceName(r.Topic),
@ -187,7 +186,7 @@ func (m *hookTracer) WithProcessSpan(r *kgo.Record) (context.Context, tracer.Spa
return m.tracer.Start(r.Context, r.Topic+" process", opts...) return m.tracer.Start(r.Context, r.Topic+" process", opts...)
} }
func maybeKeyAttr(attrs []attribute.KeyValue, r *kgo.Record) []attribute.KeyValue { func maybeKeyAttr(attrs []interface{}, r *kgo.Record) []interface{} {
if r.Key == nil { if r.Key == nil {
return attrs return attrs
} }