Cannot render the DAG details page when a DAG run is triggered by a run configuration with a circular reference

[molcay]

Hi,
I would like to take the community's opinion about a bug. We have some ideas about why it is happening and how we can fix it, but it is tightly related with the airflow core. So we would like to start a discussion here and decide on a solution and hopefully implement it.

Details

I have two different DAGs called basic_dag and basic_dag_triggerer. The basic_dag can be as simple as the following:

task = BashOperator(task_id="normal", bash_command="echo 'I am just a normal task'")

and the basic_dag_triggerer is only responsible for triggering the basic_dag via the airflow.api.common.trigger_dag.trigger_dag function but we are passing a circular configuration by mistake.

When we run the basic_dag_triggerer in the UI (or any other method like using the CLI: airflow dags trigger basic_dag_triggerer), the task and dag run for the basic_dag_triggerer can successfully finish and everything looks OK.
But if you try to navigate the details page for the basic_dag, you will be getting some errors on the UI and the grid data will not be loading:

The reason for this, we are trying to get the grid data from the backend as JSON to show it in the UI and the backend tries to serialize (and encode) the dag run config to return to the UI by calling the json.dumps. It is failing due to the circular reference. Here is the webserver log for this:

webserver  | 192.168.8.1 - - [15/May/2024:13:51:01 +0000] "GET /api/v1/dags/basic_dag HTTP/1.1" 200 1088 "http://localhost:28080/dags/basic_dag/grid" "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
webserver  | 192.168.8.1 - - [15/May/2024:13:51:01 +0000] "GET /object/grid_data?dag_id=basic_dag&num_runs=25 HTTP/1.1" 500 1593 "http://localhost:28080/dags/basic_dag/grid" "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
webserver  | [2024-05-15T13:51:01.925+0000] {app.py:1744} ERROR - Exception on /object/grid_data [GET]

.... # deducted because it is just more traceback

