Message Queue Trace Propagation Support

Primary Change

Add message queue propagation

• added trace propagation and trace context extraction support for NATS
• added trace propagation and trace context extraction support for Kafka
• added trace propagation and trace context extraction support for RabbitMQ

Example

Add a trace context into a NATS header

package main

import (
	"context"
	"fmt"

	"github.com/nats-io/nats.go"
	"github.com/wendall-robinson/flowmaster/traceflow"
)

func main() {
    ctx := context.Background()

    // Create a new span for the nats publish event itself (not the propagation)
    span := traceflow.New(ctx, "nats").Producer()

    header := nats.Header{}

    // will inject the NATS header into the current trace context
    traceflow.PropagateNats(trace.GetContext(), nats.Header{})

    // Publish the Request
    < NATS IMPLEMENTATION >

}

Extract trace context from a NATS header

package main

import (
	"context"
	"fmt"

	"github.com/nats-io/nats.go"
	"github.com/wendall-robinson/flowmaster/traceflow"
)

func main() {
   // incoming NATS message
   < NATS INBOUND MESSAGE HANDLING IMPLEMENTATION >

    // Extract the trace context from the incoming message and forwarding it
    newCtx := traceflow.ExtractNats(ctx, nats.Header)

    // Create a new span for the nats subscribe event
    span := traceflow.New(newCtx, "nats").Consumer()

   defer span.Start("nats-event").End()
}