[TWTR][[AIRFLOW-4939]] Add Default Retries and fix a small DAG refresh bug (twitter-forks#3)

Author: msumit

airflow/config_templates/default_airflow.cfg

@@ -173,6 +173,9 @@ killed_task_cleanup_time = 60
 # `airflow trigger_dag -c`, the key-value pairs will override the existing ones in params.
 dag_run_conf_overrides_params = False
 
+# The number of retries each task is going to have by default. Can be overridden at dag or task level.
+default_task_retries = 0
+
 [cli]
 # In what way should the cli access the API. The LocalClient will use the
 # database directly, while the json_client will use the api running on the

airflow/config_templates/default_test.cfg

@@ -51,6 +51,7 @@ enable_xcom_pickling = False
 killed_task_cleanup_time = 5
 secure_mode = False
 hostname_callable = socket:getfqdn
+default_task_retries = 0
 
 [cli]
 api_client = airflow.api.client.local_client

airflow/models.py

@@ -297,9 +297,12 @@ def get_dag(self, dag_id):
                     dag.last_loaded < orm_dag.last_expired
                 )
         ):
+
             # Reprocess source file
+            # TODO: remove the below hack to find relative dag location in webserver
+            filepath = dag.fileloc if dag else orm_dag.fileloc
             found_dags = self.process_file(
-                filepath=orm_dag.fileloc, only_if_updated=False)
+                filepath=filepath, only_if_updated=False)
 
             # If the source file no longer exports `dag_id`, delete it from self.dags
             if found_dags and dag_id in [found_dag.dag_id for found_dag in found_dags]:
@@ -2342,7 +2345,7 @@ def __init__(
             email=None,
             email_on_retry=True,
             email_on_failure=True,
-            retries=0,
+            retries=None,
             retry_delay=timedelta(seconds=300),
             retry_exponential_backoff=False,
             max_retry_delay=None,
@@ -2416,7 +2419,8 @@ def __init__(
                 self
             )
         self._schedule_interval = schedule_interval
-        self.retries = retries
+        self.retries = retries if retries is not None else \
+            configuration.conf.getint('core', 'default_task_retries', fallback=0)
         self.queue = queue
         self.pool = pool
         self.sla = sla