[OpenTelemetry] Observability of Async Processes with Custom Propagator
Tags: opentelemetry, go, observability, microservices
Published: 2022-12-22 23:57:21

It’s the 23rd day of [Makuake Advent Calendar 2022](https://adventar.org/calendars/8496) .

As we know OpenTelemetry is an observability framework to generate and export telemetry data. As more companies adopt microservices and SLI/SLOs, we need it to answer new, never-before-seen (and complex) questions.

In that context, observability of systems that communicate using asynchronous messaging is as important as  systems communicates using synchronous messagin like HTTP or gRPC.

How about a notification system as an example:

- How long does it take for the end-user to receive the notification after the notification event is fired?
- What’s the error rate access the notification lifecycle?

To answer those questions, we may need expensive and unique structures.

Instead, in this article, I’ll demonstrate an easy way to do that with OpenTelemetry.

## Example Case

All source code is here:

[https://github.com/ymtdzzz/batch-tracing-sample](https://github.com/ymtdzzz/batch-tracing-sample)

![example case](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/vaflfeydc9jxd5s9jie3.png)

Processing flow:

- Enqueue the notification content as a message into a queue (Rabbit MQ) in batch
- A worker (consumer) asynchronously receives the message and send a request to the notification server (/email or /push)
- In the notification server, it responses 200 or 500

It’s assumed that the instrumentation of each component has been completed, and the HTTP communication has also been instrumented by [net/http auto instrumentation library](https://github.com/open-telemetry/opentelemetry-go-contrib/tree/main/instrumentation/net/http/otelhttp).

### Problem

![problem](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/qnfiz1rooed6t2jkygpa.png)

In the current state, batch processing and subsequent processing (worker) cannot be traced.

![trace a](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/4zb840bt52j1fv3l1u94.png)

![trace b](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/c3lh8rrf11lsn6lla528.png)

Moreover, there doesn’t seem to be an instrumentation library for RabbitMQ in Golang.

[https://opentelemetry.io/registry/?s=rabbitmq&component=&language=](https://opentelemetry.io/registry/?s=rabbitmq&component=&language=)

### What Should We Do?

To propagate context, we can use OpenTelemetry `Propagator` for both sync and async types of messaging!

## Implement Custom Propagator for RabbitMQ

### What’s Propagator?

[https://opentelemetry.io/docs/reference/specification/context/api-propagators/](https://opentelemetry.io/docs/reference/specification/context/api-propagators/)

Propagator API is interface definitions for propagating contexts across process - how sender *Inject*s context into message and how receiver *Extract*s it from message. Propagator has Carrier which has a responsibility to actual injection and extraction from any type of messages.

Fortunately, RabbitMQ allows to put `key-value` format `Headers` in messages ([docs](https://www.rabbitmq.com/publishers.html#message-properties)), so we can use `TextMapPropagator`.

![propagation flow](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/yy6p88ucojjyfdace1a3.png)

### Propagator Implementation

Actually, since it is not Propagator but the Carrier that manipulates the `TextMap`, all we have to do is implementing the struct that satisfies the `TextMapCarrier` interface!

`TextMapCarrier` interface ([doc](https://pkg.go.dev/go.opentelemetry.io/otel@v1.11.2/propagation#TextMapCarrier)):

```go
type TextMapCarrier interface {

	// Get returns the value associated with the passed key.
	Get(key string) string

	// Set stores the key-value pair.
	Set(key string, value string)

	// Keys lists the keys stored in this carrier.
	Keys() []string
}
```

Carrier implementation for this interface ([source code](https://github.com/ymtdzzz/batch-tracing-sample/blob/main/notification-manager/internal/carrier.go)):

```go
type AMQPCarrier struct {
	headers amqp.Table
}

func (c *AMQPCarrier) Get(key string) string {
	return fmt.Sprintf("%s", c.headers[key])
}

func (c *AMQPCarrier) Set(key string, value string) {
	c.headers[key] = value
}

func (c *AMQPCarrier) Keys() []string {
	keys := make([]string, len(c.headers))
	for k := range c.headers {
		keys = append(keys, k)
	}
	return keys
}
```

`amqp.Table` is just `map[string]interface{}` . `Get()` implementation is a little rough but it’s enough for example… ;)

### Sender Side Implementation

At the sender side, we can inject context into header and just send the message ([source code](https://github.com/ymtdzzz/batch-tracing-sample/blob/main/notification-manager/cmd/batch/main.go#L124-L142)).

```go
	// Create an empty amqp.Tables
	headers := amqp.NewConnectionProperties()	
	// Assign it to custom Carrier
	carrier := internal.NewAMQPCarrier(headers)
	// Inject the context
	otel.GetTextMapPropagator().Inject(ctx, carrier)
	err = ch.PublishWithContext(
		ctx,
		"",
		q.Name,
		false,
		false,
		amqp.Publishing{
			ContentType: "application/octet-stream",
			Body:        msg,
			Headers:     headers, // Assign the context injected headers
		},
	)
	if err != nil {
		panic(err)
	}
	log.Println("Message has been sent")
```

### Receiver Side Implementation

Receiver side is the same.

```go
		// Assign the received headers to custom Carrier
		carrier := internal.NewAMQPCarrier(d.Headers)
		// Extract the context
		ctx := otel.GetTextMapPropagator().Extract(context.Background(), carrier)
		// Generate child Span with received context as parent Span
		ctx, span := otel.Tracer("notification").Start(ctx, "consume")

		msg, err := internal.DecodeNotificationMessage(d.Body)
		if err != nil {
			panic(err)
		}
		log.Printf("received msg: %v\n", msg)

		internal.CallServer(ctx, &client, msg)

		span.End()
```

## We’re All Set 🎉

Now, let’s start the apps and check the Jaeger UI endpoint ([http://localhost:16686/](http://localhost:16686/)).

![final result a](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/a9ro4bu6cn4rf70la79m.png)

By connecting the traces, we can now investigate any errors throughout the notification lifecycle easily.

![final result b](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/4jz1ussjrxhfbmgvs26c.png)

Moreover, since the duration of the entire Trace is able to be measured, we can analyze bottlenecks of performance decreasing and notice slow notifications based on user experience.

I hope you learned something new from this post, please let me know if you have any feedback in the comments or [Twitter](https://twitter.com/ymtdzzz)!