Is there support for Custom DSM Streams?

[darylrobbins]

Is there currently a way to mark a custom span as belonging to a DSM Stream (or is there any plan to support in the future)? We are using a custom messaging stack that I'd love to use with DSM. Thanks!

[darylrobbins]

I see datadog.trace.api.experimental.DataStreamsCheckpointer, which seems to be an experimental interface for this use case. Is there any specific data that's expected from the DataStreamsContextCarrier? How is it used?

[kr-igor]

Hi Daryl,

Could you provide a bit more details on what you're trying to achieve? DSM doesn't rely on Spans, however they are linked with "pathway.hash" span tag.

datadog.trace.api.experimental.DataStreamsCheckpointer is designed for cases when you want to support Data Streams for queuing technologies which are not currently automatically instrumented (for instance: Kinesis, SNS).
DSM passes context (~24 bytes) along with the message, typically in message headers. DataStreamsContextCarrier is a wrapper allowing us to read / write date to these headers. Please, see the example below.

import datadog.trace.api.experimental.*;

----

Carrier headersAdapter = new Carrier(kinesisHeaders);

# before calling Kinesis PUT
DataStreamsCheckpointer.get().setProduceCheckpoint("kinesis", "my-stream", headersAdapter);

# after calling Kinesis GET
DataStreamsCheckpointer.get().setConsumeCheckpoint("kinesis", "my-stream", headersAdapter);


------


# replace KinesisHeaders with whatever you're using to pass the context
private class Carrier implements DataStreamsContextCarrier {
	private KinesisHeaders headers;
	
	
	public Carrier(KinesisHeaders headers) {
		this.headers = headers;
	}

	public Set<Entry<String, Object>> entries() {
		return this.headers.entrySet();
	}

	public void set(String key, String value){
		this.headers.put(key, value);
	}
}

[darylrobbins]

Thanks, this is helpful. I am adding support for a proprietary message broker we are using that is not currently instrumented. I had it showing the production side but not the consumption. I see from your example that my problem was I wasn't allowing the DSM context to be set as a message header. Let me update my test and try again.

[darylrobbins]

I now have my producer and consumer connected up in the DSM view but the only metrics I see is throughput. Is there any way to report lag metrics and end-to-end time? I tried adding a record.queue_time_ms tag with the queue time like I saw for the standard instrumentations but it doesn't seem to have any effect.