feat: lambda support for DSM #672
Conversation
| } | ||
|
|
||
| try { | ||
| this.tracer.dataStreamsCheckpointer.setConsumeCheckpoint(eventType, arn, contextJson, false); |
There was a problem hiding this comment.
Need to wait for a tracer release that has this manual_checkpoint parameter
| const kinesisData = event?.Records?.[0]?.kinesis.data; | ||
| if (kinesisData === undefined) return null; | ||
|
|
||
| sourceARN = event.Records[0].eventSourceARN; |
There was a problem hiding this comment.
what happens if event is null or Records is null or it has no records? i would use optional chaining like above.
There was a problem hiding this comment.
We are using records[0] (so the first message). Can a lambda be called with many messages? What do we want to do in that case? Set many checkpoints or just one for the first checkpoint ? What do we do in the Python lambda?
| try { | ||
| const decodedData = Buffer.from(kinesisData, "base64").toString("ascii"); | ||
| const parsedBody = JSON.parse(decodedData); | ||
| const headers = parsedBody?._datadog; | ||
| if (headers) { | ||
| const traceContext = this.tracerWrapper.extract(headers); | ||
| this.tracerWrapper.setConsumeCheckpoint(headers, "kinesis", sourceARN); |
There was a problem hiding this comment.
if we fail to capture the sourceARN, we probably want a debug log & not set a checkpoint
| try { | ||
| // First try to extract trace context from message attributes | ||
| if (event?.Records?.[0]?.body) { | ||
| sourceARN = event.Records[0].eventSourceARN; |
src/trace/context/extractors/sns.ts
Outdated
| try { | ||
| // First try to extract trace context from message attributes | ||
| const messageAttribute = event?.Records?.[0]?.Sns?.MessageAttributes?._datadog; | ||
| sourceARN = event.Records[0].Sns.TopicArn; |
| return; | ||
| } | ||
|
|
||
| if (getEnvValue("DD_DATA_STREAMS_ENABLED", "false").toLowerCase() !== "true") { |
There was a problem hiding this comment.
is that the way javascript tests for env variables? In other languages, we sometimes have a helper function. (so you can set: DD_DATA_STREAMS_ENABLED=1 for example
There was a problem hiding this comment.
Is this behavior we want?
VARIABLE = result === "true" || result === "1"; is a pattern used, but not sure we want this.
| @@ -98,4 +99,22 @@ export class TracerWrapper { | |||
| this.tracer.inject(span, "text_map", dest); | |||
| return dest; | |||
| } | |||
|
|
|||
| public setConsumeCheckpoint(contextJson: any, eventType: string, arn: string): void { | |||
| if (!arn) { | |||
There was a problem hiding this comment.
I see we don't set a checkpoint if arn is not set. Let's just add a debug log in this case.
What does this PR do?
This PR adds lambda support for Data Streams Monitoring (DSM), sets a checkpoint after extracting context from trace propagation headers, DSM context gets packaged in with the trace prop headers. Queues supported by DSM include SQS, SNS, SNS->SQS, and Kinesis
Motivation
Consume calls of lambdas are set as triggers to queue and do not have any explicit calls the tracers can hook into to set a checkpoint
Testing Guidelines
Tried all configurations with sandbox AWS account. Wrote unit tests ensuring all cases were covered.
Additional Notes
Types of Changes
Check all that apply