Add force node stopping using SIGKILL in case of unsuccessful pg_ctl stop

Author: vshepard

testgres/node.py

@@ -330,23 +330,43 @@ def version(self):
         """
         return self._pg_version
 
-    def _try_shutdown(self, max_attempts):
+    def _try_shutdown(self, max_attempts, with_force=False):
         attempts = 0
+        node_pid = self.pid
 
         # try stopping server N times
         while attempts < max_attempts:
             try:
-                self.stop()
+                self.stop(wait=False)
                 break    # OK
             except ExecUtilException:
                 pass    # one more time
             except Exception:
-                # TODO: probably should kill stray instance
                 eprint('cannot stop node {}'.format(self.name))
                 break
 
             attempts += 1
 
+        # If force stopping is enabled and PID is valid
+        if with_force and node_pid !=0:
+            # If we couldn't stop the node
+            p_status_output = self.os_ops.exec_command(cmd=f'ps -p {node_pid}', shell=True).decode('utf-8')
+            if self.status() != NodeStatus.Stopped and p_status_output and str(node_pid) in p_status_output:
+                try:
+                    eprint(f'Force stopping node {self.name} with PID {node_pid}')
+                    self.os_ops.kill(node_pid, signal.SIGKILL, expect_error=False)
+                except Exception:
+                    # The node has already stopped
+                    pass
+
+            # Check that node stopped
+            p_status_output = self.os_ops.exec_command(f'ps -p {node_pid}', shell=True, expect_error=True).decode('utf-8')
+            if p_status_output and str(node_pid) in p_status_output:
+                eprint(f'Failed to stop node {self.name}.')
+            else:
+                eprint(f'Node {self.name} has been stopped successfully.')
+
+
     def _assign_master(self, master):
         """NOTE: this is a private method!"""
 
@@ -935,6 +955,11 @@ def cleanup(self, max_attempts=3, full=False):
             rm_dir = self.data_dir    # just data, save logs
 
         self.os_ops.rmdirs(rm_dir, ignore_errors=False)
+        is_not_removed = self.os_ops.isdir(rm_dir)
+        while is_not_removed:
+            time.sleep(1)
+            self.os_ops.rmdirs(rm_dir, ignore_errors=False)
+            is_not_removed = self.os_ops.isdir(rm_dir)
 
         return self
 

testgres/operations/local_ops.py

@@ -293,10 +293,10 @@ def remove_file(self, filename):
         return os.remove(filename)
 
     # Processes control
-    def kill(self, pid, signal):
+    def kill(self, pid, signal, expect_error=False):
         # Kill the process
         cmd = "kill -{} {}".format(signal, pid)
-        return self.exec_command(cmd)
+        return self.exec_command(cmd, expect_error=expect_error)
 
     def get_pid(self):
         # Get current process id