Micrometer trace id propagation does not work when using StreamBridge

[kabennett]

I have been discussing this with @artembilan on stackoverflow, and he encouraged me to go ahead and open an issue here regarding the inability to propagate micrometer trace ids when using a StreamBridge to produce a message. I've been trying to get trace id propagation working across the Kinesis messaging boundary between a producer and a consumer microservice. According to Artem, this is not possible as "SteamBridge does not expose its internal DirectWithAttributesChannel instances as beans. So, we cannot apply observation instrumentation from the framework perspective."

You can read more about the history of what we've discussed so far here, but as there is currently no way to get a trace id propagated from producer to consumer when using a StreamBridge, I want to open this issue so that a common solution could be built into the framework so that this is possible. I see this as a critical missing capability, as there is no way to correspond logged trace ids across the Spring Cloud Stream messaging boundary in this particular scenario (i.e., when using a StreamBridge to originate a message in a producer).

[patpatpat123]

Encountering this issue as well

[magnus-larsson]

This seems to work fine, at least in my code base, with Spring Boot 3.0.3 + usage of Hooks.enableAutomaticContextPropagation(); as described here: spring-projects/spring-boot#33372 (comment)

Sample trace tree visualized using Zipkin. The three send operations are done using the StreamBridge:

[olegz]

So, StreamBridge (SB) is the beginning of stream. In other words it is where things would start. It is not a mechanism to inherit or continue other flows.
On top of that, channels used by SB as channels used by the framework are internal implementation detail which in theory could change in the future (entirely or part of it), so the fact that we are even mentioning them here in the context of tracing is wrong.
Message on the other hand would be more appropriate abstraction as it is exposed to the end user.

That said, @magnus-larsson seems to show that he has a way of making it work. Can you please provide more details/samples etc?

[magnus-larsson]

Hello @olegz, and thanks for helping out and clarifying!

Yes, the StreamBridge is the beginning of a stream, but it can be part of a larger processing scope you want to trace with a TraceId.

In my example, the processing flow is as follows:

1. A Gateway service receives an external HTTP API request
2. The Gateway service calls an HTTP API exposed by a Product Composite service
3. The Product Composite service sends three messages using the StreamBridge to three other services
4. The three other services process the messages.

I want this processing to be described as one trace tree, i.e. using one and the same Trace Id.
Therefore, I think it is important that the StreamBridge can pick up the current Trace Id and use it.

Do you agree?

Note: This worked out of the box with Spring Cloud Sleuth using Spring Boot v2, and from what I can see in my tests, it now also works with Micrometer Tracing as described above.

[kabennett]

According to @artembilan here, this feature has been implemented and will be available in a future release.

[artembilan]

So, Kinesis Binder does now this (after spring.cloud.stream.kinesis.binder.enable-observation=true):

