Application Real-Time Monitoring Service:Maintain trace continuity in Asynq versions earlier than v0.26.0

Solution

The Task struct in Asynq versions earlier than v0.26.0 does not include a carrier to propagate the SpanContext. To maintain trace continuity, embed the trace context directly in the task payload.

// Task represents a unit of work to be performed.
type Task struct {
    // typename indicates the type of task to be performed.
    typename string

    // payload holds data needed to perform the task.
    payload []byte

    // opts holds options for the task.
    opts []Option

    // w is the ResultWriter for the task.
    w *ResultWriter
}

For example, assume the Task payload struct is defined as follows:

type WelcomeEmailPayload struct {
	UserID   int    `json:"user_id"`
	Email    string `json:"email"`
	Username string `json:"username"`
}

Add a Header field to carry the trace context:

type WelcomeEmailPayload struct {
	UserID   int    `json:"user_id"`
	Email    string `json:"email"`
	Username string `json:"username"`
    Header   map[string]string `json:"header"`
}

Before creating a task, create a span and inject the trace context into the payload:

var task asynq.Task
tracer := otel.GetTracerProvider().Tracer("")
opts := append([]tracex.SpanStartOption{}, tracex.WithSpanKind(tracex.SpanKindClient))
// This is a demo. Write the trace information of the created span to the body and send it to the server-side. Adjust the code as needed.
ctx, span := tracer.Start(context.Background(), "Push Task", opts...)
var headerMap propagation.MapCarrier
headerMap = make(map[string]string)
otel.GetTextMapPropagator().Inject(ctx, headerMap)
// Set the span context in the header.
for k, v := range headerMap {
  task.Header[k] = v
}
defer span.End()
//... push task to server

When the task is retrieved, extract and restore the trace context:

var headerMap propagation.MapCarrier
headerMap = make(map[string]string)
ctxRequest := context.Background()
// Get task header.
var task asynq.Task
for k, v := range task.Header {
  headerMap[k] = v
}
xxCtx := otel.GetTextMapPropagator().Extract(ctxRequest, headerMap)
tracer := otel.GetTracerProvider().Tracer("")
opts := append([]trace.SpanStartOption{}, trace.WithSpanKind(trace.SpanKindServer))
_, span := tracer.Start(xxCtx, "Recv Task", opts...)
defer span
//... other

This ensures the span context propagates between the task producer and consumer.

After you compile the application, enable OpenTelemetry for the application probe and restart the application.

In the ARMS console, go to the Probe Settings page. In the Plugin Switch section, select the opentelemetry-plugin checkbox and click Save. Restart your application for the changes to take effect.