aws-xray-sdk

import os from aws_xray_sdk.core import xray_recorder, patch_all import boto3 # Configure the X-Ray recorder globally # For local development without a daemon, set sampling=False to prevent errors # In deployed AWS environments, the daemon address is often discovered automatically. xray_recorder.configure( service='MyPythonApp', sampling=False, # Set to True or rely on daemon for actual sampling rules in production context_missing='LOG_ERROR', # Other options: 'RUNTIME_ERROR', 'IGNORE' daemon_address=os.environ.get('AWS_XRAY_DAEMON_ADDRESS', '127.0.0.1:2000') ) # Patch supported libraries to automatically instrument calls (e.g., boto3) patch_all() # Example: Instrumenting an AWS SDK call with a decorator @xray_recorder.capture('s3_list_buckets') def list_s3_buckets(): # Ensure AWS credentials/config are set in environment or ~/.aws/config s3_client = boto3.client('s3', region_name='us-east-1') try: response = s3_client.list_buckets() bucket_names = [bucket['Name'] for bucket in response.get('Buckets', [])] # Add custom metadata to the current subsegment xray_recorder.current_subsegment().put_metadata('bucket_count', len(bucket_names)) return bucket_names except Exception as e: # Record exceptions to the current subsegment xray_recorder.current_subsegment().add_exception(e) raise # Manual segment creation (if not using web framework middleware or a top-level decorator) # This mimics an incoming request context. with xray_recorder.in_segment(xray_recorder.begin_segment('MyApplicationRootTrace')): print("Starting application trace...") try: buckets = list_s3_buckets() print(f"Successfully listed {len(buckets)} S3 buckets.") # Add an annotation to the root segment xray_recorder.current_segment().put_annotation('result', 'success') except Exception as e: print(f"An error occurred: {e}") xray_recorder.current_segment().put_annotation('result', 'failure') xray_recorder.current_segment().add_exception(e) print("Trace completed.")