opentracing-contrib / java-rabbitmq-client

OpenTracing Instrumentation for RabbitMQ Client

OpenTracing RabbitMQ Client Instrumentation

OpenTracing instrumentation for RabbitMQ Client.

Installation

pom.xml

<dependency>
    <groupId>io.opentracing.contrib</groupId>
    <artifactId>opentracing-rabbitmq-client</artifactId>
    <version>VERSION</version>
</dependency>

Usage

// Instantiate tracer
Tracer tracer = ...

// Optionally register tracer with GlobalTracer
GlobalTracer.register(tracer);

// Decorate RabbitMQ Channel with TracingChannel
TracingChannel tracingChannel = new TracingChannel(channel, tracer);

// Send
tracingChannel.basicPublish(...);

// Get
GetResponse response = tracingChannel.basicGet(queueName, false);

// Consume
tracingChannel.basicConsume(...);

// Factory
ConnectionFactory factory = new TracingConnectionFactory(tracer);
Connection connection = factory.newConnection();

License

Apache 2.0 License.

java-rabbitmq-client's Issues

TracingUtils.buildChildSpan creates a noop span, when trace-id is not in headers

Hi,

I have been using your library for some time already - it behaves really nice and saved me a lot of coding - thanks! I do have one question / issue with it though. It relates to handleDelivery in TracingConsumer.

Context:

My application is a set of interconnected microservices and an initial call comes typically from integrations tests in python, or another java client.
Point is, in both of these situations, the client doesn't create a span himself. This is for a reason, we don't want to complicate things for them - they just want to execute a request and that's fine.
Instead, first span (trace) is created on the side of application, precisely in the first service that is hit with a request. And this is where the issue happens.

Issue:

In handleDelivery, there is a:
Span child = TracingUtils.buildChildSpan(properties, queue, tracer);
This tries to locate a trace in headers, but since it is a first service in chain of calls, it won't be there.
So, a noop span will be created instead of a normal one, representing 'receive' operation.
And since it is a noop span, it won't be a part of the whole trace, because the next span I'm creating in my code, can't reference (be a child) of a noop span.

Possible solutions:

Currently can't think of anything other than creating a span on the client side, but really don't wanna go this way. Alternative is, to modify TracingUtils.buildChildSpan and there, on some conditions create a regular span, when trace-id is not in headers. Don't know though if this is the right way to go.

Do you have any suggestions?
It's not a severe issue, but I reckon it's nice to have a full picture of how long the call is served.

Thanks!
Konrad

NoSuchMethodError Exception 0.1.5

after updating version from 0.1.4 to 0.1.5 we staring to receive exceptions:

[ERROR][][reactor-http-nio-3]r.n.h.s.HttpServer : [id: 0x2851fafe, L:/172.177.0.3:8888 - R:/10.1.1.30:54308] java.lang.NoSuchMethodError: io.opentracing.ScopeManager.activate(Lio/opentracing/Span;Z)Lio/opentracing/Scope;
at io.opentracing.contrib.spring.web.webfilter.TracingOperator.subscribe(TracingOperator.java:73)
at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
at reactor.core.publisher.MonoOnAssembly.subscribe(MonoOnAssembly.java:61)

NullPointerException in TracingConsumer

Receiving NullPointer exception in consumer, when headers are null. This may happen when message is being sent from Producers who are not using headers or opentracing rmq client.

