Make OpenTelemetry imports conditional

src/prefect/main.py

@@ -58,9 +58,9 @@
 )
 
 # Configure telemetry
-import prefect.telemetry.instrumentation
+import prefect.telemetry.bootstrap
 
-prefect.telemetry.instrumentation.setup_telemetry()
+prefect.telemetry.bootstrap.setup_telemetry()
 
 
 from prefect._internal.compatibility.deprecated import (

src/prefect/telemetry/__init__.py

@@ -1 +0,0 @@
-

src/prefect/telemetry/bootstrap.py

@@ -0,0 +1,32 @@
+from typing import TYPE_CHECKING, Union
+
+import prefect.settings
+from prefect.client.base import ServerType, determine_server_type
+
+if TYPE_CHECKING:
+    from opentelemetry.sdk._logs import LoggerProvider
+    from opentelemetry.sdk.metrics import MeterProvider
+    from opentelemetry.sdk.trace import TracerProvider
+
+
+def setup_telemetry() -> (
+    Union[
+        tuple["TracerProvider", "MeterProvider", "LoggerProvider"],
+        tuple[None, None, None],
+    ]
+):
+    settings = prefect.settings.get_current_settings()
+    if not settings.experiments.telemetry_enabled:
+        return None, None, None
+
+    server_type = determine_server_type()
+    if server_type != ServerType.CLOUD:
+        return None, None, None
+
+    assert settings.api.key
+    assert settings.api.url
+
+    # This import is here to defer importing of the `opentelemetry` packages.
+    from .instrumentation import setup_exporters
+
+    return setup_exporters(settings.api.url, settings.api.key.get_secret_value())

src/prefect/telemetry/instrumentation.py

@@ -1,7 +1,7 @@
 import logging
 import os
 import re
-from typing import TYPE_CHECKING, Union
+from typing import TYPE_CHECKING
 from urllib.parse import urljoin
 from uuid import UUID
 
@@ -17,9 +17,6 @@
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
 
-import prefect.settings
-from prefect.client.base import ServerType, determine_server_type
-
 from .logging import set_log_handler
 from .processors import InFlightSpanProcessor
 
@@ -56,26 +53,11 @@ def _url_join(base_url: str, path: str) -> str:
     return urljoin(base_url.rstrip("/") + "/", path.lstrip("/"))
 
 
-def setup_telemetry() -> (
-    Union[
-        tuple[TracerProvider, MeterProvider, "LoggerProvider"], tuple[None, None, None]
-    ]
-):
-    """Configure OpenTelemetry exporters for Prefect telemetry."""
-    settings = prefect.settings.get_current_settings()
-    if not settings.experiments.telemetry_enabled:
-        return None, None, None
-
-    server_type = determine_server_type()
-    if server_type != ServerType.CLOUD:
-        return None, None, None
-
-    assert settings.api.key
-    assert settings.api.url
-
-    api_key = settings.api.key.get_secret_value()
-    account_id, workspace_id = extract_account_and_workspace_id(settings.api.url)
-    telemetry_url = _url_join(settings.api.url, "telemetry/")
+def setup_exporters(
+    api_url: str, api_key: str
+) -> tuple[TracerProvider, MeterProvider, "LoggerProvider"]:
+    account_id, workspace_id = extract_account_and_workspace_id(api_url)
+    telemetry_url = _url_join(api_url, "telemetry/")
 
     headers = {
         "Authorization": f"Bearer {api_key}",

src/prefect/telemetry/logging.py

@@ -1,18 +1,19 @@
 import logging
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
 
-from opentelemetry.sdk._logs import LoggingHandler
+if TYPE_CHECKING:
+    from opentelemetry.sdk._logs import LoggingHandler
 
-_log_handler: Optional[LoggingHandler] = None
+_log_handler: Optional["LoggingHandler"] = None
 
 
-def set_log_handler(log_handler: Optional[LoggingHandler]) -> None:
+def set_log_handler(log_handler: Optional["LoggingHandler"]) -> None:
     """Set the OTLP log handler."""
     global _log_handler
     _log_handler = log_handler
 
 
-def get_log_handler() -> Optional[LoggingHandler]:
+def get_log_handler() -> Optional["LoggingHandler"]:
     """Get the OTLP log handler."""
     global _log_handler
     return _log_handler

tests/telemetry/test_instrumentation.py

@@ -11,10 +11,8 @@
 from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
 from opentelemetry.sdk.trace import TracerProvider
 
-from prefect.telemetry.instrumentation import (
-    extract_account_and_workspace_id,
-    setup_telemetry,
-)
+from prefect.telemetry.bootstrap import setup_telemetry
+from prefect.telemetry.instrumentation import extract_account_and_workspace_id
 from prefect.telemetry.logging import get_log_handler
 from prefect.telemetry.processors import InFlightSpanProcessor
 

tests/telemetry/test_logging.py

@@ -1,6 +1,6 @@
 import logging
 
-from prefect.telemetry.instrumentation import setup_telemetry
+from prefect.telemetry.bootstrap import setup_telemetry
 from prefect.telemetry.logging import (
     add_telemetry_log_handler,
     get_log_handler,