feat: bunny instrumentation

[johanneswuerbach]

Basic instrumentation for bunny based on the existing ruby_kafka instrumentation.

The instrumentation provides:

• context propagation through rabbitmq stored in the message header
• span generation for publishing, manual consumption (pop) and consumption using a consumer (subscibe)

This basic instrumentation should also be sufficient when using higher level libraries based on bunny like https://github.com/jondot/sneakers

[johanneswuerbach]

Thanks for the feedback, I pushed a round of fixes which should address all your comments.

[johanneswuerbach]

@fbogsany rebased and adopted to latest main changes. Let me know if you have any other comments or concerns.

[mwear]

LGTM. Did you want to take a final look @fbogsany or @robertlaurin?

[robertlaurin]

Looks good, just a couple questions.

[robertlaurin]

Similar to the above, it looks like a uncaught_exception_handler could be provided that raises the exception, in which case we would not capture that event here either.

Not being familiar with rabbitmq, is this call to super intentionally not being wrapped with the trace block for any particular reason?

[johanneswuerbach]

super only enqueues the message into the internal worker pool in this case and the actual consumer call is wrapped into a process span, which should handle the exception in this case.

[fbogsany]

It looks to me like this method wraps consumers, so we can wrap a process span around it:

Suggested change

	# This method is called when rabbitmq pushes messages to subscribed consumers
	def handle_frameset(basic_deliver, properties, content)
	OpenTelemetry::Instrumentation::Bunny::PatchHelpers.with_receive_span(self, tracer, basic_deliver, properties) do
	properties[:headers] ||= {}
	OpenTelemetry.propagation.inject(properties[:headers])
	end
	super
	end
	# This method is called when rabbitmq pushes messages to subscribed consumers
	def handle_frameset(basic_deliver, properties, content)
	OpenTelemetry::Instrumentation::Bunny::PatchHelpers.with_process_span(self, tracer, basic_deliver, properties) do
	super
	end
	end

[fbogsany]

Hmm, this is interesting because you're also wrapping the Consumer.call. Maybe we should just not wrap handle_frameset? We don't have to have a receive span in every case - the process span is much more interesting.

[johanneswuerbach]

The handle_frameset submits the message to a worker pool, so there is a time between receiving the message and eventually processing it in a consumer thread.

Should that time be part of process or receive or not be tracked at all?

[fbogsany]

Is Queue typically used directly by clients or is it only used internally? It feels like we're going to end up with nested process spans in some cases, but I may be misunderstanding the gem.

[johanneswuerbach]

.pop can be used by clients http://rubybunny.info/articles/queues.html#fetching_messages_when_needed_pull_api

[johanneswuerbach]

Thank you for the all the feedback, I made a couple of changes to be more spec compliant around receive vs. process.

Now the push & pull api are modelled as https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/messaging.md#batch-receiving.

basic_get, used by Queue.pop starts a receive span when the receiving is started and in queue pop we link that span to the process span if a block is defined.

handle_frameset is doing the same before submitting the message into internal consumer pool, where it will eventually be processed by a consumer. Technically the receive should already wrap https://github.com/ruby-amqp/bunny/blob/0b5a6f2778f64f3c4bdf8b3063c6bdcbecfc8123/lib/bunny/reader_loop.rb#L77, but I wasn't sure how to integrate there so receive now only starts once the message was read from the socket. Resolved.

[johanneswuerbach]

@fbogsany could I get another review? :-)

[fbogsany]

🤦 sorry! 👀 now.

[johanneswuerbach]

@fbogsany friendly ping :-)

[fbogsany]

🤦 👀

[fbogsany]

Thanks @johanneswuerbach - sorry this took so long to review and merge 😞

[johanneswuerbach]

No worries, we are all busy people. Thank you for working on this 🙇

[indrekj]

@johanneswuerbach shouldn't this be .extract(properties[:headers] || {})?

If I have two services A & B. The first service is instrumented but the second one is not, then I get this exception in service A:

Uncaught exception from consumer #<Bunny::Consumer:70980 @channel_id=2 @queue=amq.gen-p2ZzQp6h4r6QWdCBIC4Pgg> @consumer_tag=bunny-1629126993000-738472400850>: #<NoMethodError: undefined method `[]' for nil:NilClass> @ /home/indrek/.rvm/gems/ruby-2.7.2@admin/gems/opentelemetry-api-1.0.0.rc2/lib/opentelemetry/context/propagation/text_map_getter.rb:16:in `get'

I think that's because properties[:headers] is nil when receiving a message from a service that is not using OTEL.

EDIT: I pinged the wrong person initially.

[johanneswuerbach]

Yes, that seems possible. Could you create a separate issue for this?