<p dir="auto">The idea is to have a Gantt graph for the DAGs in the UI, so that you can see at what hours there are more overlaps, when you have a gap without any dag running, which ones are the most time consuming etc.</p>
<p dir="auto">I think it could be done by taking the average execution time of the dag in the last X days from the metadata database or something like that. Also, for further precision, an outlier processing could be done as well.</p>
<p dir="auto">From my point of view, is a very nice feature to be added to the new UI</p>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.2.5 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">Customer has a dag that generates around 2500 tasks dynamically using a task group. While running the dag, a subset of the tasks (~1000) run successfully with no issue and (~1500) of the tasks are getting "skipped", and the dag fails. The same DAG runs successfully in Airflow v2.1.3 with same Airflow configuration.</p>
<p dir="auto">While investigating the Airflow processes, We found that both the scheduler got restarted with below error during the DAG execution.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2022-04-27 20:42:44,347] {scheduler_job.py:742} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1256, in _execute_context
    self.dialect.do_executemany(
  File &quot;/usr/local/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py&quot;, line 912, in do_executemany
    cursor.executemany(statement, parameters)
psycopg2.errors.DeadlockDetected: deadlock detected
DETAIL:  Process 1646244 waits for ShareLock on transaction 3915993452; blocked by process 1640692.
Process 1640692 waits for ShareLock on transaction 3915992745; blocked by process 1646244.
HINT:  See server log for query details.
CONTEXT:  while updating tuple (189873,4) in relation &quot;task_instance&quot;"><pre class="notranslate"><code class="notranslate">[2022-04-27 20:42:44,347] {scheduler_job.py:742} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1256, in _execute_context
    self.dialect.do_executemany(
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 912, in do_executemany
    cursor.executemany(statement, parameters)
psycopg2.errors.DeadlockDetected: deadlock detected
DETAIL:  Process 1646244 waits for ShareLock on transaction 3915993452; blocked by process 1640692.
Process 1640692 waits for ShareLock on transaction 3915992745; blocked by process 1646244.
HINT:  See server log for query details.
CONTEXT:  while updating tuple (189873,4) in relation "task_instance"
</code></pre></div>
<p dir="auto">This issue seems to be related to <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1069154380" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/19957" data-hovercard-type="issue" data-hovercard-url="/apache/airflow/issues/19957/hovercard" href="https://github.com/apache/airflow/issues/19957">#19957</a></p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">This issue was observed while running huge number of concurrent task created dynamically by a DAG. Some of the tasks are getting skipped due to restart of scheduler with Deadlock exception.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">DAG file:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from propmix_listings_details import BUCKET, ZIPS_FOLDER, CITIES_ZIP_COL_NAME, DETAILS_DEV_LIMIT, DETAILS_RETRY, DETAILS_CONCURRENCY, get_api_token, get_values, process_listing_ids_based_zip
from airflow.utils.task_group import TaskGroup
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
}

date = '{{ execution_date }}'
email_to = ['example@airflow.com']
# Using a DAG context manager, you don't have to specify the dag property of each task

state = 'Maha'
with DAG('listings_details_generator_{0}'.format(state),
        start_date=datetime(2021, 11, 18),
        schedule_interval=None,
        max_active_runs=1,
        concurrency=DETAILS_CONCURRENCY,
        dagrun_timeout=timedelta(minutes=10),
        catchup=False  # enable if you don't want historical dag runs to run
        ) as dag:
    t0 = DummyOperator(task_id='start')

    with TaskGroup(group_id='group_1') as tg1:
        token = get_api_token()
        zip_list = get_values(BUCKET, ZIPS_FOLDER+state, CITIES_ZIP_COL_NAME)
        for zip in zip_list[0:DETAILS_DEV_LIMIT]:
            details_operator = PythonOperator(
                task_id='details_{0}_{1}'.format(state, zip),  # task id is generated dynamically
                pool='pm_details_pool',
                python_callable=process_listing_ids_based_zip,
                task_concurrency=40,
                retries=3,
                retry_delay=timedelta(seconds=10),
                op_kwargs={'zip': zip, 'date': date, 'token':token, 'state':state}
            )
            
    t0 &gt;&gt; tg1"><pre class="notranslate"><code class="notranslate">from propmix_listings_details import BUCKET, ZIPS_FOLDER, CITIES_ZIP_COL_NAME, DETAILS_DEV_LIMIT, DETAILS_RETRY, DETAILS_CONCURRENCY, get_api_token, get_values, process_listing_ids_based_zip
from airflow.utils.task_group import TaskGroup
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
}

date = '{{ execution_date }}'
email_to = ['example@airflow.com']
# Using a DAG context manager, you don't have to specify the dag property of each task

state = 'Maha'
with DAG('listings_details_generator_{0}'.format(state),
        start_date=datetime(2021, 11, 18),
        schedule_interval=None,
        max_active_runs=1,
        concurrency=DETAILS_CONCURRENCY,
        dagrun_timeout=timedelta(minutes=10),
        catchup=False  # enable if you don't want historical dag runs to run
        ) as dag:
    t0 = DummyOperator(task_id='start')

    with TaskGroup(group_id='group_1') as tg1:
        token = get_api_token()
        zip_list = get_values(BUCKET, ZIPS_FOLDER+state, CITIES_ZIP_COL_NAME)
        for zip in zip_list[0:DETAILS_DEV_LIMIT]:
            details_operator = PythonOperator(
                task_id='details_{0}_{1}'.format(state, zip),  # task id is generated dynamically
                pool='pm_details_pool',
                python_callable=process_listing_ids_based_zip,
                task_concurrency=40,
                retries=3,
                retry_delay=timedelta(seconds=10),
                op_kwargs={'zip': zip, 'date': date, 'token':token, 'state':state}
            )
            
    t0 &gt;&gt; tg1
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">kubernetes cluster running on GCP  linux (amd64)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">pip freeze | grep apache-airflow-providers</p>
<p dir="auto">apache-airflow-providers-amazon==1!3.2.0<br>
apache-airflow-providers-cncf-kubernetes==1!3.0.0<br>
apache-airflow-providers-elasticsearch==1!2.2.0<br>
apache-airflow-providers-ftp==1!2.1.2<br>
apache-airflow-providers-google==1!6.7.0<br>
apache-airflow-providers-http==1!2.1.2<br>
apache-airflow-providers-imap==1!2.2.3<br>
apache-airflow-providers-microsoft-azure==1!3.7.2<br>
apache-airflow-providers-mysql==1!2.2.3<br>
apache-airflow-providers-postgres==1!4.1.0<br>
apache-airflow-providers-redis==1!2.0.4<br>
apache-airflow-providers-slack==1!4.2.3<br>
apache-airflow-providers-snowflake==2.6.0<br>
apache-airflow-providers-sqlite==1!2.1.3<br>
apache-airflow-providers-ssh==1!2.4.3</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Astronomer</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Airflow v2.2.5-2<br>
Scheduler count: 2<br>
Scheduler resources: 20AU (2CPU and 7.5GB)<br>
Executor used: Celery<br>
Worker count : 2<br>
Worker resources: 24AU (2.4 CPU and 9GB)<br>
Termination grace period : 2mins</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">This issue happens in all the dag runs. Some of the tasks are getting skipped and some are getting succeeded and the scheduler fails with the Deadlock exception error.</p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>