Revert "Revert "k8s : better terminate""

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

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(

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(

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}")

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: