feat(logs): [SVLS-8582] Hold logs and add durable context to durable function logs

[lym953]

Summary

If the function is a durable function, then add two attributes to every log:

• lambda.durable_execution_id
• lambda.durable_execution_name

Background

• In previous PRs (feat: add durable function execution tags to Lambda spans datadog-lambda-python#728, [SVLS-8538] add durable function tags datadog-lambda-js#730), tracer adds attributes durable_function_execution_id and durable_function_execution_name to the aws.lambda span

Details

Data flow

1. TraceAgent::handle_traces() detects an aws.lambda span carrying request_id, durable_function_execution_id, and durable_function_execution_name in its meta tags
2. It sends a ProcessorCommand::ForwardDurableContext { request_id, execution_id, execution_name } to InvocationProcessorService
3. Processor::forward_durable_context() in the lifecycle processor relays this as a DurableContextUpdate to the logs pipeline via an mpsc channel
4. LogsAgent::spin() receives the update and calls LogsProcessor::process_durable_context_update(), which inserts the entry into LambdaProcessor::durable_context_map and drains any held logs for that request_id

Log holding and draining

• After cold start, the logs processor holds all incoming logs without flushing them, because it does not yet know whether this is a durable function
• Held logs are stored in held_logs: HashMap<String, Vec<IntakeLog>>, keyed by request_id
• durable_context_map: HashMap<String, DurableExecutionContext> maps request_id to (execution_id, execution_name). It has a fixed capacity (500 entries) with FIFO eviction.
• When the logs processor receives a PlatformInitStart event, it learns whether the function is a durable function:
  • If not a durable function: drain all held logs (mark them ready for aggregation and flush)
  • If durable: drain held logs whose request_id is already in durable_context_map (tag them with lambda.durable_execution_id and lambda.durable_execution_name); keep the rest held until their context arrives
• When an entry is inserted into durable_context_map, any held logs for that request_id are drained immediately

Types

• DurableContextUpdate { request_id, execution_id, execution_name } — message sent from trace agent through lifecycle processor to logs pipeline
• DurableExecutionContext { execution_id, execution_name } — value type stored in durable_context_map

Test plan

Manual test

Steps

Build a layer, install it on a function, and invoke it.

Result

1. In Datadog, all the logs for this durable execution have the two new attributes

2. The logs query

source:lambda @lambda.arn:"arn:aws:lambda:us-east-2:425362996713:function:yiming-durable" @lambda.durable_execution_name:2f492839-75df-4acb-9f2a-30b1b36d5c8f

returns all the logs for two invocations of this durable execution. It returns 98 logs, equal to 49 logs for the first invocation + 49 logs for the second invocation. (query link)

Integration test

Passed the added integration test

Unit tests

Passed the added unit tests

[jchrostek-dd]

Can come in a later PR, but I think we should also add Durable Function Operation Name as an attribute in the logs when it is available.

[lym953]

Can come in a later PR, but I think we should also add Durable Function Operation Name as an attribute in the logs when it is available.

@jchrostek-dd For SDK logs, operation name and operation id already exist as log attributes:

For non-SDK logs, we are not planning to add them to logs because it’s hard for the extension to get them.

[duncanista]

This looks kinda dangerous, why are we doing this for?

[lym953]

You are absolutely right to question this! I agree it would be better to add a separate ProcessorCommand to pass durable context instead of reusing an existing one. Let's talk in the upcoming meeting.

[lym953]

Will rename it to process_telemetry_event() to be parallel to process_durable_context_update(). Not in this PR because this PR is already very big.