A 新增终止任务执行功能

spug_api/apps/exec/executors.py

@@ -16,13 +16,13 @@ def exec_worker_handler(job):
 
 
 class Job:
-    def __init__(self, key, name, hostname, port, username, pkey, command, interpreter, params=None, token=None,
+    def __init__(self, token, key, name, hostname, port, username, pkey, command, interpreter, params=None, term=None):
-                 term=None):
         self.ssh = SSH(hostname, port, username, pkey, term=term)
         self.key = key
         self.command = self._handle_command(command, interpreter)
         self.token = token
         self.rds = get_redis_connection()
+        self.rds_key = f'PID:{self.token}:{self.key}'
         self.env = dict(
             SPUG_HOST_ID=str(self.key),
             SPUG_HOST_NAME=name,
@@ -50,14 +50,12 @@ class Job:
         self._send({'key': self.key, 'status': code})
 
     def run(self):
-        if not self.token:
-            with self.ssh:
-                return self.ssh.exec_command(self.command, self.env)
         flag = time.time()
         self.send('\r\n\x1b[36m### Executing ...\x1b[0m\r\n')
         code = -1
         try:
             with self.ssh:
+                self.rds.set(self.rds_key, self.ssh.get_pid(), 7200)
                 for code, out in self.ssh.exec_command_with_stream(self.command, self.env):
                     self.send(out)
             human_time = human_seconds_time(time.time() - flag)
@@ -70,4 +68,5 @@ class Job:
             self.send(f'\r\n\x1b[31m### Exception {e}\x1b[0m')
             raise e
         finally:
+            self.rds.delete(self.rds_key)
             self.send_status(code)

spug_api/apps/exec/urls.py

@@ -3,11 +3,12 @@
 # Released under the AGPL-3.0 License.
 from django.conf.urls import url
 
-from apps.exec.views import *
+from apps.exec.views import TemplateView, TaskView, handle_terminate
 from apps.exec.transfer import TransferView
 
 urlpatterns = [
     url(r'template/$', TemplateView.as_view()),
     url(r'do/$', TaskView.as_view()),
     url(r'transfer/$', TransferView.as_view()),
+    url(r'terminate/$', handle_terminate),
 ]

spug_api/apps/exec/views.py

@@ -120,4 +120,19 @@ class TaskView(View):
         return json_response(error=error)
 
 
-
+@auth('exec.task.do')
+def handle_terminate(request):
+    form, error = JsonParser(
+        Argument('token', help='参数错误'),
+        Argument('host_id', type=int, help='参数错误')
+    ).parse(request.body)
+    if error is None:
+        host = Host.objects.get(pk=form.host_id)
+        rds = get_redis_connection()
+        rds_key = f'PID:{form.token}:{host.id}'
+        pid = rds.get(rds_key)
+        if pid:
+            with host.get_ssh() as ssh:
+                ssh.exec_command_raw(f'kill -9 {pid.decode()}')
+                rds.delete(rds_key)
+    return json_response(error=error)

spug_api/libs/ssh.py

@@ -58,6 +58,7 @@ class SSH:
         self.sftp = None
         self.exec_file = None
         self.term = term or {}
+        self.pid = None
         self.eof = 'Spug EOF 2108111926'
         self.default_env = default_env
         self.regex = re.compile(r'Spug EOF 2108111926 (-?\d+)[\r\n]?')
@@ -174,6 +175,12 @@ class SSH:
         sftp = self._get_sftp()
         sftp.remove(path)
 
+    def get_pid(self):
+        if self.pid:
+            return self.pid
+        self._get_channel()
+        return self.pid
+
     def _get_channel(self):
         if self.channel:
             return self.channel
@@ -183,13 +190,17 @@ class SSH:
         command = '[ -n "$BASH_VERSION" ] && set +o history\n'
         command += '[ -n "$ZSH_VERSION" ] && set +o zle && set -o no_nomatch\n'
         command += 'export PS1= && stty -echo\n'
-        command = self._handle_command(command, self.default_env)
+        command += f'echo {self.eof} $$\n'
         self.channel.sendall(command)
         out = ''
         while True:
             if self.channel.recv_ready():
                 out += self._decode(self.channel.recv(8196))
-                if self.regex.search(out):
+                match = self.regex.search(out)
+                if match:
+                    self.pid = int(match.group(1))
+                    if self.pid <= 1:
+                        raise Exception('Failed to get process pid')
                     self.stdout = self.channel.makefile('r')
                     break
             elif counter >= 100: