feat: add instrument_asyncio() method for asyncio instrumentation

docs/integrations/asyncio.md

@@ -0,0 +1,67 @@
+---
+title: "Logfire Asyncio Integration: Monitor Async Operations"
+description: Learn how to use logfire.instrument_asyncio() to trace and monitor asyncio-based operations.
+integration: otel
+---
+# Asyncio
+
+The [`logfire.instrument_asyncio()`][logfire.Logfire.instrument_asyncio] method can be used to instrument
+`asyncio`-based operations with **Logfire**, including tracing coroutines, futures, and detecting slow
+event loop callbacks.
+
+## Installation
+
+Install `logfire` with the `asyncio` extra:
+
+{{ install_logfire(extras=['asyncio']) }}
+
+## Usage
+
+```py title="main.py" skip-run="true" skip-reason="external-connection"
+import asyncio
+import os
+
+import logfire
+
+logfire.configure()
+os.environ['OTEL_PYTHON_ASYNCIO_COROUTINE_NAMES_TO_TRACE'] = 'my_coro'
+
+logfire.instrument_asyncio()
+
+
+async def my_coro():
+    await asyncio.sleep(0.1)
+
+
+async def main():
+    await asyncio.create_task(my_coro())
+
+
+asyncio.run(main())
+```
+
+The `OTEL_PYTHON_ASYNCIO_COROUTINE_NAMES_TO_TRACE` environment variable specifies which coroutine names
+to trace. Set it to a comma-separated list of coroutine function names.
+
+You can find more configuration options in the
+[OpenTelemetry asyncio instrumentation documentation][opentelemetry-asyncio].
+
+## Slow Callback Detection
+
+[`logfire.instrument_asyncio()`][logfire.Logfire.instrument_asyncio] also includes Logfire's built-in slow
+callback detection. It logs a warning whenever a function running in the asyncio event loop blocks for
+longer than `slow_duration` seconds (default: 0.1s):
+
+```py title="main.py" skip-run="true" skip-reason="external-connection"
+import logfire
+
+logfire.configure()
+logfire.instrument_asyncio(slow_duration=0.5)
+```
+
+[`logfire.instrument_asyncio()`][logfire.Logfire.instrument_asyncio] uses the
+**OpenTelemetry Asyncio Instrumentation** package,
+which you can find more information about [here][opentelemetry-asyncio].
+
+[opentelemetry-asyncio]: https://opentelemetry-python-contrib.readthedocs.io/en/latest/instrumentation/asyncio/asyncio.html
+[opentelemetry]: https://opentelemetry.io/

docs/integrations/index.md

@@ -36,7 +36,7 @@ If a package you are using is not listed in this documentation, please let us kn
 - _Task Queues and Schedulers_: Airflow, FastStream, Celery
 - _Logging Libraries_: Standard Library Logging, Loguru, Structlog
 - _Testing_: Pytest
-- and more, such as Stripe, AWS Lambda, and system metrics.
+- and more, such as Stripe, AWS Lambda, asyncio, and system metrics.
 
 The below table lists these integrations and any corresponding `logfire.instrument_<package>()` calls:
 
