diff --git a/src/drunc/connectivity_service/client.py b/src/drunc/connectivity_service/client.py index d1182bb2d..368fb6de7 100644 --- a/src/drunc/connectivity_service/client.py +++ b/src/drunc/connectivity_service/client.py @@ -92,7 +92,7 @@ def retract(self, uid, fail_quickly=False): ignore_errors=True, ) if r.status_code == 404: - self.log.warning( + self.log.debug( f"Connection '{uid}' not found on the connectivity service" ) break diff --git a/src/drunc/controller/children_interface/child_node.py b/src/drunc/controller/children_interface/child_node.py index 2eeaa0a6a..552f86417 100644 --- a/src/drunc/controller/children_interface/child_node.py +++ b/src/drunc/controller/children_interface/child_node.py @@ -52,6 +52,7 @@ def get_endpoint(self) -> str: return "" def terminate(self): + self.log.info(f"Terminating {self.name}") pass def propagate_command( diff --git a/src/drunc/controller/children_interface/grpc_child.py b/src/drunc/controller/children_interface/grpc_child.py index dc98f4d24..7de703afa 100644 --- a/src/drunc/controller/children_interface/grpc_child.py +++ b/src/drunc/controller/children_interface/grpc_child.py @@ -208,7 +208,7 @@ def propagate_command( try: self.handle_child_grpc_error(error) except ServerUnreachable: - self.log.warning( + self.log.info( f"Connection to {self.name} at {self.uri} failed, attempting to reconnect..." ) response = self._attempt_reconnection(lambda: cmd(packed_request)) @@ -235,7 +235,7 @@ def status( try: self.handle_child_grpc_error(error) except ServerUnreachable: - self.log.warning( + self.log.info( f"Connection to {self.name} at {self.uri} failed during status check, attempting to reconnect..." ) response = self._attempt_reconnection(lambda: self.stub.status(request)) @@ -262,7 +262,7 @@ def describe( try: self.handle_child_grpc_error(error) except ServerUnreachable: - self.log.warning( + self.log.info( f"Connection to {self.name} at {self.uri} failed during describe check, attempting to reconnect..." ) response = self._attempt_reconnection( @@ -293,7 +293,7 @@ def describe_fsm( try: self.handle_child_grpc_error(error) except ServerUnreachable: - self.log.warning( + self.log.info( f"Connection to {self.name} at {self.uri} failed during describe_fsm check, attempting to reconnect..." ) response = self._attempt_reconnection( @@ -323,7 +323,7 @@ def execute_fsm_command( try: self.handle_child_grpc_error(error) except ServerUnreachable: - self.log.warning( + self.log.info( f"Connection to {self.name} at {self.uri} failed, attempting to reconnect..." ) response = self._attempt_reconnection( @@ -353,7 +353,7 @@ def execute_expert_command( try: self.handle_child_grpc_error(error) except ServerUnreachable: - self.log.warning( + self.log.info( f"Connection to {self.name} at {self.uri} failed, attempting to reconnect..." ) response = self._attempt_reconnection( @@ -382,7 +382,7 @@ def recompute_status( try: self.handle_child_grpc_error(e) except ServerUnreachable: - self.log.warning( + self.log.info( f"Connection to {self.name} at {self.uri} failed during recompute_status check, attempting to reconnect..." ) response = self._attempt_reconnection( diff --git a/src/drunc/controller/controller.py b/src/drunc/controller/controller.py index deb147495..b2fe56e78 100644 --- a/src/drunc/controller/controller.py +++ b/src/drunc/controller/controller.py @@ -446,7 +446,9 @@ def threading_publish_state(self, interval_s: float = 10.0): ) except Exception as e: self.log.exception(f"Error while publishing periodic status: {e}") - time.sleep(interval_s) + + if self.stop_event.wait(timeout=interval_s): + break def advertise_control_address(self, address): self.uri = address @@ -479,16 +481,14 @@ def update_connectivity_service(ctrler, connectivity_service, interval): self.connectivity_service_thread.start() def terminate(self): + self.log.info(f"Terminating controller {self.name}") self.running = False - if self.opmon_publisher is not None: - self.stop_event.set() - self.thread.join() if hasattr(self, "connectivity_service") and self.connectivity_service: if self.connectivity_service_thread: self.connectivity_service_thread.join() self.log.info("Unregistering from the connectivity service") - self.connectivity_service.retract(self.name + "_control") + self.connectivity_service.retract(self.name + "_control", fail_quickly=True) if self.can_broadcast(): self.broadcast( @@ -505,6 +505,17 @@ def terminate(self): if ResponseListener.exists(): ResponseListener.get().terminate() + if self.opmon_publisher is not None: + self.log.debug("Stopping opmon publisher") + self.stop_event.set() + self.thread.join(timeout=1.0) + if self.thread.is_alive(): + self.log.warning( + "OpMon publisher thread did not stop within timeout, continuing shutdown" + ) + else: + self.log.debug("opmon publisher stopped") + self.log.debug("Threading threads") for t in threading.enumerate(): self.log.debug(f"{t.name} TID: {t.native_id} is_alive: {t.is_alive}") diff --git a/src/drunc/controller/interface/controller.py b/src/drunc/controller/interface/controller.py index a7c2dd1f8..15d72874d 100644 --- a/src/drunc/controller/interface/controller.py +++ b/src/drunc/controller/interface/controller.py @@ -116,8 +116,10 @@ def serve(listen_addr: str) -> None: return server, port def controller_shutdown(): - log.warning("Requested termination") + log.info("Requested termination") + log.info("Calling ctrlr.terminate()") ctrlr.terminate() + log.info("ctrlr.terminate() completed") def kill_me(sig, frame): l = get_logger("controller.kill_me") @@ -128,21 +130,39 @@ def kill_me(sig, frame): os.killpg(pgrp, signal.SIGKILL) def shutdown(sig, frame): - log.info("Shutting down gracefully") + log.info(f"Shutting down gracefully (received signal: {sig})") try: controller_shutdown() except Exception as e: log.exception(e) kill_me(sig, frame) - signal.signal(signal.SIGHUP, kill_me) - signal.signal(signal.SIGINT, shutdown) - try: server, port = serve(commandfacility) server_name = commandfacility.split(":")[0] ctrlr.advertise_control_address(f"grpc://{server_name}:{port}") ctrlr.init_controller() + + # Add signal handling for gRPC server + def signal_handler(signum, frame): + log.info(f"Received signal {signum}, shutting down gRPC server") + server.stop(grace=2.0) # Give 2 seconds for graceful shutdown + log.info("gRPC server shutdown completed") + + try: + shutdown(signum, frame) + log.info("shutdown() completed") + except Exception as e: + log.exception(e) + finally: + log.info("Exiting...") + os._exit(0) + + # Register signal handlers for the server + signal.signal(signal.SIGTERM, signal_handler) + signal.signal(signal.SIGQUIT, signal_handler) + signal.signal(signal.SIGHUP, signal_handler) + server.wait_for_termination(timeout=None) except Exception as e: diff --git a/src/drunc/process_manager/k8s_process_manager.py b/src/drunc/process_manager/k8s_process_manager.py index 3c9a67372..584796beb 100644 --- a/src/drunc/process_manager/k8s_process_manager.py +++ b/src/drunc/process_manager/k8s_process_manager.py @@ -2,6 +2,7 @@ import getpass import os import re +import signal import threading import uuid from time import sleep, time @@ -31,7 +32,7 @@ DruncK8sPodException, ) from drunc.process_manager.process_manager import ProcessManager -from drunc.process_manager.utils import validate_k8s_session_name +from drunc.process_manager.utils import on_parent_exit, validate_k8s_session_name from drunc.utils.utils import get_logger, resolve_localhost_to_hostname @@ -75,7 +76,14 @@ def run(self) -> None: if is_terminal_phase or is_deleted_event: exit_code = -1 reason = "Unknown" - if ( + + self.pm.log.debug( + f"Pod {proc_uuid} terminated: phase={phase}, is_terminal={is_terminal_phase}, is_deleted={is_deleted_event}" + ) + if phase == "Succeeded": + exit_code = 0 + reason = "GracefulShutdown" + elif ( status.container_statuses and status.container_statuses[0].state.terminated ): @@ -83,9 +91,20 @@ def run(self) -> None: 0 ].state.terminated exit_code = terminated_state.exit_code - reason = terminated_state.reason + reason = ( + terminated_state.reason + ) # Finally, handle deleted events elif is_deleted_event: - reason = "PodDeleted" + if phase == "Succeeded": + exit_code = 0 + reason = "GracefulShutdown" + else: + exit_code = -1 + reason = "PodDeleted" + + self.pm.log.debug( + f"Final result for pod {proc_uuid}: exit_code={exit_code}, reason={reason}" + ) self.processed_uuids.add(proc_uuid) self.pm.notify_termination( @@ -186,6 +205,9 @@ def __init__(self, configuration, **kwargs) -> None: else: self.log.info("No active namespace created by drunc") + # Set up signal handlers for cleanup when parent process dies + self._setup_signal_handlers() + def _start_watcher(self) -> None: """Starts the background thread that watches for Pod status changes.""" self.log.debug("Starting K8s pod watcher thread") @@ -193,6 +215,32 @@ def _start_watcher(self) -> None: t.start() self.watchers.append(t) + def _setup_signal_handlers(self) -> None: + """Set up signal handlers to clean up pods when the process manager is terminated.""" + + def signal_handler(signum, frame): + self.log.info(f"Received signal {signum}, cleaning up all pods...") + try: + self._terminate_impl() + except Exception as e: + self.log.error(f"Error during signal cleanup: {e}") + finally: + # Exit the process + os._exit(0) + + # Register signal handlers for common termination signals + signal.signal(signal.SIGTERM, signal_handler) + signal.signal(signal.SIGHUP, signal_handler) + signal.signal(signal.SIGQUIT, signal_handler) + + # Set up parent death signal (Linux only) + try: + on_parent_exit(signal.SIGTERM)() + except Exception as e: + self.log.debug( + f"Could not set parent death signal (may not be supported on this platform): {e}" + ) + def notify_termination(self, proc_uuid, exit_code, reason, session) -> None: """Callback for when a pod terminates.""" self.log.debug( @@ -458,49 +506,51 @@ def _create_pod(self, podname, session, boot_request: BootRequest) -> None: command_parts.append(prefix + " ".join([e_and_a.exec] + list(e_and_a.args))) main_command_str = " && ".join(command_parts) - # Determine the correct shutdown command for the preStop hook - shutdown_command = "" - if "controller" in podname or podname == self.connection_server_name: + # Only add preStop hook for C++ applications (non-controllers) + lifecycle_hook = None + if "controller" not in podname and podname != self.connection_server_name: self.log.debug( - f"'{podname}' identified as a Python app, using manual PID discovery with SIGINT." + f"'{podname}' identified as a C++ app, adding preStop hook with SIGQUIT." ) - shutdown_command = """ -for p in /proc/[0-9]*; do - if [ -f "$p/cmdline" ] && grep -a "drunc-controller" "$p/cmdline" > /dev/null; then - kill -SIGINT $(basename "$p"); - fi -done -""" - else: # C++ Applications - self.log.debug(f"'{podname}' identified as a C++ app, using SIGQUIT.") shutdown_command = "kill -QUIT 1" - - lifecycle_hook = client.V1Lifecycle( - pre_stop=client.V1LifecycleHandler( - _exec=client.V1ExecAction(command=["/bin/sh", "-c", shutdown_command]) + lifecycle_hook = client.V1Lifecycle( + pre_stop=client.V1LifecycleHandler( + _exec=client.V1ExecAction( + command=["/bin/sh", "-c", shutdown_command] + ) + ) + ) + else: + self.log.debug( + f"'{podname}' identified as a Python app, no preStop hook needed." ) - ) - main_container = client.V1Container( - name=podname, - image=pod_image, - command=["/bin/sh", "-c"], - args=[main_command_str], - env=[ + # Create container with conditional lifecycle hook + container_kwargs = { + "name": podname, + "image": pod_image, + "command": ["/bin/sh", "-c"], + "args": [main_command_str], + "env": [ client.V1EnvVar(name=k, value=v) for k, v in boot_request.process_description.env.items() ], - lifecycle=lifecycle_hook, - ports=[], - volume_mounts=[ + "ports": [], + "volume_mounts": [ client.V1VolumeMount(name="nfs", mount_path="/nfs"), client.V1VolumeMount(name="cvmfs", mount_path="/cvmfs"), ], - working_dir=boot_request.process_description.process_execution_directory, - security_context=client.V1SecurityContext( + "working_dir": boot_request.process_description.process_execution_directory, + "security_context": client.V1SecurityContext( run_as_user=os.getuid(), run_as_group=os.getgid() ), - ) + } + + # Only add lifecycle hook for C++ applications + if lifecycle_hook is not None: + container_kwargs["lifecycle"] = lifecycle_hook + + main_container = client.V1Container(**container_kwargs) all_containers = [main_container] @@ -993,23 +1043,14 @@ def _kill_impl(self, query: ProcessQuery) -> ProcessInstanceList: self.log.info(f"Starting termination of {len(uuids_to_kill)} pods...") - graceful_apps, forced_apps = [], [] + apps = [] for uuid_str in uuids_to_kill: if uuid_str not in self.boot_request: continue - pd = self.boot_request[uuid_str].process_description - is_controller = ( - "controller" in pd.metadata.name - or pd.metadata.name == self.connection_server_name - ) - - if is_controller: - forced_apps.append(uuid_str) - else: - graceful_apps.append(uuid_str) + apps.append(uuid_str) - def kill_and_wait(uuids, stage_name, grace_period=None) -> None: + def kill_and_wait(uuids, grace_period=None) -> None: if not uuids: return action = ( @@ -1017,7 +1058,7 @@ def kill_and_wait(uuids, stage_name, grace_period=None) -> None: if grace_period == 0 else "Gracefully terminating" ) - self.log.info(f"Stage '{stage_name}': {action} {len(uuids)} pod(s)...") + self.log.info(f"{action} {len(uuids)} pod(s)...") self.termination_complete_event.clear() self.uuids_pending_deletion.update(uuids) @@ -1035,17 +1076,15 @@ def kill_and_wait(uuids, stage_name, grace_period=None) -> None: grace_period_seconds=grace_period, ) - wait_timeout = self.kill_timeout if grace_period is None else 15 + wait_timeout = ( + self.kill_timeout if grace_period is None else grace_period + 5 + ) if not self.termination_complete_event.wait(timeout=wait_timeout): - self.log.warning( - f"Timeout in stage '{stage_name}'. Remaining: {self.uuids_pending_deletion}" - ) + self.log.warning(f"Timeout. Remaining: {self.uuids_pending_deletion}") self.uuids_pending_deletion.clear() - kill_and_wait(graceful_apps, "Standalone C++ Applications") - - kill_and_wait(forced_apps, "Controllers & Local Session Apps", grace_period=0) + kill_and_wait(apps) final_ret = [] for proc_uuid in uuids_to_kill: diff --git a/src/drunc/utils/flask_manager.py b/src/drunc/utils/flask_manager.py index bb47ec1b8..9d34fb003 100644 --- a/src/drunc/utils/flask_manager.py +++ b/src/drunc/utils/flask_manager.py @@ -1,3 +1,4 @@ +import os import signal import threading import time @@ -112,9 +113,25 @@ def get_ready_status(): }, ) + def run_gunicorn_with_signal_handling(): + """Run gunicorn with SIGHUP ignored to prevent reload on shutdown. + + This prevents gunicorn from reloading when the parent process receives SIGHUP. + We only want graceful shutdown via SIGTERM from FlaskManager.stop(). + """ + # Create new process group first to isolate from parent's signal propagation + # This prevents SIGHUP from being sent to this process when parent receives it + try: + os.setpgid(0, 0) # Create new process group (safer than setsid) + except (OSError, PermissionError): + # May fail if already in a process group or on some systems, ignore + pass + + self.prod_app.run() + thread_name = f"{self.name}_thread" flask_srv = Process( # Indeed, we've just forked this sucker - target=self.prod_app.run, name=thread_name, daemon=True + target=run_gunicorn_with_signal_handling, name=thread_name, daemon=True ) flask_srv.start()