webserver  | File "/opt/airflow/airflow/www/views.py", line 3269, in grid_data
webserver  | encoded_runs = [wwwutils.encode_dag_run(dr, json_encoder=utils_json.WebEncoder) for dr in dag_runs]
webserver  | File "/opt/airflow/airflow/www/views.py", line 3269, in <listcomp>
webserver  | encoded_runs = [wwwutils.encode_dag_run(dr, json_encoder=utils_json.WebEncoder) for dr in dag_runs]
webserver  | File "/opt/airflow/airflow/www/utils.py", line 176, in encode_dag_run
webserver  | dag_run_conf, conf_is_json = get_dag_run_conf(dag_run.conf, json_encoder=json_encoder)
webserver  | File "/opt/airflow/airflow/www/utils.py", line 164, in get_dag_run_conf
webserver  | result = json.dumps(dag_run_conf, sort_keys=True, cls=json_encoder, ensure_ascii=False)
webserver  | File "/usr/local/lib/python3.8/json/__init__.py", line 234, in dumps
webserver  | return cls(
webserver  | File "/usr/local/lib/python3.8/json/encoder.py", line 199, in encode
webserver  | chunks = self.iterencode(o, _one_shot=True)
webserver  | File "/usr/local/lib/python3.8/json/encoder.py", line 257, in iterencode
webserver  | return _iterencode(o, 0)
webserver  | ValueError: Circular reference detected

Also, in the UI, if we click the Run button on the DAG list page for the basic_dag, we are seeing the generic traceback.html page like the following:

The same thing happens for this. json.dumps related exception. Here is the webserver log for this:

webserver  | [2024-05-15T14:18:43.535+0000] {app.py:1744} ERROR - Exception on /dags/basic_dag/trigger [POST]

.... # deducted because it is just more traceback

webserver  | File "/opt/airflow/airflow/www/views.py", line 2069, in trigger
webserver  | recent_confs = {
webserver  | File "/opt/airflow/airflow/www/views.py", line 2070, in <dictcomp>
webserver  | run_id: json.dumps(run_conf)
webserver  | File "/usr/local/lib/python3.8/json/__init__.py", line 231, in dumps
webserver  | return _default_encoder.encode(obj)
webserver  | File "/usr/local/lib/python3.8/json/encoder.py", line 199, in encode
webserver  | chunks = self.iterencode(o, _one_shot=True)
webserver  | File "/usr/local/lib/python3.8/json/encoder.py", line 257, in iterencode
webserver  | return _iterencode(o, 0)
webserver  | ValueError: Circular reference detected
webserver  | 192.168.8.1 - - [15/May/2024:14:18:43 +0000] "POST /dags/basic_dag/trigger?redirect_url=%2Fhome HTTP/1.1" 500 1593 "http://localhost:28080/home?tags=basic" "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"

If we run basic_dag_triggerer there is no error or if we want to navigate the details page for basic_dag_triggerer everything looks normal.

Apart from this, the scheduler is also failing but we only have the following warning

The scheduler does not appear to be running. Last heartbeat was received x minutes ago.
The DAGs list may not update, and new tasks will not be scheduled.

The scheduler exception is like the following:

[2024-05-15 13:33:27 +0000] [44068] [INFO] Worker exiting (pid: 44068)
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/opt/airflow/airflow/__main__.py", line 58, in main
    args.func(args)
  File "/opt/airflow/airflow/cli/cli_config.py", line 49, in command
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/utils/cli.py", line 115, in wrapper
    return f(*args, **kwargs)
  File "/opt/airflow/airflow/utils/providers_configuration_loader.py", line 55, in wrapped_function
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/cli/commands/scheduler_command.py", line 58, in scheduler
    run_command_with_daemon_option(
  File "/opt/airflow/airflow/cli/commands/daemon_utils.py", line 85, in run_command_with_daemon_option
    callback()
  File "/opt/airflow/airflow/cli/commands/scheduler_command.py", line 61, in <lambda>
    callback=lambda: _run_scheduler_job(args),
  File "/opt/airflow/airflow/cli/commands/scheduler_command.py", line 49, in _run_scheduler_job
    run_job(job=job_runner.job, execute_callable=job_runner._execute)
  File "/opt/airflow/airflow/utils/session.py", line 84, in wrapper
    return func(*args, session=session, **kwargs)
  File "/opt/airflow/airflow/jobs/job.py", line 410, in run_job
    return execute_job(job, execute_callable=execute_callable)
  File "/opt/airflow/airflow/jobs/job.py", line 439, in execute_job
    ret = execute_callable()
  File "/opt/airflow/airflow/jobs/scheduler_job_runner.py", line 856, in _execute
    self._run_scheduler_loop()
  File "/opt/airflow/airflow/jobs/scheduler_job_runner.py", line 993, in _run_scheduler_loop
    num_finished_events = self._process_executor_events(session=session)
  File "/opt/airflow/airflow/jobs/scheduler_job_runner.py", line 800, in _process_executor_events
    ti.handle_failure(error=msg % (ti, state, ti.state, info), session=session)
  File "/opt/airflow/airflow/utils/session.py", line 81, in wrapper
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/models/taskinstance.py", line 3021, in handle_failure
    _handle_failure(
  File "/opt/airflow/airflow/models/taskinstance.py", line 928, in _handle_failure
    TaskInstance.save_to_db(failure_context["ti"], session)
  File "/opt/airflow/airflow/api_internal/internal_api_call.py", line 127, in wrapper
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/utils/session.py", line 81, in wrapper
    return func(*args, **kwargs)
  File "/opt/airflow/airflow/models/taskinstance.py", line 2994, in save_to_db
    session.flush()
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3449, in flush
    self._flush(objects)
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3589, in _flush
    transaction.rollback(_capture_exception=True)
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3549, in _flush
    flush_context.execute()
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 456, in execute
    rec.execute(self)
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 630, in execute
    util.preloaded.orm_persistence.save_obj(
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 237, in save_obj
    _emit_update_statements(
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 886, in _emit_update_statements
    for (
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 626, in _collect_update_commands
    state.manager[propkey].impl.is_equal(
  File "/usr/local/lib/python3.8/site-packages/sqlalchemy/sql/sqltypes.py", line 1885, in compare_values
    return x == y
RecursionError: maximum recursion depth exceeded in comparison

While execution is happening for basic_dag_triggerer, we are inserting new DAG run to the DB and when the scheduler is trying to read the config for basic_dag and start to execute, we are having this RecursionError.

Since the scheduler is failing with this DAG run, we cannot start the scheduler. It is because every time we try to start the scheduler, it will try to read the last not executed DAG run and try to execute it. If the DAG run does not have any timeout, then the scheduler cannot start at all (we just need to remove the DAG run from DB).

See the how to reproduce section, we give dagrun_timeout as 10 minutes
After this timeout time, the scheduler finally can be started.

How to reproduce

1. Save the following 2 DAGs.

• basic_dag.py

from datetime import timedelta

import airflow
import pendulum
from airflow.operators.bash import BashOperator

default_args = {
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with airflow.DAG(
    dag_id="basic_dag",
    default_args=default_args,
    start_date=pendulum.datetime(2024, 5, 14, tz="UTC"),
    catchup=False,
    schedule="* */1 * * *",
    tags=["basic"],
    max_active_runs=2,
    dagrun_timeout=timedelta(minutes=10),
):
    task = BashOperator(task_id="normal", bash_command="echo 'I am just a normal task'")

• basic_dag_triggerer.py

"""
Triggers a DAG with a config that has a circular reference.
"""

import pendulum
import uuid

from airflow.models.dag import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.api.common.trigger_dag import trigger_dag
from airflow.utils import timezone


def my_trigger():
    circular = {}
    circular["a"] = circular
    trigger_dag(
        dag_id="basic_dag",
        run_id=str(uuid.uuid4()),
        conf=circular,
        execution_date=timezone.utcnow(),
        replace_microseconds=False,
    )


with DAG(
    dag_id="basic_dag_triggerer",
    start_date=pendulum.datetime(2024, 5, 14, tz="UTC"),
    catchup=False,
    schedule="@once",
    tags=["basic"],
) as dag:
    trigger = PythonOperator(
        task_id="trigger_dagrun",
        python_callable=my_trigger,
    )

2. Open the UI and check they are imported and validate you can navigate the details page (without starting the DAGs) for both of the DAGs.
3. Start the basic_dag, and wait a little bit to see it is working correctly and you can also test the Run button
4. Run the basic_dag_triggerer and see if it is working correctly.
5. Go to the details page for basic_dag and you will see the error on the UI and also in the webserver logs.
6. Also, you can check the scheduler logs to see that it failed

Possible Solutions

For the page rendering side, we can include some try-except to give a meaningful message to the user. It is relatively easy to do but for the scheduler problem it is tricky.

We have two different opinions on how we can resolve this:

1. We can somehow¹ try to detect this circular configuration while importing the DAG in the DAG list
   This is like the error box for DAG imports: if there is a funky thing about DAG, we are currently flashing a message to the user that you have DAG import errors.
   We can give some kind of error to say you have a problem in your DAG and it cannot be run.

2. We can handle this serialization error inside of the scheduler or right before the execution we can try to serialize and see if it has some circular reference, if it has, than we can marked the DAG Run as failed (or some other status) to give the information to the user.

If you have any other ideas for this, please feel free to share the ideas.

Footnotes

1. To be able to do this, we might create a DAG Run on the fly while importing the DAG and try to serialize it. If we are seeing an error then it means we cannot execute this DAG because it might cause problems. ↩

[moiseenkov]

Thank you for the research and very detailed description! In my opinion we should implement validation for this use case and raise an appropriate exception so users could see the actual reason and fix the config. Perhaps, we could add validation here, but I'm not sure if it is a right place.

[VladaZakharova]

Hi @eladkal @potiuk @hussein-awala @Taragolis
Do you have some ideas regarding proposed solutions?

[Taragolis]

airflow.api.common.trigger_dag.trigger_dag it is not a part of Public Interface of Airflow so if it uses in user code and lead to some abnormal errors than it is fine from the contributors perspective

[molcay]

Hi @Taragolis, your comment makes sense. Since the trigger_dag is not a part of the Public interface, using it means taking risks.
Thank you for the answer!

I also have one follow-up:
When something bad happened somewhere (not only for the case above), and while trying to render the pages (like DAG details pages, Dag Run functionality, etc.), airflow shows not very user-friendly pages/warnings.
I think we can improve the error handling on the backend side to give more user-friendly errors on the UI.
For example; We are sending the traceback.html page for the following requests:

• GET /object/graph_data API endpoint:
  The error is happening on this line while encoding the dag run (which has faulty dag run configuration).
• POST /dags/<string:dag_id>/trigger endpoint:
  The error is happening on these lines while serializing the run configs (which has faulty dag run configuration)

We can add error handling while creating the data, and if we encounter an error then we can flash some error on the UI side and give a bit more user-friendly errors/warnings to the user (like in this line). WDYT?