airflow-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (AIRFLOW-1013) airflow/jobs.py:manage_slas() exception for @once dag
Date Thu, 06 Sep 2018 18:50:00 GMT

    [ 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)

Mime
View raw message