[AIRFLOW-6730] Use total_seconds instead of seconds (#7363)

saguziel · web-flow · commit 008b4bab1422 · 2020-02-27T17:46:49.000-08:00
* [AIRFLOW-6730] Use total_seconds instead of seconds * adds tests and fixes types issue * fix test
diff --git a/airflow/jobs/base_job.py b/airflow/jobs/base_job.py
@@ -119,7 +119,7 @@ def is_alive(self, grace_multiplier=2.1):
         """
         return (
             self.state == State.RUNNING and
-            (timezone.utcnow() - self.latest_heartbeat).seconds < self.heartrate * grace_multiplier
+            (timezone.utcnow() - self.latest_heartbeat).total_seconds() < self.heartrate * grace_multiplier
         )
 
     @provide_session
diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
@@ -989,7 +989,7 @@ def is_alive(self, grace_multiplier=None):
         scheduler_health_check_threshold = conf.getint('scheduler', 'scheduler_health_check_threshold')
         return (
             self.state == State.RUNNING and
-            (timezone.utcnow() - self.latest_heartbeat).seconds < scheduler_health_check_threshold
+            (timezone.utcnow() - self.latest_heartbeat).total_seconds() < scheduler_health_check_threshold
         )
 
     @provide_session
diff --git a/airflow/kubernetes/pod_launcher.py b/airflow/kubernetes/pod_launcher.py
@@ -111,7 +111,7 @@ def run_pod(
         if resp.status.start_time is None:
             while self.pod_not_started(pod):
                 delta = dt.now() - curr_time
-                if delta.seconds >= startup_timeout:
+                if delta.total_seconds() >= startup_timeout:
                     raise AirflowException("Pod took too long to start")
                 time.sleep(1)
             self.log.debug('Pod not yet started')
diff --git a/airflow/providers/google/cloud/operators/dataproc.py b/airflow/providers/google/cloud/operators/dataproc.py
@@ -244,7 +244,7 @@ def _get_init_action_timeout(self):
                 return self.init_action_timeout
             elif match.group(2) == "m":
                 val = float(match.group(1))
-                return "{}s".format(timedelta(minutes=val).seconds)
+                return "{}s".format(int(timedelta(minutes=val).total_seconds()))
 
         raise AirflowException(
             "DataprocClusterCreateOperator init_action_timeout"
@@ -621,13 +621,13 @@ def _graceful_decommission_timeout_object(self) -> Optional[Dict]:
                 timeout = int(match.group(1))
             elif match.group(2) == "m":
                 val = float(match.group(1))
-                timeout = timedelta(minutes=val).seconds
+                timeout = int(timedelta(minutes=val).total_seconds())
             elif match.group(2) == "h":
                 val = float(match.group(1))
-                timeout = timedelta(hours=val).seconds
+                timeout = int(timedelta(hours=val).total_seconds())
             elif match.group(2) == "d":
                 val = float(match.group(1))
-                timeout = timedelta(days=val).seconds
+                timeout = int(timedelta(days=val).total_seconds())
 
         if not timeout:
             raise AirflowException(
diff --git a/airflow/providers/google/cloud/operators/postgres_to_gcs.py b/airflow/providers/google/cloud/operators/postgres_to_gcs.py
@@ -91,10 +91,10 @@ def convert_type(self, value, schema_type):
             return time.mktime(value.timetuple())
         if isinstance(value, datetime.time):
             formated_time = time.strptime(str(value), "%H:%M:%S")
-            return datetime.timedelta(
+            return int(datetime.timedelta(
                 hours=formated_time.tm_hour,
                 minutes=formated_time.tm_min,
-                seconds=formated_time.tm_sec).seconds
+                seconds=formated_time.tm_sec).total_seconds())
         if isinstance(value, dict):
             return json.dumps(value)
         if isinstance(value, Decimal):
diff --git a/tests/jobs/test_base_job.py b/tests/jobs/test_base_job.py
@@ -95,6 +95,11 @@ def test_is_alive(self):
         job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=21)
         self.assertFalse(job.is_alive())
 
+        # test because .seconds was used before instead of total_seconds
+        # internal repr of datetime is (days, seconds)
+        job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(days=1)
+        self.assertFalse(job.is_alive())
+
         job.state = State.SUCCESS
         job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=10)
         self.assertFalse(job.is_alive(), "Completed jobs even with recent heartbeat should not be alive")
diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py
@@ -1066,6 +1066,11 @@ def test_is_alive(self):
         job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=31)
         self.assertFalse(job.is_alive())
 
+        # test because .seconds was used before instead of total_seconds
+        # internal repr of datetime is (days, seconds)
+        job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(days=1)
+        self.assertFalse(job.is_alive())
+
         job.state = State.SUCCESS
         job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=10)
         self.assertFalse(job.is_alive(), "Completed jobs even with recent heartbeat should not be alive")

Original file line number	Diff line number	Diff line change
`@@ -119,7 +119,7 @@ def is_alive(self, grace_multiplier=2.1):`
`119`	`119`	`"""`
`120`	`120`	`return (`
`121`	`121`	`self.state == State.RUNNING and`
`122`		`- (timezone.utcnow() - self.latest_heartbeat).seconds < self.heartrate * grace_multiplier`
	`122`	`+ (timezone.utcnow() - self.latest_heartbeat).total_seconds() < self.heartrate * grace_multiplier`
`123`	`123`	`)`
`124`	`124`
`125`	`125`	`@provide_session`
Original file line number	Diff line number	Diff line change
`@@ -989,7 +989,7 @@ def is_alive(self, grace_multiplier=None):`
`989`	`989`	`scheduler_health_check_threshold = conf.getint('scheduler', 'scheduler_health_check_threshold')`
`990`	`990`	`return (`
`991`	`991`	`self.state == State.RUNNING and`
`992`		`- (timezone.utcnow() - self.latest_heartbeat).seconds < scheduler_health_check_threshold`
	`992`	`+ (timezone.utcnow() - self.latest_heartbeat).total_seconds() < scheduler_health_check_threshold`
`993`	`993`	`)`
`994`	`994`
`995`	`995`	`@provide_session`