opentelemetry-instrumentation-asyncclick

import asyncio from opentelemetry import trace from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor from opentelemetry.instrumentation.asyncclick import AsyncClickInstrumentor import asyncclick as click # Configure OpenTelemetry SDK resource = Resource.create({"service.name": "asyncclick-app"}) tracer_provider = TracerProvider(resource=resource) span_processor = SimpleSpanProcessor(ConsoleSpanExporter()) tracer_provider.add_span_processor(span_processor) trace.set_tracer_provider(tracer_provider) # Enable asyncclick instrumentation AsyncClickInstrumentor().instrument() # Define an async click command @click.group() async def cli(): "A simple async Click CLI." pass @cli.command() @click.option('--name', default='World', help='Name to greet.') async def hello(name): "Greets the given name." current_span = trace.get_current_span() current_span.set_attribute("greeting.name", name) click.echo(f"Hello, {name}!") await asyncio.sleep(0.05) click.echo("Greeting complete.") @cli.command() async def goodbye(): "Says goodbye." click.echo("Goodbye!") await asyncio.sleep(0.01) async def main(): # Simulate running a command, e.g., 'python your_app.py hello --name Alice' # In a real scenario, click.Group.main() would handle argv parsing. # For this example, we'll manually invoke the command. ctx = click.Context(cli, info_name='cli') await cli.invoke(ctx) # You can also simulate calling specific commands directly # await hello.callback(name="Bob") # await goodbye.callback() if __name__ == '__main__': # This part typically handles the command line invocation # For demonstration, we'll run the 'hello' command as if it was passed via argv # Note: asyncclick.main() internally uses asyncio.run() or anyio.run() # We are directly running an async function, so manually setup invocation. # This mimics `cli(['hello', '--name', 'OpenTelemetry'])` but for async context async def _run_cli_command(): ctx = click.Context(cli, obj={}, auto_envvar_prefix='OTEL') await cli.invoke(ctx) await hello.invoke(ctx, name='OpenTelemetry') await goodbye.invoke(ctx) asyncio.run(_run_cli_command()) # Ensure all spans are exported before exiting tracer_provider.shutdown()