[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=16639442#comment-16639442
]
Iuliia Volkova commented on AIRFLOW-1013:
-
[~ashb], [~sanand], [~Tagar], Is it fixed? Or it's not needed? (based on what
already 1.10 exist)
> 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)
[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:
us...@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)
[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=15999665#comment-15999665
]
Ruslan Dautkhanov commented on AIRFLOW-1013:
https://github.com/apache/incubator-airflow/pull/2203
> 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: Airflow 1.8, 1.8.1, 1.8.0
>Reporter: Ruslan Dautkhanov
>Assignee: Siddharth Anand
>Priority: Critical
> Labels: dagrun, once, scheduler, sla
> Fix For: 1.8.1
>
>
> 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
(v6.3.15#6346)
[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=15948274#comment-15948274
]
Siddharth Anand commented on AIRFLOW-1013:
--
[~Tagar]
Do you mind providing a sample DAG that I can test with.
BTW, in trying to reproduce this problem, I have run into multiple places in
the jobs.py code where '@once' no longer seems to work. Since I don't use
"@once" in any staging or production workflows, I have not run into this issue
during the 1.8 upgrade. I suspect this is actually more broken than first
reported.
Are you running on the 1.8 released version or off one of the release
candidates?
> 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: Airflow 1.8
>Reporter: Ruslan Dautkhanov
>Assignee: Siddharth Anand
>Priority: Blocker
> Labels: dagrun, once, scheduler, sla
> Fix For: 1.8.1
>
>
> 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
(v6.3.15#6346)
[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=15932080#comment-15932080
]
Ruslan Dautkhanov commented on AIRFLOW-1013:
This error exceptions goes away after we unset default_args's SLA parameter.
SLAs still should work for tasks in a `@once` dag.
> 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: Airflow 1.8
>Reporter: Ruslan Dautkhanov
> Labels: dagrun, once, scheduler, sla
>
> 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
(v6.3.15#6346)
