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

* [AIRFLOW-6730] Use total_seconds instead of seconds * adds tests and fixes types issue * fix test
apache · Feb 28, 2020 · 008b4ba · 008b4ba
1 parent ee16d30
commit 008b4ba
Show file tree

Hide file tree

Showing 7 changed files with 19 additions and 9 deletions.
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")