Skip to content

Commit

Permalink
fix: use topic attribute function in manager (#614)
Browse files Browse the repository at this point in the history 
## ❓ Why is this being changed

kafka manager is not using topic attribute function causing some labels
to be missing when reporting consumer lag
 
## 🧑‍💻 What is being changed

use the topic attribute function in the config if available
 
## ✅ How to validate the change

tests have been updated to ensure the label is there
  • Loading branch information
@kruskall
kruskall authored Jan 13, 2025
1 parent 28ffad4 commit 5b48d80
Show file tree
Hide file tree
Showing 2 changed files with 50 additions and 20 deletions.
56 changes: 36 additions & 20 deletions kafka/manager.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -136,22 +136,30 @@ func (m *Manager) DeleteTopics(ctx context.Context, topics ...apmqueue.Topic) er
deleteErrors = append(deleteErrors,
fmt.Errorf("failed to delete topic %q: %w", topic, err),
)
attrs := []attribute.KeyValue{
semconv.MessagingSystemKey.String("kafka"),
attribute.String("outcome", "failure"),
attribute.String("topic", topic),
}
if kv := m.cfg.TopicAttributeFunc(topic); kv != (attribute.KeyValue{}) {
attrs = append(attrs, kv)
}
m.deleted.Add(context.Background(), 1, metric.WithAttributeSet(
attribute.NewSet(
semconv.MessagingSystemKey.String("kafka"),
attribute.String("outcome", "failure"),
attribute.String("topic", topic),
),
attribute.NewSet(attrs...),
))
}
continue
}
attrs := []attribute.KeyValue{
semconv.MessagingSystemKey.String("kafka"),
attribute.String("outcome", "success"),
attribute.String("topic", topic),
}
if kv := m.cfg.TopicAttributeFunc(topic); kv != (attribute.KeyValue{}) {
attrs = append(attrs, kv)
}
m.deleted.Add(context.Background(), 1, metric.WithAttributeSet(
attribute.NewSet(
semconv.MessagingSystemKey.String("kafka"),
attribute.String("outcome", "success"),
attribute.String("topic", topic),
),
attribute.NewSet(attrs...),
))
logger.Info("deleted kafka topic")
}
Expand Down Expand Up @@ -284,23 +292,31 @@ func (m *Manager) MonitorConsumerLag(topicConsumers []apmqueue.TopicConsumer) (m
count := memberAssignments[key]
count++
memberAssignments[key] = count
attrs := []attribute.KeyValue{
attribute.String("group", l.Group),
attribute.String("topic", topic),
attribute.Int("partition", int(partition)),
}
if kv := m.cfg.TopicAttributeFunc(topic); kv != (attribute.KeyValue{}) {
attrs = append(attrs, kv)
}
o.ObserveInt64(
consumerGroupLagMetric, lag.Lag,
metric.WithAttributeSet(attribute.NewSet(
attribute.String("group", l.Group),
attribute.String("topic", topic),
attribute.Int("partition", int(partition)),
)),
metric.WithAttributeSet(attribute.NewSet(attrs...)),
)
}
}
for key, count := range memberAssignments {
attrs := []attribute.KeyValue{
attribute.String("group", l.Group),
attribute.String("topic", key.topic),
attribute.String("client_id", key.clientID),
}
if kv := m.cfg.TopicAttributeFunc(key.topic); kv != (attribute.KeyValue{}) {
attrs = append(attrs, kv)
}
o.ObserveInt64(assignmentMetric, count, metric.WithAttributeSet(
attribute.NewSet(
attribute.String("group", l.Group),
attribute.String("topic", key.topic),
attribute.String("client_id", key.clientID),
),
attribute.NewSet(attrs...),
))
}
})
Expand Down
14 changes: 14 additions & 0 deletions kafka/manager_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ func TestManagerDeleteTopics(t *testing.T) {
commonConfig.Logger = zap.New(core)
commonConfig.TracerProvider = tp
commonConfig.MeterProvider = mt.MeterProvider
commonConfig.TopicAttributeFunc = func(topic string) attribute.KeyValue { return attribute.KeyValue{} }
m, err := NewManager(ManagerConfig{CommonConfig: commonConfig})
require.NoError(t, err)
t.Cleanup(func() { m.Close() })
Expand Down Expand Up @@ -180,6 +181,9 @@ func TestManagerMetrics(t *testing.T) {
commonConfig.Logger = zap.New(core)
commonConfig.TracerProvider = tp
commonConfig.MeterProvider = mp
commonConfig.TopicAttributeFunc = func(topic string) attribute.KeyValue {
return attribute.Bool("foo", true)
}
m, err := NewManager(ManagerConfig{CommonConfig: commonConfig})
require.NoError(t, err)
t.Cleanup(func() { m.Close() })
Expand Down Expand Up @@ -459,34 +463,39 @@ func TestManagerMetrics(t *testing.T) {
attribute.String("group", "consumer1"),
attribute.String("topic", "topic1"),
attribute.Int("partition", 1),
attribute.Bool("foo", true),
),
Value: 0, // end offset = 1, committed = 1
}, {
Attributes: attribute.NewSet(
attribute.String("group", "consumer2"),
attribute.String("topic", "topic1"),
attribute.Int("partition", 2),
attribute.Bool("foo", true),
),
Value: 1, // end offset = 2, committed = 1
}, {
Attributes: attribute.NewSet(
attribute.String("group", "consumer2"),
attribute.String("topic", "topic2"),
attribute.Int("partition", 3),
attribute.Bool("foo", true),
),
Value: 2, // end offset = 3, committed = 1
}, {
Attributes: attribute.NewSet(
attribute.String("group", "consumer3"),
attribute.String("topic", "topic3"),
attribute.Int("partition", 4),
attribute.Bool("foo", true),
),
Value: 4, // end offset = 4, nothing committed
}, {
Attributes: attribute.NewSet(
attribute.String("group", "consumer3"),
attribute.String("topic", "mytopic"),
attribute.Int("partition", 1),
attribute.Bool("foo", true),
),
Value: 1, // end offset = 1, nothing committed
}},
Expand All @@ -498,34 +507,39 @@ func TestManagerMetrics(t *testing.T) {
attribute.String("client_id", "client_id"),
attribute.String("group", "consumer1"),
attribute.String("topic", "topic1"),
attribute.Bool("foo", true),
),
Value: 1,
}, {
Attributes: attribute.NewSet(
attribute.String("client_id", "client_id"),
attribute.String("group", "consumer2"),
attribute.String("topic", "topic2"),
attribute.Bool("foo", true),
),
Value: 1,
}, {
Attributes: attribute.NewSet(
attribute.String("client_id", "client_id"),
attribute.String("group", "consumer2"),
attribute.String("topic", "topic1"),
attribute.Bool("foo", true),
),
Value: 1,
}, {
Attributes: attribute.NewSet(
attribute.String("client_id", "client_id"),
attribute.String("group", "consumer3"),
attribute.String("topic", "topic3"),
attribute.Bool("foo", true),
),
Value: 1,
}, {
Attributes: attribute.NewSet(
attribute.String("client_id", "client_id"),
attribute.String("group", "consumer3"),
attribute.String("topic", "mytopic"),
attribute.Bool("foo", true),
),
Value: 1,
}},
Expand Down

0 comments on commit 5b48d80

Please sign in to comment.