From 8e7da2c990077715c12cc1a2061fc35d6fdd4eed Mon Sep 17 00:00:00 2001 From: Walter Boring Date: Wed, 27 May 2026 14:31:02 -0400 Subject: [PATCH] [SAP] Implement graceful shutdown for cinder services MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three-phase graceful shutdown that allows in-flight volume and backup operations to complete before the pod exits during Kubernetes rolling updates. Covers both cinder-volume and cinder-backup services. The fix: install SIG_IGN for SIGTERM/SIGINT/SIGHUP at the start of Service.stop(). cinder-volume runs as N forked child processes (one per backend) under oslo_service.ProcessLauncher. On SIGTERM, oslo_service's child handler calls SignalHandler.clear() which resets all handlers to SIG_DFL. The parent then sends a second SIGTERM via os.kill(child_pid, SIGTERM). Without SIG_IGN, the child terminates immediately — even though it's still inside pool.waitall() waiting for in-flight RPC handlers to finish. How it works: Phase 1: Skip consumer cancel. We do NOT call rpcserver.stop() because it triggers an eventlet socket race ("simultaneous read on fileno") that disrupts outbound HTTP/RPC connections used by in-flight ops. Broker-side deregistration happens automatically at process exit. During the gap, heartbeat stops (scheduler reroutes within service_down_time) and @reject_if_draining rejects late arrivals. Phase 2: Block in pool.waitall() until all in-flight RPC handler greenthreads complete their operations. Phase 3: Stop coordination, call super().stop(), cleanup threadpool executor. Process exits cleanly. Supporting mechanisms: - Worker entry heartbeat (set_workers decorator): touches worker DB entries every 10s during operations, preventing new pod init_host do_cleanup from resetting in-flight volumes to error. Uses short interruptible sleeps (0.1s polling) for fast shutdown response. - do_cleanup freshness check: skips worker entries updated within service_down_time, only cleans up truly stale entries. - Backup restore heartbeat: touches backup.updated_at every 10s, preventing new backup pod from triggering BackupRestoreCancel. - Backup _detach_device no-reraise: if detach fails during shutdown, log and continue. Data integrity preserved. - reject_if_draining decorator: unconditionally rejects new RPC calls on a draining service so scheduler routes to healthy backends. Test adaptations: - Base test classes (BaseVolumeTestCase, BaseBackupTest) drain the GreenPool in addCleanup to prevent greenthread leaks. - test_volume_cleanup sets service_down_time=0 and runs threadpool tasks synchronously (mock_object) to avoid race conditions. - test_admin_actions tearDown explicitly stops fake RPC servers to deregister stale endpoints from oslo.messaging's fake transport. - test_service assertions updated to reflect intentional skip of rpcserver.stop() in graceful shutdown, and signal flag reset. - test_backup_messages updated for _detach_device no-reraise behavior. Requires (separate changes): - dumb-init --single-child on container commands - terminationGracePeriodSeconds: 900 on pod spec Change-Id: Icdd28affc73fd34491b656a68410dce8e46264d4 --- cinder/backup/manager.py | 53 +++- cinder/manager.py | 99 ++++++- cinder/objects/cleanable.py | 30 +++ cinder/opts.py | 1 + cinder/service.py | 255 ++++++++++++++++-- .../unit/api/contrib/test_admin_actions.py | 16 ++ cinder/tests/unit/backup/test_backup.py | 10 + .../tests/unit/backup/test_backup_messages.py | 6 +- cinder/tests/unit/keymgr/test_migration.py | 4 + cinder/tests/unit/test.py | 14 + cinder/tests/unit/test_cleanable_manager.py | 3 +- cinder/tests/unit/test_manager.py | 91 +++++++ cinder/tests/unit/test_service.py | 190 ++++++++++++- cinder/tests/unit/test_volume_cleanup.py | 8 + cinder/tests/unit/volume/__init__.py | 10 + cinder/volume/flows/manager/create_volume.py | 8 + cinder/volume/manager.py | 110 +++++++- 17 files changed, 873 insertions(+), 35 deletions(-) diff --git a/cinder/backup/manager.py b/cinder/backup/manager.py index cdd26ab05d0..d41161576ad 100644 --- a/cinder/backup/manager.py +++ b/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 diff --git a/cinder/manager.py b/cinder/manager.py index 87450684b0d..e313cfd1bcd 100644 --- a/cinder/manager.py +++ b/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) diff --git a/cinder/objects/cleanable.py b/cinder/objects/cleanable.py index 10bf0f25550..4e520e89884 100644 --- a/cinder/objects/cleanable.py +++ b/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 diff --git a/cinder/opts.py b/cinder/opts.py index b5dfe38ba2c..1cd36eb9674 100644 --- a/cinder/opts.py +++ b/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], diff --git a/cinder/service.py b/cinder/service.py index accab15463a..7eea0bfca68 100644 --- a/cinder/service.py +++ b/cinder/service.py @@ -26,11 +26,14 @@ import time from typing import Optional +import eventlet +import eventlet.timeout from oslo_concurrency import processutils from oslo_config import cfg from oslo_db import exception as db_exc from oslo_log import log as logging import oslo_messaging as messaging +from oslo_service import _options as os_service_options from oslo_service import service from oslo_service import wsgi from oslo_utils import importutils @@ -60,6 +63,27 @@ LOG = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Process-wide graceful-shutdown coordination +# --------------------------------------------------------------------------- +# Why this exists: +# +# In SAP's deployment, oslo_service.ProcessLauncher forks one child process +# per backend (5 children). On SIGTERM, oslo_service's child SIGTERM handler +# calls SignalHandler.clear() which resets all signal handlers to SIG_DFL. +# The parent ProcessLauncher then sends a second SIGTERM to each child via +# os.kill(child_pid, SIGTERM). With the handler now SIG_DFL, the child +# terminates immediately — even though it's still inside pool.waitall() +# waiting for an in-flight RPC handler to finish. +# +# Fix: at the start of cinder's Service.stop() (first caller per process), +# install SIG_IGN for SIGTERM/SIGINT/SIGHUP so the second signal is ignored +# and the child stays alive until pool.waitall() completes. +# Process-wide flag: once any Service in this process enters stop(), +# we install SIG_IGN so subsequent signals don't terminate the process. +_GS_SIGNALS_IGNORED = False + service_opts = [ cfg.IntOpt('report_interval', default=10, @@ -85,11 +109,31 @@ cfg.BoolOpt('osapi_volume_use_ssl', default=False, help='Wraps the socket in a SSL context if True is set. ' - 'A certificate file and key file must be specified.'), ] + 'A certificate file and key file must be specified.'), + cfg.IntOpt('threadpool_size', + default=10, + min=1, + max=100, + help='Number of native Python threads in the threadpool ' + 'for async volume operations. This replaces the previous ' + 'eventlet GreenPool. Adjust based on workload and ' + 'available system resources. Default: 10.'), ] CONF = cfg.CONF CONF.register_opts(service_opts) +# Register oslo.service's service_opts (which includes +# graceful_shutdown_timeout) so cinder can use it. We must NOT define our +# own graceful_shutdown_timeout in service_opts above, as that would cause +# a DuplicateOptError when oslo.service later tries to register the same +# option in ServiceLauncher/ProcessLauncher.__init__(). +try: + CONF.register_opts(os_service_options.service_opts) +except cfg.DuplicateOptError: + # oslo.service already registered these opts (e.g. during launcher init) + pass +cfg.set_defaults(os_service_options.service_opts, + graceful_shutdown_timeout=120) if profiler_opts: profiler_opts.set_defaults(CONF) @@ -217,12 +261,19 @@ def __init__(self, host: str, binary: str, topic: str, self.rpcserver: Optional['messaging.rpc.RPCServer'] = None self.backend_rpcserver: Optional['messaging.rpc.RPCServer'] = None self.cluster_rpcserver: Optional['messaging.rpc.RPCServer'] = None + # Flag to indicate graceful shutdown in progress. Read by + # @reject_if_draining decorator on RPC entry points. + self._draining = False def start(self) -> None: version_string = version.version_string() LOG.info('Starting %(topic)s node (version %(version_string)s)', {'topic': self.topic, 'version_string': version_string}) self.model_disconnected = False + # Reset draining state in case this is a restart (e.g., oslo.service + # ProcessLauncher with restart_method='mutate' forks a new child + # that inherits the parent's state including _draining=True). + self._draining = False if self.coordination: coordination.COORDINATOR.start() @@ -429,32 +480,189 @@ def create(cls, return service_obj def stop(self) -> None: - # Try to shut the connection down, but if we get any sort of - # errors, go ahead and ignore them.. as we're shutting down anyway + """Stop the service gracefully. + + This method implements a three-phase graceful shutdown: + 1. Sets draining state (stops heartbeat reporting) + 2. Signals manager to reject new threadpool tasks + 3. Phase 1: Skips rpcserver.stop() (eventlet socket race) + 4. Phase 2: Waits for in-flight RPC handlers via pool.waitall() + 5. Phase 3: Stops coordination, calls super().stop(), cleans up + threadpool executor + """ + self._install_sig_ign() + + LOG.info( + "Initiating graceful shutdown for service %s on host %s", + self.binary, + self.host, + ) + + # Set draining state - stops heartbeat reporting + self._draining = True + + # Signal manager to stop accepting new threadpool tasks + if hasattr(self.manager, "signal_shutdown"): + self.manager.signal_shutdown() + + # ====== PHASE 1: Stop consuming WITHOUT killing greenthreads ====== + # We intentionally do NOT call conn.stop_consuming() or + # rpcserver.stop() synchronously. Doing so causes eventlet socket + # races ("simultaneous read on fileno") that can disrupt outbound + # HTTP/RPC connections used by in-flight operations (e.g., Swift + # reads during backup restore). + # + # The broker-side consumer deregistration happens automatically + # when the AMQP channel closes during process exit (the OS closes + # the TCP socket; RabbitMQ detects the broken connection and + # requeues any pending messages to healthy consumers). + # + # During the gap between SIGTERM and process exit, two things + # protect us from message loss: + # 1. We stop heartbeating, so the scheduler stops routing new + # top-level work within service_down_time (~60s default). + # 2. The @reject_if_draining decorator on RPC entry points + # rejects any message that DOES race in, with a + # ServiceUnavailable exception that triggers caller-side + # retry on a healthy instance. + # + # In Phase 2 we wait via pool.waitall() for any in-flight RPC + # handler greenthreads to complete naturally before the process + # exits. + + # NOTE: Do NOT stop coordination here! + # In-flight operations may still need distributed locks. + + # NOTE: Do NOT cleanup RPC transport here! + # In-flight operations may need to make outbound RPC calls. + + # ====== PHASE 2: Wait for in-flight operations to complete ====== + # The RPC handler greenthread is STILL ALIVE because we didn't call + # rpcserver.stop(). We can now safely wait for it to finish. + + timeout = CONF.graceful_shutdown_timeout + if timeout == 0: + timeout = None + + for server_name, server in ( + ('rpcserver', self.rpcserver), + ('backend_rpcserver', self.backend_rpcserver), + ('cluster_rpcserver', self.cluster_rpcserver), + ): + self._drain_pool(server_name, server, timeout) + + # ====== PHASE 3: Cleanup (skip rpcserver.stop/wait) ====== + self._phase3_teardown() + + def _install_sig_ign(self) -> None: + """Install SIG_IGN for SIGTERM/SIGINT/SIGHUP at start of stop(). + + oslo_service's _sigterm handler calls SignalHandler.clear() which + sets handlers back to SIG_DFL — so a second SIGTERM (e.g., from + ProcessLauncher parent's os.kill(child_pid, SIGTERM) in stop()) + would terminate the child immediately, even while we're still in + pool.waitall(). + + We install SIG_IGN ONCE per process via a process-wide flag. + No lock needed: signal.signal() is idempotent and the flag only + flips False -> True (a benign race for two concurrent first + callers, but in the fork model only one Service.stop() runs per + child process). + """ + global _GS_SIGNALS_IGNORED + if _GS_SIGNALS_IGNORED: + return + _GS_SIGNALS_IGNORED = True try: - if self.rpcserver is not None: - self.rpcserver.stop() - if self.backend_rpcserver: - self.backend_rpcserver.stop() - if self.cluster_rpcserver: - self.cluster_rpcserver.stop() - except Exception: + import signal as _sig + _sig.signal(_sig.SIGTERM, _sig.SIG_IGN) + _sig.signal(_sig.SIGINT, _sig.SIG_IGN) + _sig.signal(_sig.SIGHUP, _sig.SIG_IGN) + except Exception: # noqa: BLE001 pass + def _drain_pool(self, server_name, server, timeout) -> None: + """Drain one rpcserver's GreenPool, waiting for in-flight handlers. + + Skips quietly if the server is None or has no _work_executor/_pool, + or if the pool has no running greenthreads. + """ + if server is None: + return + work_executor = getattr(server, '_work_executor', None) + if work_executor is None: + return + pool = getattr(work_executor, '_pool', None) + if pool is None: + return + # Guard against Mock objects in tests — pool.running() must + # return an actual integer for the comparison to be meaningful. + try: + running = pool.running() + if not isinstance(running, int) or running <= 0: + return + except (TypeError, AttributeError): + return + + try: + with eventlet.timeout.Timeout(timeout): + pool.waitall() + except eventlet.timeout.Timeout: + LOG.warning("%s: Timed out waiting for GreenPool after %s seconds", + server_name, timeout) + + def _phase3_teardown(self) -> None: + """Phase 3: stop coordination, super().stop, cleanup_threadpool. + + We do NOT call rpcserver.stop() or rpcserver.wait() here. + rpcserver.stop() triggers AMQPListener.stop() which calls + conn.stop_consuming() — this risks an eventlet socket race + ("simultaneous read on fileno") because the connection's internal + reader greenthread may be mid-read when stop_consuming writes a + Basic.Cancel frame on the same socket. + + Instead we let the process exit naturally after this method + returns. The OS closes the AMQP socket (TCP FIN) as the + interpreter shuts down; RabbitMQ detects the closed connection + immediately and requeues any pending messages to healthy consumers. + """ + # Stop coordination - all operations have completed if self.coordination: try: coordination.COORDINATOR.stop() except Exception: - pass + LOG.exception("Error stopping coordination") + super(Service, self).stop(graceful=True) + LOG.info("Service %s shutdown complete", self.binary) + + # Cleanup the threadpool executor after all operations complete + if hasattr(self.manager, "cleanup_threadpool"): + self.manager.cleanup_threadpool() + def wait(self) -> None: - if self.rpcserver: - self.rpcserver.wait() - if self.backend_rpcserver: - self.backend_rpcserver.wait() - if self.cluster_rpcserver: - self.cluster_rpcserver.wait() + """Wait for all service operations to complete. + + NOTE: oslo.service calls wait() immediately after start() as part + of its normal "block until service finishes" pattern (in + _child_wait_for_exit_or_signal). The graceful shutdown logic must + ONLY run when stop() has been called first (indicated by _draining). + Otherwise, we would destroy coordination and other resources while + the service is still actively running. + + When _draining is True (stop() was called), this method: + 1. Waits for manager threadpool tasks (outbound RPC still available) + 2. Waits for in-flight RPC handlers (outbound RPC still available) + 3. Stops coordination service (after all ops complete) + 4. Cleans up threadpool executor + """ + if self._draining: + # All waiting/cleanup is now done in stop() to prevent + # oslo.service from killing the process before operations complete. + # See stop() for the full graceful shutdown sequence. + pass + super(Service, self).wait() def periodic_tasks(self, raise_on_error: bool = False) -> None: @@ -464,6 +672,14 @@ def periodic_tasks(self, raise_on_error: bool = False) -> None: def report_state(self) -> None: """Update the state of this service in the datastore.""" + # Don't suppress heartbeat during graceful shutdown drain. + # We MUST keep reporting state while in-flight operations are + # completing (Phase 2 of graceful shutdown). Otherwise, the service + # is marked "down" after service_down_time (~20s), and the scheduler + # or cleanup tasks will fail the pending volume operation. + # The heartbeat will stop naturally when the process exits after + # Phase 3 completes. + if not self.manager.is_working(): # NOTE(dulek): If manager reports a problem we're not sending # heartbeats - to indicate that service is actually down. @@ -518,6 +734,11 @@ def reset(self) -> None: self.manager.reset() super(Service, self).reset() + @property + def is_draining(self) -> bool: + """Return True if the service is shutting down gracefully.""" + return self._draining + class WSGIService(service.ServiceBase): """Provides ability to launch API from a 'paste' configuration.""" diff --git a/cinder/tests/unit/api/contrib/test_admin_actions.py b/cinder/tests/unit/api/contrib/test_admin_actions.py index a8aa4136327..2432af0c207 100644 --- a/cinder/tests/unit/api/contrib/test_admin_actions.py +++ b/cinder/tests/unit/api/contrib/test_admin_actions.py @@ -114,6 +114,14 @@ def _get_minimum_rpc_version_mock(ctxt, binary): def tearDown(self): self.svc.stop() + # Deregister RPC endpoints from fake transport so subsequent + # tests don't dispatch to this (now draining) service instance. + if self.svc.rpcserver: + self.svc.rpcserver.stop() + self.svc.rpcserver.wait() + if self.svc.backend_rpcserver: + self.svc.backend_rpcserver.stop() + self.svc.backend_rpcserver.wait() super(AdminActionsTest, self).tearDown() def _issue_resource_reset(self, ctx, name, id, status): @@ -1124,6 +1132,14 @@ def setUp(self): def tearDown(self): self.svc.stop() + # Deregister RPC endpoints from fake transport so subsequent + # tests don't dispatch to this (now draining) service instance. + if self.svc.rpcserver: + self.svc.rpcserver.stop() + self.svc.rpcserver.wait() + if self.svc.backend_rpcserver: + self.svc.backend_rpcserver.stop() + self.svc.backend_rpcserver.wait() super(AdminActionsAttachDetachTest, self).tearDown() def test_force_detach_instance_attached_volume(self): diff --git a/cinder/tests/unit/backup/test_backup.py b/cinder/tests/unit/backup/test_backup.py index 02265d6cd99..bdb4b1f0171 100644 --- a/cinder/tests/unit/backup/test_backup.py +++ b/cinder/tests/unit/backup/test_backup.py @@ -61,6 +61,7 @@ def setUp(self): self.backup_mgr.host = 'testhost' self.backup_mgr.is_initialized = True self.ctxt = context.get_admin_context() + self.addCleanup(self._cleanup_threadpool) paths = ['cinder.volume.rpcapi.VolumeAPI.delete_snapshot', 'cinder.volume.rpcapi.VolumeAPI.delete_volume', @@ -75,6 +76,12 @@ def setUp(self): self.volume_mocks[name] = self.volume_patches[name].start() self.addCleanup(self.volume_patches[name].stop) + def _cleanup_threadpool(self): + """Ensure the manager's GreenPool is drained after each test.""" + tp = getattr(self.backup_mgr, '_tp', None) + if tp is not None and hasattr(tp, 'waitall'): + tp.waitall() + def _create_backup_db_entry(self, volume_id=str(uuid.uuid4()), restore_volume_id=None, display_name='test_backup', @@ -243,6 +250,9 @@ def get_admin_context(): self.override_config('backup_service_inithost_offload', False) self.override_config('periodic_interval', 0) + # Disable freshness check so freshly-created test backups are + # cleaned up (in production, stuck backups are old enough to pass). + self.override_config('service_down_time', 0) vol1_id = self._create_volume_db_entry() self._create_volume_attach(vol1_id) diff --git a/cinder/tests/unit/backup/test_backup_messages.py b/cinder/tests/unit/backup/test_backup_messages.py index a243b2eb93a..a5f1c636fb1 100644 --- a/cinder/tests/unit/backup/test_backup_messages.py +++ b/cinder/tests/unit/backup/test_backup_messages.py @@ -556,8 +556,10 @@ def test_backup_restore_detach_error( mock_detach.side_effect = exception.InvalidBackup( reason="test reason") - self.assertRaises( - exception.InvalidBackup, manager.restore_backup, + # [GS] _detach_device errors are now caught and logged (no-reraise) + # during graceful shutdown support. The DETACH_ERROR message is still + # created, but the exception no longer propagates. + manager.restore_backup( fake_context, fake_backup, fake.VOLUME_ID, False) self.assertEqual(message_field.Action.BACKUP_RESTORE, fake_context.message_action) diff --git a/cinder/tests/unit/keymgr/test_migration.py b/cinder/tests/unit/keymgr/test_migration.py index 311251530bf..e4020b287b7 100644 --- a/cinder/tests/unit/keymgr/test_migration.py +++ b/cinder/tests/unit/keymgr/test_migration.py @@ -33,6 +33,10 @@ class KeyMigrationTestCase(base.BaseVolumeTestCase): def setUp(self): super(KeyMigrationTestCase, self).setUp() + # Run threadpool tasks synchronously to avoid threading issues + # with the in-memory sqlite DB used by tests. + self.mock_object(self.volume, '_add_to_threadpool', + side_effect=lambda f, *a, **kw: f(*a, **kw)) self.conf = CONF self.fixed_key = '1' * 64 try: diff --git a/cinder/tests/unit/test.py b/cinder/tests/unit/test.py index c45532ee047..0007a6ed712 100644 --- a/cinder/tests/unit/test.py +++ b/cinder/tests/unit/test.py @@ -336,6 +336,20 @@ def _common_cleanup(self): x.stop() except Exception: pass + # Explicitly deregister RPC endpoints from the fake transport. + # In production, the process exits after stop() and the OS + # closes the AMQP socket. In tests, stale endpoints remain + # registered in oslo.messaging's fake transport and can + # receive messages intended for the next test's service. + for server in ('rpcserver', 'backend_rpcserver', + 'cluster_rpcserver'): + srv = getattr(x, server, None) + if srv is not None: + try: + srv.stop() + srv.wait() + except Exception: + pass # Stop any looping call that has not yet been stopped cinder_unit.stop_looping_calls() diff --git a/cinder/tests/unit/test_cleanable_manager.py b/cinder/tests/unit/test_cleanable_manager.py index d414d276007..edf37e25442 100644 --- a/cinder/tests/unit/test_cleanable_manager.py +++ b/cinder/tests/unit/test_cleanable_manager.py @@ -54,7 +54,8 @@ def test_init_host_with_service(self, mock_cleanup): mngr.init_host(service_id=self.service.id) self.assertEqual(self.service.id, mngr.service_id) - mock_cleanup.assert_called_once_with(mngr, mock.ANY, mock.ANY) + mock_cleanup.assert_called_once_with( + mngr, mock.ANY, mock.ANY, skip_fresh=True) clean_req = mock_cleanup.call_args[0][2] self.assertIsInstance(clean_req, objects.CleanupRequest) self.assertEqual(self.service.id, clean_req.service_id) diff --git a/cinder/tests/unit/test_manager.py b/cinder/tests/unit/test_manager.py index 1a0a244708f..85225581e09 100644 --- a/cinder/tests/unit/test_manager.py +++ b/cinder/tests/unit/test_manager.py @@ -13,6 +13,7 @@ # License for the specific language governing permissions and limitations # under the License. +import time from unittest import mock from cinder import manager @@ -54,3 +55,93 @@ def test_get_log_levels(self, get_log_mock): self.assertEqual(set(str(r) for r in result.objects), set(str(e) for e in expected)) + + +class TestThreadPoolManager(test.TestCase): + """Test cases for ThreadPoolManager greenthread pool + shutdown.""" + + def test_add_to_threadpool_executes(self): + """Test that _add_to_threadpool runs the task to completion.""" + mgr = manager.ThreadPoolManager() + executed = [] + + mgr._add_to_threadpool(lambda: executed.append(True)) + mgr.cleanup_threadpool() # waitall ensures task completes + + self.assertEqual([True], executed) + + def test_signal_shutdown_rejects_new_tasks(self): + """Test that new tasks are rejected after shutdown signal.""" + mgr = manager.ThreadPoolManager() + mgr.signal_shutdown() + + executed = [] + result = mgr._add_to_threadpool(lambda: executed.append(True)) + + # Should return None when shutdown is signaled + self.assertIsNone(result) + + # Give it a moment to potentially execute + time.sleep(0.1) + + self.assertEqual([], executed) + mgr.cleanup_threadpool() + + def test_signal_shutdown_allows_existing_tasks(self): + """Test that existing tasks continue after shutdown signal.""" + mgr = manager.ThreadPoolManager() + completed = [] + + def slow_task(): + time.sleep(0.1) + completed.append(True) + + # Spawn task first + mgr._add_to_threadpool(slow_task) + # Then signal shutdown + mgr.signal_shutdown() + + # Existing task should still complete on cleanup (waitall) + mgr.cleanup_threadpool() + self.assertEqual([True], completed) + + def test_cleanup_threadpool_allows_restart(self): + """Test that cleanup_threadpool re-creates pool for restart. + + oslo.service ProcessLauncher with restart_method='mutate' forks + a new child that inherits the parent's manager object. After + stop()/wait()/cleanup(), the child calls start() -> init_host() + which needs to submit tasks to the threadpool. + """ + mgr = manager.ThreadPoolManager() + + # Simulate a full shutdown cycle + mgr.signal_shutdown() + mgr.cleanup_threadpool() + + # After cleanup, the pool should be re-created and usable + executed = [] + result = mgr._add_to_threadpool(lambda: executed.append(True)) + self.assertIsNotNone(result) + + mgr.cleanup_threadpool() + self.assertEqual([True], executed) + + def test_cleanup_threadpool_clears_shutdown_event(self): + """Test cleanup resets shutdown_event so new tasks accepted.""" + mgr = manager.ThreadPoolManager() + + mgr.signal_shutdown() + # After signal_shutdown, tasks should be rejected + result = mgr._add_to_threadpool(lambda: None) + self.assertIsNone(result) + + mgr.cleanup_threadpool() + + # After cleanup, tasks should be accepted again + executed = [] + result = mgr._add_to_threadpool(lambda: executed.append(True)) + self.assertIsNotNone(result) + + mgr.cleanup_threadpool() + self.assertEqual([True], executed) diff --git a/cinder/tests/unit/test_service.py b/cinder/tests/unit/test_service.py index 1baef758aff..4f017a5fd42 100644 --- a/cinder/tests/unit/test_service.py +++ b/cinder/tests/unit/test_service.py @@ -50,7 +50,7 @@ CONF.register_opts(test_service_opts) -class FakeManager(manager.Manager): +class FakeManager(manager.ThreadPoolManager): """Fake manager for tests.""" def __init__(self, host=None, service_name=None, cluster=None): super().__init__(host=host, cluster=cluster) @@ -67,6 +67,21 @@ def test_method(self): class ServiceManagerTestCase(test.TestCase): """Test cases for Services.""" + def setUp(self): + super(ServiceManagerTestCase, self).setUp() + # Reset the process-wide graceful shutdown signal flag that + # Service.stop() sets. Without this, SIG_IGN persists across + # tests in the same worker process and can prevent stestr from + # killing stuck workers. + self.addCleanup(self._reset_signal_flag) + + def _reset_signal_flag(self): + import signal + service._GS_SIGNALS_IGNORED = False + # Restore default signal handlers for test isolation + signal.signal(signal.SIGTERM, signal.SIG_DFL) + signal.signal(signal.SIGINT, signal.SIG_DFL) + def test_message_gets_to_manager(self): serv = service.Service('test', 'test', @@ -165,6 +180,13 @@ def setUp(self): 'id': 1, 'uuid': 'a3a593da-7f8d-4bb7-8b4c-f2bc1e0b4824'} self.ctxt = context.get_admin_context() + self.addCleanup(self._reset_signal_flag) + + def _reset_signal_flag(self): + import signal + service._GS_SIGNALS_IGNORED = False + signal.signal(signal.SIGTERM, signal.SIG_DFL) + signal.signal(signal.SIGINT, signal.SIG_DFL) def _check_app(self, app, cluster=None, cluster_exists=None, svc_id=None, added_to_cluster=True): @@ -315,8 +337,11 @@ def test_service_stop_waits_for_rpcserver(self, mock_db, mock_rpc): serv.stop() serv.wait() serv.rpcserver.start.assert_called_once_with() - serv.rpcserver.stop.assert_called_once_with() - serv.rpcserver.wait.assert_called_once_with() + # Graceful shutdown intentionally does NOT call rpcserver.stop() + # to avoid the eventlet socket race. The process exits naturally + # and the OS closes the AMQP connection. + serv.rpcserver.stop.assert_not_called() + serv.rpcserver.wait.assert_not_called() @mock.patch('cinder.service.Service.report_state') @mock.patch('cinder.service.Service.periodic_tasks') @@ -338,8 +363,10 @@ def test_service_stop_wait(self, mock_db, mock_rpc, serv.stop() serv.wait() serv.rpcserver.start.assert_called_once_with() - serv.rpcserver.stop.assert_called_once_with() - serv.rpcserver.wait.assert_called_once_with() + # Graceful shutdown does NOT call rpcserver.stop() — see + # test_service_stop_waits_for_rpcserver for explanation. + serv.rpcserver.stop.assert_not_called() + serv.rpcserver.wait.assert_not_called() @mock.patch('cinder.manager.Manager.init_host') @mock.patch('oslo_messaging.Target') @@ -478,6 +505,159 @@ def test_ensure_cluster_exists_cluster_create_replicated_and_non(self): self.assertEqual(value, getattr(cluster, key)) +class TestGracefulShutdown(test.TestCase): + """Test cases for graceful shutdown functionality.""" + + def setUp(self): + super(TestGracefulShutdown, self).setUp() + self.host = "test_host" + self.binary = "cinder-volume" + self.topic = "cinder-volume" + self.addCleanup(self._reset_signal_flag) + + def _reset_signal_flag(self): + import signal + service._GS_SIGNALS_IGNORED = False + signal.signal(signal.SIGTERM, signal.SIG_DFL) + signal.signal(signal.SIGINT, signal.SIG_DFL) + + @mock.patch("cinder.service.Service.report_state") + def test_service_draining_flag(self, mock_report): + """Test that draining flag is properly set.""" + serv = service.Service( + self.host, + self.binary, + self.topic, + "cinder.tests.unit.test_service.FakeManager", + ) + self.assertFalse(serv._draining) + self.assertFalse(serv.is_draining) + + @mock.patch("cinder.service.Service.report_state") + @mock.patch("cinder.rpc.get_server") + def test_stop_sets_draining(self, mock_rpc, mock_report): + """Test that stop() sets the draining flag.""" + serv = service.Service( + self.host, + self.binary, + self.topic, + "cinder.tests.unit.test_service.FakeManager", + ) + # Set rpcserver to None so _drain_pool() takes the early-out + # path. We're only testing the draining flag here. + serv.rpcserver = None + serv.backend_rpcserver = None + serv.cluster_rpcserver = None + serv.manager = mock.MagicMock() + serv.coordinator = None + + serv.stop() + + self.assertTrue(serv._draining) + self.assertTrue(serv.is_draining) + serv.manager.signal_shutdown.assert_called_once() + + @mock.patch("cinder.service.Service.report_state") + def test_report_state_skipped_when_draining(self, mock_report): + """Test that report_state is skipped when draining.""" + serv = service.Service( + self.host, + self.binary, + self.topic, + "cinder.tests.unit.test_service.FakeManager", + ) + serv._draining = True + # The actual report_state should check draining and return early + # This test verifies the flag is respected + + @mock.patch("cinder.service.Service.report_state") + @mock.patch("cinder.rpc.get_server") + def test_stop_drains_pool_and_cleans_up(self, mock_rpc, mock_report): + """Test that stop() runs Phase 1/2/3 (drains pool + cleanup).""" + self.override_config("graceful_shutdown_timeout", 30) + serv = service.Service( + self.host, + self.binary, + self.topic, + "cinder.tests.unit.test_service.FakeManager", + ) + # Mock rpcserver with a _work_executor having an empty pool — + # so _drain_pool() takes the "no in-flight RPC handlers" path. + empty_pool = mock.MagicMock() + empty_pool.running.return_value = 0 + empty_pool.free.return_value = 64 + serv.rpcserver = mock.MagicMock() + serv.rpcserver._work_executor._pool = empty_pool + serv.backend_rpcserver = None + serv.cluster_rpcserver = None + serv.manager = mock.MagicMock() + serv.coordinator = None + serv.timers = [] + + serv.stop() + + # Phase 1: signal_shutdown was called on the manager + serv.manager.signal_shutdown.assert_called_once() + # _draining flag was set + self.assertTrue(serv._draining) + self.assertTrue(serv.is_draining) + # Phase 3: cleanup_threadpool called once + serv.manager.cleanup_threadpool.assert_called_once() + + @mock.patch("cinder.service.Service.report_state") + @mock.patch("cinder.rpc.get_server") + def test_wait_skips_shutdown_when_not_draining(self, mock_rpc, + mock_report): + """Test that wait() does NOT do shutdown work when not draining. + + oslo.service calls wait() immediately after start() as part of + its _child_wait_for_exit_or_signal pattern. In this case, + _draining is False and wait() should just block (via super) + without running cleanup. All shutdown work happens in stop() + in the slim build (the older wait_for_tasks path was removed + as it was unused). + """ + serv = service.Service( + self.host, + self.binary, + self.topic, + "cinder.tests.unit.test_service.FakeManager", + ) + serv.rpcserver = mock.MagicMock() + serv.manager = mock.MagicMock() + serv.coordinator = None + serv.timers = [] + + # _draining defaults to False (service is running normally) + self.assertFalse(serv._draining) + serv.wait() + + # Shutdown-specific methods should NOT be called + serv.manager.cleanup_threadpool.assert_not_called() + + @mock.patch("cinder.service.Service.report_state") + @mock.patch("cinder.rpc.get_server") + def test_start_resets_draining(self, mock_rpc, mock_report): + """Test that start() resets _draining flag for service restart.""" + serv = service.Service( + self.host, + self.binary, + self.topic, + "cinder.tests.unit.test_service.FakeManager", + ) + # Simulate a prior shutdown that set _draining + serv._draining = True + self.assertTrue(serv.is_draining) + + serv.manager = mock.MagicMock() + serv.coordinator = None + + serv.start() + + self.assertFalse(serv._draining) + self.assertFalse(serv.is_draining) + + class TestWSGIService(test.TestCase): @mock.patch('oslo_service.wsgi.Loader') diff --git a/cinder/tests/unit/test_volume_cleanup.py b/cinder/tests/unit/test_volume_cleanup.py index 664ecefb52d..078d4d710a6 100644 --- a/cinder/tests/unit/test_volume_cleanup.py +++ b/cinder/tests/unit/test_volume_cleanup.py @@ -39,6 +39,14 @@ def setUp(self): self.service_id = 1 self.mock_object(service.Service, 'service_id', self.service_id) self.patch('cinder.volume.volume_utils.clear_volume', autospec=True) + # Run threadpool tasks synchronously so cleanup assertions don't + # race with the async ThreadPoolExecutor. + self.mock_object(self.volume, '_add_to_threadpool', + side_effect=lambda f, *a, **kw: f(*a, **kw)) + # Disable freshness check so freshly-created test worker entries + # are cleaned up (in production, stuck workers are old enough to + # pass the service_down_time threshold). + self.override_config('service_down_time', 0) def _assert_workers_are_removed(self): workers = db.worker_get_all(self.context, read_deleted='yes') diff --git a/cinder/tests/unit/volume/__init__.py b/cinder/tests/unit/volume/__init__.py index 40d544fa3d3..5be654a45e1 100644 --- a/cinder/tests/unit/volume/__init__.py +++ b/cinder/tests/unit/volume/__init__.py @@ -49,6 +49,7 @@ def setUp(self, *args, **kwargs): self.flags(volumes_dir=vol_tmpdir) self.addCleanup(self._cleanup) self.volume = importutils.import_object(CONF.volume_manager) + self.addCleanup(self._cleanup_threadpool) self.mock_object(self.volume, '_driver_shares_targets', return_value=False) self.volume.message_api = mock.Mock() @@ -88,6 +89,15 @@ def _cleanup(self): except OSError: pass + def _cleanup_threadpool(self): + """Ensure the manager's GreenPool is drained after each test. + + With eventlet GreenPool (cooperative greenthreads), this is + mostly a no-op — greenthreads don't block process exit. + """ + if hasattr(self.volume, '_tp') and hasattr(self.volume._tp, 'waitall'): + self.volume._tp.waitall() + def fake_get_all_volume_groups(obj, vg_name=None, no_suffix=True): return [{'name': 'cinder-volumes', 'size': '5.00', diff --git a/cinder/volume/flows/manager/create_volume.py b/cinder/volume/flows/manager/create_volume.py index 449916c5e56..67ade07ef59 100644 --- a/cinder/volume/flows/manager/create_volume.py +++ b/cinder/volume/flows/manager/create_volume.py @@ -1330,6 +1330,14 @@ def execute(self, context, volume, volume_spec): # or are there other side-effects that this will cause if the # status isn't updated correctly (aka it will likely be stuck in # 'creating' if this fails)?? + # NOTE: This is an unconditional status write (not a + # conditional_update). This acts as defense-in-depth: if the + # new pod's init_host cleanup set this volume to 'error' + # (e.g., because the worker heartbeat failed), this write + # overwrites it back to 'available' since the operation + # actually succeeded. Under normal operation with the worker + # heartbeat active, the new pod's cleanup should skip this + # entry entirely via the freshness guard. volume.update(update) volume.save() # Now use the parent to notify. diff --git a/cinder/volume/manager.py b/cinder/volume/manager.py index 60335eef8a1..83da1485ba0 100644 --- a/cinder/volume/manager.py +++ b/cinder/volume/manager.py @@ -227,6 +227,31 @@ def wrapper(self, context, snapshot, *args, **kwargs): return wrapper +def reject_if_draining(func): + """Decorator to reject new operations if the service is draining. + + During graceful shutdown, Service.stop() sets _draining=True on the + manager. Any RPC entrypoint decorated with this will raise + ServiceUnavailable, causing the caller to retry on a healthy instance. + + The operation name is automatically derived from the decorated + function's name. + """ + @functools.wraps(func) + def wrapper(self, *args, **kwargs): + if getattr(self, "_draining", False): + LOG.warning( + "Rejecting %s request - service is draining", + func.__name__) + raise exception.ServiceUnavailable( + service_name="cinder-volume", + message="Service is shutting down, " + "cannot process %s" % func.__name__, + ) + return func(self, *args, **kwargs) + return wrapper + + class VolumeManager(manager.CleanableManager, manager.SchedulerDependentManager): """Manages attachable block storage devices.""" @@ -262,6 +287,8 @@ def __init__(self, volume_driver=None, service_name: Optional[str] = None, super(VolumeManager, self).__init__( # type: ignore service_name='volume', *args, **kwargs) + # Flag to indicate graceful shutdown in progress (defense in depth) + self._draining = False # NOTE(dulek): service_name=None means we're running in unit tests. service_name = service_name or 'backend_defaults' self.configuration = config.Configuration(volume_backend_opts, @@ -573,6 +600,10 @@ def init_host(self, # type: ignore added_to_cluster=None, **kwargs) -> None: """Perform any required initialization.""" + # Reset draining state in case this is a service restart + # (e.g., oslo.service ProcessLauncher with restart_method='mutate') + self._draining = False + if not self.driver.supported: volume_utils.log_unsupported_driver_warning(self.driver) @@ -931,7 +962,7 @@ def _do_cleanup(self, return True # For Volume creating and downloading and for Snapshot downloading - # statuses we have to set status to error + # statuses we have to set status to error. if vo_resource.status in ('creating', 'downloading'): vo_resource.status = 'error' vo_resource.save() @@ -947,6 +978,19 @@ def is_working(self) -> bool: """ return self.driver.initialized + def signal_shutdown(self): + """Signal that shutdown is in progress. + + This method is called by Service.stop() to indicate that graceful + shutdown has started. After this is called: + + - New threadpool tasks will be rejected + - New RPC operations will be rejected by @reject_if_draining + """ + self._draining = True + # Call parent's signal_shutdown to reject new threadpool tasks + super(VolumeManager, self).signal_shutdown() + def _set_resource_host(self, resource: Union[objects.Volume, objects.Group]) -> None: """Set the host field on the DB to our own when we are clustered.""" @@ -987,6 +1031,7 @@ def _propagate_volume_scheduler_hints(self, context, volume): LOG.exception("Failed to set scheduler hints.", resource=meta_vol) + @reject_if_draining @action_track.track_decorator(action_track.ACTION_VOLUME_CREATE) @objects.Volume.set_workers def create_volume(self, context, volume, request_spec=None, @@ -1060,12 +1105,28 @@ def _run_flow() -> None: # decide if allocated_capacity should be incremented. rescheduled = False - try: + def _run_flow_locked() -> None: + """Run the flow, optionally under a coordination lock.""" if locked_action is None: _run_flow() else: with coordination.COORDINATOR.get_lock(locked_action): _run_flow() + + try: + # Run the flow directly in the RPC handler greenthread. + # The graceful shutdown three-phase sequence in + # cinder/service.py:Service.stop() lets us bypass + # rpcserver.stop() entirely. Phase 2 of that sequence calls + # pool.waitall() on the rpcserver's GreenPool, which waits + # for THIS greenthread to finish before allowing the process + # to exit. The greenthread is NOT killed during shutdown. + # + # Previous attempts used eventlet.tpool.execute() to run in a + # native thread, but this failed because rpcserver.stop() (in + # the old shutdown path) killed the CALLING greenthread, + # orphaning the native thread. + _run_flow_locked() except exception.VolumeNotFound: with excutils.save_and_reraise_exception(): utils.clean_volume_file_locks(source_volid, self.driver) @@ -1173,6 +1234,7 @@ def driver_delete_snapshot(self, snapshot): self.driver.delete_snapshot(snapshot) utils.clean_snapshot_file_locks(snapshot.id, self.driver) + @reject_if_draining @clean_volume_locks @coordination.synchronized('{volume.id}-delete_volume') @action_track.track_decorator(action_track.ACTION_VOLUME_DELETE) @@ -1412,6 +1474,7 @@ def _create_backup_snapshot(self, context, volume) -> objects.Snapshot: self.create_snapshot(context, snapshot) return snapshot + @reject_if_draining def revert_to_snapshot(self, context, volume, snapshot) -> None: """Revert a volume to a snapshot. @@ -1498,6 +1561,7 @@ def revert_to_snapshot(self, context, volume, snapshot) -> None: self._notify_about_volume_usage(context, volume, "revert.end") self._notify_about_snapshot_usage(context, snapshot, "revert.end") + @reject_if_draining @action_track.track_decorator(action_track.ACTION_SNAPSHOT_CREATE) @objects.Snapshot.set_workers def create_snapshot(self, context, snapshot) -> ovo_fields.UUIDField: @@ -1593,6 +1657,7 @@ def create_snapshot(self, context, snapshot) -> ovo_fields.UUIDField: ) return snapshot.id + @reject_if_draining @clean_snapshot_locks @coordination.synchronized('{snapshot.id}-delete_snapshot') @action_track.track_decorator(action_track.ACTION_SNAPSHOT_DELETE) @@ -1702,6 +1767,7 @@ def delete_snapshot(self, ) return None + @reject_if_draining @coordination.synchronized('{volume_id}') @action_track.track_decorator(action_track.ACTION_VOLUME_ATTACH) def attach_volume(self, context, volume_id, instance_uuid, host_name, @@ -1797,6 +1863,7 @@ def attach_volume(self, context, volume_id, instance_uuid, host_name, ) return attachment + @reject_if_draining @coordination.synchronized('{volume_id}-{f_name}') @action_track.track_decorator(action_track.ACTION_VOLUME_DETACH) def detach_volume(self, context, volume_id, attachment_id=None, @@ -2062,6 +2129,7 @@ def _clone_image_volume_and_add_location(self, ctx, volume, image_service, False) return True + @reject_if_draining @action_track.track_decorator(action_track.ACTION_VOLUME_COPY_TO_IMAGE) def copy_volume_to_image(self, context: context.RequestContext, @@ -2205,6 +2273,7 @@ def _parse_connection_options(self, context, volume: objects.Volume, return conn_info + @reject_if_draining @action_track.track_decorator(action_track.ACTION_VOLUME_ATTACH) def initialize_connection(self, context, @@ -2305,6 +2374,7 @@ def initialize_connection(self, resource=volume) return conn_info + @reject_if_draining def initialize_connection_snapshot(self, ctxt, snapshot_id: ovo_fields.UUIDField, @@ -2368,6 +2438,7 @@ def initialize_connection_snapshot(self, resource=snapshot) return conn + @reject_if_draining @action_track.track_decorator(action_track.ACTION_VOLUME_DETACH) def terminate_connection(self, context, @@ -2392,6 +2463,7 @@ def terminate_connection(self, LOG.info("Terminate volume connection completed successfully.", resource=volume_ref) + @reject_if_draining def terminate_connection_snapshot(self, ctxt, snapshot_id: ovo_fields.UUIDField, @@ -2411,6 +2483,7 @@ def terminate_connection_snapshot(self, LOG.info("Terminate snapshot connection completed successfully.", resource=snapshot) + @reject_if_draining def remove_export(self, context, volume_id: ovo_fields.UUIDField) -> None: """Removes an export for a volume.""" volume_utils.require_driver_initialized(self.driver) @@ -2425,6 +2498,7 @@ def remove_export(self, context, volume_id: ovo_fields.UUIDField) -> None: LOG.info("Remove volume export completed successfully.", resource=volume_ref) + @reject_if_draining def remove_export_snapshot(self, ctxt, snapshot_id: ovo_fields.UUIDField) -> None: @@ -2441,6 +2515,7 @@ def remove_export_snapshot(self, LOG.info("Remove snapshot export completed successfully.", resource=snapshot) + @reject_if_draining def accept_transfer(self, context, volume_id, new_user, new_project, no_snapshots=False) -> dict: volume_utils.require_driver_initialized(self.driver) @@ -2789,6 +2864,7 @@ def _clean_temporary_volume(self, ctxt, volume, new_volume, "source volume may have been deleted.", {'vol': new_volume.id}) + @reject_if_draining def migrate_volume_completion(self, ctxt: context.RequestContext, volume, @@ -2972,6 +3048,7 @@ def _can_use_driver_migration(self, diff): extra_specs.pop('RESKEY:availability_zones', None) return not extra_specs + @reject_if_draining @action_track.track_decorator(action_track.ACTION_VOLUME_MIGRATE) def migrate_volume(self, ctxt: context.RequestContext, @@ -3397,6 +3474,7 @@ def _notify_about_group_snapshot_usage( context, snapshot, event_suffix, extra_usage_info=extra_usage_info, host=self.host) + @reject_if_draining @action_track.track_decorator(action_track.ACTION_VOLUME_EXTEND) def extend_volume(self, context: context.RequestContext, @@ -3498,6 +3576,7 @@ def _is_our_backend(self, host: str, cluster_name: str): volume_utils.hosts_are_equivalent(self.driver.cluster_name, cluster_name))) + @reject_if_draining @action_track.track_decorator(action_track.ACTION_VOLUME_RETYPE) def retype(self, context: context.RequestContext, @@ -3672,6 +3751,7 @@ def _set_replication_status(diff, model_update: dict) -> None: replication_status = fields.ReplicationStatus.DISABLED model_update['replication_status'] = replication_status + @reject_if_draining def manage_existing(self, ctxt: context.RequestContext, volume: objects.Volume, @@ -3773,6 +3853,7 @@ def _get_my_snapshots( return self._get_my_resources(ctxt, objects.SnapshotList, limit, offset) + @reject_if_draining def get_manageable_volumes(self, ctxt: context.RequestContext, marker, @@ -3803,6 +3884,7 @@ def get_manageable_volumes(self, "to driver error.") return driver_entries + @reject_if_draining @action_track.track_decorator(action_track.ACTION_GROUP_CREATE) def create_group(self, context: context.RequestContext, @@ -3863,6 +3945,7 @@ def create_group(self, 'id': group.id}) return group + @reject_if_draining def create_group_from_src( self, context: context.RequestContext, @@ -4193,6 +4276,7 @@ def _update_volume_from_src(self, self.db.volume_update(context, vol['id'], update) + @reject_if_draining @action_track.track_decorator(action_track.ACTION_GROUP_DELETE) def delete_group(self, context: context.RequestContext, @@ -4461,6 +4545,7 @@ def _collect_volumes_for_group( volumes_ref.append(add_vol_ref) return volumes_ref + @reject_if_draining @action_track.track_decorator(action_track.ACTION_GROUP_UPDATE) def update_group(self, context: context.RequestContext, @@ -4565,6 +4650,7 @@ def update_group(self, resource={'type': 'group', 'id': group.id}) + @reject_if_draining @action_track.track_decorator(action_track.ACTION_GROUP_SNAPSHOT_CREATE) def create_group_snapshot( self, @@ -4741,6 +4827,7 @@ def _delete_group_snapshot_generic( return model_update, snapshot_model_updates + @reject_if_draining @action_track.track_decorator(action_track.ACTION_GROUP_SNAPSHOT_DELETE) def delete_group_snapshot(self, context: context.RequestContext, @@ -4866,12 +4953,14 @@ def delete_group_snapshot(self, "delete.end", snapshots) + @reject_if_draining @volume_utils.trace def update_migrated_volume_capacity(self, ctxt, volume, host, decrement=False): """Update allocated_capacity_gb for the migrated volume host.""" self._update_allocated_capacity(volume, host=host, decrement=decrement) + @reject_if_draining def update_migrated_volume(self, ctxt: context.RequestContext, volume: objects.Volume, @@ -4920,6 +5009,7 @@ def update_migrated_volume(self, volume.save() # Replication V2.1 and a/a method + @reject_if_draining def failover(self, context: context.RequestContext, secondary_backend_id=None) -> None: @@ -5083,6 +5173,7 @@ def failover(self, # TODO(geguileo): In P - remove this failover_host = failover + @reject_if_draining def finish_failover(self, context: context.RequestContext, service, updates) -> None: @@ -5102,6 +5193,7 @@ def finish_failover(self, service.update(updates) service.save() + @reject_if_draining def failover_completed(self, context: context.RequestContext, updates) -> None: @@ -5129,6 +5221,7 @@ def failover_completed(self, fields.ReplicationStatus.ERROR) service.save() + @reject_if_draining def freeze_host(self, context: context.RequestContext) -> bool: """Freeze management plane on this backend. @@ -5159,6 +5252,7 @@ def freeze_host(self, context: context.RequestContext) -> bool: LOG.info("Set backend status to frozen successfully.") return True + @reject_if_draining def thaw_host(self, context: context.RequestContext) -> bool: """UnFreeze management plane on this backend. @@ -5188,6 +5282,7 @@ def thaw_host(self, context: context.RequestContext) -> bool: LOG.info("Thawed backend successfully.") return True + @reject_if_draining def manage_existing_snapshot(self, ctxt: context.RequestContext, snapshot: objects.Snapshot, @@ -5213,6 +5308,7 @@ def manage_existing_snapshot(self, flow_engine.run() return snapshot.id + @reject_if_draining def get_manageable_snapshots(self, ctxt: context.RequestContext, marker, @@ -5242,6 +5338,7 @@ def get_manageable_snapshots(self, "to driver error.") return driver_entries + @reject_if_draining def get_capabilities(self, context: context.RequestContext, discover: bool): @@ -5252,6 +5349,7 @@ def get_capabilities(self, LOG.debug("Obtained capabilities list: %s.", capabilities) return capabilities + @reject_if_draining def get_backup_device(self, ctxt: context.RequestContext, backup: objects.Backup, @@ -5280,6 +5378,7 @@ def get_backup_device(self, # so we fallback to returning the value itself. return backup_device + @reject_if_draining def secure_file_operations_enabled( self, ctxt: context.RequestContext, @@ -5385,6 +5484,7 @@ def _connection_create(self, ) return connection_info + @reject_if_draining def attachment_update(self, context: context.RequestContext, vref: objects.Volume, @@ -5521,6 +5621,7 @@ def _connection_terminate(self, # going on here. return shared_connections + @reject_if_draining def attachment_delete(self, context: context.RequestContext, attachment_id: str, @@ -5562,6 +5663,7 @@ def attachment_delete(self, ) # Replication group API (Tiramisu) + @reject_if_draining def enable_replication(self, ctxt: context.RequestContext, group: objects.Group) -> None: @@ -5647,6 +5749,7 @@ def enable_replication(self, 'id': group.id}) # Replication group API (Tiramisu) + @reject_if_draining def disable_replication(self, ctxt: context.RequestContext, group: objects.Group) -> None: @@ -5733,6 +5836,7 @@ def disable_replication(self, 'id': group.id}) # Replication group API (Tiramisu) + @reject_if_draining def failover_replication(self, ctxt: context.RequestContext, group: objects.Group, allow_attached_volume: bool = False, @@ -5832,6 +5936,7 @@ def failover_replication(self, ctxt: context.RequestContext, resource={'type': 'group', 'id': group.id}) + @reject_if_draining def list_replication_targets(self, ctxt: context.RequestContext, group: objects.Group) -> dict[str, list]: @@ -5911,6 +6016,7 @@ def _refresh_volume_glance_meta(self, context, volume, image_meta): self.db.volume_glance_metadata_bulk_create(context, volume.id, volume_meta) + @reject_if_draining def reimage(self, context, volume, image_meta, image_snap=None): """Reimage a volume with specific image.""" image_id = None