================================
2018-05-02 14:40:58 ERROR ForgivingExceptionHandler:119 - Consumer io.opentracing.contrib.rabbitmq.TracingConsumer@74c62dfb (amq.ctag-7GgIK06zrLFkQLDACUWgRg) method handleDelivery for channel AMQChannel(amqp://[email protected]:5672/,1) threw an exception for channel AMQChannel(amqp://[email protected]:5672/,1)
java.lang.NullPointerException
at io.opentracing.contrib.rabbitmq.HeadersMapExtractAdapter.(HeadersMapExtractAdapter.java:27)
at io.opentracing.contrib.rabbitmq.TracingUtils.extract(TracingUtils.java:29)
at io.opentracing.contrib.rabbitmq.TracingUtils.buildChildSpan(TracingUtils.java:49)
at io.opentracing.contrib.rabbitmq.TracingConsumer.handleDelivery(TracingConsumer.java:65)
at com.rabbitmq.client.impl.ConsumerDispatcher$5.run(ConsumerDispatcher.java:149)
at com.rabbitmq.client.impl.ConsumerWorkService$WorkPoolRunnable.run(ConsumerWorkService.java:104)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

Tracing AMPQ workflow

Assume that we have three microservices, say A, B and C. The microservice A declares a queue, say QUEUE_1 and performs a basicPublish(...). B calls basicConsume(...) to retrieve the messages from QUEUE_1 and then uses another queue: QUEUE_2 to publish a message to C. Finally, C consumes the message from QUEUE_2.

How can we perform distributed tracing for this case?

I read the documentation given on the following link:

https://github.com/opentracing-contrib/java-rabbitmq-client/blob/master/README.md

and came up with some solutions which doesn't work as intended. What I basically need is that just like the commonly used distributed tracing in microservice communications, is it possible to get a trace (traceID) associated with a certain send and receive in RabbitMQ for a certain workflow A -> B -> C?

Here are the codes for reference:

For B:

@Service
public class B {
	public static String RECV_QUEUE_NAME = "QUEUE_1";
	public static String SEND_QUEUE_NAME = "QUEUE_2";
	
	
	@Autowired
	private JaegerTracer tracer;
	
	public boolean sendMessage(String message) throws Exception {
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		
		boolean status = true;
		
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		
		TracingChannel tracingChannel = new TracingChannel(channel, tracer);
		
		try  {
			tracingChannel.queueDeclare(SEND_QUEUE_NAME, true, false, true, null);
			tracingChannel.basicPublish("", SEND_QUEUE_NAME, null, message.getBytes("UTF-8"));
			
		} catch (Exception e) {
			status = false;
		}
		return status;
	}
	
	public void consumeAll(List<String> res) throws Exception {
		
		Span span = tracer.buildSpan("B-ConsumeAll").start();
		
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		
		TracingChannel tracingChannel = new TracingChannel(channel, tracer);
	
		tracingChannel.queueDeclare(RECV_QUEUE_NAME, true, false, true, null);
		
		
		DeliverCallback deliverCallback = (consumerTag, delivery) -> {
			
	        String message = new String(delivery.getBody(), "UTF-8");
	        res.add(message);
	        
	        try {
			sendMessage( ("SENDING AGAIN : " + message));
		} catch (Exception e) {
			e.printStackTrace();
		}
	        
	        System.out.println(" [x] B Received '" + message + "'");
	    };
	    
	    tracingChannel.basicConsume(RECV_QUEUE_NAME, true, deliverCallback, consumerTag -> { });
	    span.finish();
	}
}

For C:

@Service
public class RabbitService {
	public static String RECV_QUEUE_NAME = "QUEUE_2";	
	
	@Autowired
	private JaegerTracer tracer;
	
	public void consumeAll(List<String> res) throws Exception {

		Span span = tracer.buildSpan("C-ConsumeAll").start();
		
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();

		TracingChannel tracingChannel = new TracingChannel(channel, tracer);
	
		tracingChannel.queueDeclare(RECV_QUEUE_NAME, true, false, true, null);
		
		DeliverCallback deliverCallback = (consumerTag, delivery) -> {
			
	        String message = new String(delivery.getBody(), "UTF-8");
	        res.add(message);
	        
	        System.out.println(" [x] C Received '" + message + "'");

	    };
	    
	    tracingChannel.basicConsume(RECV_QUEUE_NAME, true, deliverCallback, consumerTag -> { });
	    span.finish();
	}
}

Baggage Item not propagated through the next SpanContext

I tried adding an item in Baggage:

tracer.activeSpan().setBaggageItem("customId" , "12345")

Unfortunately it was not propagated to next spanContext in the same trace. And in the same way it should be propagated through the complete chain of spans.

I want to know whether this feature is supported at present? And if not, are there any future plans?