[SAP] Implement graceful shutdown for cinder services

cinder/backup/manager.py

@@ -32,6 +32,7 @@
 """
 
 import contextlib
+import datetime
 import os
 import typing
 
@@ -203,7 +204,7 @@ def _cleanup_incomplete_backup_operations(self, ctxt):
         backups = objects.BackupList.get_all_by_host(ctxt, self.host)
         for backup in backups:
             try:
-                self._cleanup_one_backup(ctxt, backup)
+                self._cleanup_one_backup(ctxt, backup, skip_fresh=True)
             except Exception:
                 LOG.exception("Problem cleaning up backup %(bkup)s.",
                               {'bkup': backup['id']})
@@ -253,8 +254,22 @@ def _cleanup_one_snapshot(self, ctxt, snapshot_id):
             snapshot.status = fields.SnapshotStatus.AVAILABLE
             snapshot.save()
 
-    def _cleanup_one_backup(self, ctxt, backup):
+    def _cleanup_one_backup(self, ctxt, backup, skip_fresh=False):
         if backup['status'] == fields.BackupStatus.CREATING:
+            # [GS] Check if the backup entry is fresh — if so, the old pod
+            # is still actively draining and we should not interfere.
+            # Only applied during init_host startup (skip_fresh=True).
+            if skip_fresh:
+                now = timeutils.utcnow(with_timezone=True)
+                updated = backup.updated_at
+                if (updated and hasattr(updated, 'tzinfo')
+                        and updated.tzinfo is None):
+                    updated = updated.replace(
+                        tzinfo=datetime.timezone.utc)
+                age = ((now - updated).total_seconds()
+                       if updated else 9999)
+                if age < CONF.service_down_time:
+                    return
             LOG.info('Resetting backup %s to error (was creating).',
                      backup['id'])
             self._cleanup_one_volume(ctxt, backup.volume_id)
@@ -263,6 +278,20 @@ def _cleanup_one_backup(self, ctxt, backup):
             err = 'incomplete backup reset on manager restart'
             volume_utils.update_backup_error(backup, err)
         elif backup['status'] == fields.BackupStatus.RESTORING:
+            # [GS] Check if the backup entry is fresh — if so, the old pod
+            # is still actively draining and we should not interfere.
+            # Only applied during init_host startup (skip_fresh=True).
+            if skip_fresh:
+                now = timeutils.utcnow(with_timezone=True)
+                updated = backup.updated_at
+                if (updated and hasattr(updated, 'tzinfo')
+                        and updated.tzinfo is None):
+                    updated = updated.replace(
+                        tzinfo=datetime.timezone.utc)
+                age = ((now - updated).total_seconds()
+                       if updated else 9999)
+                if age < CONF.service_down_time:
+                    return
             LOG.info('Resetting backup %s to '
                      'available (was restoring).',
                      backup['id'])
@@ -756,6 +785,21 @@ def restore_backup(self, context, backup, volume_id, volume_is_new):
             raise exception.InvalidBackup(reason=err)
 
         canceled = False
+        # [GS] Spawn a heartbeat greenthread that periodically touches the
+        # backup's updated_at field. This prevents the new pod's init_host
+        # from resetting the backup while the old pod is still draining.
+        import eventlet
+        _hb_stop = eventlet.event.Event()
+
+        def _backup_restore_heartbeat():
+            while not _hb_stop.ready():
+                try:
+                    backup.save()
+                except Exception:
+                    pass
+                eventlet.sleep(10)
+
+        hb_thread = eventlet.spawn(_backup_restore_heartbeat)
         try:
             self._run_restore(context, backup, volume, volume_is_new)
         except exception.BackupRestoreCancel:
@@ -773,6 +817,9 @@ def restore_backup(self, context, backup, volume_id, volume_is_new):
                                 else fields.VolumeStatus.ERROR_RESTORING)})
                 backup.status = fields.BackupStatus.AVAILABLE
                 backup.save()
+        finally:
+            _hb_stop.send()
+            hb_thread.wait()
 
         if canceled:
             volume.status = fields.VolumeStatus.ERROR
@@ -886,11 +933,11 @@ def _run_restore(self, context, backup, volume, volume_is_new):
                 self._detach_device(context, attach_info, volume, properties,
                                     force=True)
             except Exception:
+                LOG.exception("_detach_device failed during restore cleanup")
                 if not message_created:
                     self.message_api.create_from_request_context(
                         context,
                         detail=message_field.Detail.DETACH_ERROR)
-                raise
 
         # Regardless of whether the restore was successful, do some
         # housekeeping to ensure the restored volume's encryption key ID is

cinder/manager.py

@@ -51,6 +51,9 @@
 
 """
 
