[WIP] handle sigterm gracefully by adding status to metadata

metaflow/task.py

@@ -4,6 +4,7 @@
 import sys
 import os
 import time
+import signal
 
 from types import MethodType, FunctionType
 
@@ -39,6 +40,42 @@
 MAX_FOREACH_PATH_LENGTH = 256
 
 
+class SystemSignalHandler:
+ def __init__(self, run_id, step_name, task_id, metadata_service, metadata_tags):
+ self.run_id = run_id
+ self.step_name = (step_name,)
+ self.task_id = task_id
+ self.metadata_service = metadata_service
+ self.metadata_tags = metadata_tags
+
+ signal.signal(signal.SIGINT, self.exit_sigint_gracefully)
+ signal.signal(signal.SIGTERM, self.exit_sigterm_gracefully)
+ print("SystemSignalHandler initialized...")
+
+ def log_exit_signal(self, signal):
+ metadata_info = [
+ MetaDatum(
+ field="signal",
+ value=signal,
+ type="signal",
+ tags=self.metadata_tags,
+ )
+ ]
+ self.metadata_service.register_metadata(
+ self.run_id, self.step_name, self.task_id, metadata_info
+ )
+
+ def exit_sigint_gracefully(self, signum, frame):
+ print("SIGINT received... ")
+ self.log_exit_signal("SIGINT")
+ self.status = signal.SIGINT
+
+ def exit_sigterm_gracefully(self, signum, frame):
+ print("SIGTERM received... ")
+ self.log_exit_signal("SIGTERM")
+ self.status = signal.SIGTERM
+
+
 class MetaflowTask(object):
  """
  MetaflowTask prepares a Flow instance for execution of a single step.
@@ -412,6 +449,7 @@ def run_step(
  )
 
  metadata_tags = ["attempt_id:{0}".format(retry_count)]
+ SystemSignalHandler(run_id, step_name, task_id, self.metadata, metadata_tags)
 
  metadata = [
  MetaDatum(