How to trace two asynchronous go routines with open telemetry

I am trying to trace a method that has two Go routines using Open Telemetry. The first Go routine reads from Kafka and creates a long-lasting job (can take anywhere from 1 second to 1 minute). Then, the second Go routine listens for the finished jobs.

What would be the correct way of doing the tracing so that we know which job result (in the second routine) corresponds to which kafka message (from the first routine)?

My guess is that the two spans created in the Go routines have to be linked via same traceId.

func startListening(ctx context.Context) {
  // initialise kafka client

  go kafkaConsumeMessages(ctx)
  go waitForJob(ctx)
}

func kafkaConsumeMessages(ctx) {
  for message := range kafkaEvents {
    // process message, create long job
    // create span here with traceID?
  }  

func waitForJobs(ctx) {
  for results := range finishedJobs
    // process result
    // create span here with traceID?
  }
}

Any suggestion is highly appreciated!

The answer was in fact easier than I thought. You need to pass further the tracing info attached to that long job and then decode it when processing the finished job.

In my case it, because I was using a text traceparent header and therefore a propagation.TraceContext{} implementation of propagation.TextMapPropagator, I decided to send the whole traceparent header (although I might need to do the same with tracestate) and then make use of the Extract method to decode the header when processing the finished job. But in order to use the Extract method, you need to implement the propagation.TextMapCarrier interface.

func startListening(ctx context.Context) {
  // initialise kafka client

  go kafkaConsumeMessages(ctx)
  go waitForJob(ctx)
}

func kafkaConsumeMessages(ctx) {
  for msg := range kafkaEvents {
    // extract incoming tracing info from traceparent header. Example at https://github.com/open-telemetry/opentelemetry-go-contrib/blob/main/instrumentation/github.com/Shopify/sarama/otelsarama/example/consumer/consumer.go#L84
    ctx := otel.GetTextMapPropagator().Extract(context.Background(), otelsarama.NewConsumerMessageCarrier(msg))

    // create span 
    tr := otel.Tracer("consumer")
	_, span := tr.Start(ctx, "consume message", trace.WithAttributes(
		semconv.MessagingOperationProcess,
	))
	defer span.End()
   
    // get just the traceparent header
    carrier := otelsarama.NewConsumerMessageCarrier(&msg)
    traceparentHeader := carrier.Get("traceparent")

    // process message, create long job and attach the header
    jobs.enqueue{TraceparentHeader: traceparentHeader}
  }  

func waitForJobs(ctx) {
  for result := range finishedJobs {
    ctx = otel.GetTextMapPropagator().Extract(ctx, models.PseudoCarrier{S: result.TraceparentHeader})
    ctx, span := tr.Start(ctx, "process result", trace.WithAttributes(
		attribute.String("jobName", result.JobName),
    ))
    defer span.End()
 
    // do more work 
  }
}

// PseudoCarrier implements the propagation.TextMapCarrier interface so we can use the propagation.Extract method when parsing the traceparent header
type PseudoCarrier struct {
	S string
}

func (c PseudoCarrier) Get(_ string) string {
	return c.S
}

func (c PseudoCarrier) Set(string, string) {}

func (c PseudoCarrier) Keys() []string {
	return []string{"traceparent"}
}