Turn TASK_GRACE_INTERVAL into a setting

The new setting is measured in seconds. Also added some clarification
around graceful shutdown.

fixes pulp#6242

Author: mdellweg

CHANGES/6242.feature

@@ -0,0 +1 @@
+Added `TASK_GRACE_INTERVAL` as a setting to specify the amount of seconds a worker gives the current task to finish when receiving SIGINT.

docs/admin/reference/settings.md

@@ -312,6 +312,16 @@ Defaults to `30` seconds.
 
 
 
+### TASK_GRACE_INTERVAL
+
+On receiving SIGHUP or SIGTERM a worker will await the currently running task forever.
+On SIGINT, this value represents the time before the worker will attempt to kill the subprocess.
+This time is only accurate to one worker heartbeat corresponding to `WORKER_TTL / 3`.
+
+Defaults to `600` seconds.
+
+
+
 ### REMOTE_USER_ENVIRON_NAME
 
 The name of the WSGI environment variable to read for `webserver authentication
@@ -417,8 +427,6 @@ up before the task finishes. Default is 1440 minutes (24 hours).
 
 
 
-
-
 ### UPLOAD_PROTECTION_TIME, TMPFILE_PROTECTION_TIME and TASK_PROTECTION_TIME
 
 Pulp uses `tasks`, `uploads` and `pulp temporary files` to pass data from the api to worker tasks.

pulpcore/app/settings.py

@@ -292,6 +292,11 @@
 CONTENT_APP_TTL = 30
 WORKER_TTL = 30
 
+# Seconds for a task to finish on semi graceful worker shutdown (approx)
+# On SIGHUP, SIGTERM the currently running task will be awaited forever.
+# On SIGINT, this value represents the time before the worker will attempt to kill the subprocess.
+TASK_GRACE_INTERVAL = 600
+
 # how long to protect ephemeral items in minutes
 ORPHAN_PROTECTION_TIME = 24 * 60
 

pulpcore/tasking/worker.py

@@ -45,9 +45,9 @@
 # Unless/until we can provide reasonable ways to decide to change their values,
 # they will live as constants instead of "proper" settings.
 
-# Number of heartbeats for a task to finish on graceful worker shutdown (approx)
-TASK_GRACE_INTERVAL = 3
-# Number of heartbeats between attempts to kill the subprocess (approx)
+# Seconds for a task to finish on semi graceful worker shutdown (approx)
+TASK_GRACE_INTERVAL = settings.TASK_GRACE_INTERVAL
+# Seconds between attempts to kill the subprocess (approx)
 TASK_KILL_INTERVAL = 1
 # Number of heartbeats between cleaning up worker processes (approx)
 WORKER_CLEANUP_INTERVAL = 100
@@ -69,7 +69,10 @@ def __init__(self):
         self.versions = {app.label: app.version for app in pulp_plugin_configs()}
         self.cursor = connection.cursor()
         self.worker = self.handle_worker_heartbeat()
-        self.task_grace_timeout = 0
+        # This defaults to immediate task cancelation.
+        # It will be set into the future on moderately graceful worker shutdown,
+        # and set to None for fully graceful shutdown.
+        self.task_grace_timeout = timezone.now()
         self.worker_cleanup_countdown = random.randint(
             int(WORKER_CLEANUP_INTERVAL / 10), WORKER_CLEANUP_INTERVAL
         )
@@ -104,7 +107,8 @@ def _init_instrumentation(self):
     def _signal_handler(self, thesignal, frame):
         if thesignal in (signal.SIGHUP, signal.SIGTERM):
             _logger.info(_("Worker %s was requested to shut down gracefully."), self.name)
-            self.task_grace_timeout = -1
+            # Wait forever...
+            self.task_grace_timeout = None
         else:
             # Reset signal handlers to default
             # If you kill the process a second time it's not graceful anymore.
@@ -113,7 +117,9 @@ def _signal_handler(self, thesignal, frame):
             signal.signal(signal.SIGHUP, signal.SIG_DFL)
 
             _logger.info(_("Worker %s was requested to shut down."), self.name)
-            self.task_grace_timeout = TASK_GRACE_INTERVAL
+            self.task_grace_timeout = timezone.now() + timezone.timedelta(
+                seconds=TASK_GRACE_INTERVAL
+            )
         self.shutdown_requested = True
 
     def _pg_notify_handler(self, notification):
@@ -173,8 +179,6 @@ def worker_cleanup(self):
     def beat(self):
         if self.worker.last_heartbeat < timezone.now() - self.heartbeat_period:
             self.worker = self.handle_worker_heartbeat()
-            if self.task_grace_timeout > 0:
-                self.task_grace_timeout -= 1
             self.worker_cleanup_countdown -= 1
             if self.worker_cleanup_countdown <= 0:
                 self.worker_cleanup_countdown = WORKER_CLEANUP_INTERVAL
@@ -390,10 +394,12 @@ def supervise_task(self, task):
             task_process.start()
             while True:
                 if cancel_state:
-                    if self.task_grace_timeout != 0:
+                    if self.task_grace_timeout is None or self.task_grace_timeout > timezone.now():
                         _logger.info("Wait for canceled task to abort.")
                     else:
-                        self.task_grace_timeout = TASK_KILL_INTERVAL
+                        self.task_grace_timeout = timezone.now() + timezone.TimeDelta(
+                            seconds=TASK_KILL_INTERVAL
+                        )
                         _logger.info(
                             "Aborting current task %s in domain: %s due to cancelation.",
                             task.pk,
@@ -430,7 +436,7 @@ def supervise_task(self, task):
                 if self.sentinel in r:
                     os.read(self.sentinel, 256)
                 if self.shutdown_requested:
-                    if self.task_grace_timeout != 0:
+                    if self.task_grace_timeout is None or self.task_grace_timeout > timezone.now():
                         msg = (
                             "Worker shutdown requested, waiting for task {pk} in domain: {name} "
                             "to finish.".format(pk=task.pk, name=domain.name)