You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ASF GitHub Bot (JIRA)" <ji...@apache.org> on 2018/09/06 18:50:00 UTC
[jira] [Commented] (AIRFLOW-1013) airflow/jobs.py:manage_slas()
exception for @once dag
[ https://issues.apache.org/jira/browse/AIRFLOW-1013?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16606255#comment-16606255 ]
ASF GitHub Bot commented on AIRFLOW-1013:
-----------------------------------------
r39132 closed pull request #2203: [AIRFLOW-1013] Skip manage_sla check for @once DAGs
URL: https://github.com/apache/incubator-airflow/pull/2203
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/airflow/jobs.py b/airflow/jobs.py
index 006a180ef5..b21c81daa6 100644
--- a/airflow/jobs.py
+++ b/airflow/jobs.py
@@ -592,6 +592,10 @@ def manage_slas(self, dag, session=None):
dttm = ti.execution_date
if task.sla:
dttm = dag.following_schedule(dttm)
+ # Skip tasks for @once dags
+ if not dttm and dag.schedule_interval == '@once':
+ self.logger.warn(' --------------> SLAs not supported in @once DAGs')
+ continue;
while dttm < datetime.now():
following_schedule = dag.following_schedule(dttm)
if following_schedule + task.sla < datetime.now():
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
> airflow/jobs.py:manage_slas() exception for @once dag
> -----------------------------------------------------
>
> Key: AIRFLOW-1013
> URL: https://issues.apache.org/jira/browse/AIRFLOW-1013
> Project: Apache Airflow
> Issue Type: Bug
> Affects Versions: 1.8.0, 1.8.1, 1.8.2
> Reporter: Ruslan Dautkhanov
> Assignee: Muhammad Ahmmad
> Priority: Critical
> Labels: dagrun, once, scheduler, sla
> Fix For: 1.10.0
>
>
> Getting following exception
> {noformat}
> [2017-03-19 20:16:25,786] {jobs.py:354} DagFileProcessor2638 ERROR - Got an exception! Propagating...
> Traceback (most recent call last):
> File "/opt/cloudera/parcels/Anaconda/lib/python2.7/site-packages/airflow/jobs.py", line 346, in helper
> pickle_dags)
> File "/opt/cloudera/parcels/Anaconda/lib/python2.7/site-packages/airflow/utils/db.py", line 53, in wrapper
> result = func(*args, **kwargs)
> File "/opt/cloudera/parcels/Anaconda/lib/python2.7/site-packages/airflow/jobs.py", line 1581, in process_file
> self._process_dags(dagbag, dags, ti_keys_to_schedule)
> File "/opt/cloudera/parcels/Anaconda/lib/python2.7/site-packages/airflow/jobs.py", line 1175, in _process_dags
> self.manage_slas(dag)
> File "/opt/cloudera/parcels/Anaconda/lib/python2.7/site-packages/airflow/utils/db.py", line 53, in wrapper
> result = func(*args, **kwargs)
> File "/opt/cloudera/parcels/Anaconda/lib/python2.7/site-packages/airflow/jobs.py", line 595, in manage_slas
> while dttm < datetime.now():
> TypeError: can't compare datetime.datetime to NoneType
> {noformat}
> Exception is in airflow/jobs.py:manage_slas() :
> https://github.com/apache/incubator-airflow/blob/v1-8-stable/airflow/jobs.py#L595
> {code}
> ts = datetime.now()
> SlaMiss = models.SlaMiss
> for ti in max_tis:
> task = dag.get_task(ti.task_id)
> dttm = ti.execution_date
> if task.sla:
> dttm = dag.following_schedule(dttm)
> >>> while dttm < datetime.now(): <<< here
> following_schedule = dag.following_schedule(dttm)
> if following_schedule + task.sla < datetime.now():
> session.merge(models.SlaMiss(
> task_id=ti.task_id,
> {code}
> It seems that dag.following_schedule() returns None for @once dag?
> Here's how dag is defined:
> {code}
> main_dag = DAG(
> dag_id = 'DISCOVER-Oracle-Load',
> default_args = default_args,
> user_defined_macros = dag_macros,
> start_date = datetime.now(),
> catchup = False,
> schedule_interval = '@once',
> concurrency = 2,
> max_active_runs = 1,
> dagrun_timeout = timedelta(days=4),
> )
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)