+import threading
+from typing import Callable
+
 from eventlet import greenpool
 from eventlet import tpool
 from oslo_config import cfg
@@ -161,12 +164,85 @@ def get_log_levels(self, context, log_request):
 
 
 class ThreadPoolManager(Manager):
+    """Manager class that provides a managed thread pool.
+
+    Tasks spawned via _add_to_threadpool() will be waited on during
+    graceful shutdown, ensuring in-flight operations complete before
+    the service terminates.
+
+    This implementation uses native Python threads via ThreadPoolExecutor
+    instead of eventlet green threads, as eventlet is being deprecated
+    and will be removed in a future OpenStack release.
+    """
+
+    # Maximum number of worker threads for async operations
+    # This replaces the eventlet GreenPool which had unlimited concurrency
+    DEFAULT_THREADPOOL_SIZE = 10
+
     def __init__(self, *args, **kwargs):
+        # Use eventlet GreenPool for async task dispatch. Green threads
+        # are cooperative (same OS process/thread) so they don't block
+        # process exit and don't require thread-safe DB access patterns.
+        # The graceful shutdown drain (pool.waitall in Service._drain_pool)
+        # waits for all spawned greenthreads to complete.
         self._tp = greenpool.GreenPool()
+        self._shutdown_event = threading.Event()
         super(ThreadPoolManager, self).__init__(*args, **kwargs)
 
-    def _add_to_threadpool(self, func, *args, **kwargs):
+    @staticmethod
+    def _set_thread_daemon():
+        pass  # Not used with GreenPool, kept for interface compatibility
+
+    def _add_to_threadpool(self, func: Callable, *args, **kwargs):
+        """Spawn a task in the green thread pool.
+
+        Tasks spawned here will be waited on during graceful shutdown
+        via pool.waitall() in Service._drain_pool().
+
+        :param func: The function to execute
+        :param args: Positional arguments to pass to the function
+        :param kwargs: Keyword arguments to pass to the function
+        :returns: True if spawned, None if rejected (shutdown in progress)
+        """
+        if self._shutdown_event.is_set():
+            LOG.warning(
+                "Rejecting threadpool task during shutdown: %s",
+                func.__name__)
+            return None
+
         self._tp.spawn_n(func, *args, **kwargs)
+        return True
+
+    def signal_shutdown(self) -> None:
+        """Signal that shutdown is in progress.
+
+        After this is called, new tasks submitted via _add_to_threadpool()
+        will be rejected. Existing tasks will continue to run.
+        """
+        LOG.info("Shutdown signaled, rejecting new threadpool tasks")
+        self._shutdown_event.set()
+
+    def cleanup_threadpool(self) -> None:
+        """Cleanup the green thread pool and prepare for potential restart.
+
+        Should be called during service shutdown after pool.waitall() has
+        drained in-flight RPC handlers. Waits for any remaining spawned
+        greenthreads and resets the pool.
+
+        After cleanup, a fresh pool is created so the manager is ready
+        for reuse if the service is restarted (e.g., oslo.service
+        ProcessLauncher with restart_method='mutate' will fork a new child
+        that inherits this object and calls start() again).
+        """
+        LOG.info("Waiting for green thread pool to drain")
+        self._tp.waitall()
+        LOG.info("Green thread pool drained")
+
+        # Re-create the pool and reset shutdown state so the manager
+        # is ready if the service is restarted.
+        self._tp = greenpool.GreenPool()
+        self._shutdown_event.clear()
+        LOG.info("Green thread pool re-initialized for potential restart")
 
 
 class SchedulerDependentManager(ThreadPoolManager):
@@ -199,8 +275,7 @@ def update_service_capabilities(self, capabilities):
     def _publish_service_capabilities(self, context):
         """Pass data back to the scheduler at a periodic interval."""
         if self.last_capabilities:
