datadog-lambda

import json import os from datadog_lambda.trigger import extract_trigger_tags from datadog_lambda.wrapper import datadog_lambda_wrapper # Datadog API Key is typically set as an environment variable or handled by the Datadog Lambda Extension. # No direct in-code API key needed for basic wrapping. @datadog_lambda_wrapper def lambda_handler(event, context): # Extract tags specific to the trigger (e.g., SQS queue name, API Gateway details) trigger_tags = extract_trigger_tags(event) print(f"Trigger tags: {trigger_tags}") # Example of adding a custom tag to the trace span if os.environ.get('DD_TRACE_ENABLED', 'true').lower() == 'true': try: from ddtrace import tracer span = tracer.current_span() if span: # Span might be None if tracing is not enabled or not yet started span.set_tag('my_custom_tag', 'hello_datadog') except ImportError: pass # ddtrace is not installed or available response_body = { "message": "Hello from Datadog Lambda!", "input": event } return { "statusCode": 200, "body": json.dumps(response_body) } # To test locally (requires dummy event and context): # if __name__ == "__main__": # class MockContext: # aws_request_id = "test-request-id-123" # function_name = "test-function" # invoked_function_arn = "arn:aws:lambda:us-east-1:123456789012:function:test-function" # memory_limit_in_mb = 128 # get_remaining_time_in_millis = lambda: 10000 # test_event = { # "httpMethod": "GET", # "path": "/test", # "queryStringParameters": {"param1": "value1"} # } # os.environ['DD_SERVICE'] = 'my-lambda-service' # os.environ['DD_ENV'] = 'dev' # os.environ['DD_VERSION'] = '1.0.0' # os.environ['DD_TRACE_ENABLED'] = 'true' # print(lambda_handler(test_event, MockContext()))