Skip to content

The scheduler does not appear to be running. Last heartbeat was received X minutes ago.  #19192

Closed

Description

@t4n1o

Apache Airflow version

2.1.4

Operating System

Linux / Ubuntu Server

Versions of Apache Airflow Providers

apache-airflow-providers-ftp==2.0.1
apache-airflow-providers-http==2.0.1
apache-airflow-providers-imap==2.0.1
apache-airflow-providers-postgres==2.3.0

Deployment

Virtualenv installation

Deployment details

Airflow v2.1.4
Postgres 14
LocalExecutor
Installed with Virtualenv / ansible - https://github.com/idealista/airflow-role

What happened

image

I run a single BashOperator (for a long running task, we have to download data for 8+ hours initially to download from the rate-limited data source API, then download more each day in small increments).

We're only using 3% CPU and 2 GB of memory (out of 64 GB) but the scheduler is unable to run any other simple task at the same time.

Currently only the long task is running, everything else is queued, even thought we have more resources:
image

What you expected to happen

I expect my long running BashOperator task to run, but for airflow to have the resources to run other tasks without getting blocked like this.

How to reproduce

I run a command with bashoperator (I use it because I have python, C, and rust programs being scheduled by airflow).
bash_command='umask 002 && cd /opt/my_code/ && /opt/my_code/venv/bin/python -m path.to.my.python.namespace'

Configuration:

airflow_executor: LocalExecutor
airflow_database_conn: 'postgresql+psycopg2://airflow:airflow_pass@localhost:5432/airflow'
airflow_database_engine_encoding: utf-8
airflow_database_engine_collation_for_ids:
airflow_database_pool_enabled: True
airflow_database_pool_size: 3
airflow_database_max_overflow: 10
airflow_database_pool_recycle: 2000
airflow_database_pool_pre_ping: True
airflow_database_schema:
airflow_database_connect_args:
airflow_parallelism: 10
airflow_dag_concurrency: 7
airflow_dags_are_paused_at_creation: True
airflow_max_active_runs_per_dag: 16
airflow_load_examples: False
airflow_load_default_connections: False
airflow_plugins_folder: "{{ airflow_app_home }}/plugins"

# [operators]
airflow_operator_default_owner: airflow
airflow_operator_default_cpus: 1
airflow_operator_default_ram: 512
airflow_operator_default_disk: 512
airflow_operator_default_gpus: 0
airflow_default_queue: default
airflow_allow_illegal_arguments: False

Anything else

This occurs every time consistently, also on 2.1.2

The other tasks have this state:
image

When the long-running task finishes, the other tasks resume normally. But I expect to be able to do some parallel execution /w LocalExecutor.

I haven't tried using pgbouncer.

Are you willing to submit PR?

  • Yes I am willing to submit a PR!

Code of Conduct

