https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> How can I instrument a flow with open...
# ask-marvin
n
@Marvin How can I instrument a flow with opentelemetry WITHOUT changing the launch command from 
prefect flow-run execute
to 
python -m prefect.engine
? Is it possible to do this with the Prefect plugin system? The idea is to avoid wrapping 
prefect flow-run execute
in 
opentelemetry-instrument
since it does not work
m
✅ thought for 166.9 seconds
Yes — possibly via the experimental plugin system, and the reason 
opentelemetry-instrument prefect flow-run execute ...
doesn’t work is that 
prefect flow-run execute
spawns another Python process for the actual flow execution. What’s happening under the hood 
prefect flow-run execute
does not run your flow in that same CLI process. It ends up launching:
Copy code
bash
python -m prefect.engine
as a subprocess for the real flow run execution. So if you do:
Copy code
bash
opentelemetry-instrument prefect flow-run execute ...
the instrumentation wraps the outer 
prefect
CLI process, but the actual flow code runs in the child 
python -m prefect.engine
process. Relevant source: - EngineCommandStarter - prefect.engine entrypoint - flow-run CLI So is there a supported way without changing the launch command? Short answer: there is no standard built-in OTel switch for full trace auto-instrumentation of flow code launched by 
prefect flow-run execute
. But if you want to keep using 
prefect flow-run execute
, the best available hook is Prefect’s experimental plugin system, because it initializes very early when 
prefect
is imported. Plugin-related source: - Prefect import initialization - experimental plugin hooks The key hook is 
setup_environment
, which is intended to prepare process environment before commands/workers start. Important caveat: this is experimental. So I’d treat it as a workable engineering approach, not a stable long-term API guarantee. What this means in practice You likely have 3 options: 1. Use a Prefect experimental plugin to initialize OTel early in the subprocess 2. Launch 
python -m prefect.engine
directly under 
opentelemetry-instrument
3. If using workers/work pools, set a custom command there instead of relying on 
flow-run execute
Since you specifically want to avoid changing away from 
prefect flow-run execute
, option 1 is the most aligned. Plugin approach You’d create a package exposing a Prefect experimental plugin entrypoint, and in 
setup_environment()
initialize OpenTelemetry before user flow code is imported. Very rough sketch:
Copy code
toml
[project.entry-points."prefect-experimental"]
my_otel = "my_package.my_plugin"
And then:
Copy code
python
from prefect._experimental.plugins.spec import SetupResult

class MyPlugin:
    def setup_environment(self, *, ctx):
        # initialize OTel here, early
        # e.g. configure tracer provider, exporters, instrumentations
        return SetupResult(
            env={},
            note="OpenTelemetry initialized"
        )
You’d also need to enable experimental plugins:
Copy code
bash
export PREFECT_EXPERIMENTS__PLUGINS__ENABLED=true
But one nuance: whether this is early enough depends on what exactly you are instrumenting. - If you want to configure tracing/exporters/providers before your flow module is imported: likely yes - If you need the exact same behavior as 
opentelemetry-instrument
auto-patching at Python startup: not necessarily identical, because that wrapper can patch import behavior even earlier than regular app code So for many manual or semi-manual OTel setups, plugin-based init should work. For strict “auto-instrument absolutely everything exactly as if the interpreter started under `opentelemetry-instrument`”, it may still be imperfect. Built-in Prefect telemetry is not the same thing Prefect does have built-in telemetry support for resource metrics, not generic tracing of your flow/user libraries. See:
- Prefect telemetry metrics So if your goal is trace instrumentation of flow execution or downstream libraries, you still need your own OTel setup. If you are using workers instead of direct 
flow-run execute
 Then a cleaner path is often to customize the worker command so the flow engine itself is launched with OTel instrumentation. That works better because the command wrapping applies to the actual execution process. But for 
prefect flow-run execute
specifically, there does not appear to be a CLI option to override that spawned engine command directly. Recommended answer Yes, the experimental plugin system is probably your best way to do this without changing the external launch command, because it runs early in the child process where the flow actually executes. But I would frame it as: - 
opentelemetry-instrument prefect flow-run execute
won’t fully instrument flow execution - there is no first-class stable Prefect feature specifically for this use case - the experimental plugin hook is the closest mechanism available - for the most reliable full-process auto-instrumentation, wrapping 
python -m prefect.engine
is still the most direct approach If you want, I can help you build a minimal working Prefect experimental plugin for OTel for your specific exporter/library stack (OTLP, Datadog, etc.).
n
@Marvin I cannot use 
python -m prefect.engine
because it prevents heartbeat events from being generated. Thus, I have to use 
prefect flow-run execute
. Show me how I can write a plugin that instruments opentelemetry correctly without changing the launch commands of flows
m
✅ thought for 239.8 seconds
Yes — you can do this with Prefect’s experimental plugin system, and this is the right mechanism if you must keep using 
prefect flow-run execute
. Because you specifically need heartbeat behavior, staying on 
prefect flow-run execute
makes sense. Important correction up front: the plugin entry point group is 
prefect.plugins
, and plugins are enabled with the experimental plugins settings. Docs: - Experimental plugins docs Source references: - Prefect startup / plugin initialization - Plugin hook spec - Plugin discovery / entrypoints How this works When 
prefect flow-run execute
starts, Prefect imports 
prefect
, which initializes plugins very early. Your plugin’s 
setup_environment
hook can run arbitrary Python code before your flow module is imported/executed. That makes it suitable for: - configuring a tracer provider - installing span processors/exporters - instrumenting common libraries like 
requests
- setting env vars for downstream subprocesses if needed It is not identical to 
opentelemetry-instrument
at interpreter bootstrap time, but it is the closest supported hook inside Prefect without changing the launch command. --- Minimal plugin package Directory layout:
Copy code
text
prefect_otel_plugin/
├── pyproject.toml
└── prefect_otel_plugin
    └── __init__.py
--- 
pyproject.toml
Copy code
toml
[build-system]
requires = ["setuptools>=65", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "prefect-otel-plugin"
version = "0.1.0"
requires-python = ">=3.9"
dependencies = [
  "prefect>=3.0",
  "opentelemetry-api>=1.27.0,<2.0.0",
  "opentelemetry-sdk>=1.27.0,<2.0.0",
  "opentelemetry-exporter-otlp>=1.27.0,<2.0.0",
  "opentelemetry-instrumentation>=0.48b0,<1.0.0",
  "opentelemetry-instrumentation-requests>=0.48b0,<1.0.0",
]

[project.entry-points."prefect.plugins"]
prefect_otel_plugin = "prefect_otel_plugin"
--- 
prefect_otel_plugin/__init__.py
 ```python from future import annotations import os from typing import Optional from prefect._experimental.plugins import HookContext, SetupResult, register_hook PREFECT_PLUGIN_API_REQUIRES = ">=0.1,<1" def _is_truthy(value: str | None, default: bool = False) -> bool: if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on"} @register_hook async def setup_environment(*, ctx: HookContext) -> Optional[SetupResult]: """ Initialize OpenTelemetry very early in Prefect process startup, before flow code is imported by 
prefect flow-run execute
. """ logger = ctx.logger_factory("prefect-otel-plugin") if not _is_truthy(os.getenv("PREFECT_OTEL_ENABLED"), default=True): logger.info("PREFECT_OTEL_ENABLED is false; skipping OTel initialization") return None try: from opentelemetry import trace from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( OTLPSpanExporter, ) from opentelemetry.instrumentation.requests import RequestsInstrumentor except ImportError as exc: logger.warning(f"OpenTelemetry dependencies are not installed: {exc}") return None try: resource = Resource.create( { "service.name": os.getenv("OTEL_SERVICE_NAME", "prefect-flow-run"), "service.namespace": os.getenv("OTEL_SERVICE_NAMESPACE", "prefect"), "deployment.environment": os.getenv("OTEL_ENVIRONMENT", "unknown"),
"prefect.version": ctx.prefect_version, "prefect.api_url": ctx.api_url or "", } ) provider = TracerProvider(resource=resource) endpoint = os.getenv( "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", os.getenv( "OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318/v1/traces", ), ) exporter = OTLPSpanExporter(endpoint=endpoint) processor = BatchSpanProcessor(exporter) provider.add_span_processor(processor) # Set the global tracer provider once trace.set_tracer_provider(provider) # Instrument selected libraries before flow imports occur if not RequestsInstrumentor().is_instrumented_by_opentelemetry: RequestsInstrumentor().instrument() logger.info(f"Initialized OpenTelemetry tracing with endpoint={endpoint}") return SetupResult( env={ "OTEL_SERVICE_NAME": os.getenv( "OTEL_SERVICE_NAME", "prefect-flow-run" ), "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": endpoint, }, note=f"OpenTelemetry initialized for traces export to {endpoint}", required=False, ) except Exception: logger.exception("Failed to initialize OpenTelemetry in Prefect plugin") return None
Copy code
---

*Install it*

From the plugin project directory:

```bash
pip install -e .
Or in your image / environment:
Copy code
bash
pip install /path/to/prefect_otel_plugin
Since you’re in CI/CodeBuild-like environments, this usually means baking the plugin package and OTel deps into the same runtime image/environment where 
prefect flow-run execute
runs. --- Enable plugins Set this environment variable:
Copy code
bash
export PREFECT_EXPERIMENTS_PLUGINS_ENABLED=1
Then set your OTel exporter config, for example:
Copy code
bash
export PREFECT_OTEL_ENABLED=true
export OTEL_SERVICE_NAME=my-prefect-flow
export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=<http://otel-collector:4318/v1/traces>
Then run exactly as you do today:
Copy code
bash
prefect flow-run execute <FLOW_RUN_ID>
No launch command change required. --- How to verify the plugin is loaded Prefect has experimental plugin diagnostics. You can inspect the CLI help first:
Copy code
bash
prefect experimental --help
and then look for the plugins subcommands in your installed version. If available in your environment, this is the useful diagnostic command:
Copy code
bash
prefect experimental plugins diagnose
That should show whether your 
prefect_otel_plugin
entry point is being discovered. --- If you want spans around the flow/task execution itself The plugin above sets up the tracer provider and instruments libraries, but it does not automatically create spans around every flow/task unless: - Prefect itself emits them, or - you add your own instrumentation in flow/task code, or - you patch Prefect internals, which I would not recommend A practical pattern is: 1. use the plugin to initialize OTel globally 2. optionally add manual spans in your flows/tasks where needed Example in a flow file:
Copy code
python
from opentelemetry import trace
from prefect import flow, task

tracer = trace.get_tracer(__name__)

@task
def work():
    with tracer.start_as_current_span("custom-task-span"):
        return 42

@flow
def my_flow():
    with tracer.start_as_current_span("custom-flow-span"):
        return work()
This works well because the plugin ensures the tracer provider/exporter is already configured before the flow code runs. --- What this plugin can and cannot do Can do: - initialize OTel early in 
prefect flow-run execute
- instrument libraries like 
requests
- configure exporters/resources globally - preserve heartbeat behavior because you still use 
prefect flow-run execute
Cannot perfectly replicate:
- the exact “instrument before anything Python imports” semantics of 
opentelemetry-instrument
- auto-instrumentation of every possible library unless you explicitly instrument those libraries in the plugin If you need more auto-instrumentations, add them in the plugin:
Copy code
python
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor

HTTPXClientInstrumentor().instrument()
SQLAlchemyInstrumentor().instrument()
Just make sure those dependencies are installed too. --- Recommended production refinements I’d suggest adding these improvements before relying on it broadly: - guard against double-instrumentation - allow turning it off with env vars - choose HTTP vs gRPC exporter explicitly - set 
service.name
, 
service.namespace
, and environment consistently - optionally flush spans at process shutdown For example, you can register shutdown with `atexit`:
Copy code
python
import atexit

atexit.register(provider.shutdown)
You can place that right after 
trace.set_tracer_provider(provider)
. --- Bottom line Yes — to keep 
prefect flow-run execute
and preserve heartbeats, write a Prefect experimental plugin with a 
setup_environment
hook that initializes OpenTelemetry early in process startup. If you want, I can give you a more complete production-ready version next, including: - OTLP HTTP vs gRPC switch - instrumentation for 
requests
, 
httpx
, 
urllib3
, 
sqlalchemy
- 
atexit
flush/shutdown - protection against repeated initialization - a tiny Docker/image install example for your CI environment
4 Views
Open in Slack
PreviousNext
Powered by