contrib/segmentio/kafka-go: add tracing for kafka writer and kafka reader #897
Conversation
There was a problem hiding this comment.
Thanks for your contribution. Can you please take a moment to read our contribution guidelines. I think we ultimately want this change, but please go ahead and open an issue so we can follow the regular process...
Thanks for reviewing this PR, created #899 issue. |
|
@gbbr CircleCI metadata is failing still, not sure what I am missing, Can you help me fix the same? |
.circleci/config.yml
Outdated
| @@ -119,6 +119,17 @@ jobs: | |||
| KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092 | |||
| KAFKA_CREATE_TOPICS: gotest:1:1 | |||
| KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "1" | |||
| - image: wurstmeister/zookeeper:3.4.6 | |||
There was a problem hiding this comment.
We are already running Kafka above (confluentinc/cp-kafka:5.0.0). Can we not use that instance instead? Why do we need a different one?
There was a problem hiding this comment.
I tried using that one, it didn't work as expected.
When using same topic, it reads the messages from confluent test producer.
Tried with different topic, added the new topic in circleci config, got the error message saying the topic doesn't exist though I see topic is created.
Due to librdkafka requirment, I couldn't test the changes in my local.
One advantage of using a separate will remove the dependency and the unexpected issues, So I added this image.
There was a problem hiding this comment.
Let's use just one of them for both, whichever works.
There was a problem hiding this comment.
using wurstmeister/zookeeper and wurstmeister/kafka for both
| } | ||
|
|
||
| // NewMessageCarrier creates a new MessageCarrier. | ||
| func NewMessageCarrier(msg *kafka.Message) MessageCarrier { |
There was a problem hiding this comment.
Does this need to be part of the public API? Probably not. We can remove the function and just use contextCarrier(msg) inline
There was a problem hiding this comment.
That works for me, I added this method here because both confuent-kafka and sarama package has this.
The reason we may need when the clients want to Extract the span context from the header.
I have added ExtractSpanContextFromMessage method todo that.
| ) | ||
|
|
||
| // A MessageCarrier injects and extracts traces from a sarama.ProducerMessage. | ||
| type MessageCarrier struct { |
There was a problem hiding this comment.
Do we need a struct? Probably not
| type MessageCarrier struct { | |
| // contextCarrier implements tracer.TextMapWriter and tracer.TextMapReader on top of a kafka.Message. | |
| type contextCarrier *kafka.Message |
There was a problem hiding this comment.
Yes, I verified it, it doesn't update the header when we use contextCarrier
| prev ddtrace.Span | ||
| } | ||
|
|
||
| func (r *Reader) startSpan(ctx context.Context, msg *kafka.Message) ddtrace.Span { |
There was a problem hiding this comment.
There seems to be a ton of duplication between here and our other Kafka integration. In order to keep the experience consistent I'd prefer it if we would factor out the duplication into another package, let's say contrib/internal/kafkautil. WDYT? I mean both code and tests.
There was a problem hiding this comment.
I also thought the same, didn't get time to do that.
There was a problem hiding this comment.
Unfortunately we'll have to do that, it's too much duplication and I don't want to risk discrepancies and unilateral bugs.
There was a problem hiding this comment.
Ok, will do it when I get time
|
Hey y'all, was curious if there were any plans to pick this back up any time soon? No worries if not, but figured I'd check! |
|
This is in our backlog now and we'll continue with it when we can schedule it. |
|
This was merged with #1152 |
Fixes #899
Currently datadog tracing wrapper is not available for segmentio/kafka-go, add segmentio/kafka-go wrapper for KafkaWriter and KafkaReader in github.com/segementio/kafka-go package to create traces when writing messages to kafka topic and reading message from kafka topic
This implemention is similar to Sarama and Confluentinc kafka.
Also added the integration test to test the Reader and Writer wrappers.