Activity

  1. potiuk commented on Oct 28, 2021

    @potiuk
    Member

    It looks like you are actually using SequentialExecutor. That would perfectly explain the behaviour. Are you sure you are using local executor and scheduler is running ? Can you run airflow info and verify it / paste the results here?

  2. t4n1o commented on Oct 28, 2021

    @t4n1o
    Author

    LocalExecutor:
    image

    In this screenshot the scheduler is running 4 of the same process / task, because max_active_runs was not set (I subsequently set it to 1, because that's the behaviour I want).

    As stated above, the issue is airflow will not run other dags and the scheduler is not responding. (Strangely, the scheduler is apparently also quite happy to run 1 task from 1 dag in 4 parallel processes.)

    I suspect some value in the configuration or not enough database connections.

  3. t4n1o commented on Oct 28, 2021

    @t4n1o
    Author
    
    Apache Airflow
    version                | 2.1.4                                                            
    executor               | LocalExecutor                                                    
    task_logging_handler   | airflow.utils.log.file_task_handler.FileTaskHandler              
    sql_alchemy_conn       | postgresql+psycopg2://airflow:airflow_pass@localhost:5432/airflow
    dags_folder            | /opt/data_workflows/dags                     
    plugins_folder         | /opt/airflow/plugins                                             
    base_log_folder        | /var/log/airflow                                                 
    remote_base_log_folder |                                                                  
                                                                                              
    
    System info
    OS              | Linux                                                                                                                                                                  
    architecture    | x86_64                                                                                                                                                                 
    uname           | uname_result(system='Linux', node='Ubuntu-2004-focal-64-minimal', release='5.4.0-89-generic', version='#100-Ubuntu SMP Fri Sep 24 14:50:10 UTC 2021', machine='x86_64')
    locale          | ('en_US', 'UTF-8')                                                                                                                                                     
    python_version  | 3.9.5 (default, May 19 2021, 11:32:47)  [GCC 9.3.0]                                                                                                                    
    python_location | /opt/airflow/bin/python                                                                                                                                                
                                                                                                                                                                                             
    
    Tools info
    git             | git version 2.25.1                                                                         
    ssh             | OpenSSH_8.2p1 Ubuntu-4ubuntu0.3, OpenSSL 1.1.1f  31 Mar 2020                               
    kubectl         | NOT AVAILABLE                                                                              
    gcloud          | NOT AVAILABLE                                                                              
    cloud_sql_proxy | NOT AVAILABLE                                                                              
    mysql           | NOT AVAILABLE                                                                              
    sqlite3         | 3.31.1 2020-01-27 19:55:54 3bfa9cc97da10598521b342961df8f5f68c7388fa117345eeb516eaa837balt1
    psql            | psql (PostgreSQL) 14.0 (Ubuntu 14.0-1.pgdg20.04+1)                                         
                                                                                                                 
    
    Paths info
    airflow_home    | /opt/airflow                                                                                                                                                                                     
    system_path     | /usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games:/snap/bin                                                                                               
    python_path     | /opt/airflow:/usr/lib/python39.zip:/usr/lib/python3.9:/usr/lib/python3.9/lib-dynload:/opt/airflow/lib/python3.9/site-packages:/opt/data_workflows/dags:/opt/airflow/config:/o
                    | pt/airflow/plugins                                                                                                                                                                               
    airflow_on_path | False                                                                                                                                                                                            
                                                                                                                                                                                                                       
    
    Providers info
    apache-airflow-providers-ftp      | 2.0.1
    apache-airflow-providers-http     | 2.0.1
    apache-airflow-providers-imap     | 2.0.1
    apache-airflow-providers-postgres | 2.3.0
    apache-airflow-providers-sqlite   | 2.0.1
    
  4. potiuk commented on Oct 28, 2021

    @potiuk
    Member

    It does not look like like Airflow problem to be honest.

    It certainly looks like your long running task blocks some resources that blocks scheduler somehow (but there is no indication how it can be blocked). There must be something in your DAGs or task that simply causes the Airflow scheduler to lock up.

    This is almost certainly something specific to your deployment (others do not experience it). But what it is, it's hard to say from the information provided.

    My best guess is that you have some lock on the database and it makes scheduler wait while running some query.

    Is it possible to dump the state of scheduler and dump generaly more of the state of your machine, resouces, sockets, DB locks while it happens (this shoudl be possible with py-spy for example). Also getting all logs of scheduler and seing if it actually "does" something might help. Unfortunately the information we have now is not enough to deduce the reason.

    Any insight of WHERE the schduler is locked might help with investigating it.

  5. potiuk commented on Oct 28, 2021

    @potiuk
    Member

    BTW. Yeah, checking the limits of connections opened in your DB might be a good idea. Especially if you are using variables in your DAGs at top level, it MIGHT lead to a significant number of connections opened, which *might eventually cause scheduler to try to open a new connection and patiently wait until the DB server will have any connection free. It might simply be that your long running tasks are written in the way that they (accidentally) open a lot of those connections and do not close them until the task completes.

    I think PGBouncer might help with that, but if too many connections are opened by a single long running task and they are not closed, that might also not help too much.

  6. t4n1o commented on Jan 2, 2022

    @t4n1o
    Author

    Well, there is nothing locking the db. This issue occurs even if there is no database being used by my application.

    Types of tasks that cause this problem:

    • download all the archives from public.bitmex.com via a python script (internet speed is the bottleneck) (takes about 3 hours the first run)
    • decompress csv.gz files into csv files (disk speed is the bottleneck) (takes about 4 hours to run the first time)
    • read csv records for each day and transform them with a custom rust parsing tool

    Any program written in rust or python that takes a long time to execute will cause this problem. We are using airflow because once we sync all the historical data, we run the task once per day each new day.

    image

    Here is a dump of what the scheduler is doing, while it's stuck.

    image

    State of the various airflow processes:
    image

    I am starting the rust/python programs in a separate process with BashOperator, and it's stuck on _recv(). Since all the tasks are limited by a rate-limit of some API, disk speed, or network speed, it would be better if airflow could actually run more than 1 task at a time. Any ideas?

  7. t4n1o commented on Jan 2, 2022

    @t4n1o
    Author

    image

  8. potiuk commented on Jan 2, 2022

    @potiuk
    Member

    I think you are misusing Airlfow. Airlfow by definition can run multiple tasks even with single local executor, so you are likely misunderstanding how airflow operators are run and run it as a top-level code, and not as task?

    Can you please copy your DAG here?

    You are not supposed to run long running operations when the DAG file is parsed - parsing should be run rather quickly and from what it looks like you execute a long running process while parsing happens rather than when tasks are executed:

    https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html#top-level-python-code

  9. t4n1o commented on Jan 2, 2022

    @t4n1o
    Author
    from datetime import timedelta
    
    from airflow import DAG
    from airflow.operators.bash import BashOperator
    from airflow.utils.dates import days_ago
    
    
    default_args = {
        "owner": "t4n1o",
        "depends_on_past": False,
        "email": ["myemail@gmail.com"],
        "email_on_failure": True,
        "email_on_retry": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=2),
        "max_active_runs_per_dag": 1,
    
    }
    with DAG(
        "Bitmex_Archives_Mirror",
        default_args=default_args,
        description="Mirror the archives from public.bitmex.com ",
        schedule_interval=timedelta(days=1),
        start_date=days_ago(2),
        tags=["raw price history"],
        catchup=False,
    ) as dag:
    
        t1 = BashOperator(
            task_id="download_csv_gz_archives",
            bash_command="umask 002 && cd /opt/soft/ &&"
            "/opt/soft/venv/bin/python -m soft.data_workflows.bitmex.archives_mirror",
        )
    
        t2 = BashOperator(
            task_id="process_archives_into_daily_csv_files",
            depends_on_past=False,
            bash_command="umask 002 && cd /opt/executables/bitmex_csvgz_parser && ./parser",
            retries=3,
        )
    
    
        t1 >> t2
    
    
  10. t4n1o commented on Jan 2, 2022

    @t4n1o
    Author

    I'm not sure how airflow is intended to be used, but sometimes people find other use cases for a tool they haven't designed.

    We run a task that can take a few hours to collect all the historical data and process it. And then we want the task to run once per day.

    It appears, from my side, that the airflow server UI can't contact the scheduler while the long task is running, and other DAGs can't be run. Perhaps the scheduler wants my code to yield control back to it frequently (once per day of data, for example), but I prefer to let my own code manage the date ranges, because that's where the unit tests are, and all the heavy lifting is in rust anyway.

  11. t4n1o commented on Jan 2, 2022

    @t4n1o
    Author

    What I do find use for in airflow

    • keeping track of tasks
    • sending emails when a task fails
    • running the tasks every day like cron + a nice ui
  12. potiuk commented on Jan 2, 2022

    @potiuk
    Member

    I'm not sure how airflow is intended to be used, but sometimes people find other use cases for a tool they haven't designed.

    We run a task that can take a few hours to collect all the historical data and process it. And then we want the task to run once per day.

    This is what airflow is designed for. I tihnk you just use it wrongly (or misconfigured it). It is supposed to handle that case perfectly (and it works this way for thousands of users. So it's your configuration/setup/way of using it is wrong.

    It appears, from my side, that the airflow server UI can't contact the scheduler while the long task is running, and other DAGs can't be run. Perhaps the scheduler wants my code to yield control back to it frequently (once per day of data, for example), but I prefer to let my own code manage the date ranges, because that's where the unit tests are, and all the heavy lifting is in rust anyway.

    No. This is no the case (unless you use Sequential Executor which is only suposed to be used for debugging). . Airflow is designed to run multiple paralllel tass at a time:. You likely have some problem in your airlfow installation/configuration

    Questions:

    1. Do you actually have scheduler running at all? Does it have coninuous access to the DB?

    2. Are you absolutely sure you are not using SequentialExecutior ? What does your airflow info say - can you paste-bin output of it ? (airlfow has built-in flag to send to pastebin). Please make sure also that you do it in exactly the way your scheduler works. Miost likely you run your scheduler with a different configuration than your webserver and that causes the problem.

    3. Are you sure you are using Postgres and not Sqlite? What does your airflow info say?

    4. Where is your Python code (non-DAG)? Did you .airflowignore non-DAG files from airflow's DAG folder?

    5. can you upgrade to Airlfow 2.2.3 (latest released) - it has built-in warnings in case you use Sequential Executor/SQLite in the UI.

    6. Can you change your DAGs to:

    default_args = {
        "owner": "t4n1o",
        "depends_on_past": False,
        "email": ["myemail@gmail.com"],
        "email_on_failure": True,
        "email_on_retry": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=2),
        "max_active_runs_per_dag": 1,
    
    }
    with DAG(
        "Bitmex_Archives_Mirror",
        default_args=default_args,
        description="Mirror the archives from public.bitmex.com ",
        schedule_interval=timedelta(days=1),
        start_date=days_ago(2),
        tags=["raw price history"],
        catchup=False,
    ) as dag:
    
        t1 = BashOperator(
            task_id="download_csv_gz_archives",
            bash_command="sleep 1000",
        )
    
        t2 = BashOperator(
            task_id="process_archives_into_daily_csv_files",
            depends_on_past=False,
            bash_command="sleep 1000",
            retries=3,
        )
    
        t1 >> t2
    

    I just run it in 2.2.3 and I was able to successuly start even 5 paralllel runs and no problems with Scheduler

    Screenshot from 2022-01-02 19-57-18

  13. t4n1o commented on Jan 2, 2022

    @t4n1o
    Author
    Apache Airflow
    version                | 2.2.3                                                            
    executor               | LocalExecutor                                                    
    task_logging_handler   | airflow.utils.log.file_task_handler.FileTaskHandler              
    sql_alchemy_conn       | postgresql+psycopg2://airflow:airflow_pass@localhost:5432/airflow
    dags_folder            | /opt/data_workflows/dags                     
    plugins_folder         | /opt/airflow/plugins                                             
    base_log_folder        | /var/log/airflow                                                 
    remote_base_log_folder |                                                                  
     
     
    System info
    OS              | Linux                                                                                                                                           
    architecture    | x86_64                                                                                                                                          
    uname           | uname_result(system='Linux', node='staging', release='5.10.0-8-amd64', version='#1 SMP Debian 5.10.46-4 (2021-08-03)', machine='x86_64')
    locale          | ('en_US', 'UTF-8')                                                                                                                              
    python_version  | 3.9.2 (default, Feb 28 2021, 17:03:44)  [GCC 10.2.1 20210110]                                                                                   
    python_location | /opt/airflow/bin/python                                                                                                                         
     
     
    Tools info
    git             | git version 2.30.2                                                                         
    ssh             | OpenSSH_8.4p1 Debian-5, OpenSSL 1.1.1k  25 Mar 2021                                        
    kubectl         | NOT AVAILABLE                                                                              
    gcloud          | NOT AVAILABLE                                                                              
    cloud_sql_proxy | NOT AVAILABLE                                                                              
    mysql           | NOT AVAILABLE                                                                              
    sqlite3         | 3.34.1 2021-01-20 14:10:07 10e20c0b43500cfb9bbc0eaa061c57514f715d87238f4d835880cd846b9ealt1
    psql            | psql (PostgreSQL) 13.4 (Debian 13.4-0+deb11u1)                                             
     
     
    Paths info
    airflow_home    | /opt/airflow                                                                                                                                                                                     
    system_path     | /usr/local/bin:/usr/bin:/bin:/usr/local/games:/usr/games:/opt/airflow/bin:/usr/lib/postgresql/13/bin                                                                                             
    python_path     | /opt/airflow/bin:/usr/lib/python39.zip:/usr/lib/python3.9:/usr/lib/python3.9/lib-dynload:/opt/airflow/lib/python3.9/site-packages:/opt/data_workflows/dags:/opt/airflow/confi
                    | g:/opt/airflow/plugins                                                                                                                                                                           
    airflow_on_path | True                                                                                                                                                                                             
     
     
    Providers info
    apache-airflow-providers-ftp      | 2.0.1
    apache-airflow-providers-http     | 2.0.1
    apache-airflow-providers-imap     | 2.0.1
    apache-airflow-providers-postgres | 2.3.0
    apache-airflow-providers-sqlite   | 2.0.1
    
  14. 26 remaining items

  15. potiuk commented on Aug 17, 2022

    @potiuk
    Member

    Do you remember which one @uranusjr ?

  16. potiuk commented on Aug 17, 2022

    @potiuk
    Member

    This one ? #25532

  17. deepchand commented on Aug 18, 2022

    @deepchand
    Contributor

    I have tried to debug it more and found its self.executor.heartbeat() function which is taking almost 95% of total time of scheduling loop

  18. uranusjr commented on Aug 18, 2022

    @uranusjr
    Member

    Yes that one

  19. deepchand commented on Aug 23, 2022

    @deepchand
    Contributor

    @potiuk i have debugged more on this issue and found like updating the task state in the scheduler loop significantly increase the total time of self.executor.heartbeat() function which is causing this problem "The scheduler does not appear to be running. Last heartbeat was received X minutes ago"
    To be specific

    self.update_task_state(key, result.state, getattr(result, 'info', None))
    the state which we are passing in this function is kind of blocking and if we stop updating task state from here significantly decrease the scheduling loop time and also not giving this warning "The scheduler does not appear to be running. Last heartbeat was received X minutes ago"
    Please have a look and let me know if any other info is needed

  20. potiuk commented on Aug 26, 2022

    @potiuk
    Member

    I do not know that well, but for me it looks like you have some bottlences with your Celery Queu. You have not specified what kind of queue you used - but I think you should look at your Redis or RabbitMQ and see if there are no problems there. Also it might be that simply your Redis or RabbitMQ is badly configured or overloaded and it is somehow blocking state update.

    Can you please set log level to debug and see if there are more information printed on what's going on in update_state method?

    (you will find how to do it in airlfow configuration docs)

  21. deepchand commented on Aug 26, 2022

    @deepchand
    Contributor

    @potiuk We are using Redis as queue, i have cross verified redis is not bottleneck and messages are consumed from redis as soon as they land in redis queue. We have enough resources available on redis side as well.
    Screenshot 2022-08-26 at 8 23 59 PM
    Screenshot 2022-08-26 at 8 25 06 PM

  22. potiuk commented on Aug 26, 2022

    @potiuk
    Member

    Any chance for debug logs ?

  23. deepchand commented on Aug 26, 2022

    @deepchand
    Contributor

    Debug logs of scheduler ?

  24. potiuk commented on Aug 26, 2022

    @potiuk
    Member

    Yep:

    Can you please set log level to debug and see if there are more information printed on what's going on in update_state method?

    (you will find how to do it in airlfow configuration docs)

  25. deepchand commented on Aug 26, 2022

    @deepchand
    Contributor

    @potiuk I have added some more debug logs to find the total time taken while fetching state of a single result and for total results in a single loop, please find below
    Screenshot 2022-08-26 at 10 04 51 PM
    Screenshot 2022-08-26 at 10 05 52 PM

  26. deepchand commented on Aug 26, 2022

    @deepchand
    Contributor

    Too much debug logs so tried to filter out only important one, let me know if more info needed

  27. potiuk commented on Aug 26, 2022

    @potiuk
    Member

    I think your problem is simply extremely slow connection to your database. 5 seconds to run single query indicate a HUGE problem you have with your database. It should take single mlilliseconds .

    THIS IS your problem. Not airflow.

    You should fix your DB/connectivity and debug why your database is 1000x slower than it should.

  28. locked and limited conversation to collaborators on Aug 26, 2022
  29. converted this issue into a discussion #25991 on Aug 26, 2022
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions