[Experimental] Pubsub distributed tracing

[davidgm0]

What does this PR do?

Adds pubsub distributed tracing, which was asked in #2556. This is an early try, but I wanted to see if there was interest in this to continue and add tests, etc. The tests were a bit complicated to set up locally, so if someone else wants to help happy to get some extra commits. Otherwise I can take a deeper look, if the PR would be accepted with tests

• Adds distributed tracing for listen. There is a 'pull' method that isn't covered, since we don't use it.
• Adds the topic as resource to the publisher span
• Adds the subscription as resource to the consumer span
• There's a special case If both grpc and pubsub integrations are on.

Motivation

Pubsub is an important way of distributed communication so it would be really good to have it.

Additional Notes

This is how it looks:

The resources are the topic and the subscription respectively.

Publisher spans info:

{
component:
pubsub,
env: somenv,
language: ruby
pubsub: {
topic: 'projects/myproject/topics/mytopicname'
},
span: {
kind: 'producer'
},
version: 'git-<>'
}

Consumer span info:

{
component: 'pubsub',
env: myenv,
language: 'ruby',
process_id: 1
pubsub: {
subscription: 'projects/myproject/subscriptions/subscription-name'
topic: 'projects/myproject/topics/mytopicname'
},
runtime-id: '1a70e630-10ad-4eba-9de3-2f616ffb3b74'
span:{
kind: 'consumer'
}
version: 'git-<>'
}

How to test the change?

I've directly tested it with the gem 'google-cloud-pubsub'. It is possible to setup locally with an emulator, which you can use with docker with the image messagebird/gcloud-pubsub-emulator.

Code could be something like:

# consumer
        sub = Google::Cloud::PubSub.new.subscription @subscription_name
        return Rails.logger.error "Couldn't find subscription #{@subscription_name}" unless sub

        # 1 thread for the callback/processing and 1 thread for the acknowledgment of the message.
        @service = sub.listen threads: {callback: 1, push: 1} do |received_message|
          process_message(received_message)
        end

        @service.on_error do |exception|
          Rails.logger.error "Exception: #{exception.class} #{exception.message}"
        end

        @service.start

# publisher elsewhere
        topic = Google::Cloud::PubSub.new.topic(topic_name)
        topic.publish(event) # event is a string in json format

[maxnorth]

+1, would love to see this added!

[marcotc]

Thank you so much for the contribution, @davidgm0! 🙇

Our team will review it shortly.

[marcotc]

As a general question, does the gRPC handling of the Pubsub message happen before or after the pubsub.consumer.receive_message event?

I ask because the distributed tracing Datadog::Tracing::Distributed::Datadog#extract only needs to be done at the very first span of a trace. If the gRPC span is created before Pubsub, then gRPC should already take care of it.

The same goes for pubsub.producer.send_messages: looking at the screenshot you posted, gRPC happens after Pubsub, meaning gRPC already handles Datadog::Tracing::Distributed::Datadog#inject!, meaning Pubsub doesn't have to do it, if I'm describing the situation correctly.

[marcotc]

It's great that there's a Docker image that can simulate the backend required for testing.

For testing, we'll probably need it.

I suggest following how our most recent integration, OpenSearch, added a docker image for their own testing: https://github.com/DataDog/dd-trace-rb/pull/2940/files#diff-e45e45baeda1c1e73482975a664062aa56f20c03dd9d64a827aba57775bed0d3

You can make similar changes to add messagebird/gcloud-pubsub-emulator to the docker-compose.yml, and then you can run Ruby tests inside the Docker environment: docker-compose run --rm tracer-3.0 /bin/bash. See our guide for full information on the setup: https://github.com/DataDog/dd-trace-rb/blob/master/docs/DevelopmentGuide.md#setting-up.