aws / aws/aws-xray-sdk-node

Creating Linked Traces between SQS Producer and Consumer on Non Serverless Compute

Open
#637 2 comments 1 reaction 0 assignees View on GitHub
Dominant language
JavaScript
Stars
280
Forks
157
PR merge metrics
No merged PRs in 30d

Description

I've been trying to get something similar to what is discussed in this [Tracing event-driven application](https://docs.aws.amazon.com/xray/latest/devguide/xray-tracelinking.html#xray-tracelinking-servicemap) documentation where the traces created upstream from the producer are able to be linked to traces from downstream consumers.

The documentation highlight the capability for Lambda and SQS but I was wondering if the same thing can be achieved outside of Lambda (EC2, containers, etc).

From my testing and following what thought to be related GitHub Issues (https://github.com/aws/aws-xray-sdk-node/issues/208, https://github.com/aws/aws-xray-sdk-node/issues/419) I feel like I'm close, but I'm not seeing the same "linking" behavior or the `This trace is poart of a linked set of traces` messages on the traces as shown in the documentation.

![image](https://github.com/aws/aws-xray-sdk-node/assets/18585217/0aaa4b18-9b67-4d02-8fad-d2c0b1a6ef7e)

I have a simple ExpressJS web app that is creating the SQS messages with XRay Tracing header as an attribute.
```javascript
const AWSXRay = require("aws-xray-sdk");
const XRayExpress = AWSXRay.express;
const express = require("express");
const { SQSClient, SendMessageCommand } = require("@aws-sdk/client-sqs");

const app = express();
const port = 3000;

app.use(XRayExpress.openSegment("simple-api"));

app.use((err, req, res, next) => {
console.error(err.stack);
res.status(500).send("Something broke!");
});

app.get("/sqs", async (req, res) => {
const sqs = AWSXRay.captureAWSv3Client(
new SQSClient({ region: "us-east-1" })
);
const params = {
QueueUrl: "",
MessageBody: JSON.stringify({ message: "body here" }),
};
await sqs.send(new SendMessageCommand(params));
res.send("ok");
});

app.use(XRayExpress.closeSegment());

app.listen(port, () => console.log(`Example app listening on port ${port}!`));
```

And a simple consumer that is pulling the messages, creating a new `segment` from the values in XRayTracing header, creating an example `subsegment` to simulate processing logic, then closing the segment.

```javascript
const AWSXRay = require("aws-xray-sdk-core");
const {
SQSClient,
ReceiveMessageCommand,
DeleteMessageCommand,
} = require("@aws-sdk/client-sqs");

// Initialize SQS client
const sqsClient = new SQSClient({ region: "us-east-1" });

async function processMessages(queueUrl) {
const receiveParams = {
QueueUrl: queueUrl,
AttributeNames: ["All"],
WaitTimeSeconds: 20, // Enable long polling
};

while (true) {
const received = await sqsClient.send(
new ReceiveMessageCommand(receiveParams)
);

if (received.Messages) {
for (const message of received.Messages) {
const traceHeaderStr = message.Attributes.AWSTraceHeader;

// Check if the traceHeaderStr is available and valid
if (traceHeaderStr) {
const traceData = AWSXRay.utils.processTraceData(traceHeaderStr);

// Inside this context, we can now work with X-Ray segments
const segment = new AWSXRay.Segment(
"SQSMessageProcessing",
traceData.root, // Root ID from the trace header
traceData.parent // Parent ID from the trace header
);

try {
const subsegment = segment.addNewSubsegment("processingFunction");
setTimeout(() => {
subsegment.close();
}, 500);
const deleteParams = {
QueueUrl: queueUrl,
ReceiptHandle: message.ReceiptHandle,
};
await sqsClient.send(new DeleteMessageCommand(deleteParams));
} catch (error) {
console.error("Error processing message:", error);
segment.addError(error); // Add error to segment
} finally {
// Close the segment after processing
segment.close();
}
}
}
}
}
}

// Replace 'YOUR_SQS_QUEUE_URL' with your actual SQS queue URL
processMessages(
""
).catch(console.error);
```
From the screenshots below, you'll see that a single trace is created containing sub/segments create by my "API" and the sub/segments create by my "consumer". I was expecting to seeing something like what is highlighted in the [Tracing event-driven application](https://docs.aws.amazon.com/xray/latest/devguide/xray-tracelinking.html#xray-tracelinking-servicemap) documentation where multiple traces are created and then "linked" together.

![image](https://github.com/aws/aws-xray-sdk-node/assets/18585217/2fc8a0bd-307f-49e6-b343-4797b5dadfd0)
![image](https://github.com/aws/aws-xray-sdk-node/assets/18585217/4d7edf5a-68a4-485a-bc85-10ab755290ce)

Another issue I'm having with this implementation is that XRay capture functions can't seem to find the current context.

If I add a little bit more logic to my consumer code to try and simulate capturing outgoing HTTPS calls from the consumer
```javascript
const AWSXRay = require("aws-xray-sdk-core");
const {
SQSClient,
ReceiveMessageCommand,
DeleteMessageCommand,
} = require("@aws-sdk/client-sqs");
const https = require("https");

// Initialize SQS client
const sqsClient = new SQSClient({ region: "us-east-1" });

async function processMessages(queueUrl) {
const receiveParams = {
QueueUrl: queueUrl,
AttributeNames: ["All"],
WaitTimeSeconds: 20, // Enable long polling
};

while (true) {
const received = await sqsClient.send(
new ReceiveMessageCommand(receiveParams)
);

if (received.Messages) {
console.log("Received messages:", received.Messages);
for (const message of received.Messages) {
const traceHeaderStr = message.Attributes.AWSTraceHeader;

// Check if the traceHeaderStr is available and valid
if (traceHeaderStr) {
const traceData = AWSXRay.utils.processTraceData(traceHeaderStr);

// Inside this context, we can now work with X-Ray segments
const segment = new AWSXRay.Segment(
"SQSMessageProcessing",
traceData.root, // Root ID from the trace header
traceData.parent // Parent ID from the trace header
);

AWSXRay.captureHTTPsGlobal(https);

try {
const subsegment = segment.addNewSubsegment("processingFunction");
setTimeout(() => {
https.get("https://amazon.com/", (response) => {
response.on("data", () => {});

response.on("error", (err) => {
console.error(err);
subsegment.close();
});

response.on("end", () => {
subsegment.close();
});
});
}, 500);
const deleteParams = {
QueueUrl: queueUrl,
ReceiptHandle: message.ReceiptHandle,
};
await sqsClient.send(new DeleteMessageCommand(deleteParams));
} catch (error) {
console.error("Error processing message:", error);
segment.addError(error); // Add error to segment
} finally {
// Close the segment after processing
segment.close();
}
}
}
}
}
}

// Replace 'YOUR_SQS_QUEUE_URL' with your actual SQS queue URL
processMessages(
"https://sqs.us-east-1.amazonaws.com/785630775706/xray-test"
).catch(console.error);
```

I get the classic `[ERROR] Error: Failed to get the current sub/segment from the context.` error.

Thank you for any support and if there is better documentation somewhere on how to implement thorough tracing for applications that use SQS on "non-serverless" compute (containers, EC2, etc), please let me know.

Contributor guide

Open the contributing guide

Research direction

Reproduce the Express producer and SQS consumer flow, starting with processTraceData, Segment construction, and captureHTTPsGlobal in the consumer example. Compare the observed trace behavior and current-context error with the linked AWS X-Ray tracing documentation and related issues. Done means the supported non-serverless linkage behavior and the cause of the context error are clearly established and documented.

Written by the indexing model from the issue text.

Assessment

Tech stack
aws, express, javascript, node.js
Domain
backend, distributed-systems, observability
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.