@@ -48,6 +48,7 @@ The below table lists these integrations and any corresponding `logfire.instrume
 | [Airflow](event-streams/airflow.md)       | Task Scheduler          | N/A (built in, config needed)                                                                                                                                          |
 | [Anthropic](llms/anthropic.md)            | AI                      | [`logfire.instrument_anthropic()`][logfire.Logfire.instrument_anthropic]                                                                                               |
 | [ASGI](web-frameworks/asgi.md)            | Web Framework Interface | [`logfire.instrument_asgi()`][logfire.Logfire.instrument_asgi]                                                                                                         |
+| [Asyncio](asyncio.md)                    | Async Runtime           | [`logfire.instrument_asyncio()`][logfire.Logfire.instrument_asyncio]                                                                                                   |
 | [AWS Lambda](aws-lambda.md)               | Cloud Function          | [`logfire.instrument_aws_lambda()`][logfire.Logfire.instrument_aws_lambda]                                                                                             |
 | [Asyncpg](databases/asyncpg.md)           | Database                | [`logfire.instrument_asyncpg()`][logfire.Logfire.instrument_asyncpg]                                                                                                   |
 | [BigQuery](databases/bigquery.md)         | Database                | N/A (built in, no config needed)                                                                                                                                       |

logfire-api/logfire_api/__init__.py

@@ -129,6 +129,9 @@ def force_flush(self, *args, **kwargs) -> None: ...
             def log_slow_async_callbacks(self, *args, **kwargs) -> None:  # pragma: no cover
                 return nullcontext()
 
+            def instrument_asyncio(self, *args, **kwargs) -> None:  # pragma: no cover
+                return nullcontext()
+
             def install_auto_tracing(self, *args, **kwargs) -> None: ...
 
             def instrument(self, *args, **kwargs):
@@ -220,6 +223,7 @@ def shutdown(self, *args, **kwargs) -> None: ...
         with_settings = DEFAULT_LOGFIRE_INSTANCE.with_settings
         force_flush = DEFAULT_LOGFIRE_INSTANCE.force_flush
         log_slow_async_callbacks = DEFAULT_LOGFIRE_INSTANCE.log_slow_async_callbacks
+        instrument_asyncio = DEFAULT_LOGFIRE_INSTANCE.instrument_asyncio
         install_auto_tracing = DEFAULT_LOGFIRE_INSTANCE.install_auto_tracing
         instrument = DEFAULT_LOGFIRE_INSTANCE.instrument
         instrument_asgi = DEFAULT_LOGFIRE_INSTANCE.instrument_asgi

logfire-api/logfire_api/_internal/main.pyi

@@ -382,6 +382,24 @@ class Logfire:
                 without waiting for the context manager to be opened,
                 i.e. it's not necessary to use this as a context manager.
         """
+    def instrument_asyncio(self, slow_duration: float = 0.1, **kwargs: Any) -> AbstractContextManager[None]:
+        """Instrument asyncio to trace coroutines, futures, and detect slow event loop callbacks.
+
+        This combines the OpenTelemetry asyncio instrumentation (for tracing coroutines, futures,
+        and `to_thread` calls) with Logfire's slow callback detection.
+
+        Args:
+            slow_duration: the threshold in seconds for when an event loop callback is considered slow.
+            **kwargs: Additional keyword arguments to pass to the OpenTelemetry `instrument` methods,
+                for future compatibility.
+
+        Returns:
+            A context manager that will revert the slow callback patch when exited.
+                This context manager doesn't take into account threads or other concurrency.
+                Calling this method will immediately apply the instrumentation
+                without waiting for the context manager to be opened,
+                i.e. it's not necessary to use this as a context manager.
+        """
     def install_auto_tracing(self, modules: Sequence[str] | Callable[[AutoTraceModule], bool], *, min_duration: float, check_imported_modules: Literal['error', 'warn', 'ignore'] = 'error') -> None:
         """Install automatic tracing.
 

logfire/__init__.py

@@ -36,6 +36,7 @@
 instrument = DEFAULT_LOGFIRE_INSTANCE.instrument
 force_flush = DEFAULT_LOGFIRE_INSTANCE.force_flush
 log_slow_async_callbacks = DEFAULT_LOGFIRE_INSTANCE.log_slow_async_callbacks
+instrument_asyncio = DEFAULT_LOGFIRE_INSTANCE.instrument_asyncio
 install_auto_tracing = DEFAULT_LOGFIRE_INSTANCE.install_auto_tracing
 instrument_pydantic = DEFAULT_LOGFIRE_INSTANCE.instrument_pydantic
 instrument_pydantic_ai = DEFAULT_LOGFIRE_INSTANCE.instrument_pydantic_ai
@@ -143,6 +144,7 @@ def loguru_handler() -> Any:
     'fatal',
     'force_flush',
     'log_slow_async_callbacks',
+    'instrument_asyncio',
     'install_auto_tracing',
     'instrument_asgi',
     'instrument_wsgi',

logfire/_internal/integrations/asyncio_.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from contextlib import AbstractContextManager
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from ..main import Logfire
+
+try:
+    from opentelemetry.instrumentation.asyncio import AsyncioInstrumentor
+except ImportError:  # pragma: no cover
+    raise RuntimeError(
+        '`logfire.instrument_asyncio()` requires the `opentelemetry-instrumentation-asyncio` package.\n'
+        'You can install this with:\n'
+        "    pip install 'logfire[asyncio]'"
+    )
+
+
+def instrument_asyncio(
+    logfire_instance: Logfire,
+    slow_duration: float = 0.1,
+    **kwargs: Any,
+) -> AbstractContextManager[None]:
+    """Instrument asyncio to trace coroutines, futures, and detect slow callbacks.
+
+    See the `Logfire.instrument_asyncio` method for details.
+    """
+    from ..async_ import log_slow_callbacks
+
+    AsyncioInstrumentor().instrument(**kwargs)
+    return log_slow_callbacks(logfire_instance, slow_duration)

logfire/_internal/main.py

@@ -892,6 +892,41 @@ def force_flush(self, timeout_millis: int = 3_000) -> bool:  # pragma: no cover
         """
         return self._config.force_flush(timeout_millis)
 
+    def instrument_asyncio(
+        self,
+        slow_duration: float = 0.1,
+        **kwargs: Any,
+    ) -> AbstractContextManager[None]:
+        """Instrument asyncio to trace coroutines, futures, and detect slow event loop callbacks.
+
+        This combines the OpenTelemetry asyncio instrumentation (for tracing coroutines, futures,
+        and `to_thread` calls) with Logfire's slow callback detection.
+
+        Args:
+            slow_duration: the threshold in seconds for when an event loop callback is considered slow.
+            **kwargs: Additional keyword arguments to pass to the OpenTelemetry `instrument` methods,
+                for future compatibility.
+
+        Returns:
+            A context manager that will revert the slow callback patch when exited.
+                This context manager doesn't take into account threads or other concurrency.
+                Calling this method will immediately apply the instrumentation
+                without waiting for the context manager to be opened,
+                i.e. it's not necessary to use this as a context manager.
+        """
+        from .integrations.asyncio_ import instrument_asyncio
+
+        self._warn_if_not_initialized_for_instrumentation()
+        return instrument_asyncio(
+            self,
+            slow_duration=slow_duration,
+            **{
+                'tracer_provider': self._config.get_tracer_provider(),
+                'meter_provider': self._config.get_meter_provider(),
+                **kwargs,
+            },
+        )
+
     def log_slow_async_callbacks(self, slow_duration: float = 0.1) -> AbstractContextManager[None]:
         """Log a warning whenever a function running in the asyncio event loop blocks for too long.
 

mkdocs.yml

@@ -180,6 +180,7 @@ nav:
               - Pytest: integrations/pytest.md
           - Pydantic: integrations/pydantic.md
           - System Metrics: integrations/system-metrics.md
+          - Asyncio: integrations/asyncio.md
           - Stripe: integrations/stripe.md
           - AWS Lambda: integrations/aws-lambda.md
       - Observe & Investigate:

pyproject.toml

@@ -64,6 +64,7 @@ wsgi = ["opentelemetry-instrumentation-wsgi >= 0.42b0"]
 aiohttp = ["opentelemetry-instrumentation-aiohttp-client >= 0.42b0"]
 aiohttp-client = ["opentelemetry-instrumentation-aiohttp-client >= 0.42b0"]
 aiohttp-server = ["opentelemetry-instrumentation-aiohttp-server >= 0.55b0"]
+asyncio = ["opentelemetry-instrumentation-asyncio >= 0.42b0"]
 celery = ["opentelemetry-instrumentation-celery >= 0.42b0"]
 django = ["opentelemetry-instrumentation-django >= 0.42b0", "opentelemetry-instrumentation-asgi >= 0.42b0"]
 fastapi = ["opentelemetry-instrumentation-fastapi >= 0.42b0"]
@@ -124,6 +125,7 @@ dev = [
     "opentelemetry-instrumentation-aiohttp-client>=0.42b0",
     "opentelemetry-instrumentation-aiohttp-server>=0.55b0",
     "opentelemetry-instrumentation-asgi>=0.42b0",
+    "opentelemetry-instrumentation-asyncio>=0.42b0",
     "opentelemetry-instrumentation-wsgi>=0.42b0",
     "opentelemetry-instrumentation-fastapi>=0.42b0",
     "opentelemetry-instrumentation-starlette>=0.42b0",

tests/otel_integrations/test_asyncio.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from asyncio.events import Handle
+
+import logfire
+from logfire.testing import TestExporter
+
+
+def test_instrument_asyncio(exporter: TestExporter) -> None:
+    """Test that instrument_asyncio patches Handle._run and can be reverted."""
+    from opentelemetry.instrumentation.asyncio import AsyncioInstrumentor
+
+    assert Handle._run.__qualname__ == 'Handle._run'
+
+    try:
+        with logfire.instrument_asyncio(slow_duration=100):
+            # Check that the slow callback patching is in effect
+            assert Handle._run.__qualname__ == 'log_slow_callbacks.<locals>.patched_run'
+
+        # Check that the patching is reverted after exiting the context manager
+        assert Handle._run.__qualname__ == 'Handle._run'
+    finally:
+        # Clean up OTel instrumentation (context manager only reverts slow callback patch)
+        AsyncioInstrumentor().uninstrument()

tests/test_logfire_api.py

@@ -157,6 +157,11 @@ def test_runtime(logfire_api_factory: Callable[[], ModuleType], module_name: str
     # NOTE: We don't call the log_slow_async_callbacks, to not give side effect to the test suite.
     logfire__all__.remove('log_slow_async_callbacks')
 
+    assert hasattr(logfire_api, 'instrument_asyncio')
+    # NOTE: We don't call instrument_asyncio, to not give side effect to the test suite
+    # (it patches asyncio.events.Handle._run via log_slow_callbacks).
+    logfire__all__.remove('instrument_asyncio')
+
     assert hasattr(logfire_api, 'install_auto_tracing')
     logfire_api.install_auto_tracing(modules=['all'], min_duration=0)
     logfire__all__.remove('install_auto_tracing')