Instrument Airflow with Opentracing/Opentelemetry

[Etienne-Carriere]

Description

I would like to instrument airflow with tracing technology like Opentracing (https://opentracing.io/) / Opentelemetry (https://opentelemetry.io/)

Use case / motivation

The motivation are to :

• have the DAG logic represented in tracing
• transfer the tracing information to downwards requests

** Implementation ideas **

For example, on Celery, we use celery signals to hook before and after each task (https://github.com/uber-common/opentracing-python-instrumentation/blob/master/opentracing_instrumentation/client_hooks/celery.py)

I am ok to contribute to the code but interested by some hints on where to look to .

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[mik-laj]

How is Opentracing / Opentelemetry different from Sentry? We have native integration with Sentry and I wonder if we should combine these two features or develop it independently.
http://apache-airflow-docs.s3-website.eu-central-1.amazonaws.com/docs/apache-airflow/latest/logging-monitoring/errors.html

[Etienne-Carriere]

Hello @mik-laj , thank you for the link about sentry. It will helps to design a hook (in a first time in an external repo and perhaps merged in a second time)
Opentracing/Opentelemetry is an open-source standard for APM (in the same level of sentry)

[potiuk]

Hello @Etienne-Carriere @mik-laj -> we started recently seeing those annoying logs in Airflow 2.0.

{opentelemetry_tracing.py:29} INFO - This service is instrumented using OpenTelemetry. OpenTelemetry could not be imported; please add opentelemetry-api and opentelemetry-instrumentation packages in order to get BigQuery Tracing data.

Is this something you know about ?

[ashb]

@potiuk Looks like it's from something BigQuery related.

[ashb]

I know @mjpieters was playing/hacking about with adding opentracing to Airflow too

[potiuk]

Yeah. It can likely be disabled by adding some custom logging configuration - possibly that's what we should do I think

[ashb]

https://github.com/googleapis/python-bigquery/blob/master/google/cloud/bigquery/opentelemetry_tracing.py

[potiuk]

Added #13131

[dm03514]

👍 For this feature.

@mik-laj

I created a little test script to illustrate building a trace from a dag run:

https://github.com/dm03514/airflow-tracer/blob/main/README.md

Tracing is a different type of telemetry (alongside statsd and sentry). It provides a "Gantt" Chart view into a dag. Some tracing providers (like LIghtstep) provide advanced dashboarding and alerting.

The POC project above builds a trace by querying airflows database. I'm hoping that there is a way to add tracing support inside airflow so that when dags are being executed they can emit open telemetry traces!

With tracing each task span needs a reference to the global span and i'm not sure how easy / hard this is to share data across tasks?

Start the root span when the dagrun starts:

    root_span = tracer.start_span(
        name=dag.dag.dag_id
    )

Start a task specific span when each task starts:

        ctx = trace.set_span_in_context(root_span)

        span = tracer.start_span(
            name=ti.task_id,
            context=ctx
        )

       ... # invoke task

      span.end() # task is complete

Close the root span when all tasks are complete

root_span.end()

---

def main(cli):
 dag = serialized_dag.SerializedDagModel.get(dag_id=cli.dag_id)
    dagrun = dag.dag.get_dagrun(execution_date=cli.execution_date)
    tis = dagrun.get_task_instances()

    root_span = tracer.start_span(
        name=dag.dag.dag_id,
        start_time=dt_to_ns_epoch(dagrun.start_date)
    )
    root_span.end(end_time=dt_to_ns_epoch(dagrun.end_date))

    for ti in tis:
        ctx = trace.set_span_in_context(root_span)

        span = tracer.start_span(
            name=ti.task_id,
            context=ctx,
            start_time=dt_to_ns_epoch(ti.start_date),
        )
        span.set_attribute('airflow.state', ti.state)
        span.set_attribute('airflow.operation', ti.operator)
        if ti.job_id is not None:
            span.set_attribute('airflow.job_id', ti.job_id)
        if ti.state != 'success':
            span.set_attribute('error', True)
        span.end(end_time=dt_to_ns_epoch(ti.end_date))

thank you!

[mjpieters]

Interesting little test; but note it generates a trace from the tasks after the fact without support for additional spans created inside each task invocation, and you the tracing context is not shared overy any RPC / REST / etc. calls to other services to inherit the tracing context.

I integrated Jaeger tracing (opentracing) support into a production Airflow setup using Celery. Specific challenges I had to overcome:

• The dagrun span is 'virtual', in that it exists from when the dagrun is created until the last task is completed. The scheduler will then update the dagrun state in the database. But tasks need a valid parent span to attach their own spans to.

  I solved this by creating a span in a monkey-patch to DAG.create_dagrun(), injecting the span info in to the dagrun configuration together with the start time, then discarding the span. Then, in DagRun.set_state(), when the state changes to a finished state, I create a jaeger_client.span.Span() object from scratch using the dagrun conf-stored data, and submit that to Jaeger.

• Tasks inherit the parent (dagrun) span context from the dagrun config; I patched Task.run_raw_task() to run the actual code under a tracer.start_active_span() context manager. This captures timing and any exceptions.

• You need an active tracer for traces to be captured and sent on to the tracer agent. So I registered code to run in the cli_action_loggers.register_pre_exec_callback() hook when the scheduler or dag_trigger sub-commands run, which then registers a closer with cli_action_loggers.register_post_exec_callback. Closing a tracer in dag_trigger takes careful work with the asyncio / tornado loop used by the Jaeger client, you'll lose traces if you don't watch out. I found that you had to go hunt for the I/O loop attached to the trace reporter object and call tracer.close() from a callback sent to that loop as the only fail-proof method of getting the last traces out. I don't know if opentracing needs this level of awareness of the implementation details.

• You generally want to start a trace when the Airflow webserver receives a trigger, so instrument the Flask layer too. This is where the sampling decision needs to be taken too; tracing often only instruments a subset of all jobs, but in this project I set the sampling frequency to 100% at this level.

• I added a custom configuration section to airflow.cfg to allow me to tweak Jaeger tracing parameters.

But, with that work in place, we now get traces in mostly real time, with full support for tracing contexts being shared with other services called from Airflow tasks. We can trace a job through the frontend, submitting a job to Airflow, then follow any calls from tasks to further REST APIs, all as one system.

I'd prefer it if the tracing context was not shoehorned into the dagrun configuration; I'd have created additional database tables or columns for this in the Airflow database model if I had to do this inside the Airflow project itself.

Note that I did not use the Celery task hooks here to track the task spans, because Celery has its own overhead that we wanted to keep separate. The opentracing_instrumentation already has Celery hooks you can use, but I needed the timings to be closer to the actual task invocation.

Anothing thing to consider is per-DAG or per-task tagging you want to add to the spans. For this project I needed to track specific data from the submitted DAG config so we can compare task runs using the same input configuration.

[WattsInABox]

@mjpieters can you share any of the code at all? I think I understand what you're saying but I think also I would be rediscovering problems you have already solved without the code.

[mjpieters]

@WattsInABox the code is part of a private project, and the majority is either project specific or specific to the jaeger_client implementation, neither of which apply here. E.g. the custom configuration section and how to close the client are tied very closely to the jaeger_client codebase, as is how you create a jaeger_client.span.Span() instance from scratch.

I can see about talking to the client about scrubbing the project-specific details from the code, however.