protected void postProcessOutputChannel(MessageChannel outputChannel,
			ExtendedProducerProperties<KinesisProducerProperties> producerProperties) {

		((IntegrationManagement) outputChannel).registerObservationRegistry(this.observationRegistry);

exactly on the channel supplied to the binding via that StreamBridge.
That MessageChannel observation is a PRODUCER kind from a trace perspective.
And if there is a current Observation in the scope, this new one on a channel is going to be included in the current trace.

The Kafka binder does something similar with the property, but it instrument a little lower around KafkaTemplate.send() with similar PRODUCER kind.
Rabbit binder at the moment does that unconditionally, however in the low level similar to Kafka binder - instruments a RabbitTemplate PRODUCER kind.

So, sounds like we have everything in hands for this issue.
Not sure how else we can improve this stuff: I really doubt that we are going to expose a dedicated observation for StreamBridge.

Can we treat this as completed?

[braunsonm]

@artembilan The Kinesis binder is not the only binder, We see this issue with the rabbit binder as well

[artembilan]

What issue do you see?
I've just explained that Kafka and Rabbit binders have an observation instrumentation.
For example Rabbit binder does this in its ObservationAutoConfiguration:

amqpOutboundEndpoint.getRabbitTemplate().setObservationEnabled(true);

When we send a message via StreamBridge it ends up in an AmqpOutboundEndpoint for the target destination and that is where that RabbitTemplate is involved.
So, any chances to have a simple project from you which claim something opposite than we see in the code?

[artembilan]

So, here is an absolutely simple Spring Cloud Stream application with Rabbit binder:

1. POM:

	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-actuator</artifactId>
		</dependency>
		<dependency>
			<groupId>io.micrometer</groupId>
			<artifactId>micrometer-tracing-bridge-brave</artifactId>
		</dependency>
		<dependency>
			<groupId>io.zipkin.reporter2</groupId>
			<artifactId>zipkin-reporter-brave</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.cloud</groupId>
			<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
		</dependency>
	</dependencies>

2. application.properties:

spring.application.name=my-application
management.tracing.sampling.probability=1
logging.pattern.level=%5p [${spring.application.name:},%X{traceId:-},%X{spanId:-}]
logging.level.com.example.stream.rabbit.binder.observation=debug
spring.cloud.stream.bindings.myRabbitConsumer-in-0.destination=myQueue

3. Code:

@SpringBootApplication
public class ObservationOverStreamRabbitBinderApplication {

	private static final Log LOGGER = LogFactory.getLog(ObservationOverStreamRabbitBinderApplication.class);

	public static void main(String[] args) {
		SpringApplication.run(ObservationOverStreamRabbitBinderApplication.class, args);
	}

	@Bean
	ApplicationRunner myApplicationRunner(StreamBridge streamBridge, ObservationRegistry observationRegistry) {
		return args ->
				Observation.createNotStarted("my parent observation", observationRegistry)
						.observe(() -> {
							String data = "my data";
							LOGGER.debug("Send data to RabbitMQ: " + data);
							streamBridge.send("myQueue", data);
						});
	}

	@Bean
	public Consumer<Message<String>> myRabbitConsumer() {
		return message -> LOGGER.debug("Received message from RabbitMQ: " + message);
	}

}

4. Log output:

2023-03-02T14:34:59.772-05:00 DEBUG [my-application,6400fa637cc78254fd1072c68f9f7ef1,fd1072c68f9f7ef1] 11344 --- [           main] rvationOverStreamRabbitBinderApplication : Send data to RabbitMQ: my data
2023-03-02T14:34:59.825-05:00  INFO [my-application,6400fa637cc78254fd1072c68f9f7ef1,fd1072c68f9f7ef1] 11344 --- [           main] o.s.c.s.m.DirectWithAttributesChannel    : Channel 'unknown.channel.name' has 1 subscriber(s).
2023-03-02T14:34:59.846-05:00  INFO [my-application,6400fa637cc78254fd1072c68f9f7ef1,fd1072c68f9f7ef1] 11344 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [localhost:5672]
2023-03-02T14:34:59.852-05:00  INFO [my-application,6400fa637cc78254fd1072c68f9f7ef1,fd1072c68f9f7ef1] 11344 --- [           main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory.publisher#2dde0a34:0/SimpleConnection@1b632442 [delegate=amqp://guest@127.0.0.1:5672/, localPort=60879]
2023-03-02T14:34:59.856-05:00  INFO [my-application,6400fa637cc78254fd1072c68f9f7ef1,fd1072c68f9f7ef1] 11344 --- [           main] o.s.amqp.rabbit.core.RabbitAdmin         : Auto-declaring a non-durable, auto-delete, or exclusive Queue (myQueue.anonymous.fqLkdQ57TGuJVLOQvfdmkg) durable:false, auto-delete:true, exclusive:true. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.
2023-03-02T14:34:59.895-05:00 DEBUG [my-application,6400fa637cc78254fd1072c68f9f7ef1,c41af37123a77963] 11344 --- [GuJVLOQvfdmkg-1] rvationOverStreamRabbitBinderApplication : Received message from RabbitMQ: GenericMessage [payload=my data, headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedExchange=myQueue, amqp_deliveryTag=1, deliveryAttempt=1, amqp_consumerQueue=myQueue.anonymous.fqLkdQ57TGuJVLOQvfdmkg, amqp_redelivered=false, amqp_receivedRoutingKey=myQueue, amqp_timestamp=Thu Mar 02 14:34:59 EST 2023, traceparent=00-6400fa637cc78254fd1072c68f9f7ef1-80011313d0d02f3f-01, source-type=amqp, amqp_messageId=2ca57851-e926-5113-62cf-0e24c1028b01, id=9b015e88-5d6c-1a23-ba80-773f71db9634, amqp_consumerTag=amq.ctag-T5J6E6RUh0WJstv8ZCC-UQ, sourceData=(Body:'my data' MessageProperties [headers={traceparent=00-6400fa637cc78254fd1072c68f9f7ef1-80011313d0d02f3f-01, target-protocol=amqp}, timestamp=Thu Mar 02 14:34:59 EST 2023, messageId=2ca57851-e926-5113-62cf-0e24c1028b01, contentType=application/json, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=myQueue, receivedRoutingKey=myQueue, deliveryTag=1, consumerTag=amq.ctag-T5J6E6RUh0WJstv8ZCC-UQ, consumerQueue=myQueue.anonymous.fqLkdQ57TGuJVLOQvfdmkg]), contentType=application/json, timestamp=1677785699886}]

5. And Zipkin screenshot:
   

Feels like everything works as advertised.

@braunsonm,
what do I miss in my investigation to understand that we have an issue with StreamBridge and Rabbit Binder?

@magnus-larsson ,
according to my sample the parent trace is really propagated down to StreamBridge and then further over RabbitMQ broker until consumer on the other side.
Probably your application with that reactive API should benefit from the context-propagation library and respective Reactor hooks: https://projectreactor.io/docs/core/release/reference/#context.propagation.
So, then you might would need to use that StreamBridge in the handle() operator which is able to propagate Reactor context to ThreadLocal automatically.

I'm closing this as completed and confirmed as working.
Anything else, please, start discussing on StackOverflow.

[Toxa45]

@artembilan Good afternoon. Here is an example, based on what you attached above, and the traceId are different
https://github.com/Toxa45/TestRabbitTraceId/tree/master/TestRabbitTraceId

I also noticed that this happens when the "environment" is filled. Initially, I made a typo and wrote "enviroment" - and to my surprise, everything worked locally.