Using custom sender on Jaeger

[theprobugmaker]

I'm trying to find an approach as mentioned in another discussion to make it possible for me to have the spans on Kafka.
I found an article that has an interesting approach that is: instead of sending the spans directly to Jaeger, why not sending it to Kafka first? https://www.confluent.io/blog/fault-tolerance-distributed-systems-tracing-with-apache-kafka-jaeger/
It's a fault tolerance approach cause if Jaeger is down I will have the spans on Kafka.

The approach consists on creating a custom sender:

https://github.com/burkaa01/jaeger-tracing-kafka-sender/blob/83ab26d7a779326d3950a55db7856b32744ba6f9/stream-app/src/main/java/com/github/burkaa01/stream/config/tracing/KafkaSender.java#L39-L55

It will receive the spans and send to a Kafka topic.

That is how he configured the custom sender:

https://github.com/burkaa01/jaeger-tracing-kafka-sender/blob/83ab26d7a779326d3950a55db7856b32744ba6f9/stream-app/src/main/java/com/github/burkaa01/stream/config/tracing/TracingConfig.java#L32-L45

It's almost similar as we do in Golang:

	config := jaegerconfig.Configuration{
		Sampler: &jaegerconfig.SamplerConfig{
			Type:  jaeger.SamplerTypeConst,
			Param: 1,
		},
		Reporter: &jaegerconfig.ReporterConfig{
			LogSpans: true,
		},
		ServiceName: "jaeger-rocks",
	}

Questions:

1 - Is it possible to pass a custom reporter as the Java example?
2 - Is there an example or something I can follow from to create a custom sender as the Java example?

Thank you very much.

[theprobugmaker]

I tried and implemented a custom reporter, a simple one like this:

type KafkaReporter struct{}

func (k *KafkaReporter) Report(s *jaeger.Span) {
	fmt.Println(s.Duration())
}

func (k *KafkaReporter) Close() {

}

func NewKafkaReporter() *KafkaReporter {
	return &KafkaReporter{}
}

and used like this:

	reporter := NewKafkaReporter()

	config := jaegerconfig.Configuration{
		Sampler: &jaegerconfig.SamplerConfig{
			Type:  jaeger.SamplerTypeConst,
			Param: 1,
		},
		ServiceName: "jaeger-rocks",
	}

	tracer, closer, err := config.NewTracer(
		jaegerconfig.Logger(jaeger.StdLogger),
		jaegerconfig.Metrics(metrics.NullFactory),
		jaegerconfig.Reporter(reporter),
	)

It's working, I can see the span duration but the Kafka example has another application running reading from the Kafka topic and sending to Jaeger using the API, the thing is that the Java sender is using Thrift so it's sending in a format that when sending it to Jaeger it will understand.

I'm confused on how I can do the same in Golang:

public class KafkaSender extends ThriftSender {

    public void send(Process process, List<Span> spans) throws SenderException {
        Batch batch = new Batch(process, spans);
        byte[] bytes;
        try {
            bytes = serialize(batch);
        } catch (Exception e) {
            throw new SenderException(String.format("Failed to serialize %d spans", spans.size()), e, spans.size());
        }
        if (bytes != null) {
            ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, bytes);
            producer.send(record, (RecordMetadata recordMetadata, Exception exception) -> {
                if (exception != null) {
                    LOGGER.error(String.format("Could not send %d spans", spans.size()), exception);
                }
            });
        }
    }

They are all using these imports:

import io.jaegertracing.thrift.internal.senders.ThriftSender;
import io.jaegertracing.thriftjava.Batch;
import io.jaegertracing.thriftjava.Span;

[theprobugmaker]

I just found this: https://github.com/jaegertracing/jaeger-client-go/blob/7aa7af5fd0410a778e14e0a50d062ce4f7cb9e73/jaeger_thrift_span.go#L28

I will try to implement something and probably put in here if I came up with something.

I'm answering with things I'm finding cause it might help someone else. If you have anything to add I'd appreciate.

[theprobugmaker]

I was able to send to Kafka by doing this, but there are information missing like the service name for example, is there a way to get that?:

func (k *KafkaReporter) Report(s *jaeger.Span) {
	thrift := jaeger.BuildJaegerThrift(s)

	bytes, err := json.Marshal(&thrift)
	if err != nil {
		log.Fatalf("failed to marshal json: %s", err.Error())
	}

	msg := &sarama.ProducerMessage{
		Topic: k.topic,
		Key:   sarama.StringEncoder(fmt.Sprintf("%d", s.SpanContext().SpanID())),
		Value: sarama.ByteEncoder(bytes),
	}

	_, _, err = k.producer.SendMessage(msg)
	if err != nil {
		log.Fatalf("failed to publish message: %s", err.Error())
	}
}