-            LOG.debug('Notifying Schedulers of capabilities for %(host)s...',
-                      {'host': self.host})
+            LOG.debug('Notifying Schedulers of capabilities ...')
             self.scheduler_rpcapi.update_service_capabilities(
                 context,
                 self.service_name,
@@ -230,7 +305,8 @@ def reset(self):
 class CleanableManager(object):
     def do_cleanup(self,
                    context: context.RequestContext,
-                   cleanup_request: objects.CleanupRequest) -> None:
+                   cleanup_request: objects.CleanupRequest,
+                   skip_fresh: bool = False) -> None:
         LOG.info('Initiating service %s cleanup',
                  cleanup_request.service_id)
 
@@ -247,6 +323,19 @@ def do_cleanup(self,
             until=until)
 
         for clean in to_clean:
+            # Skip worker entries that are being actively heartbeated.
+            # During graceful shutdown, the set_workers decorator keeps
+            # touching updated_at every 10s. If the entry is fresh
+            # (< service_down_time), the operation is still in progress
+            # on the draining pod — don't reset it.
+            # Only applied during init_host startup (skip_fresh=True) to
+            # avoid interfering with a draining pod's in-flight operations.
+            if skip_fresh:
+                age = (timeutils.utcnow()
+                       - clean.updated_at).total_seconds()
+                if age < CONF.service_down_time:
+                    continue
+
             original_service_id = clean.service_id
             original_time = clean.updated_at
             # Try to do a soft delete to mark the entry as being cleaned up
@@ -320,4 +409,4 @@ def init_host(self, service_id, added_to_cluster=None, **kwargs):
         ctxt = context.get_admin_context()
         self.service_id = service_id
         cleanup_request = objects.CleanupRequest(service_id=service_id)
-        self.do_cleanup(ctxt, cleanup_request)
+        self.do_cleanup(ctxt, cleanup_request, skip_fresh=True)

cinder/objects/cleanable.py

@@ -17,6 +17,7 @@
 import inspect
 
 import decorator
+import eventlet
 from oslo_utils import versionutils
 
 from cinder import db
@@ -199,14 +200,43 @@ def wrapper(f, *args, **kwargs):
                 cleanables = [cand for cand in candidates
                               if (isinstance(cand, CinderCleanableObject)
                                   and cand.is_cleanable(pinned=False))]
+                hb = None
+                stop_heartbeat = eventlet.event.Event()
                 try:
                     # Create the entries in the workers table
                     for cleanable in cleanables:
                         cleanable.set_worker()
 
+                    # Spawn a greenthread that periodically touches worker
+                    # entries to keep them fresh. This prevents a new pod's
+                    # init_host -> _do_cleanup from resetting resources that
+                    # are actively being processed during graceful shutdown.
+                    def _worker_heartbeat():
+                        while not stop_heartbeat.ready():
+                            for cleanable in cleanables:
+                                if cleanable.worker:
+                                    try:
+                                        db.worker_update(
+                                            cleanable._context,
+                                            cleanable.worker.id)
+                                    except Exception:
+                                        pass
+                            # Use event.ready() + sleep instead of a long
+                            # uninterruptible sleep. Short sleeps allow the
+                            # loop to notice stop_heartbeat quickly.
+                            for _ in range(100):
+                                if stop_heartbeat.ready():
+                                    break
+                                eventlet.sleep(0.1)
+                    hb = eventlet.spawn(_worker_heartbeat)
+
                     # Call the function
                     result = f(*args, **kwargs)
                 finally:
+                    # Stop the heartbeat greenthread
+                    stop_heartbeat.send()
+                    if hb is not None:
+                        hb.wait()
                     # Remove entries from the workers table
                     for cleanable in cleanables:
                         # NOTE(geguileo): We check that the status has changed

cinder/opts.py

@@ -284,6 +284,7 @@ def list_opts():
                 cinder_scheduler_weights_volumenumber.
                 volume_number_weight_opts,
                 cinder_service.service_opts,
+                cinder_service.os_service_options.service_opts,
                 cinder_sshutils.ssh_opts,
                 cinder_transfer_api.volume_transfer_opts,
                 [cinder_volume_api.allow_force_upload_opt],