aiobotocore-otel

import asyncio import os from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor from opentelemetry.instrumentation.botocore import AiobotocoreInstrumentor import aiobotocore.session async def main(): # 1. Setup OpenTelemetry TracerProvider provider = TracerProvider() processor = SimpleSpanProcessor(ConsoleSpanExporter()) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # 2. Instrument aiobotocore AiobotocoreInstrumentor().instrument() # 3. Use aiobotocore client (e.g., S3) session = aiobotocore.session.get_session() # Use environment variables for AWS credentials in a real scenario aws_access_key_id = os.environ.get('AWS_ACCESS_KEY_ID', 'YOUR_ACCESS_KEY') aws_secret_access_key = os.environ.get('AWS_SECRET_ACCESS_KEY', 'YOUR_SECRET_KEY') # Ensure a proper region and dummy credentials for a runnable example if aws_access_key_id == 'YOUR_ACCESS_KEY' or aws_secret_access_key == 'YOUR_SECRET_KEY': print("WARNING: Using dummy AWS credentials. Replace with actual credentials or environment variables.") async with session.create_client( 's3', region_name='us-east-1', aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key ) as client: try: # This call will be traced if instrumentation is active response = await client.list_buckets() print("Successfully listed S3 buckets (or attempted to if credentials are dummy).") except Exception as e: print(f"Error listing S3 buckets: {e}. If using dummy credentials, this is expected.") if __name__ == "__main__": asyncio.run(main())