Dynamically generated schedule interval?

276 views Asked by At

I create a DAG to do monitoring of other applications, when the DAG does not fulfill the monitoring task, it sends an alert email. My DAG has a ten minutes interval scheduler, that is, if the task fails, it sends emails every ten minutes, but what I want to achieve is to send me the first email at ten minutes, the second email at fifteen minutes, the third email at thirty minutes and continue as long as the condition of the previous task is failed, if the task is successful continue monitoring every ten minutes.


current_date = datetime.now()
default_args = {
    'owner': 'my_dag',
    'start_date':  datetime(2021, 3, 25, current_date.hour),
    'timeout': 300,  --> seconds
    'poke_interval': 20, --> seconds
    'exponential_backoff': True
    }


with DAG(
    dag_id="Health_Check",
    schedule_interval="*/10 8-19 * * *", # At every 10th minutes past every hour from 8 through 20.
    tags=["Health_Check"],
    default_args=default_args,
    catchup=False
        ) as dag

    do_request = CustomSensor(
        task_id='do_request',
        endpoint='/Url',
        method='GET',
        http_conn_id='conn_api'
        )

I try implement exponential retracement at the DAG level because in the tasks I use exponential_backoff this attribute of the sensor operator so that when it tries to generate a request to a URL it tries it the first time if it fails it tries 20 seconds later if it fails it tries 40s later if it fails 80s after and continue in this way until it falls on time out.

I would like to implement the same, but at the DAG level, any idea??

0

There are 0 answers