feat: propagate trace context from user app

[fatih-acar]

We may want to propagate the traceparent from an instrumented app running Prefect deployments so that we get end-to-end tracing from the app to the Prefect workers.

Feel free to rework this but the main use case is about getting end-to-end traces for apps relying on Prefect deployments.

Checklist

• This pull request references any related issue by including "closes <link to issue>"
  • If no issue exists and your change is not a small fix, please create an issue first.
• If this pull request adds new functionality, it includes unit tests that cover the changes
• If this pull request removes docs files, it includes redirect settings in mint.json.
• If this pull request adds functions or classes, it includes helpful docstrings.

[codspeed-hq]

CodSpeed Performance Report

Merging #16812 will not alter performance

_{Comparing fatih-acar:fac-add-e2e-tracing (c59e6a0) with main (bf451fb)}

Summary

✅ 2 untouched benchmarks

[desertaxle]

Thanks for the PR @fatih-acar! Can you provide a script showing how this new functionality would be used and add some unit tests? Both would be very valuable in verifying these changes.

[fatih-acar]

Not sure if a unit test is relevant here, but I'll describe my use case using code snippets and screenshots:

Without this patch, I get two separate traces: one for the API call and one for the Prefect flow run.

With this patch, I get a single trace spanning both services:

import fastapi
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

from prefect import State
from prefect.deployments import run_deployment

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

app = fastapi.FastAPI()

@app.get("/foobar")
async def foobar():
    await run_deployment(name="test-worker-flow/worker", timeout=0)
    return {"message": "hello world"}

FastAPIInstrumentor.instrument_app(app)

OTEL_EXPORTER_OTLP_PROTOCOL=grpc OTEL_EXPORTER_OTLP_INSECURE=true OTEL_SERVICE_NAME=api-server poetry run fastapi dev api.py

import asyncio

from prefect import serve
from prefect import flow

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

@flow(flow_run_name="test_worker_subflow")
async def test_worker_subflow():
    print("test subflow")
    await asyncio.sleep(3)

@flow(flow_run_name="test_worker_flow")
async def test_worker_flow():
    print("test flow")
    await asyncio.sleep(2)
    await test_worker_subflow()

if __name__ == "__main__":
    serve(test_worker_flow.to_deployment(name="worker"))

OTEL_EXPORTER_OTLP_PROTOCOL=grpc OTEL_EXPORTER_OTLP_INSECURE=true OTEL_SERVICE_NAME=task-worker poetry run python worker.py

[desertaxle]

Thanks for the info @fatih-acar! I think it'd be worthwhile to add a test to prevent regression after the PR is merged. I think a test similar to this one would work well:

prefect/tests/deployment/test_flow_runs.py

Line 440 in c685760

async def test_propagates_otel_trace_to_deployment_flow_run(

[github-actions]

This pull request is stale because it has been open 14 days with no activity. To keep this pull request open remove stale label or comment.

[fatih-acar]

Thanks for the info @fatih-acar! I think it'd be worthwhile to add a test to prevent regression after the PR is merged. I think a test similar to this one would work well:

prefect/tests/deployment/test_flow_runs.py

Line 440 in c685760

async def test_propagates_otel_trace_to_deployment_flow_run(

Thanks for the pointer, I have added a test that should be enough.