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

minimize memory usage, name all traces with sdk.broker
Some checks failed
build / lint (push) Failing after 8s
Details
build / test (push) Failing after 7s
Details
codeql / analyze (go) (push) Failing after 11s
Details

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов 2024-09-27 13:28:17 +03:00
parent 8e3f2c67d7
commit c1fa2f639d
1 changed files with 10 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
tracer.go
View File

@ -16,6 +16,8 @@ type hookTracer struct {
tracer tracer.Tracer
}
var messagingSystem = semconv.MessagingSystemKey.String("kafka")
var (
_ kgo.HookProduceRecordBuffered = (*hookTracer)(nil)
_ kgo.HookProduceRecordUnbuffered = (*hookTracer)(nil)
@ -32,7 +34,7 @@ var (
func (m *hookTracer) OnProduceRecordBuffered(r *kgo.Record) {
// Set up span options.
attrs := []interface{}{
semconv.MessagingSystemKey.String("kafka"),
messagingSystem,
semconv.MessagingDestinationKindTopic,
semconv.MessagingDestinationName(r.Topic),
semconv.MessagingOperationPublish,
@ -59,9 +61,9 @@ func (m *hookTracer) OnProduceRecordBuffered(r *kgo.Record) {
}
if !ok {
r.Context, _ = m.tracer.Start(metadata.NewOutgoingContext(r.Context, md), r.Topic+" publish", opts...)
r.Context, _ = m.tracer.Start(metadata.NewOutgoingContext(r.Context, md), "sdk.broker", opts...)
} else {
r.Context, _ = m.tracer.Start(r.Context, r.Topic+" publish", opts...)
r.Context, _ = m.tracer.Start(r.Context, "sdk.broker", opts...)
}
md, _ = metadata.FromOutgoingContext(r.Context)
@ -96,7 +98,7 @@ func (m *hookTracer) OnProduceRecordUnbuffered(r *kgo.Record, err error) {
func (m *hookTracer) OnFetchRecordBuffered(r *kgo.Record) {
// Set up the span options.
attrs := []interface{}{
semconv.MessagingSystemKey.String("kafka"),
messagingSystem,
semconv.MessagingSourceKindTopic,
semconv.MessagingSourceName(r.Topic),
semconv.MessagingOperationReceive,
@ -126,9 +128,9 @@ func (m *hookTracer) OnFetchRecordBuffered(r *kgo.Record) {
}
if !ok {
r.Context, _ = m.tracer.Start(metadata.NewIncomingContext(r.Context, md), r.Topic+" receive", opts...)
r.Context, _ = m.tracer.Start(metadata.NewIncomingContext(r.Context, md), "sdk.broker", opts...)
} else {
r.Context, _ = m.tracer.Start(r.Context, r.Topic+" receive", opts...)
r.Context, _ = m.tracer.Start(r.Context, "sdk.broker", opts...)
}
md, _ = metadata.FromIncomingContext(r.Context)
@ -155,7 +157,7 @@ func (m *hookTracer) OnFetchRecordUnbuffered(r *kgo.Record, _ bool) {
func (m *hookTracer) WithProcessSpan(r *kgo.Record) (context.Context, tracer.Span) {
// Set up the span options.
attrs := []interface{}{
semconv.MessagingSystemKey.String("kafka"),
messagingSystem,
semconv.MessagingSourceKindTopic,
semconv.MessagingSourceName(r.Topic),
semconv.MessagingOperationProcess,
@ -186,7 +188,7 @@ func (m *hookTracer) WithProcessSpan(r *kgo.Record) (context.Context, tracer.Spa
}
// Start a new span using the provided context and options.
return m.tracer.Start(r.Context, r.Topic+" process", opts...)
return m.tracer.Start(r.Context, "sdk.broker", opts...)
}
func maybeKeyAttr(attrs []interface{}, r *kgo.Record) []interface{} {